Add dag_bundle_name option to subprocess coordinators

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

@@ -409,14 +409,32 @@ All ``kwargs`` in the ``coordinators`` config entry are passed to the
      - Default
      - Description
    * - ``executables_root``
-     - *(required)*
+     - *(optional)*
      - One or more directories scanned recursively for executable bundles. Accepts a string,
-       a path, or a list of strings/paths.
+       a path, or a list of strings/paths. When omitted, bundles are 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 executable bundles from. Mutually exclusive
+       with ``executables_root``.
    * - ``task_startup_timeout``
      - ``10.0``
      - Seconds to wait for the bundle subprocess to connect after launch. Increase this if your
        bundle startup is slow (e.g. on constrained hardware).
 
+.. note::
+
+  **Locating bundles.** ``executables_root`` and ``dag_bundle_name`` are mutually exclusive,
+  and both are optional:
+
+  * Set ``executables_root`` to scan explicit filesystem directories you manage yourself.
+  * Set ``dag_bundle_name`` to load bundles from a configured Dag bundle, so they are delivered
+    and versioned through the same bundle machinery as your Dags. The task uses the version that
+    bundle is on when it starts, pinned for the whole task.
+  * Leave both unset (the default) to load bundles from the **task's own** Dag bundle, pinned
+    to the version the run was created with.
+
 .. _go-sdk/limitations:
 
 Limitations

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

@@ -663,9 +663,15 @@ All ``kwargs`` in the ``coordinators`` config entry are passed to the
      - Default
      - Description
    * - ``jars_root``
-     - *(required)*
+     - *(optional)*
      - One or more directories scanned recursively for ``.jar`` files. Accepts a string,
-       a path, or a list of strings/paths.
+       a path, or a list of strings/paths. When omitted, JARs are 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 JARs from. Mutually exclusive with
+       ``jars_root``.
    * - ``java_executable``
      - ``"java"``
      - Path to the ``java`` binary.  Defaults to ``java`` on ``$PATH``.
@@ -674,15 +680,27 @@ All ``kwargs`` in the ``coordinators`` config entry are passed to the
      - Extra JVM arguments such as ``["-Xmx1g", "-Dsome.property=value"]``.
    * - ``main_class``
      - *(auto-detect)*
-     - Explicit entry-point class. If omitted,
-       :class:`~airflow.sdk.coordinators.java.JavaCoordinator` scans ``jars_root`` for a
-       JAR whose manifest sets ``Main-Class``. If multiple executable JARs are found the
-       result is non-deterministic; set ``main_class`` explicitly in that case.
+     - Explicit entry-point class. If omitted, the coordinator scans for a JAR whose
+       manifest sets ``Main-Class`` — in ``jars_root`` when set, otherwise across the
+       resolved Dag bundle. If multiple executable JARs match the result is
+       non-deterministic; set ``main_class`` explicitly in that case.
    * - ``task_startup_timeout``
      - ``10.0``
      - Seconds to wait for the JVM subprocess to connect after launch.  Increase this if your
        JVM startup is slow (e.g. on constrained hardware or with a large classpath).
 
+.. note::
+
+  **Locating JARs.** ``jars_root`` and ``dag_bundle_name`` are mutually exclusive, and both
+  are optional:
+
+  * Set ``jars_root`` to scan explicit filesystem directories you manage yourself.
+  * Set ``dag_bundle_name`` to load JARs from a configured Dag bundle, so they are delivered and
+    versioned through the same bundle machinery as your Dags. The task uses the version that bundle
+    is on when it starts, pinned for the whole task.
+  * Leave both unset (the default) to load JARs from the **task's own** Dag bundle, pinned to
+    the version the run was created with.
+
 .. note::
 
   The ``[sdk]`` configuration is read at startup, so changes to ``coordinators`` or

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

@@ -289,9 +289,15 @@ All ``kwargs`` in the ``coordinators`` config entry are passed to the
      - Default
      - Description
    * - ``bundles_root``
-     - *(required)*
+     - *(optional)*
      - One or more directories searched, in order, for a ``bundle.mjs`` (with embedded metadata, or with an
-       ``airflow-metadata.yaml`` sidecar). Accepts a string, a path, or a list of strings/paths.
+       ``airflow-metadata.yaml`` sidecar). 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). 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 ``bundle.mjs`` from. Mutually exclusive with
+       ``bundles_root``.
    * - ``node_executable``
      - ``"node"``
      - Path to the ``node`` binary. Defaults to ``node`` on ``$PATH``.
