Preserve custom subprocess coordinator compatibility

External language SDK coordinators can implement the documented command hook. Keeping that contract stable avoids unexpected-keyword failures when upgrading while allowing Dag-bundle-backed artifact discovery.

Author: jason810496

contributing-docs/30_new_language_sdk.rst

@@ -107,9 +107,7 @@ the subprocess understands:
 
 .. code-block:: python
 
-    def _build_execute_task_command(
-        self, *, what: TaskInstanceDTO, roots: list[pathlib.Path]
-    ) -> tuple[list[str], str]: ...
+    def _build_execute_task_command(self, *, what: TaskInstanceDTO) -> tuple[list[str], str]: ...
 
 The method returns a ``(command, subprocess_schema_version)`` pair:
 
@@ -119,10 +117,11 @@ The method returns a ``(command, subprocess_schema_version)`` pair:
   subprocess understands, used by the supervisor to negotiate message formats
   across SDK versions. See `Supervisor Schema`_ below.
 
-``roots`` are the artifact directories the base class has already resolved from
-the coordinator's configured source — an explicit filesystem root, a named Dag
-bundle (``dag_bundle_name``), or the task's own bundle — so subclasses scan
-``roots`` rather than reading the configured root directly.
+Call ``self._get_scan_roots()`` to retrieve the artifact directories the base
+class has already resolved from the coordinator's configured source — an
+explicit filesystem root, a named Dag bundle (``dag_bundle_name``), or the
+task's own bundle. Subclasses should scan those roots rather than reading the
+configured root directly.
 
 Supervisor Schema
 ~~~~~~~~~~~~~~~~~

task-sdk/src/airflow/sdk/coordinators/_subprocess.py

@@ -447,6 +447,7 @@ class SubprocessCoordinator(BaseCoordinator):
     # resolve roots without knowing the subclass field name.
     _configured_roots: list[pathlib.Path] = attrs.field(init=False, factory=list)
     _active_bundle_info: BundleInfo | None = attrs.field(init=False, default=None)
+    _active_scan_roots: tuple[pathlib.Path, ...] | None = attrs.field(init=False, default=None)
 
     @property
     def _explicit_artifact_roots(self) -> tuple[str, Sequence[pathlib.Path]]:
