Move Dag cycle detection from the Task SDK to shared

[jason810496]

Stack (bottom to top): #74034, #74040, #74041, #74042, #74035, #74043, #74036, #74037, #73841, #73845, #73846, #73847

• related: Add ADR for Lang SDK Dag and mixed-language Task processing flow #71929
• Part of the native Dag e2e stack. This is the bottom layer and builds on main.
• next: Validate a serialized Dag and fill its unset settings from the Airflow config #74041 uses it to reject a native Dag whose task graph has a cycle.

Why

Move the Cycle detection part to share so that all the further SDKs native Dag can be validated by the same single source.

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

---

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

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

[jason810496]

The current static check failure is not related and will be resolved once #73963 (review) finish.

[uranusjr]

From what I can tell this does not make anything in core call the shared code yet. Is this planned to be done in a later PR? (The PR title probably should be adjusted slightly if that’s the case.)

[pierrejeambrun]

LGTM.

From what I can tell this does not make anything in core call the shared code yet. Is this planned to be done in a later PR? (The PR title probably should be adjusted slightly if that’s the case.)

I updated the description which is misleading at this point. Core calling this is done in #74041 I believe and then wired in the dag processor in #74035 (by calling validate_serialized_dag)