UI: Fix grid columns vanishing when a Dag version has been cleaned

airflow-core/src/airflow/api_fastapi/core_api/routes/ui/grid.py

@@ -104,7 +104,7 @@ def _get_serdag(
     if dag_version_id is not None:
         serdag = dag_bag.get_dag(dag_version_id, session=session)
         if serdag is None:
-            log.error("No serialized dag found", dag_id=dag_id, version_id=dag_version_id)
+            log.warning("No serialized dag found", dag_id=dag_id, version_id=dag_version_id)
         return serdag
 
     # Fallback: pre-3.0 upgrade — pick the oldest DagVersion for this dag_id.
@@ -115,7 +115,7 @@ def _get_serdag(
         return None
     serdag = dag_bag.get_dag(oldest_version_id, session=session)
     if serdag is None:
-        log.error("No serialized dag found", dag_id=dag_id, version_id=oldest_version_id)
+        log.warning("No serialized dag found", dag_id=dag_id, version_id=oldest_version_id)
     return serdag
 
 
@@ -429,21 +429,22 @@ def _build_ti_summaries(
         return None
 
     serdag = _get_serdag(dag_bag, dag_id, dag_version_id, session)
-    if TYPE_CHECKING:
-        assert serdag
 
     def get_node_summaries() -> Iterable[dict[str, Any]]:
         yielded_task_ids: set[str] = set()
-        for node, _ in _find_aggregates(
-            node=serdag.task_group,
-            parent_node=None,
-            ti_details=ti_details,
-        ):
-            if node["type"] in {"task", "mapped_task"}:
-                yielded_task_ids.add(node["task_id"])
-                if node["type"] == "task":
-                    node["child_states"] = None
-            yield node
+        # serdag is None once `airflow db clean` removes the version's serialized Dag row. Leaving
+        # yielded_task_ids empty then routes every task through the flat-node path below.
+        if serdag is not None:
+            for node, _ in _find_aggregates(
+                node=serdag.task_group,
+                parent_node=None,
+                ti_details=ti_details,
+            ):
+                if node["type"] in {"task", "mapped_task"}:
+                    yielded_task_ids.add(node["task_id"])
+                    if node["type"] == "task":
+                        node["child_states"] = None
+                yield node
         missing_task_ids = set(ti_details.keys()) - yielded_task_ids
         for task_id in sorted(missing_task_ids):
             detail = ti_details[task_id]

airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_grid.py

@@ -23,14 +23,15 @@
 
 import pendulum
 import pytest
-from sqlalchemy import select
+from sqlalchemy import delete, select
 from sqlalchemy.orm import Session
 
 from airflow._shared.timezones import timezone
 from airflow.models.dag import DagModel
 from airflow.models.dag_version import DagVersion
 from airflow.models.dagbag import DBDagBag
 from airflow.models.dagrun import DagRun, DagRunNote
+from airflow.models.serialized_dag import SerializedDagModel
 from airflow.models.taskinstance import TaskInstance, TaskInstanceNote
 from airflow.providers.standard.operators.empty import EmptyOperator
 from airflow.providers.standard.operators.python import PythonOperator
@@ -1260,6 +1261,55 @@ def test_grid_ti_summaries_stream_skips_missing_runs(self, session, test_client)
         assert len(summaries) == 1
         assert summaries[0]["run_id"] == "run_1"
 
+    def test_grid_ti_summaries_stream_continues_past_cleaned_dag_version(self, session, test_client):
+        """A run whose version was cleaned still streams, and does not cut the runs after it."""
+        cleaned_version = session.scalar(
+            select(DagVersion).where(DagVersion.dag_id == DAG_ID_5, DagVersion.version_number == 1)
+        )
+        session.execute(
+            delete(SerializedDagModel).where(SerializedDagModel.dag_version_id == cleaned_version.id)
+        )
+        session.commit()
+
+        response = test_client.get(
+            f"/grid/ti_summaries/{DAG_ID_5}", params={"run_ids": ["run_5_1", "run_5_2"]}
+        )
+        assert response.status_code == 200
+        summaries = {summary["run_id"]: summary for summary in self._parse_ndjson(response)}
+        assert set(summaries) == {"run_5_1", "run_5_2"}
+        assert {node["task_id"] for node in summaries["run_5_1"]["task_instances"]} == {
+            "task_a",
+            "task_b",
+            "task_c",
+            "task_d",
+            "task_f",
+        }
+        # The run on the surviving version is unaffected.
+        assert {node["task_id"] for node in summaries["run_5_2"]["task_instances"]} == {
+            "task_a",
+            "task_b",
+            "task_c",
+            "task_d",
+            "task_e",
+        }
+
+    def test_grid_ti_summaries_stream_without_serialized_dag_flattens_groups(self, session, test_client):
+        """With no structure to walk, a grouped Dag's tasks come back as flat leaf nodes."""
+        session.execute(delete(SerializedDagModel).where(SerializedDagModel.dag_id == DAG_ID))
+        session.commit()
+
+        response = test_client.get(f"/grid/ti_summaries/{DAG_ID}", params={"run_ids": ["run_1"]})
+        assert response.status_code == 200
+        [summary] = self._parse_ndjson(response)
+        assert {node["task_id"]: node["state"] for node in summary["task_instances"]} == {
+            TASK_ID: "success",
+            "mapped_task_group.subtask": "success",
+            f"{TASK_GROUP_ID}.{MAPPED_TASK_ID}": "success",
+            f"{TASK_GROUP_ID}.{INNER_TASK_GROUP}.{INNER_TASK_GROUP_SUB_TASK}": "success",
+            MAPPED_TASK_ID_2: "success",
+        }
+        assert all(node["child_states"] is None for node in summary["task_instances"])
+
     def test_grid_ti_summaries_stream_empty_run_ids(self, session, test_client):
         """Streaming endpoint with no run_ids returns an empty body."""
         session.commit()