@@ -300,6 +306,18 @@ All ``kwargs`` in the ``coordinators`` config entry are passed to the
      - Seconds to wait for the Node.js subprocess to connect after launch. Increase this if your bundle
        startup is slow (e.g. on constrained hardware).
 
+.. note::
+
+  **Locating the bundle.** ``bundles_root`` and ``dag_bundle_name`` are mutually exclusive, and both
+  are optional:
+
+  * Set ``bundles_root`` to scan explicit filesystem directories you manage yourself.
+  * Set ``dag_bundle_name`` to load the bundle from a configured Dag bundle, so it is delivered
+    and versioned through the same bundle machinery as your Dags. The task uses the version that
+    bundle is on when it starts, pinned for the whole task.
+  * Leave both unset (the default) to load the bundle from the **task's own** Dag bundle, pinned to the
+    version the run was created with.
+
 Limitations
 -----------
 

airflow-core/src/airflow/dag_processing/bundles/manager.py

@@ -16,13 +16,14 @@
 # under the License.
 from __future__ import annotations
 
+import functools
 import importlib
 import logging
 import os
 import warnings
 from collections import defaultdict
 from pathlib import Path
-from typing import TYPE_CHECKING, Any, cast
+from typing import TYPE_CHECKING, Any, NamedTuple, cast
 
 from itsdangerous import URLSafeSerializer
 from pydantic import BaseModel, ValidationError
@@ -107,6 +108,34 @@ def _bundle_item_exc(msg):
     )
 
 
+class _BundleConfigSnapshot(NamedTuple):
+    """The configured Dag bundles, as names only and as full configs."""
+
+    configs: tuple[_ExternalBundleConfig, ...]
+    names: frozenset[str]
+
+
+@functools.cache
+def _load_bundle_config_snapshot() -> _BundleConfigSnapshot:
+    """Read and validate the configured Dag bundles, without importing their classes."""
+    config_list = conf.getjson("dag_processor", "dag_bundle_config_list")
+    if not config_list:
+        return _BundleConfigSnapshot(configs=(), names=frozenset())
+    if not isinstance(config_list, list):
+        raise AirflowConfigException(
+            "Section `dag_processor` key `dag_bundle_config_list` "
+            f"must be list but got {config_list.__class__}"
+        )
+    bundle_config_list = _parse_bundle_config(config_list)
+    if conf.getboolean("core", "LOAD_EXAMPLES"):
+        _add_example_dag_bundle(bundle_config_list)
+        _add_provider_example_dags_to_bundle(bundle_config_list)
+    return _BundleConfigSnapshot(
+        configs=tuple(bundle_config_list),
+        names=frozenset(cfg.name for cfg in bundle_config_list),
+    )
+
+
 def _parse_bundle_config(config_list) -> list[_ExternalBundleConfig]:
     bundles = {}
     for item in config_list:
@@ -274,18 +303,9 @@ def parse_config(self) -> None:
         if self._bundle_config:
             return
 
-        config_list = conf.getjson("dag_processor", "dag_bundle_config_list")
-        if not config_list:
+        bundle_config_list = _load_bundle_config_snapshot().configs
+        if not bundle_config_list:
             return
-        if not isinstance(config_list, list):
-            raise AirflowConfigException(
-                "Section `dag_processor` key `dag_bundle_config_list` "
-                f"must be list but got {config_list.__class__}"
-            )
-        bundle_config_list = _parse_bundle_config(config_list)
-        if conf.getboolean("core", "LOAD_EXAMPLES"):
-            _add_example_dag_bundle(bundle_config_list)
-            _add_provider_example_dags_to_bundle(bundle_config_list)
 
         for bundle_config in bundle_config_list:
             if bundle_config.team_name and not conf.getboolean("core", "multi_team"):
@@ -653,6 +673,16 @@ def get_bundle(
             name=name, version=version, version_data=version_data, **cfg_bundle.kwargs
         )
 
