Move the Dag file processor's shared plumbing into a base class

.agents/skills/airflow-java-sdk/SKILL.md

@@ -50,8 +50,9 @@ subclasses must only import from `org.apache.airflow.sdk`; any import of
 
 ## Bundle composition and coordinator discovery
 
-A **bundle** is a directory of JAR files (typically `build/bundle/`) placed on the coordinator's
-`jars_root`. The coordinator scans the directory at task-dispatch time to find:
+A **bundle** is a directory of JAR files (typically `build/bundle/`) placed in the Dag bundle named
+by the coordinator's `task_handler_bundle_name` (the task's own Dag bundle when unset). The
+coordinator scans that Dag bundle at task-dispatch time to find:
 
 1. **`Main-Class`** (standard JAR manifest attribute) — the fully-qualified class name of the
    entry point that the coordinator invokes with `java -classpath … <Main-Class> --comm … --logs …`.
@@ -65,7 +66,7 @@ A **bundle** is a directory of JAR files (typically `build/bundle/`) placed on t
    `runtimeClasspath` and copies it into the shadow JAR manifest. In thin-JAR mode (`fatJar =
    false`), the value stays in the `airflow-sdk` JAR deployed alongside the bundle JAR.
 
-The Python coordinator (`JavaCoordinator`) scans every JAR under `jars_root` with
+The Python coordinator (`JavaCoordinator`) scans every JAR in that Dag bundle with
 `_JarInfo.find()`, reads `META-INF/MANIFEST.MF` out of each ZIP, and collects `Main-Class` and
 `Airflow-Supervisor-Schema-Version` from whichever JARs carry them. The resolved schema version
 is then passed as the `schema_version` return value from `_build_execute_task_command`, which
@@ -74,7 +75,7 @@ the base `SubprocessCoordinator` uses to negotiate the supervisor wire protocol.
 If `main_class` is set explicitly on the `JavaCoordinator` instance (via `[sdk] coordinators`
 kwargs), the scan uses it as a filter; otherwise the first JAR with a `Main-Class` attribute
 wins. Either way, `Airflow-Supervisor-Schema-Version` must be present in at least one JAR in
-`jars_root` or startup fails.
+the Dag bundle or startup fails. Every JAR in the Dag bundle goes on one classpath.
 
 ---
 
@@ -117,7 +118,7 @@ E2E_TEST_MODE=java_sdk uv run --project airflow-e2e-tests pytest \
 
 `coordinator.py` extends `SubprocessCoordinator`. The only method subclasses must implement is
 `_build_execute_task_command`, which returns `(argv, schema_version)`. Look at the existing
-implementation for how `jars_root`, `java_executable`, `jvm_args`, and `main_class` are
+implementation for how the scanned Dag bundle, `java_executable`, `jvm_args`, and `main_class` are
 assembled into the command. Do not reach into the JVM process from Python beyond what this
 method provides.
 

airflow-core/adr/lang-sdk/0013-persisted-task-handler-bindings.md

@@ -21,7 +21,7 @@
 
 ## Status
 
-Proposed
+Accepted
 
 ## Context
 
