Deprecate cyclic TaskGroup dependencies

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

@@ -624,6 +624,78 @@ If you want to see a more advanced use of TaskGroup, you can look at the ``examp
 
     When using the ``@task_group`` decorator, the decorated-function's docstring will be used as the TaskGroups tooltip in the UI except when a ``tooltip`` value is explicitly supplied.
 
+.. _concepts:taskgroup-cycles:
+
+Cyclic TaskGroup dependencies
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+.. deprecated:: 3.4.0
+    Dags with cyclic TaskGroup dependencies are planned to fail Dag parsing from Airflow 3.5.
+
+When each TaskGroup is treated as a single unit, a TaskGroup and its siblings must not depend on each other
+in a cycle. A dependency into or out of any task in a group counts as a dependency of the whole group. This
+can make a group both upstream and downstream of a sibling, although no task-level dependency forms a cycle:
+
+.. code-block:: python
+
+    with TaskGroup("group1"):
+        a1 = EmptyOperator(task_id="a1")
+        a2 = EmptyOperator(task_id="a2")
+
+    with TaskGroup("group2"):
+        b1 = EmptyOperator(task_id="b1")
+        b2 = EmptyOperator(task_id="b2")
+
+    a1 >> b1  # group2 depends on group1
+    b2 >> a2  # group1 depends on group2
+
+A path that leaves a TaskGroup and comes back into it also forms a cycle, even when the tasks inside the
+group are ordered directly as well:
+
+.. code-block:: python
+
+    with TaskGroup("group"):
+        a = EmptyOperator(task_id="a")
+        b = EmptyOperator(task_id="b")
+        a >> b
+
+    bridge = EmptyOperator(task_id="bridge")
+
+    a >> bridge >> b  # group -> bridge -> group
+
+This includes a setup and teardown pair inside a TaskGroup with the work they wrap outside it. This is
+intended: a TaskGroup that holds a setup and its teardown must also hold the tasks between them.
+
+.. code-block:: python
+
+    with TaskGroup("cluster"):
+        create = EmptyOperator(task_id="create")
+        delete = EmptyOperator(task_id="delete")
+
+    work = EmptyOperator(task_id="work")
+
+    create >> work >> delete.as_teardown(setups=create)  # cluster -> work -> cluster
+
+These Dags still parse and run, but features that act on a TaskGroup as a whole need an unambiguous order
+between groups. Parsing them emits a ``TaskGroupCycleDeprecationWarning`` and a Dag warning in the UI that
+name the TaskGroups and tasks involved. The warning lists at most five members of each cycle and at most ten
+cycles, with a count of the ones left out.
+
+To remove the cycle, move tasks between TaskGroups, or out of them, so that each group depends on the others
+in one direction only. In the second example, move ``bridge`` into ``group``, or move ``b`` out of it. In the
+setup and teardown example, move ``work`` into ``cluster``.
+
+Only Dags authored with the Python Task SDK are checked. Dags from the Go, Java and TypeScript SDKs are not
+checked yet.
+
+To catch these Dags in CI, turn the warning into an error. The check runs when a Dag is added to a
+``DagBag``, so this works for a pytest test that loads your Dags into a ``DagBag`` and asserts there are no
+import errors, but not for one that only imports a Dag file:
+
+.. code-block:: bash
+
+    pytest -W error::airflow.sdk.exceptions.TaskGroupCycleDeprecationWarning tests/test_dags.py
+
 .. _concepts:edge-labels:
 
 Edge Labels

airflow-core/newsfragments/73746.significant.rst

