diff --git a/airflow-core/src/airflow/serialization/definitions/taskgroup.py b/airflow-core/src/airflow/serialization/definitions/taskgroup.py index 65d59cb15f17e..1b353b4c7d910 100644 --- a/airflow-core/src/airflow/serialization/definitions/taskgroup.py +++ b/airflow-core/src/airflow/serialization/definitions/taskgroup.py @@ -251,6 +251,11 @@ def _project_child_deps( self, child_idx: int, child: DAGNode, id_to_idx: dict[str, int] ) -> tuple[int, ...]: upstream_ids = child.upstream_task_ids + if isinstance(child, SerializedTaskGroup): + root_upstream_ids = set() + for root_task in child.get_roots(): + root_upstream_ids.update(root_task.upstream_task_ids) + upstream_ids = upstream_ids | root_upstream_ids if not upstream_ids: return () sib_deps: set[int] = set() diff --git a/airflow-core/tests/unit/utils/test_task_group.py b/airflow-core/tests/unit/utils/test_task_group.py index 2d1458e95fbd1..d3708bbde0a6c 100644 --- a/airflow-core/tests/unit/utils/test_task_group.py +++ b/airflow-core/tests/unit/utils/test_task_group.py @@ -1150,6 +1150,31 @@ def nested_topo(group): ] +def test_topological_sort_serialized_task_level_cross_group_dep(): + """Task-level deps between groups are respected for ordering after serialization. + + When a task inside group_b depends on a task inside group_a, the serialized + topological sort must place group_a before group_b. + """ + with DAG("test_cross_group_task_dep", schedule=None, start_date=DEFAULT_DATE) as dag: + with TaskGroup("stage_b"): + b_start = EmptyOperator(task_id="b_start") + b_end = EmptyOperator(task_id="b_end") + b_start >> b_end + + with TaskGroup("stage_a"): + a_start = EmptyOperator(task_id="a_start") + a_end = EmptyOperator(task_id="a_end") + a_start >> a_end + + b_end >> a_start + + serialized = create_scheduler_dag(dag) + order = [node.node_id for node in serialized.task_group.topological_sort()] + + assert order.index("stage_b") < order.index("stage_a") + + def test_topological_sort_serialized_layered(): """SerializedTaskGroup.topological_sort emits a valid order after DAG round-trip.