@@ -157,31 +157,34 @@ backs many handlers and its fingerprint must have exactly one value.
 
 ```sql
 CREATE TABLE lang_sdk_task_handler_artifact (
-    id                UUID          NOT NULL,
-    bundle_name       VARCHAR(250)  NOT NULL,   -- the task_handler_bundle_name it was found in
-    relative_fileloc  VARCHAR(2000) NOT NULL,   -- path within that bundle
-    size_bytes        BIGINT        NOT NULL,   -- cheap fingerprint tier
-    cache_digest      VARCHAR(64)   NOT NULL,   -- content fingerprint tier; see "The fast path"
-    last_probed_at    TIMESTAMP     NOT NULL,
-    PRIMARY KEY (id),
-    CONSTRAINT lstha_bundle_fileloc_uq UNIQUE (bundle_name, relative_fileloc)
+    id                     UUID          NOT NULL,
+    bundle_name            VARCHAR(250)  NOT NULL,   -- the task_handler_bundle_name it was found in
+    relative_fileloc       VARCHAR(2000) NOT NULL,   -- path within that bundle
+    relative_fileloc_hash  VARCHAR(32)   NOT NULL,   -- md5 of relative_fileloc; the path is too long to index
+    size_bytes             BIGINT        NOT NULL,   -- cheap fingerprint tier
+    cache_digest           VARCHAR(128)  NULL,       -- content fingerprint tier, NULL if none is stored; see "The fast path"
+    task_handlers          JSON          NOT NULL,   -- {dag_id: [TaskHandlerDeclaration, ...]}, the probe answer
+    last_probed_at         TIMESTAMP     NOT NULL,
+    CONSTRAINT lang_sdk_task_handler_artifact_pkey PRIMARY KEY (id),
+    CONSTRAINT lang_sdk_task_handler_artifact_bundle_fileloc_uq UNIQUE (bundle_name, relative_fileloc_hash)
 );
 
 CREATE TABLE lang_sdk_task_handler (
-    dag_id                VARCHAR(250)  NOT NULL,
-    task_id               VARCHAR(250)  NOT NULL,
-    artifact_id           UUID          NOT NULL,
-    dag_bundle_name       VARCHAR(250)  NOT NULL,   -- the *Python* file that owns this row
-    dag_relative_fileloc  VARCHAR(2000) NOT NULL,   -- ditto
-    handler_params        JSON          NOT NULL,   -- list[TaskHandlerParam], ordered
-    PRIMARY KEY (dag_id, task_id),
-    CONSTRAINT lsth_dag_fkey FOREIGN KEY (dag_id)
+    dag_id                     VARCHAR(250)  NOT NULL,
+    task_id                    VARCHAR(250)  NOT NULL,
+    artifact_id                UUID          NOT NULL,
+    dag_bundle_name            VARCHAR(250)  NOT NULL,   -- the *Python* file that owns this row
+    dag_relative_fileloc       VARCHAR(2000) NOT NULL,   -- ditto
+    dag_relative_fileloc_hash  VARCHAR(32)   NOT NULL,   -- md5 of dag_relative_fileloc
+    CONSTRAINT lang_sdk_task_handler_pkey PRIMARY KEY (dag_id, task_id),
+    CONSTRAINT lang_sdk_task_handler_dag_id_fkey FOREIGN KEY (dag_id)
         REFERENCES dag (dag_id) ON DELETE CASCADE,
-    CONSTRAINT lsth_artifact_fkey FOREIGN KEY (artifact_id)
+    CONSTRAINT lang_sdk_task_handler_artifact_id_fkey FOREIGN KEY (artifact_id)
         REFERENCES lang_sdk_task_handler_artifact (id)
 );
-CREATE INDEX idx_lsth_dag_file ON lang_sdk_task_handler (dag_bundle_name, dag_relative_fileloc);
-CREATE INDEX idx_lsth_artifact_id ON lang_sdk_task_handler (artifact_id);
+CREATE INDEX idx_lang_sdk_task_handler_dag_file
+    ON lang_sdk_task_handler (dag_bundle_name, dag_relative_fileloc_hash);
+CREATE INDEX idx_lang_sdk_task_handler_artifact_id ON lang_sdk_task_handler (artifact_id);
 ```
 
 `PRIMARY KEY (dag_id, task_id)` is the conflict guard: two artifacts claiming the same task cannot