@@ -0,0 +1,22 @@
+Deprecate cyclic TaskGroup dependencies
+
+A Dag whose TaskGroups depend on each other in a cycle, when each TaskGroup is treated as a single unit,
+now issues a ``TaskGroupCycleDeprecationWarning`` at parse time and shows a Dag warning in the UI that
+names the TaskGroups and tasks involved. These Dags still parse and run, because their task dependencies
+are acyclic, but Airflow 3.5 is planned to reject them at parse time.
+
+A dependency into or out of any task in a TaskGroup counts as a dependency of the whole group, so a path
+that leaves a TaskGroup and comes back into it is a cycle. For example, ``a >> bridge >> b``, with ``a``
+and ``b`` in the same TaskGroup and ``bridge`` outside it, makes the group depend on ``bridge`` and
+``bridge`` depend on the group, even if ``a >> b`` is also set.
+
+**Migration:**
+
+- Move tasks between TaskGroups, or out of them, so that each group depends on the others in one direction
+  only. In the example above, move ``bridge`` into the group, or move ``b`` out of it.
+- To catch these Dags in CI, run your Dag tests with
+  ``-W error::airflow.sdk.exceptions.TaskGroupCycleDeprecationWarning``. The check runs when a Dag is
+  added to a ``DagBag``, so only tests that load your Dags into a ``DagBag`` and assert there are no import
+  errors catch them; a test that only imports a Dag file does not.
+
+See "Cyclic TaskGroup dependencies" in the Dags documentation.

airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml

@@ -16250,6 +16250,7 @@ components:
       - duplicate dag id
       - non-existent pool
       - runtime varying value
+      - task group cycle
       title: DagWarningType
       description: 'Enum for DAG warning types.
 

airflow-core/src/airflow/dag_processing/collection.py

@@ -587,6 +587,7 @@ def update_dag_parsing_results_in_db(
         DagWarningType.DUPLICATE_DAG_ID,
         DagWarningType.NONEXISTENT_POOL,
         DagWarningType.RUNTIME_VARYING_VALUE,
+        DagWarningType.TASK_GROUP_CYCLE,
     ),
     files_parsed: set[tuple[str, str]] | None = None,
     dag_source_codes: dict[str, DagSourceCode] | None = None,

airflow-core/src/airflow/dag_processing/dagbag.py

@@ -46,6 +46,7 @@
 from airflow.executors.executor_loader import ExecutorLoader
 from airflow.listeners.listener import get_listener_manager
 from airflow.models.pool import Pool
+from airflow.sdk.exceptions import AirflowDagCycleException, TaskGroupCycleDeprecationWarning
 from airflow.sdk.importers import DagImportError, get_importer_registry
 from airflow.serialization.definitions.dag import SerializedDAG
 from airflow.serialization.definitions.notset import NOTSET, ArgNotSet, is_arg_set
