Share Dag cycle detection between the Task SDK and core

[jason810496]

Stack (bottom to top): #73841, #74004, #73842, #73843, #73844, #73845, #73846, #73847

• related: Add ADR for Lang SDK Dag and mixed-language Task processing flow #71929
• Part of the native Dag e2e stack. Builds on [DO NOT MERGE] Mirror of #74020 for the native Dag stack #73841 (layer diff).
• next: Parse native Lang-SDK Dags with their coordinator's runtime #73842 uses it to reject a native Dag whose task graph has a cycle.

Why

Cycle detection only runs for Dags that go through the Python SDK's DAG object. A Dag from another language SDK reaches core already serialized, so nothing rejects a cycle in it.

How

Core cannot reuse DAG.check_cycle as it is: it is a method on the authoring class and reads task objects, while core holds serialized data it should not hydrate just to run this check.

• The traversal moves to shared/dagnode, which task-sdk and airflow-core already depend on.
• The graph is described by callbacks, not node objects, so a caller answers from whatever data it holds.
• It returns the offending task id instead of raising. AirflowDagCycleException lives in the Task SDK, and importing it here would invert the dependency, so each caller raises its own error.

from airflow._shared.dagnode.cycle import detect_cycle

downstream = {"extract": ["load"], "load": ["extract"]}
detect_cycle(downstream, downstream.__getitem__)  # "load", or None when there is no cycle

What

• detect_cycle(node_ids, downstream_of) in shared/dagnode/src/airflow_shared/dagnode/cycle.py, with its unit tests.
• DAG.check_cycle() delegates to it and keeps its message, Cycle detected in Dag: <dag_id>. Faulty task: <task_id>.

No behavior change: the walk still follows downstream edges only and stays iterative, and the existing TestCycleTester tests pass unchanged. No newsfragment, since nothing user-visible changes.

How to test

uv run --project task-sdk pytest task-sdk/tests/task_sdk/definitions/test_dag.py -k cycle -xvs

shared/dagnode/tests/dagnode/test_cycle.py covers detect_cycle itself.

---

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

• Yes, with help of Claude Code (Claude Opus 4.5) following the guidelines

[jason810496]

Replaced by #74034. GitHub marked this PR as merged into a stack branch when the stack was reordered; nothing was merged into main.