Reject cyclic TaskGroup dependencies during Dag parsing

airflow-core/docs/core-concepts/dags.rst

@@ -585,6 +585,48 @@ 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. 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.
+
+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

airflow-core/newsfragments/73087.significant.rst

@@ -0,0 +1,19 @@
+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. 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 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.

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

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,22 @@ 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 as cycle_exc:
+                group_id = task_group.group_id or "<root>"
+                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}. "
+                    f"Faulty TaskGroup: {group_id}.{nodes_detail}"
+                ) from None
+
     def cli(self):
         """Exposes a CLI specific to this Dag."""
         self.check_cycle()

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]

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."""

task-sdk/tests/task_sdk/definitions/test_dag.py

@@ -997,6 +997,64 @@ def test_cycle_task_group_with_edge_labels(self):
 
         assert not dag.check_cycle()
 
+    @pytest.mark.parametrize(
+        ("nested", "faulty_group", "faulty_nodes"),
+        [
+            pytest.param(False, "<root>", "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, faulty_nodes):
+        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}\. "
+            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: <root>\. "
+            r"Nodes involved: bridge, group1",
+        ):
+            dag.check_cycle()
+
 
 class TestDagGetItem:
     def test_getitem_returns_task(self):