@@ -213,6 +214,7 @@ def __init__(
         self.import_errors: dict[str, str] = {}
         self.captured_warnings: dict[str, tuple[str, ...]] = {}
         self._import_warnings: dict[str, list[DagImportWarning]] = {}
+        self.task_group_cycle_warnings: dict[str, str] = {}
         # The source code of each definition that produced Dags, keyed by its fileloc
         self.dag_source_codes: dict[str, DagSourceCode] = {}
         # Only used by SchedulerJob to compare the dag_hash to identify change in DAGs
@@ -446,7 +448,11 @@ def dag_warnings(self) -> set[DagWarning]:
         """Get the set of DagWarnings for the bagged dags."""
         from airflow.models.dagwarning import DagWarning, DagWarningType
 
-        dag_warnings: set[DagWarning] = set()
+        dag_warnings: set[DagWarning] = {
+            DagWarning(dag_id, DagWarningType.TASK_GROUP_CYCLE, message)
+            for dag_id, message in self.task_group_cycle_warnings.items()
+            if dag_id in self.dags
+        }
         for dag in self.dags.values():
             for import_warning in self._import_warnings.get(dag.fileloc, ()):
                 # Only importer-namespaced types (``yaml:deprecated_field``) are Dag warnings;
@@ -491,7 +497,21 @@ def bag_dag(self, dag: DAG):
         :raises: AirflowDagCycleException if a cycle is detected.
         :raises: AirflowDagDuplicatedIdException if this dag already exists in the bag.
         """
-        dag.check_cycle()
+        task_group_cycle_warning = None
+        with warnings.catch_warnings(record=True) as captured_warnings:
+            # DeprecationWarning is ignored by default outside __main__, which would hide it here too.
+            warnings.simplefilter("always", TaskGroupCycleDeprecationWarning)
+            dag.check_cycle()
+        for captured in captured_warnings:
+            if issubclass(captured.category, TaskGroupCycleDeprecationWarning):
+                task_group_cycle_warning = str(captured.message)
+            warnings.warn_explicit(
+                message=captured.message,
+                category=captured.category,
+                filename=captured.filename,
+                lineno=captured.lineno,
+                source=captured.source,
+            )
         dag.resolve_template_files()
         dag.last_loaded = timezone.utcnow()
 
@@ -515,15 +535,18 @@ def bag_dag(self, dag: DAG):
             self.log.exception(e)
             raise AirflowClusterPolicyError(e)
         self._add_to_bag(dag)
+        # Only once bagged: a rejected duplicate must not touch the warning of the Dag it lost to.
+        if task_group_cycle_warning is None:
+            self.task_group_cycle_warnings.pop(dag.dag_id, None)
+        else:
+            self.task_group_cycle_warnings[dag.dag_id] = task_group_cycle_warning
 
     def _add_to_bag(self, dag: BaggedDAG) -> None:
         """
         Add *dag* to the bag, unless another file already defines its id.
 
         :raises: AirflowDagDuplicatedIdException if this dag already exists in the bag.
         """
-        from airflow.sdk.exceptions import AirflowDagCycleException
-
         try:
             prev_dag = self.dags.get(dag.dag_id)
             if prev_dag and prev_dag.fileloc != dag.fileloc:

airflow-core/src/airflow/dag_processing/processor.py

@@ -254,6 +254,8 @@ def _parse_file(msg: DagFileParseRequest, log: FilteringBoundLogger) -> DagFileP
 
     serialized_dags, serialization_import_errors = _serialize_dags(bag, log)
     bag.import_errors.update(serialization_import_errors)
+    # Dag warnings reference the dag table, so only Dags that serialized can carry one.
+    serialized_dag_ids = {dag.dag_id for dag in serialized_dags}
     result = DagFileParsingResult(
         fileloc=msg.file,
         serialized_dags=serialized_dags,
@@ -263,6 +265,7 @@ def _parse_file(msg: DagFileParseRequest, log: FilteringBoundLogger) -> DagFileP
             *(
                 {"dag_id": w.dag_id, "warning_type": w.warning_type, "message": w.message}
                 for w in bag.dag_warnings
+                if w.dag_id in serialized_dag_ids
             ),
         ],
         parsed_definitions=bag.parsed_definitions,

airflow-core/src/airflow/models/dagwarning.py

@@ -108,6 +108,7 @@ class DagWarningType(str, Enum):
     DUPLICATE_DAG_ID = "duplicate dag id"
     NONEXISTENT_POOL = "non-existent pool"
     RUNTIME_VARYING_VALUE = "runtime varying value"
+    TASK_GROUP_CYCLE = "task group cycle"
 
 
 ImporterWarningType = Annotated[

airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts

@@ -5618,7 +5618,7 @@ export const $DagVersionResponse = {
 
 export const $DagWarningType = {
     type: 'string',
-    enum: ['asset conflict', 'duplicate dag id', 'non-existent pool', 'runtime varying value'],
+    enum: ['asset conflict', 'duplicate dag id', 'non-existent pool', 'runtime varying value', 'task group cycle'],
     title: 'DagWarningType',
     description: `Enum for DAG warning types.
 

airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts

@@ -1555,7 +1555,7 @@ export type DagVersionResponse = {
  * This is the set of allowable values for the ``warning_type`` field
  * in the DagWarning model.
  */
-export type DagWarningType = 'asset conflict' | 'duplicate dag id' | 'non-existent pool' | 'runtime varying value';
+export type DagWarningType = 'asset conflict' | 'duplicate dag id' | 'non-existent pool' | 'runtime varying value' | 'task group cycle';
 
 /**
  * How much of a component's work has a live instance covering it.

airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_warning.py

@@ -115,9 +115,9 @@ def test_get_dag_warnings_bad_request(self, test_client):
         response = test_client.get("/dagWarnings", params={"warning_type": "invalid"})
         response_json = response.json()
         assert response.status_code == 422
-        assert (
-            response_json["detail"][0]["msg"]
-            == "Input should be 'asset conflict', 'duplicate dag id', 'non-existent pool' or 'runtime varying value'"
+        assert response_json["detail"][0]["msg"] == (
+            "Input should be 'asset conflict', 'duplicate dag id', 'non-existent pool', "
+            "'runtime varying value' or 'task group cycle'"
         )
 
     def test_get_dag_warnings_non_matching_type(self, test_client):

airflow-core/tests/unit/dag_processing/test_collection.py

@@ -994,6 +994,30 @@ def test_dag_source_codes_are_written_to_dag_code(self, testing_dag_bundle, sess
         dag_code = DagCode.get_latest_dagcode("yaml_dag", session=session)
         assert (dag_code.source_code, dag_code.language) == ("dag_id: yaml_dag\n", "yaml")
 
+    def test_task_group_cycle_warning_is_removed_when_dag_is_fixed(self, testing_dag_bundle, session):
+        dag = DAG(dag_id="fixed_dag")
+        session.add(DagModel(dag_id="fixed_dag", bundle_name="testing", is_stale=False))
+        session.add(
+            DagWarning(
+                dag_id="fixed_dag",
+                warning_type=DagWarningType.TASK_GROUP_CYCLE,
+                message="Dag 'fixed_dag': group and bridge depend on each other in a cycle.",
+            )
+        )
+        session.flush()
+
+        update_dag_parsing_results_in_db(
+            bundle_name="testing",
+            bundle_version=None,
+            dags=[LazyDeserializedDAG.from_dag(dag)],
+            import_errors={},
+            parse_duration=None,
+            warnings=set(),
+            session=session,
+        )
+
+        assert session.scalar(select(DagWarning).where(DagWarning.dag_id == "fixed_dag")) is None
+
     def test_parse_time_written_to_db_on_sync(self, testing_dag_bundle, session):
         """Test that the parse time is correctly written to the DB after parsing"""
 

airflow-core/tests/unit/dag_processing/test_dagbag.py

@@ -44,15 +44,15 @@
 )
 from airflow.dag_processing.lang_sdk_processor import LangSDKDagFileProcessorProcess
 from airflow.dag_processing.processor import DagFileParsingResult
-from airflow.exceptions import UnknownExecutorException
+from airflow.exceptions import AirflowDagDuplicatedIdException, UnknownExecutorException
 from airflow.executors.executor_loader import ExecutorLoader
 from airflow.models.dag import DagModel
 from airflow.models.dagwarning import DagWarning, DagWarningType
 from airflow.models.errors import ParseImportError
 from airflow.models.pool import Pool
 from airflow.models.serialized_dag import SerializedDagModel
-from airflow.sdk import DAG, BaseOperator
-from airflow.sdk.exceptions import AirflowConfigException
+from airflow.sdk import DAG, BaseOperator, TaskGroup
+from airflow.sdk.exceptions import AirflowConfigException, TaskGroupCycleDeprecationWarning
 from airflow.sdk.importers import (
     AbstractDagImporter,
     DagImportResult,
@@ -1402,6 +1402,58 @@ def test_custom_importer_dags_are_bound_to_their_definition(self, tmp_path):
             os.fspath(dag_file): (f"{dag_file}:1: test:deprecated_field: Deprecated field",)
         }
 
+    @staticmethod
+    def _make_task_group_cycle_dag(*, cyclic: bool) -> DAG:
+        with DAG(dag_id="test") as dag:
+            with TaskGroup("group") as group:
+                a = BaseOperator(task_id="a")
+                b = BaseOperator(task_id="b")
+            a >> BaseOperator(task_id="bridge", task_group=None if cyclic else group) >> b
+        return dag
+
+    def test_dag_warnings_task_group_cycle(self):
+        dagbag = DagBag(dag_folder="", collect_dags=False)
+
+        # Python ignores DeprecationWarning outside __main__, as in the Dag processor.
+        with warnings.catch_warnings():
+            warnings.simplefilter("ignore")
+            dagbag.bag_dag(self._make_task_group_cycle_dag(cyclic=True))
+
+        (warning,) = dagbag.dag_warnings
+        assert (warning.dag_id, warning.warning_type) == ("test", DagWarningType.TASK_GROUP_CYCLE)
+        assert "group and bridge depend on each other in a cycle." in warning.message
+
+        with pytest.warns(TaskGroupCycleDeprecationWarning):
+            dagbag.bag_dag(self._make_task_group_cycle_dag(cyclic=True))
+
+        dagbag.bag_dag(self._make_task_group_cycle_dag(cyclic=False))
+        assert dagbag.dag_warnings == set()
+
+    @pytest.mark.parametrize(
+        ("bagged_cyclic", "duplicate_cyclic"),
+        [
+            pytest.param(False, True, id="cyclic-duplicate-of-acyclic"),
+            pytest.param(True, False, id="acyclic-duplicate-of-cyclic"),
+        ],
+    )
+    def test_dag_warnings_task_group_cycle_ignores_rejected_duplicate(self, bagged_cyclic, duplicate_cyclic):
+        dagbag = DagBag(dag_folder="", collect_dags=False)
+        bagged = self._make_task_group_cycle_dag(cyclic=bagged_cyclic)
+        bagged.fileloc = "/dags/bagged.py"
+        duplicate = self._make_task_group_cycle_dag(cyclic=duplicate_cyclic)
+        duplicate.fileloc = "/dags/duplicate.py"
+
+        with warnings.catch_warnings():
+            warnings.simplefilter("ignore")
+            dagbag.bag_dag(bagged)
+            with pytest.raises(AirflowDagDuplicatedIdException):
+                dagbag.bag_dag(duplicate)
+
+        assert dagbag.dags["test"] is bagged
+        assert {w.warning_type for w in dagbag.dag_warnings} == (
+            {DagWarningType.TASK_GROUP_CYCLE} if bagged_cyclic else set()
+        )
+
     def test_sigsegv_handling(self, tmp_path, caplog):
         """
         Test that a SIGSEGV in a DAG file is handled gracefully and does not crash the process.

airflow-core/tests/unit/dag_processing/test_processor.py

@@ -794,6 +794,41 @@ def test_parse_file_reports_definitions_source_codes_and_dag_warnings(mock_dag_w
     ]
 
 
+TASK_GROUP_CYCLE_DAGS = """
+from airflow.sdk import DAG, BaseOperator, TaskGroup
+
+for dag_id in ("cyclic", "acyclic", "unserializable"):
+    with DAG(dag_id, schedule=None):
+        with TaskGroup("group") as group:
+            a = BaseOperator(task_id="a")
+            b = BaseOperator(task_id="b")
+        a >> BaseOperator(task_id="bridge", task_group=group if dag_id == "acyclic" else None) >> b
+"""
+
+
+def test_parse_file_task_group_cycle_warning(tmp_path):
+    dag_file = tmp_path / "task_group_cycle_dags.py"
+    dag_file.write_text(TASK_GROUP_CYCLE_DAGS)
+    to_dict = DagSerialization.to_dict
+
+    def fail_to_serialize_unserializable(dag):
+        if dag.dag_id == "unserializable":
+            raise ValueError("serialization failed")
+        return to_dict(dag)
+
+    with patch.object(
+        DagSerialization, "to_dict", autospec=True, side_effect=fail_to_serialize_unserializable
+    ):
+        result = _parse_file(
+            DagFileParseRequest(file=str(dag_file), bundle_path=tmp_path, bundle_name="testing"),
+            log=structlog.get_logger(),
+        )
+
+    assert [(warning["dag_id"], warning["warning_type"]) for warning in result.warnings] == [
+        ("cyclic", "task group cycle")
+    ]
+
+
 def test_callback_processing_does_not_update_timestamps():
     """Callback processing should not update last_finish_time to prevent stale DAG detection."""
     stat = process_parse_results(