@@ -194,11 +197,10 @@ joining through `DagModel`: `dag.relative_fileloc` is not indexed, and the codeb
 that querying it means "a sequential scan of dag"
 ([`reassign_dags_with_unconfigured_bundles`](https://github.com/apache/airflow/blob/79991cd4db0c9346a28b23c453377f6df0c6b4ed/airflow-core/src/airflow/dag_processing/bundles/manager.py#L478)).
 
-`handler_params` stores what the runtime declared, so a changed Python file can be re-validated
-against a cached declaration with no subprocess. It is deliberately **not** called `arg_bindings`:
-that name already denotes the Python side of the comparison (`XComArgBinding` / `LiteralArgBinding`,
-carrying wiring and values, [`build_arg_bindings`](https://github.com/apache/airflow/blob/79991cd4db0c9346a28b23c453377f6df0c6b4ed/airflow-core/src/airflow/serialization/stub_arg_bindings.py#L221-L286)),
-and reusing it would make the validation read as comparing a thing to itself.
+`task_handlers` caches the artifact's whole probe answer, every handler it registers keyed by
+`dag_id`, so a changed Python file can be re-validated against it with no subprocess. It is written
+only from a probe, together with `size_bytes` and `cache_digest`, so the answer always belongs to the
+fingerprint beside it. A binding row is then only a mapping from a stub task to its artifact.
 
 `cache_digest` is **opaque and coordinator-defined**, not "SHA-256 of the file".
 

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

@@ -48,7 +48,8 @@ Prerequisites
 
 * Go 1.24 or later to build and pack bundles. This is a build-time requirement only; the worker that runs a
   packed bundle needs no Go toolchain, because the bundle is a self-contained native executable.
-* The packed bundle must be accessible from the Airflow worker, under a directory the coordinator scans.
+* The packed bundle must be accessible from the Airflow worker and the Dag processor, in the Dag bundle the
+  coordinator scans, and built for the operating system and CPU architecture of both.
 * The ``apache-airflow-task-sdk`` package (installed with Airflow) provides the coordinator; no additional
   Python packages are needed.
 
@@ -178,33 +179,46 @@ to pass on with ``bundle.Register(reports.Handlers()...)``.
 Coordinator configuration
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~
 
-Register the coordinator and route the queue to it under ``[sdk]`` in ``airflow.cfg`` (or the equivalent
-``AIRFLOW__SDK__*`` environment variables):
+Register a Dag bundle for the packed bundles, register the coordinator, and route the queue to it in
+``airflow.cfg`` (or the equivalent ``AIRFLOW__*`` environment variables):
 
 .. code-block:: ini
 
+    [dag_processor]
+    dag_bundle_config_list = [
+        {"name": "dags-folder", "classpath": "airflow.dag_processing.bundles.local.LocalDagBundle", "kwargs": {}},
+        {
+          "name": "go-task-handlers",
+          "classpath": "airflow.dag_processing.bundles.local.LocalDagBundle",
+          "kwargs": {"path": "/opt/airflow/go-task-handlers"}
+        }
+      ]
+
     [sdk]
     coordinators = {
       "go": {
         "classpath": "airflow.sdk.coordinators.executable.ExecutableCoordinator",
-        "kwargs": {"executables_root": ["~/airflow/executable-bundles"]}
+        "kwargs": {"task_handler_bundle_name": "go-task-handlers"}
       }
     }
     queue_to_coordinator = {"golang": "go"}
 
-``executables_root`` is one or more directories the coordinator scans for bundles; ``queue_to_coordinator``
-routes stub tasks with ``queue="golang"`` to this Go coordinator. See :ref:`go-sdk/coordinator-config` for
-the full list of accepted ``kwargs``.
+``task_handler_bundle_name`` names the Dag bundle the coordinator scans for packed bundles;
+``queue_to_coordinator`` routes stub tasks with ``queue="golang"`` to this Go coordinator. See
+:ref:`go-sdk/coordinator-config` for the full list of accepted ``kwargs`` and how bundles are located.
 
 There is no separate Go worker to run: the Airflow worker forks the bundle binary once per task instance.
 
 .. note::
 
-  The coordinator is part of the Airflow worker, so the ``[sdk]`` config (and the bundle files in
-  ``executables_root``) only need to be present wherever tasks actually execute. With ``CeleryExecutor``,
-  setting it on the Celery workers is sufficient. With ``LocalExecutor``, tasks run inside the scheduler
-  process, so it must be set where the scheduler can read it. The API server and Dag processor do not need
-  it.
+  The ``[sdk]`` config and the packed bundle files must be present wherever tasks execute and on the Dag
+  processor. With ``CeleryExecutor``, tasks execute on the Celery workers; with ``LocalExecutor``, they run
+  inside the scheduler process. The Dag processor checks the stub tasks of each Python Dag against the task
+  handlers the packed bundles register, so it runs them too and needs bundles built for its operating
+  system and CPU architecture. The API server does not need any of it. Register the Dag bundle in
+  ``[dag_processor] dag_bundle_config_list`` on every component, like your other Dag bundles: the worker
+  and the Dag processor resolve ``task_handler_bundle_name`` through it, and wherever the ``[sdk]`` config
+  is read it is rejected if the name is missing there.
 
 Writing tasks
 -------------
@@ -444,24 +458,24 @@ Build and pack in one step; any flags after ``--`` are forwarded verbatim to ``g
 
     go tool airflow-go-pack ./example/bundle -- -trimpath -tags=prod
 
-Use ``--output <path>`` to write the packed bundle straight into a directory the coordinator scans
-(``executables_root``):
+Use ``--output <path>`` to write the packed bundle straight into the directory of the Dag bundle the
+coordinator scans (see `Deploying`_):
 
 .. code-block:: bash
 
-    go tool airflow-go-pack --output ~/airflow/executable-bundles/sample-dag-bundle ./example/bundle
+    go tool airflow-go-pack --output /opt/airflow/go-task-handlers/sample-dag-bundle ./example/bundle
 
 Cross-platform builds
 ~~~~~~~~~~~~~~~~~~~~~~~
 
-The worker that runs a bundle often uses a different operating system or CPU architecture than your build
-machine (for example, deploying to a Linux host from an Apple-silicon ``darwin/arm64`` laptop). Pass
-``--goos`` / ``--goarch`` and the packer cross-builds for you:
+The worker and the Dag processor that run a bundle often use a different operating system or CPU
+architecture than your build machine (for example, deploying to a Linux host from an Apple-silicon
+``darwin/arm64`` laptop). Pass ``--goos`` / ``--goarch`` and the packer cross-builds for you:
 
 .. code-block:: bash
 
     go tool airflow-go-pack --goos linux --goarch amd64 \
-      --output ~/airflow/executable-bundles/sample-dag-bundle \
+      --output /opt/airflow/go-task-handlers/sample-dag-bundle \
       ./example/bundle
 
 Alternatively, pack a pre-built binary with ``--executable`` / ``--source``. The packer normally execs the
@@ -485,12 +499,16 @@ with ``--airflow-metadata``:
 Deploying
 ~~~~~~~~~
 
-Copy or mount the packed bundle into a directory listed in the coordinator's ``executables_root``. The
-:class:`~airflow.sdk.coordinators.executable.ExecutableCoordinator` scans those directories recursively,
+Copy or mount the packed bundle into the Dag bundle named by the coordinator's ``task_handler_bundle_name``.
+The :class:`~airflow.sdk.coordinators.executable.ExecutableCoordinator` scans that Dag bundle recursively,
 matches the incoming ``dag_id`` against each bundle's manifest, verifies the bundle's integrity hash, and
 launches the matching bundle. Bundles are identified by the trailer magic, not by filename (no extension on
 Linux/macOS, ``.exe`` on Windows), so the file name on the worker is irrelevant.
 
+Only files with the executable bit set are considered, so the packed bundles need a Dag bundle that keeps
+it. A ``LocalDagBundle`` over a directory you manage does; object-store Dag bundles such as ``S3DagBundle``
+do not.
+
 .. _go-sdk/coordinator-config:
 
 :class:`~airflow.sdk.coordinators.executable.ExecutableCoordinator` configuration
@@ -506,32 +524,29 @@ All ``kwargs`` in the ``coordinators`` config entry are passed to the
    * - Parameter
      - Default
      - Description
-   * - ``executables_root``
-     - *(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). 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_handler_bundle_name``
+     - *(task's own Dag bundle)*
+     - Name of the Dag bundle scanned recursively for executable bundles. It is used only by
+       mixed-language Dags, to locate the task handlers for the ``@task.stub`` tasks of a Python Dag;
+       Dags defined natively in a language SDK do not use it. It must be registered in
+       ``[dag_processor] dag_bundle_config_list``. It is checked when the ``[sdk]`` configuration is
+       loaded, so a typo fails there rather than on the first task.
    * - ``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:
+  **Locating bundles.** The packed bundles for the ``@task.stub`` tasks of a Python Dag are read from a Dag
+  bundle, so they are delivered, refreshed and versioned by the same machinery as your Dags.
 
-  * 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.
+  * The expected layout is a separate Dag bundle for the packed bundles, named by
+    ``task_handler_bundle_name``, rather than the Dag bundle that holds your ``.py`` files. The task
+    uses the version that Dag bundle is on when it starts, pinned for the whole task.
+  * If ``task_handler_bundle_name`` is unset, packed bundles are read from the **task's own** Dag
+    bundle, pinned to the version the run was created with.
+  * The Dag bundle must keep the executable bit (see `Deploying`_).
 
 .. _go-sdk/limitations:
 

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

@@ -155,9 +155,9 @@ Coordinators are registered in ``airflow.cfg`` (or via environment variables) un
             }
         }
 
-    The ``classpath`` value must be importable by the worker.  The ``kwargs`` are passed directly
-    to the coordinator's constructor.  See the language-specific guide for the accepted kwargs
-    of each coordinator (e.g. :ref:`java-sdk/coordinator-config` for
+    The ``classpath`` value must be importable by the worker and the Dag processor.  The ``kwargs``
+    are passed directly to the coordinator's constructor.  See the language-specific guide for the
+    accepted kwargs of each coordinator (e.g. :ref:`java-sdk/coordinator-config` for
     :class:`~airflow.sdk.coordinators.java.JavaCoordinator`).
 
     ``extra`` is an optional object for any additional information you want to associate with a