+    @classmethod
+    def is_bundle_configured(cls, name: str) -> bool:
+        """
+        Return whether *name* is a configured Dag bundle.
+
+        Deliberately reads configured names only: a caller validating a bundle name
+        must not depend on every *other* configured bundle's class being importable.
+        """
+        return name in _load_bundle_config_snapshot().names
+
     def get_all_dag_bundles(self) -> Iterable[BaseDagBundle]:
         """
         Get all DAG bundles.

airflow-core/tests/integration/otel/test_otel.py

@@ -330,7 +330,7 @@ def test_export_metrics_during_process_shutdown(self, capfd):
                 # Additional detail spans are deferred to follow-up PRs; tracked
                 # at https://linear.app/astronomer/issue/ACD-157.
                 {
-                    "_verify_bundle_access": "parse",
+                    "verify_bundle_access": "parse",
                     "parse": "startup",
                     "get_template_context": "startup",
                     "startup": "worker.task1",

airflow-core/tests/unit/dag_processing/bundles/test_dag_bundle_manager.py

@@ -29,7 +29,11 @@
 from sqlalchemy import func, select, update
 
 from airflow.dag_processing.bundles.base import BaseDagBundle
-from airflow.dag_processing.bundles.manager import DagBundlesManager, _guess_best_bundle_for_fileloc
+from airflow.dag_processing.bundles.manager import (
+    DagBundlesManager,
+    _guess_best_bundle_for_fileloc,
+    _load_bundle_config_snapshot,
+)
 from airflow.exceptions import AirflowConfigException
 from airflow.models.dag import DagModel
 from airflow.models.dag_version import DagVersion
@@ -167,6 +171,55 @@ def test_get_bundle():
     assert bundle.version is None
 
 
+@pytest.fixture(autouse=True)
+def _clear_bundle_config_snapshot_cache():
+    _load_bundle_config_snapshot.cache_clear()
+    yield
+    _load_bundle_config_snapshot.cache_clear()
+
+
+def _bundle_config_env(config) -> dict[str, str]:
+    return {
+        "AIRFLOW__CORE__LOAD_EXAMPLES": "False",
+        "AIRFLOW__DAG_PROCESSOR__DAG_BUNDLE_CONFIG_LIST": json.dumps(config),
+    }
+
+
+def test_is_bundle_configured():
+    """is_bundle_configured reports membership without constructing the bundle."""
+    with patch.dict(os.environ, _bundle_config_env(BASIC_BUNDLE_CONFIG)):
+        assert DagBundlesManager.is_bundle_configured("my-test-bundle") is True
+        assert DagBundlesManager.is_bundle_configured("bundle-that-doesn't-exist") is False
+
+
+def test_is_bundle_configured_does_not_import_bundle_class():
+    """A validly-named bundle whose class is unimportable is still reported as configured.
+
+    The check reads names only; the class is imported lazily at materialization,
+    so an unimportable classpath must not make a coordinator fail at construction.
+    """
+    config = [{"name": "artifacts", "classpath": "does.not.exist.Bundle", "kwargs": {}}]
+    with patch.dict(os.environ, _bundle_config_env(config)):
+        assert DagBundlesManager.is_bundle_configured("artifacts") is True
+
+
+def test_is_bundle_configured_empty_config():
+    with patch.dict(os.environ, _bundle_config_env([])):
+        assert DagBundlesManager.is_bundle_configured("anything") is False
+
+
+def test_is_bundle_configured_sees_implicitly_added_bundles():
+    """Bundles parse_config adds implicitly are configured for the name check too."""
+    with patch.dict(
+        os.environ,
+        {
+            "AIRFLOW__CORE__LOAD_EXAMPLES": "True",
+            "AIRFLOW__DAG_PROCESSOR__DAG_BUNDLE_CONFIG_LIST": json.dumps(BASIC_BUNDLE_CONFIG),
+        },
+    ):
+        assert DagBundlesManager.is_bundle_configured("example_dags") is True
+
+
 @pytest.fixture
 def clear_db():
     clear_db_dag_bundles()
@@ -183,9 +236,7 @@ def _get_bundle_names_and_active():
         ).all()
 
     # Initial add
-    with patch.dict(
-        os.environ, {"AIRFLOW__DAG_PROCESSOR__DAG_BUNDLE_CONFIG_LIST": json.dumps(BASIC_BUNDLE_CONFIG)}
-    ):
+    with conf_vars({("dag_processor", "dag_bundle_config_list"): json.dumps(BASIC_BUNDLE_CONFIG)}):
         manager = DagBundlesManager()
         manager.sync_bundles_to_db()
     assert _get_bundle_names_and_active() == [("my-test-bundle", True)]
@@ -210,9 +261,7 @@ def _get_bundle_names_and_active():
     assert session.scalar(select(func.count(ParseImportError.id))) == 0
 
     # Re-enable one that reappears in config
-    with patch.dict(
-        os.environ, {"AIRFLOW__DAG_PROCESSOR__DAG_BUNDLE_CONFIG_LIST": json.dumps(BASIC_BUNDLE_CONFIG)}
-    ):
+    with conf_vars({("dag_processor", "dag_bundle_config_list"): json.dumps(BASIC_BUNDLE_CONFIG)}):
         manager = DagBundlesManager()
         manager.sync_bundles_to_db()
     assert _get_bundle_names_and_active() == [
@@ -260,26 +309,20 @@ def _get_bundle_names_and_active():
         ).all()
 
     # Processor A: only knows "my-test-bundle".
-    with patch.dict(
-        os.environ, {"AIRFLOW__DAG_PROCESSOR__DAG_BUNDLE_CONFIG_LIST": json.dumps(BASIC_BUNDLE_CONFIG)}
-    ):
+    with conf_vars({("dag_processor", "dag_bundle_config_list"): json.dumps(BASIC_BUNDLE_CONFIG)}):
         DagBundlesManager().sync_bundles_to_db(deactivate_missing=False)
     assert _get_bundle_names_and_active() == [("my-test-bundle", True)]
 
     # Processor B: only knows "other-test-bundle". It must not touch A's bundle.
-    with patch.dict(
-        os.environ, {"AIRFLOW__DAG_PROCESSOR__DAG_BUNDLE_CONFIG_LIST": json.dumps(OTHER_BUNDLE_CONFIG)}
-    ):
+    with conf_vars({("dag_processor", "dag_bundle_config_list"): json.dumps(OTHER_BUNDLE_CONFIG)}):
         DagBundlesManager().sync_bundles_to_db(deactivate_missing=False)
     assert _get_bundle_names_and_active() == [
         ("my-test-bundle", True),
         ("other-test-bundle", True),
     ]
 
     # Processor A runs again: still must not disable B's bundle.
-    with patch.dict(
-        os.environ, {"AIRFLOW__DAG_PROCESSOR__DAG_BUNDLE_CONFIG_LIST": json.dumps(BASIC_BUNDLE_CONFIG)}
-    ):
+    with conf_vars({("dag_processor", "dag_bundle_config_list"): json.dumps(BASIC_BUNDLE_CONFIG)}):
         DagBundlesManager().sync_bundles_to_db(deactivate_missing=False)
     assert _get_bundle_names_and_active() == [
         ("my-test-bundle", True),
@@ -377,9 +420,7 @@ def test_bundle_model_render_url(clear_db, session):
 @conf_vars({("core", "LOAD_EXAMPLES"): "False"})
 def test_template_params_update_on_sync(clear_db, session):
     """Test that template parameters are updated when bundle configuration changes."""
-    with patch.dict(
-        os.environ, {"AIRFLOW__DAG_PROCESSOR__DAG_BUNDLE_CONFIG_LIST": json.dumps(TEMPLATE_BUNDLE_CONFIG)}
-    ):
+    with conf_vars({("dag_processor", "dag_bundle_config_list"): json.dumps(TEMPLATE_BUNDLE_CONFIG)}):
         manager = DagBundlesManager()
         manager.sync_bundles_to_db()
 
@@ -402,9 +443,7 @@ def test_template_params_update_on_sync(clear_db, session):
         }
     ]
 
-    with patch.dict(
-        os.environ, {"AIRFLOW__DAG_PROCESSOR__DAG_BUNDLE_CONFIG_LIST": json.dumps(updated_config)}
-    ):
+    with conf_vars({("dag_processor", "dag_bundle_config_list"): json.dumps(updated_config)}):
         manager = DagBundlesManager()
         manager.sync_bundles_to_db()
 
@@ -421,9 +460,7 @@ def test_template_params_update_on_sync(clear_db, session):
 def test_template_update_on_sync(clear_db, session):
     """Test that templates are updated when bundle configuration changes."""
     # First, sync with initial template
-    with patch.dict(
-        os.environ, {"AIRFLOW__DAG_PROCESSOR__DAG_BUNDLE_CONFIG_LIST": json.dumps(TEMPLATE_BUNDLE_CONFIG)}
-    ):
+    with conf_vars({("dag_processor", "dag_bundle_config_list"): json.dumps(TEMPLATE_BUNDLE_CONFIG)}):
         manager = DagBundlesManager()
         manager.sync_bundles_to_db()
 
@@ -446,9 +483,7 @@ def test_template_update_on_sync(clear_db, session):
         }
     ]
 
-    with patch.dict(
-        os.environ, {"AIRFLOW__DAG_PROCESSOR__DAG_BUNDLE_CONFIG_LIST": json.dumps(updated_config)}
-    ):
+    with conf_vars({("dag_processor", "dag_bundle_config_list"): json.dumps(updated_config)}):
         manager = DagBundlesManager()
         manager.sync_bundles_to_db()
 

contributing-docs/30_new_language_sdk.rst

@@ -120,6 +120,12 @@ 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.
 
+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
 ~~~~~~~~~~~~~~~~~