From 0d27df7aa2da4b6d19d99bf7dfa32356067a9428 Mon Sep 17 00:00:00 2001 From: Dheeraj Turaga Date: Fri, 11 Sep 2026 14:19:28 -0500 Subject: [PATCH 1/3] Reject cyclic TaskGroup dependencies during Dag parsing TaskGroup dependency projections must remain acyclic because downstream consumers require an unambiguous group ordering. --- airflow-core/docs/core-concepts/dags.rst | 5 +++ .../tests/unit/dag_processing/test_dagbag.py | 31 +++++++++++++++++ task-sdk/src/airflow/sdk/definitions/dag.py | 12 ++++++- .../tests/task_sdk/definitions/test_dag.py | 34 +++++++++++++++++++ 4 files changed, 81 insertions(+), 1 deletion(-) diff --git a/airflow-core/docs/core-concepts/dags.rst b/airflow-core/docs/core-concepts/dags.rst index ab397f3532c26..d3fd36c0f8c18 100644 --- a/airflow-core/docs/core-concepts/dags.rst +++ b/airflow-core/docs/core-concepts/dags.rst @@ -585,6 +585,11 @@ Dependency relationships can be applied across all tasks in a TaskGroup with the group1() >> task3 +Dependencies between sibling tasks and TaskGroups must remain acyclic when each TaskGroup is treated as a +single unit. For example, if any task in ``group1`` is upstream of a task in ``group2``, another task in +``group2`` cannot be upstream of a task in ``group1``. Airflow reports this as a Dag parsing error even when +the individual task dependencies do not form a cycle. + TaskGroup also supports ``default_args`` like Dag, it will overwrite the ``default_args`` in Dag level: .. code-block:: python diff --git a/airflow-core/tests/unit/dag_processing/test_dagbag.py b/airflow-core/tests/unit/dag_processing/test_dagbag.py index cb110ef5c7fda..0856f6e304fa5 100644 --- a/airflow-core/tests/unit/dag_processing/test_dagbag.py +++ b/airflow-core/tests/unit/dag_processing/test_dagbag.py @@ -1025,6 +1025,37 @@ def basic_cycle(): self.validate_dags(test_dag, found_dags, dagbag, should_be_found=False) assert file_path in dagbag.import_errors + def test_skip_task_group_dependency_cycle_dags(self, tmp_path): + def task_group_dependency_cycle(): + import datetime + + from airflow.providers.standard.operators.empty import EmptyOperator + from airflow.sdk import DAG, TaskGroup + + with DAG( + "task_group_dependency_cycle", + schedule=None, + start_date=datetime.datetime(2016, 1, 1), + ) as dag: + with TaskGroup("left"): + left_source = EmptyOperator(task_id="left_source") + left_sink = EmptyOperator(task_id="left_sink") + with TaskGroup("right"): + right_source = EmptyOperator(task_id="right_source") + right_sink = EmptyOperator(task_id="right_sink") + + left_source >> right_sink + right_source >> left_sink + + return dag + + test_dag = task_group_dependency_cycle() + + dagbag, found_dags, file_path = self.process_dag(task_group_dependency_cycle, tmp_path) + + self.validate_dags(test_dag, found_dags, dagbag, should_be_found=False) + assert "TaskGroup dependency cycle detected" in dagbag.import_errors[file_path] + def test_process_file_with_none(self, tmp_path): """ test that process_file can handle Nones diff --git a/task-sdk/src/airflow/sdk/definitions/dag.py b/task-sdk/src/airflow/sdk/definitions/dag.py index 633feccb0483a..2c08c1ebcf071 100644 --- a/task-sdk/src/airflow/sdk/definitions/dag.py +++ b/task-sdk/src/airflow/sdk/definitions/dag.py @@ -1138,7 +1138,7 @@ def add_result(self, xcom_arg: X) -> X: def check_cycle(self) -> None: """ - Check to see if there are any cycles in the Dag. + Check to see if there are any task or TaskGroup dependency cycles in the Dag. :raises AirflowDagCycleException: If cycle is found in the Dag. """ @@ -1177,6 +1177,16 @@ def _check_adjacent_tasks(task_id, current_task): else: path_stack.append(child_to_check) + task_group_dict = self.task_group.get_task_group_dict() + for task_group in task_group_dict.values(): + try: + task_group.topological_sort(group_dict=task_group_dict) + except AirflowDagCycleException: + group_id = task_group.group_id or "" + raise AirflowDagCycleException( + f"TaskGroup dependency cycle detected in Dag: {self.dag_id}. Faulty TaskGroup: {group_id}" + ) from None + def cli(self): """Exposes a CLI specific to this Dag.""" self.check_cycle() diff --git a/task-sdk/tests/task_sdk/definitions/test_dag.py b/task-sdk/tests/task_sdk/definitions/test_dag.py index 07b8c8186c8e5..560a945cf790a 100644 --- a/task-sdk/tests/task_sdk/definitions/test_dag.py +++ b/task-sdk/tests/task_sdk/definitions/test_dag.py @@ -997,6 +997,40 @@ def test_cycle_task_group_with_edge_labels(self): assert not dag.check_cycle() + @pytest.mark.parametrize( + ("nested", "faulty_group"), + [ + pytest.param(False, "", id="root"), + pytest.param(True, "parent", id="nested"), + ], + ) + def test_cycle_between_sibling_task_groups(self, nested, faulty_group): + dag = DAG("dag", schedule=None, start_date=DEFAULT_DATE, default_args={"owner": "owner1"}) + + def add_groups(): + with TaskGroup("left"): + left_source = DoNothingOperator(task_id="left_source") + left_sink = DoNothingOperator(task_id="left_sink") + with TaskGroup("right"): + right_source = DoNothingOperator(task_id="right_source") + right_sink = DoNothingOperator(task_id="right_sink") + + left_source >> right_sink + right_source >> left_sink + + with dag: + if nested: + with TaskGroup("parent"): + add_groups() + else: + add_groups() + + with pytest.raises( + AirflowDagCycleException, + match=rf"TaskGroup dependency cycle detected in Dag: dag\. Faulty TaskGroup: {faulty_group}", + ): + dag.check_cycle() + class TestDagGetItem: def test_getitem_returns_task(self): From b3c467d381d7c41f7dad6f7db56cad0880e765b9 Mon Sep 17 00:00:00 2001 From: Dheeraj Turaga Date: Mon, 14 Sep 2026 15:44:52 -0500 Subject: [PATCH 2/3] Clarify which TaskGroup edges trigger cycle rejection and add newsfragment Review feedback on the parse-time TaskGroup cycle check: the doc's example claimed any cross-group task pairing triggers rejection, but the check only projects edges landing on a group's root tasks, so it understated what is and isn't caught. Reworded with an accurate example, and added the newsfragment the check's user-facing behavior change was missing, noting that #69933 is what turned a pre-existing cyclic-TaskGroup Dag from a silent rendering quirk into an unhandled Grid/Graph error, which is what this PR now catches earlier at parse time instead. --- airflow-core/docs/core-concepts/dags.rst | 23 ++++++++++++++++--- .../newsfragments/73087.significant.rst | 17 ++++++++++++++ 2 files changed, 37 insertions(+), 3 deletions(-) create mode 100644 airflow-core/newsfragments/73087.significant.rst diff --git a/airflow-core/docs/core-concepts/dags.rst b/airflow-core/docs/core-concepts/dags.rst index d3fd36c0f8c18..3b2fb983ddbe0 100644 --- a/airflow-core/docs/core-concepts/dags.rst +++ b/airflow-core/docs/core-concepts/dags.rst @@ -586,9 +586,26 @@ Dependency relationships can be applied across all tasks in a TaskGroup with the group1() >> task3 Dependencies between sibling tasks and TaskGroups must remain acyclic when each TaskGroup is treated as a -single unit. For example, if any task in ``group1`` is upstream of a task in ``group2``, another task in -``group2`` cannot be upstream of a task in ``group1``. Airflow reports this as a Dag parsing error even when -the individual task dependencies do not form a cycle. +single unit. This is evaluated using only edges that land on a TaskGroup's root tasks -- tasks with no +upstream task inside that group -- so a cycle can exist at the group level even when the individual task +dependencies form no cycle among themselves: + +.. code-block:: python + + with TaskGroup("group1"): + source1 = EmptyOperator(task_id="source1") + sink1 = EmptyOperator(task_id="sink1") + + with TaskGroup("group2"): + source2 = EmptyOperator(task_id="source2") + sink2 = EmptyOperator(task_id="sink2") + + source1 >> sink2 + source2 >> sink1 + +``sink1`` and ``sink2`` are each a root of their own group (neither has an upstream task inside its own +group), so this places ``group1`` both upstream and downstream of ``group2``. Airflow reports this as a Dag +parsing error even though no individual task-to-task dependency forms a cycle. TaskGroup also supports ``default_args`` like Dag, it will overwrite the ``default_args`` in Dag level: diff --git a/airflow-core/newsfragments/73087.significant.rst b/airflow-core/newsfragments/73087.significant.rst new file mode 100644 index 0000000000000..60a6119377f12 --- /dev/null +++ b/airflow-core/newsfragments/73087.significant.rst @@ -0,0 +1,17 @@ +Cyclic TaskGroup dependencies now fail Dag parsing instead of failing later in Grid or Graph + +Dag parsing now rejects TaskGroups whose dependencies form a cycle when each group is treated as a single +unit -- for example, a task in ``group1`` upstream of ``group2``'s root task while a task in ``group2`` is +upstream of ``group1``'s root task. Task-level dependencies can be acyclic while this group-level +projection still cycles, and downstream consumers such as Grid and Graph need an unambiguous group +ordering. + +Before this change, a Dag with cyclic TaskGroup dependencies could parse and run its tasks normally. #69933 +fixed the group-level topological sort used by Grid and Graph to actually detect group-to-group and +cross-group edges -- it previously ignored them and rendered an arbitrary, incorrect order instead of +failing. That made a pre-existing cycle in a Dag's TaskGroups surface as an unhandled error from the Grid +and Graph API endpoints rather than a rendering quirk. This change moves that failure to Dag parsing time +instead, where it raises an import error naming the offending TaskGroup rather than a 500 response from the +UI. + +Dags that previously parsed with cyclic TaskGroup dependencies must be restructured to remove the cycle. From f0677f67705ba44d85da49999958d97b4d47f65e Mon Sep 17 00:00:00 2001 From: Dheeraj Turaga Date: Mon, 14 Sep 2026 17:09:40 -0500 Subject: [PATCH 3/3] Name faulty tasks/TaskGroups in cycle errors, document bridged cycles Review feedback on the parse-time TaskGroup cycle check: the exception only named the containing TaskGroup, which collapses to the uninformative for a cycle between top-level siblings and gives no hint which nodes to fix on a Dag with many groups. The sweep and pass-numbering sort passes already know which nodes never got processed, so surface those ids on the exception and include them in the message. Also documented and tested a second cycle shape the docs missed: a task with no upstream inside its own TaskGroup counts as a root of that group, so an external task can bridge two of a group's own tasks and close a cycle without a second TaskGroup in sight -- something that reads like an ordinary Dag until it fails to parse. --- airflow-core/docs/core-concepts/dags.rst | 20 +++++++++++ .../newsfragments/73087.significant.rst | 8 +++-- task-sdk/src/airflow/sdk/definitions/dag.py | 10 ++++-- .../src/airflow/sdk/definitions/taskgroup.py | 8 +++-- task-sdk/src/airflow/sdk/exceptions.py | 3 ++ .../tests/task_sdk/definitions/test_dag.py | 34 ++++++++++++++++--- 6 files changed, 71 insertions(+), 12 deletions(-) diff --git a/airflow-core/docs/core-concepts/dags.rst b/airflow-core/docs/core-concepts/dags.rst index 3b2fb983ddbe0..63447a0ff5321 100644 --- a/airflow-core/docs/core-concepts/dags.rst +++ b/airflow-core/docs/core-concepts/dags.rst @@ -607,6 +607,26 @@ dependencies form no cycle among themselves: group), so this places ``group1`` both upstream and downstream of ``group2``. Airflow reports this as a Dag parsing error even though no individual task-to-task dependency forms a cycle. +The same "root task" rule can trip up a single group, without a second TaskGroup in sight. A task counts +as a root of its group as soon as it has no upstream task *inside* that group -- even if it has an upstream +task *outside* the group. Routing between two such tasks through an external task closes a cycle on the +group itself: + +.. code-block:: python + + with TaskGroup("group1"): + first = EmptyOperator(task_id="first") + second = EmptyOperator(task_id="second") + + bridge = EmptyOperator(task_id="bridge") + + first >> bridge >> second + +``second`` has no upstream task inside ``group1``, so it is a root of the group even though ``bridge`` sits +outside it. ``group1`` is upstream of ``bridge`` (via ``first``) and downstream of ``bridge`` (via +``second``'s root edge), which Airflow rejects the same way as the sibling-group example above -- even +though this reads like an ordinary Dag with a single TaskGroup. + TaskGroup also supports ``default_args`` like Dag, it will overwrite the ``default_args`` in Dag level: .. code-block:: python diff --git a/airflow-core/newsfragments/73087.significant.rst b/airflow-core/newsfragments/73087.significant.rst index 60a6119377f12..a42e7cd319555 100644 --- a/airflow-core/newsfragments/73087.significant.rst +++ b/airflow-core/newsfragments/73087.significant.rst @@ -4,14 +4,16 @@ Dag parsing now rejects TaskGroups whose dependencies form a cycle when each gro unit -- for example, a task in ``group1`` upstream of ``group2``'s root task while a task in ``group2`` is upstream of ``group1``'s root task. Task-level dependencies can be acyclic while this group-level projection still cycles, and downstream consumers such as Grid and Graph need an unambiguous group -ordering. +ordering. The same root-task rule can also close a cycle on a single group: a task with no upstream task +inside its own group counts as a root of that group even when an external task provides its only upstream, +so routing between two such tasks through an outside task closes a cycle on the group itself. Before this change, a Dag with cyclic TaskGroup dependencies could parse and run its tasks normally. #69933 fixed the group-level topological sort used by Grid and Graph to actually detect group-to-group and cross-group edges -- it previously ignored them and rendered an arbitrary, incorrect order instead of failing. That made a pre-existing cycle in a Dag's TaskGroups surface as an unhandled error from the Grid and Graph API endpoints rather than a rendering quirk. This change moves that failure to Dag parsing time -instead, where it raises an import error naming the offending TaskGroup rather than a 500 response from the -UI. +instead, where it raises an import error naming the offending TaskGroup and the specific task/TaskGroup ids +involved in the cycle, rather than a 500 response from the UI. Dags that previously parsed with cyclic TaskGroup dependencies must be restructured to remove the cycle. diff --git a/task-sdk/src/airflow/sdk/definitions/dag.py b/task-sdk/src/airflow/sdk/definitions/dag.py index 2c08c1ebcf071..f02841ce4d449 100644 --- a/task-sdk/src/airflow/sdk/definitions/dag.py +++ b/task-sdk/src/airflow/sdk/definitions/dag.py @@ -1181,10 +1181,16 @@ def _check_adjacent_tasks(task_id, current_task): for task_group in task_group_dict.values(): try: task_group.topological_sort(group_dict=task_group_dict) - except AirflowDagCycleException: + except AirflowDagCycleException as cycle_exc: group_id = task_group.group_id or "" + nodes_detail = ( + f" Nodes involved: {', '.join(cycle_exc.cyclic_node_ids)}" + if cycle_exc.cyclic_node_ids + else "" + ) raise AirflowDagCycleException( - f"TaskGroup dependency cycle detected in Dag: {self.dag_id}. Faulty TaskGroup: {group_id}" + f"TaskGroup dependency cycle detected in Dag: {self.dag_id}. " + f"Faulty TaskGroup: {group_id}.{nodes_detail}" ) from None def cli(self): diff --git a/task-sdk/src/airflow/sdk/definitions/taskgroup.py b/task-sdk/src/airflow/sdk/definitions/taskgroup.py index 89bc37d7127bd..df441d2bd1684 100644 --- a/task-sdk/src/airflow/sdk/definitions/taskgroup.py +++ b/task-sdk/src/airflow/sdk/definitions/taskgroup.py @@ -657,7 +657,9 @@ def _sweep_projection(self, nodes: list[DAGNode], projected: list[tuple[int, ... emitted[i] = 1 order_append(nodes[i]) if len(next_pending) == len(pending): - raise AirflowDagCycleException(f"A cyclic dependency occurred in dag: {self.dag_id}") + exc = AirflowDagCycleException(f"A cyclic dependency occurred in dag: {self.dag_id}") + exc.cyclic_node_ids = tuple(sorted(nodes[i].node_id for i in pending)) + raise exc pending = next_pending return order @@ -697,7 +699,9 @@ def _sort_via_pass_numbering( queue.append(s) if processed != n: - raise AirflowDagCycleException(f"A cyclic dependency occurred in dag: {self.dag_id}") + exc = AirflowDagCycleException(f"A cyclic dependency occurred in dag: {self.dag_id}") + exc.cyclic_node_ids = tuple(sorted(nodes[i].node_id for i in range(n) if in_degree[i] != 0)) + raise exc sorted_indices = sorted(range(n), key=lambda i: (pass_of[i], i)) return [nodes[i] for i in sorted_indices] diff --git a/task-sdk/src/airflow/sdk/exceptions.py b/task-sdk/src/airflow/sdk/exceptions.py index 6f43d5421ecf2..2f9f3b8d17712 100644 --- a/task-sdk/src/airflow/sdk/exceptions.py +++ b/task-sdk/src/airflow/sdk/exceptions.py @@ -71,6 +71,9 @@ class AirflowSecretsBackendAccessDenied(PermissionError): class AirflowDagCycleException(AirflowException): """Raise when there is a cycle in Dag definition.""" + #: node ids (tasks and/or TaskGroups) identified as part of the cycle, when known. + cyclic_node_ids: tuple[str, ...] = () + class AirflowRuntimeError(Exception): """Generic Airflow error raised by runtime functions.""" diff --git a/task-sdk/tests/task_sdk/definitions/test_dag.py b/task-sdk/tests/task_sdk/definitions/test_dag.py index 560a945cf790a..05c44af420108 100644 --- a/task-sdk/tests/task_sdk/definitions/test_dag.py +++ b/task-sdk/tests/task_sdk/definitions/test_dag.py @@ -998,13 +998,13 @@ def test_cycle_task_group_with_edge_labels(self): assert not dag.check_cycle() @pytest.mark.parametrize( - ("nested", "faulty_group"), + ("nested", "faulty_group", "faulty_nodes"), [ - pytest.param(False, "", id="root"), - pytest.param(True, "parent", id="nested"), + pytest.param(False, "", "left, right", id="root"), + pytest.param(True, "parent", "parent.left, parent.right", id="nested"), ], ) - def test_cycle_between_sibling_task_groups(self, nested, faulty_group): + def test_cycle_between_sibling_task_groups(self, nested, faulty_group, faulty_nodes): dag = DAG("dag", schedule=None, start_date=DEFAULT_DATE, default_args={"owner": "owner1"}) def add_groups(): @@ -1027,7 +1027,31 @@ def add_groups(): with pytest.raises( AirflowDagCycleException, - match=rf"TaskGroup dependency cycle detected in Dag: dag\. Faulty TaskGroup: {faulty_group}", + match=rf"TaskGroup dependency cycle detected in Dag: dag\. Faulty TaskGroup: {faulty_group}\. " + rf"Nodes involved: {faulty_nodes}", + ): + dag.check_cycle() + + def test_cycle_between_group_root_bridged_by_external_task(self): + """A task outside a TaskGroup can create a cycle by bridging two of the group's own tasks. + + ``second`` has no upstream task inside ``group1``, so it counts as a root of the group even + though its only upstream (``bridge``) sits outside it. + """ + dag = DAG("dag", schedule=None, start_date=DEFAULT_DATE, default_args={"owner": "owner1"}) + + with dag: + with TaskGroup("group1"): + first = DoNothingOperator(task_id="first") + second = DoNothingOperator(task_id="second") + bridge = DoNothingOperator(task_id="bridge") + + first >> bridge >> second + + with pytest.raises( + AirflowDagCycleException, + match=r"TaskGroup dependency cycle detected in Dag: dag\. Faulty TaskGroup: \. " + r"Nodes involved: bridge, group1", ): dag.check_cycle()