@@ -526,14 +527,18 @@ def _init_root_source(
             raise FileNotFoundError(f"Dag bundle {target.name!r} resolved to {path}, which does not exist.")
         return [path], bundle
 
-    def _build_execute_task_command(
-        self, *, what: TaskInstance, roots: list[pathlib.Path]
-    ) -> tuple[list[str], str | None]:
+    def _get_scan_roots(self) -> tuple[pathlib.Path, ...]:
+        """Return the artifact roots resolved for the active task."""
+        if self._active_scan_roots is None:
+            raise RuntimeError("_get_scan_roots requires an active task; call it during execute_task.")
+        return self._active_scan_roots
+
+    def _build_execute_task_command(self, *, what: TaskInstance) -> tuple[list[str], str | None]:
         """
         Build the subprocess command and resolve its supervisor wire-schema version for *what*.
 
-        *roots* are the directories to scan for artifacts, already resolved by
-        :meth:`_init_root_source` from the coordinator's configured source.
+        Subclasses can retrieve the directories to scan for artifacts with
+        :meth:`_get_scan_roots`.
         Returns a ``(command, subprocess_schema_version)`` pair. *command* MUST
         NOT include the ``--comm`` / ``--logs`` flags — those are appended by
         :class:`_PopenActivitySubprocess` once the listening sockets have been
@@ -558,6 +563,15 @@ def _set_current_bundle(self, bundle_info: BundleInfo):
         finally:
             self._active_bundle_info = None
 
+    @contextlib.contextmanager
+    def _set_scan_roots(self, roots: Sequence[pathlib.Path]):
+        """Expose *roots* to the command builder for the duration of the task."""
+        self._active_scan_roots = tuple(roots)
+        try:
+            yield
+        finally:
+            self._active_scan_roots = None
+
     def execute_task(
         self,
         *,
@@ -584,7 +598,8 @@ def execute_task(
                         bundle_version=resolved_bundle.version,
                     )
                 )
-            command, subprocess_schema_version = self._build_execute_task_command(what=what, roots=roots)
+            stack.enter_context(self._set_scan_roots(roots))
+            command, subprocess_schema_version = self._build_execute_task_command(what=what)
             process = _PopenActivitySubprocess.start(
                 what=what,
                 dag_rel_path=dag_rel_path,

task-sdk/src/airflow/sdk/coordinators/executable/coordinator.py

@@ -354,8 +354,7 @@ class ExecutableCoordinator(SubprocessCoordinator):
     def _explicit_artifact_roots(self) -> tuple[str, list[pathlib.Path]]:
         return "executables_root", self.executables_root
 
-    def _build_execute_task_command(
-        self, *, what: TaskInstance, roots: list[pathlib.Path]
-    ) -> tuple[list[str], str | None]:
+    def _build_execute_task_command(self, *, what: TaskInstance) -> tuple[list[str], str | None]:
+        roots = self._get_scan_roots()
         bundle = _Bundle.find(roots, what.dag_id)
         return [str(bundle.path)], bundle.schema_version

task-sdk/src/airflow/sdk/coordinators/java/coordinator.py

@@ -209,11 +209,10 @@ class JavaCoordinator(SubprocessCoordinator):
     def _explicit_artifact_roots(self) -> tuple[str, list[pathlib.Path]]:
         return "jars_root", self.jars_root
 
-    def _build_execute_task_command(
-        self, *, what: TaskInstance, roots: list[pathlib.Path]
-    ) -> tuple[list[str], str | None]:
+    def _build_execute_task_command(self, *, what: TaskInstance) -> tuple[list[str], str | None]:
         # Without main_class, the first executable JAR in walk order wins; tracked at
         # https://github.com/apache/airflow/issues/71134
+        roots = self._get_scan_roots()
         jar = _JarInfo.find(roots, self.main_class)
         command = [
             self.java_executable,

task-sdk/src/airflow/sdk/coordinators/node/coordinator.py

@@ -161,10 +161,9 @@ class NodeCoordinator(SubprocessCoordinator):
     def _explicit_artifact_roots(self) -> tuple[str, list[pathlib.Path]]:
         return "bundles_root", self.bundles_root
 
-    def _build_execute_task_command(
-        self, *, what: TaskInstance, roots: list[pathlib.Path]
-    ) -> tuple[list[str], str | None]:
+    def _build_execute_task_command(self, *, what: TaskInstance) -> tuple[list[str], str | None]:
         # Multi-bundle routing should be added here by passing `what.dag_id` and
         # `what.task_id` into bundle selection and matching against metadata["dags"].
+        roots = self._get_scan_roots()
         bundle = _find_bundle(roots)
         return [self.node_executable, os.fspath(bundle.path)], bundle.schema_version

task-sdk/tests/task_sdk/coordinators/executable/test_coordinator.py

@@ -399,9 +399,10 @@ def test_configured_dag_bundle_name_accepted(self, mock_manager):
     def test_build_command_scans_passed_roots_in_colocated_mode(self, tmp_path):
         binary = _build_bundle(tmp_path / "my_bundle", dag_ids=["tutorial_dag"])
         coordinator = ExecutableCoordinator()
-        command, schema_version = coordinator._build_execute_task_command(
-            what=_make_ti(dag_id="tutorial_dag"), roots=[tmp_path]
-        )
+        with coordinator._set_scan_roots([tmp_path]):
+            command, schema_version = coordinator._build_execute_task_command(
+                what=_make_ti(dag_id="tutorial_dag")
+            )
         assert command == [str(binary.resolve())]
         assert schema_version == "2026-06-16"
 
@@ -412,7 +413,8 @@ def test_returns_resolved_executable_and_schema_version(self, tmp_path):
         ti = _make_ti(dag_id="tutorial_dag")
 
         coordinator = ExecutableCoordinator(executables_root=[tmp_path])
-        command, schema_version = coordinator._build_execute_task_command(what=ti, roots=[tmp_path])
+        with coordinator._set_scan_roots([tmp_path]):
+            command, schema_version = coordinator._build_execute_task_command(what=ti)
         assert command == [str(binary.resolve())]
         assert schema_version == "2026-06-16"
 
@@ -423,16 +425,22 @@ def test_raises_when_bundle_omits_schema_version(self, tmp_path):
         ti = _make_ti(dag_id="tutorial_dag")
 
         coordinator = ExecutableCoordinator(executables_root=[tmp_path])
-        with pytest.raises(FileNotFoundError, match="matching bundles were rejected"):
-            coordinator._build_execute_task_command(what=ti, roots=[tmp_path])
+        with (
+            coordinator._set_scan_roots([tmp_path]),
+            pytest.raises(FileNotFoundError, match="matching bundles were rejected"),
+        ):
+            coordinator._build_execute_task_command(what=ti)
 
     def test_raises_when_dag_id_not_found(self, tmp_path):
         _build_bundle(tmp_path / "my_bundle", dag_ids=["other_dag"])
         ti = _make_ti(dag_id="tutorial_dag")
 
         coordinator = ExecutableCoordinator(executables_root=[tmp_path])
-        with pytest.raises(FileNotFoundError, match="cannot find executable bundle"):
-            coordinator._build_execute_task_command(what=ti, roots=[tmp_path])
+        with (
+            coordinator._set_scan_roots([tmp_path]),
+            pytest.raises(FileNotFoundError, match="cannot find executable bundle"),
+        ):
+            coordinator._build_execute_task_command(what=ti)
 
 
 @pytest.fixture

task-sdk/tests/task_sdk/coordinators/java/test_coordinator.py

@@ -261,7 +261,8 @@ def test_unconfigured_dag_bundle_name_raises(self, mock_manager):
     def test_build_command_scans_passed_roots_in_colocated_mode(self, tmp_path):
         _make_jar(tmp_path / "app.jar", main_class="com.example.TaskRunner", schema_version="2026-06-16")
         coordinator = JavaCoordinator(main_class="com.example.TaskRunner")
-        command, schema_version = coordinator._build_execute_task_command(what=_make_ti(), roots=[tmp_path])
+        with coordinator._set_scan_roots([tmp_path]):
+            command, schema_version = coordinator._build_execute_task_command(what=_make_ti())
         assert command[0] == "java"
         assert command[-1] == "com.example.TaskRunner"
         assert schema_version == "2026-06-16"

task-sdk/tests/task_sdk/coordinators/node/test_coordinator.py

@@ -117,7 +117,8 @@ def test_unconfigured_dag_bundle_name_raises(self, mock_manager):
     def test_build_command_scans_passed_roots_in_colocated_mode(self, tmp_path):
         bundle = write_bundle(tmp_path)
         coordinator = NodeCoordinator()
-        command, schema_version = coordinator._build_execute_task_command(what=_make_ti(), roots=[tmp_path])
+        with coordinator._set_scan_roots([tmp_path]):
+            command, schema_version = coordinator._build_execute_task_command(what=_make_ti())
         assert command == ["node", str(bundle)]
         assert schema_version == SCHEMA_VERSION
 
@@ -240,7 +241,8 @@ def test_build_execute_task_command_returns_node_bundle_and_schema_version(self,
             bundles_root=tmp_path,
         )
 
-        command, schema_version = coordinator._build_execute_task_command(what=_make_ti(), roots=[tmp_path])
+        with coordinator._set_scan_roots([tmp_path]):
+            command, schema_version = coordinator._build_execute_task_command(what=_make_ti())
 
         assert command == ["/opt/node/bin/node", str(bundle)]
         assert schema_version == SCHEMA_VERSION

task-sdk/tests/task_sdk/coordinators/test_subprocess.py

@@ -588,8 +588,8 @@ class _StubSubprocessCoordinator(SubprocessCoordinator):
     def _explicit_artifact_roots(self) -> tuple[str, list[pathlib.Path]]:
         return "explicit_roots", self.explicit_roots
 
-    def _build_execute_task_command(self, *, what, roots):
-        self.recorded_roots.append(roots)
+    def _build_execute_task_command(self, *, what):
+        self.recorded_roots.append(list(self._get_scan_roots()))
         return list(self.command), self.schema_version
 
 
@@ -615,7 +615,7 @@ class _Plain(SubprocessCoordinator):
             pass
 
         with pytest.raises(NotImplementedError):
-            _Plain()._build_execute_task_command(what=_make_ti(), roots=[])
+            _Plain()._build_execute_task_command(what=_make_ti())
 
 
 class TestSubprocessCoordinatorExecuteTask:
@@ -885,7 +885,7 @@ def test_subclass_without_overrides_defaults_to_task_bundle(self):
 
         @attrs.define(kw_only=True)
         class _Bare(SubprocessCoordinator):
-            def _build_execute_task_command(self, *, what, roots):
+            def _build_execute_task_command(self, *, what):
                 return [], None
 
         assert _Bare()._artifact_source is _ArtifactSource.TASK_BUNDLE
@@ -1085,6 +1085,17 @@ def test_named_bundle_mode_locks_the_tree_it_scans(
         mock_lock.assert_called_once_with(bundle_name="artifacts", bundle_version="sha-abc")
 
 
+class TestGetScanRoots:
+    def test_returns_bound_roots_and_clears_them_on_exit(self, tmp_path):
+        coordinator = _StubSubprocessCoordinator(command=["x"])
+
+        with coordinator._set_scan_roots([tmp_path]):
+            assert coordinator._get_scan_roots() == (tmp_path,)
+
+        with pytest.raises(RuntimeError, match="requires an active task"):
+            coordinator._get_scan_roots()
+
+
 class TestSetCurrentBundle:
     def test_binds_for_the_block_and_clears_on_exit_allowing_reuse(self):
         coordinator = _StubSubprocessCoordinator(command=["/bin/true"])