[jason810496]

Stack (bottom to top): #73970, #73971, #73972, #73973, #73974, #73975, #73976, #73977, #74030, #74031, #74032, #74033, #74067

Why

The Dag-file parse will resolve each @task.stub task against the artifacts in its coordinator's Dag bundle (ADR-0013 Flow 1, steps 4 and 5). To do that it must find the artifacts without running them, and skip the probe subprocess for an artifact whose recorded answer still holds. This PR adds both. Nothing calls them yet; a later PR wires them into _parse_file.

What changes

coordinator = get_coordinator_manager().get_coordinator("jdk-17")
candidates = coordinator.list_task_handler_candidates(bundle_path)
# [TaskHandlerCandidate(rel_path="airflow-java-sdk-scala-spark-example.jar", size_bytes=21135,
#                       cache_digest="067c98e6...", error=None)]

plan = plan_task_handler_probes(
    bundle_name="java-task-handlers", candidates=candidates, known_artifacts=request.known_artifacts
)
plan.cached    # recorded answers (TaskHandlerArtifact) that still hold
plan.probe     # candidates to probe
plan.rejected  # candidates the listing found unusable

• Listing. BaseCoordinator.list_task_handler_candidates(bundle_path) raises NotImplementedError by default. SubprocessCoordinator walks the bundle in sorted order and asks a new per-language hook, _read_task_handler_candidate(path, *, rel_path), about each file. A file reached through symlinks is listed once, under the first path walked, and a coordinator without the hook raises NotImplementedError before walking. The hook reads what marks the artifact and its stored digest, takes the size from the same open file, and never hashes the artifact.
  • Go: any file that ends in the AFBNDL01 trailer, whatever its executable bit. A bundle without the bit, or with a trailer this runtime cannot run, is listed with an error. Object-store Dag bundles drop the bit, and such a bundle used to be skipped silently.
  • Java: a .jar whose manifest carries Airflow-Cache-Digest. In the Scala Spark example bundle, 6 of 167 JARs declare Main-Class, 5 of them dependencies, and only the bundle JAR carries the digest. A digest JAR without Main-Class is listed with an error. A JAR whose manifest cannot be read (an encrypted entry, an unsupported compression, corrupt LZMA data) is skipped, like any other unreadable JAR.
  • TS: a .min.mjs whose first line starts with the //# airflowBundle= marker. One whose layout line is invalid is listed with an error. A .min.mjs without the marker may be an unrelated minified module, so it is skipped.
• Digests. Go's read_cache_digest now shares its reader with the listing. The new TS read_cache_digest is the SHA-256 of the whole layout line, which records every region's SHA-256, so head -n1 bundle.min.mjs | sha256sum reproduces it. Java reads the manifest attribute. A stored digest says "unchanged", not "intact", so the size is compared too, and execution's integrity check is unchanged. A digest longer than 128 characters, the column's length, becomes None, so that artifact is probed on every parse.
• Decision. plan_task_handler_probes decides each candidate on its own, with no database:
  • a candidate with a listing error is rejected, even when a recorded answer matches it;
  • a candidate with the same path, size and non-null stored digest as a known artifact of its bundle reuses that artifact's whole answer;
  • any other candidate, including one that stores no digest, is probed.
• Docs. A contributor-guide subsection for the hook, a cache digest paragraph in ts-bundle-spec.rst, one sentence each in the JavaCoordinator and ExecutableCoordinator docstrings, and ADR-0013's fast path, Flow 1 steps 4 to 8 and Consequences, plus the Java marker in ADR-0014.

How to test

uv run --project task-sdk pytest task-sdk/tests/task_sdk/coordinators task-sdk/tests/task_sdk/execution_time/test_coordinator.py -q
uv run --project airflow-core --with-editable shared/secrets_masker pytest airflow-core/tests/unit/dag_processing/test_task_handler_fast_path.py -q

Ran:

• task-sdk coordinator and test_coordinator.py tests: 299 passed.
• The airflow-core dag_processing suite on SQLite: 680 passed, 13 skipped (the opt-in real probes and a Postgres/MySQL-only lock test).
• Listing the locally built bundles: the Scala Spark example (167 JARs) gives one candidate in 182 ms, each fat-JAR bundle (example, e2e, k8s) one, and the TS fixture bundle-v1.min.mjs one, whose digest equals head -n1 | sha256sum.
• mypy-task-sdk and mypy-airflow-core on all files, and the prek pre-commit hooks on the changed files, passed.

---

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)

Generated-by: Claude Code (Opus 5.5) following the guidelines