Reject empty subprocess coordinator artifact roots

An explicitly empty root is likely a configuration or templating error. Treating it as omitted can silently scan the task's Dag bundle and run unintended artifacts.

Author: jason810496

airflow-core/docs/authoring-and-scheduling/language-sdks/go.rst

@@ -510,7 +510,8 @@ All ``kwargs`` in the ``coordinators`` config entry are passed to the
      - *(optional)*
      - One or more directories scanned recursively for executable bundles. Accepts a string,
        a path, or a list of strings/paths. When omitted, bundles are located through a Dag
-       bundle instead (see the note below).
+       bundle instead (see the note below). Explicitly setting this option to ``null`` or
+       an empty list is invalid.
    * - ``dag_bundle_name``
      - *(auto: task's own bundle)*
      - Name of a configured Dag bundle to load executable bundles from. Mutually exclusive

airflow-core/docs/authoring-and-scheduling/language-sdks/java.rst

@@ -1025,7 +1025,8 @@ All ``kwargs`` in the ``coordinators`` config entry are passed to the
      - *(optional)*
      - One or more directories scanned recursively for ``.jar`` files. Accepts a string,
        a path, or a list of strings/paths. When omitted, JARs are located through a Dag
-       bundle instead (see the note below).
+       bundle instead (see the note below). Explicitly setting this option to ``null`` or
+       an empty list is invalid.
    * - ``dag_bundle_name``
      - *(auto: task's own bundle)*
      - Name of a configured Dag bundle to load JARs from. Mutually exclusive with

airflow-core/docs/authoring-and-scheduling/language-sdks/typescript.rst

@@ -449,7 +449,8 @@ All ``kwargs`` in the ``coordinators`` config entry are passed to the
      - *(optional)*
      - One or more directories searched recursively, in order, for an integrity-verified ``*.min.mjs``
        bundle that declares the requested Dag. Accepts a string, a path, or a list of strings/paths. When
-       omitted, the bundle is located through a Dag bundle instead (see the note below).
+       omitted, the bundle is located through a Dag bundle instead (see the note below). Explicitly setting
+       this option to ``null`` or an empty list is invalid.
    * - ``dag_bundle_name``
      - *(auto: task's own bundle)*
      - Name of a configured Dag bundle to load the ``*.min.mjs`` bundle from. Mutually exclusive with

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

@@ -22,7 +22,7 @@
 import os
 import pathlib
 import stat
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, Final
 
 import attrs
 import structlog
@@ -38,6 +38,13 @@
 log: FilteringBoundLogger = structlog.get_logger(logger_name="coordinators")
 
 
+class _UnsetArtifactRoots:
+    """Sentinel distinguishing an omitted artifact-roots option from an empty value."""
+
+
+ARTIFACT_ROOTS_NOT_CONFIGURED: Final = _UnsetArtifactRoots()
+
+
 def convert_roots(
     value: None | os.PathLike[str] | pathlib.Path | list[os.PathLike[str] | pathlib.Path],
 ) -> list[pathlib.Path]:
@@ -96,6 +103,25 @@ def _sorted_children(directory: pathlib.Path) -> list[pathlib.Path]:
         return []
 
 
+def convert_configured_roots(
+    value: _UnsetArtifactRoots
+    | None
+    | os.PathLike[str]
+    | pathlib.Path
+    | list[os.PathLike[str] | pathlib.Path],
+) -> list[pathlib.Path]:
+    """Normalize configured roots while rejecting explicitly empty values."""
+    if isinstance(value, _UnsetArtifactRoots):
+        return []
+    roots = convert_roots(value)
+    if not roots:
+        raise ValueError(
+            "Artifact roots must contain at least one path when provided; "
+            "omit the option to use the task's Dag bundle."
+        )
+    return roots
+
+
 def validate_schema_version(instance, _, value) -> str:
     """Attrs validator resolving a bundle's supervisor schema version to a known one."""
     return get_schema_version_migrator().resolve_version(str(value))

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

@@ -31,8 +31,9 @@
 import structlog
 
 from airflow.sdk.coordinators._bundle_metadata import (
+    ARTIFACT_ROOTS_NOT_CONFIGURED,
     ResolvedBundle,
-    convert_roots,
+    convert_configured_roots,
     extract_supervisor_schema_version,
     parse_metadata_mapping,
 )
@@ -346,8 +347,8 @@ class ExecutableCoordinator(SubprocessCoordinator):
     """
 
     executables_root: list[pathlib.Path] = attrs.field(
-        converter=convert_roots,
-        factory=list,
+        default=ARTIFACT_ROOTS_NOT_CONFIGURED,
+        converter=convert_configured_roots,
     )
 
     @property

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

@@ -29,7 +29,11 @@
 import attrs
 import structlog
 
-from airflow.sdk.coordinators._bundle_metadata import convert_roots, validate_schema_version
+from airflow.sdk.coordinators._bundle_metadata import (
+    ARTIFACT_ROOTS_NOT_CONFIGURED,
+    convert_configured_roots,
+    validate_schema_version,
+)
 from airflow.sdk.coordinators._subprocess import SubprocessCoordinator
 
 if TYPE_CHECKING:
@@ -200,8 +204,8 @@ class JavaCoordinator(SubprocessCoordinator):
     java_executable: str = "java"
     jvm_args: list[str] = attrs.field(factory=list)
     jars_root: list[pathlib.Path] = attrs.field(
-        converter=convert_roots,
-        factory=list,
+        default=ARTIFACT_ROOTS_NOT_CONFIGURED,
+        converter=convert_configured_roots,
     )
     main_class: str = ""
 

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

@@ -26,7 +26,12 @@
 import attrs
 import structlog
 
-from airflow.sdk.coordinators._bundle_metadata import ResolvedBundle, convert_roots, walk_files
+from airflow.sdk.coordinators._bundle_metadata import (
+    ARTIFACT_ROOTS_NOT_CONFIGURED,
+    ResolvedBundle,
+    convert_configured_roots,
+    walk_files,
+)
 from airflow.sdk.coordinators._subprocess import SubprocessCoordinator
 from airflow.sdk.coordinators.node._bundle_reader import read_bundle
 
@@ -119,8 +124,8 @@ class NodeCoordinator(SubprocessCoordinator):
 
     node_executable: str = "node"
     bundles_root: list[pathlib.Path] = attrs.field(
-        converter=convert_roots,
-        factory=list,
+        default=ARTIFACT_ROOTS_NOT_CONFIGURED,
+        converter=convert_configured_roots,
     )
 
     @property

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

@@ -375,9 +375,10 @@ def test_executables_root_optional_defaults_to_empty(self):
         assert coordinator.executables_root == []
         assert coordinator.dag_bundle_name is None
 
-    def test_none_executables_root_normalized_to_empty(self):
-        coordinator = ExecutableCoordinator(executables_root=None)
-        assert coordinator.executables_root == []
+    @pytest.mark.parametrize("executables_root", [None, []], ids=["none", "empty-list"])
+    def test_explicit_empty_executables_root_raises(self, executables_root):
+        with pytest.raises(ValueError, match="must contain at least one path when provided"):
+            ExecutableCoordinator(executables_root=executables_root)
 
     def test_root_and_dag_bundle_name_are_mutually_exclusive(self, tmp_path):
         with pytest.raises(ValueError, match="at most one of 'executables_root' or 'dag_bundle_name'"):

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

@@ -244,6 +244,11 @@ def test_jars_root_optional_defaults_to_empty(self):
         assert coordinator.dag_bundle_name is None
         assert coordinator.main_class == ""
 
+    @pytest.mark.parametrize("jars_root", [None, []], ids=["none", "empty-list"])
+    def test_explicit_empty_jars_root_raises(self, jars_root):
+        with pytest.raises(ValueError, match="must contain at least one path when provided"):
+            JavaCoordinator(jars_root=jars_root)
+
     def test_explicit_root_does_not_require_main_class(self, tmp_path):
         coordinator = JavaCoordinator(jars_root=tmp_path)
         assert coordinator.main_class == ""

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

@@ -85,9 +85,10 @@ def test_bundles_root_optional_defaults_to_empty(self):
         assert coordinator.bundles_root == []
         assert coordinator.dag_bundle_name is None
 
-    def test_none_bundles_root_normalized_to_empty(self):
-        coordinator = NodeCoordinator(bundles_root=None)
-        assert coordinator.bundles_root == []
+    @pytest.mark.parametrize("bundles_root", [None, []], ids=["none", "empty-list"])
+    def test_explicit_empty_bundles_root_raises(self, bundles_root):
+        with pytest.raises(ValueError, match="must contain at least one path when provided"):
+            NodeCoordinator(bundles_root=bundles_root)
 
     def test_root_and_dag_bundle_name_are_mutually_exclusive(self):
         with pytest.raises(ValueError, match="at most one of 'bundles_root' or 'dag_bundle_name'"):
@@ -116,9 +117,8 @@ def test_selects_bundle_by_dag_id(self, tmp_path):
             bundles_root=tmp_path,
         )
 
-        command, schema_version = coordinator._build_execute_task_command(
-            what=_make_ti(dag_id="sales"), roots=[tmp_path]
-        )
+        with coordinator._set_scan_roots([tmp_path]):
+            command, schema_version = coordinator._build_execute_task_command(what=_make_ti(dag_id="sales"))
 
         assert command == ["/opt/node/bin/node", str(selected)]
         assert schema_version == SCHEMA_VERSION