From caee4b79badfcc6c494d8b7d0ce98d498da0a31a Mon Sep 17 00:00:00 2001
From: Anas Khan <83116240+anxkhn@users.noreply.github.com>
Date: Fri, 3 Jul 2026 04:05:42 +0530
Subject: [PATCH] Fix Dag serialization crash for Task SDK EventsTimetable

A Dag scheduled with the Task SDK EventsTimetable failed to serialize
because the SDK timetable serializer omits the _summary key, while the
core EventsTimetable.deserialize required it unconditionally and raised
KeyError. Fall back to the summary computed by __init__ when the key is
absent, so Dags using the SDK timetable serialize and schedule again
while existing serialized data keeps its stored summary.

Signed-off-by: Anas Khan <83116240+anxkhn@users.noreply.github.com>
---
 airflow-core/newsfragments/69401.bugfix.rst   |  1 +
 airflow-core/src/airflow/timetables/events.py |  4 ++-
 .../serialization/test_dag_serialization.py   | 34 ++++++++++++++++++-
 3 files changed, 37 insertions(+), 2 deletions(-)
 create mode 100644 airflow-core/newsfragments/69401.bugfix.rst

diff --git a/airflow-core/newsfragments/69401.bugfix.rst b/airflow-core/newsfragments/69401.bugfix.rst
new file mode 100644
index 0000000000000..ad2f9f00243ce
--- /dev/null
+++ b/airflow-core/newsfragments/69401.bugfix.rst
@@ -0,0 +1 @@
+Fix Dag serialization crash for Dags scheduled with the Task SDK ``EventsTimetable``.
diff --git a/airflow-core/src/airflow/timetables/events.py b/airflow-core/src/airflow/timetables/events.py
index 9fedc476c1cda..6c4f6fd6feb7d 100644
--- a/airflow-core/src/airflow/timetables/events.py
+++ b/airflow-core/src/airflow/timetables/events.py
@@ -16,6 +16,7 @@
 # under the License.
 from __future__ import annotations
 
+import contextlib
 import itertools
 from typing import TYPE_CHECKING, Any
 
@@ -137,5 +138,6 @@ def deserialize(cls, data: dict[str, Any]) -> Timetable:
             presorted=True,
             description=data["description"],
         )
-        timetable._summary = data["_summary"]
+        with contextlib.suppress(KeyError):
+            timetable._summary = data["_summary"]
         return timetable
diff --git a/airflow-core/tests/unit/serialization/test_dag_serialization.py b/airflow-core/tests/unit/serialization/test_dag_serialization.py
index 4708365e846e2..b4d5705f2506c 100644
--- a/airflow-core/tests/unit/serialization/test_dag_serialization.py
+++ b/airflow-core/tests/unit/serialization/test_dag_serialization.py
@@ -61,7 +61,17 @@
 from airflow.models.xcom import XCOM_RETURN_KEY, XComModel
 from airflow.providers.cncf.kubernetes.pod_generator import PodGenerator
 from airflow.providers.standard.operators.bash import BashOperator
-from airflow.sdk import DAG, Asset, AssetAlias, BaseHook, TaskGroup, WeightRule, XComArg, teardown
+from airflow.sdk import (
+    DAG,
+    Asset,
+    AssetAlias,
+    BaseHook,
+    EventsTimetable,
+    TaskGroup,
+    WeightRule,
+    XComArg,
+    teardown,
+)
 from airflow.sdk.bases.decorator import DecoratedOperator
 from airflow.sdk.bases.operator import OPERATOR_DEFAULTS, BaseOperator
 from airflow.sdk.definitions._internal.expandinput import EXPAND_INPUT_EMPTY
@@ -781,6 +791,28 @@ def test_dag_roundtrip_from_timetable(self, timetable):
         roundtripped = DagSerialization.from_json(DagSerialization.to_json(dag))
         self.validate_deserialized_dag(roundtripped, dag)
 
+    @pytest.mark.db_test
+    @pytest.mark.parametrize(
+        ("description", "expected_summary"),
+        [
+            pytest.param(None, "2 events", id="no-description"),
+            pytest.param("World Cup", "World Cup", id="with-description"),
+        ],
+    )
+    def test_dag_roundtrip_from_sdk_events_timetable(self, description, expected_summary):
+        """Round-trip a Dag scheduled with the SDK ``EventsTimetable``.
+
+        The SDK timetable serializer does not emit ``_summary``, so the core
+        ``EventsTimetable.deserialize`` must tolerate its absence instead of
+        raising ``KeyError``.
+        """
+        event_dates = [pendulum.datetime(2025, 1, 1), pendulum.datetime(2025, 6, 1)]
+        dag = get_timetable_based_simple_dag(
+            EventsTimetable(event_dates=event_dates, description=description)
+        )
+        roundtripped = DagSerialization.from_json(DagSerialization.to_json(dag))
+        assert roundtripped.timetable.summary == expected_summary
+
     def validate_deserialized_dag(self, serialized_dag: SerializedDAG, dag: DAG):
         """
         Verify that all example DAGs work with DAG Serialization by