Keep the task instance in dag callbacks when its Dag version is missing

airflow-core/src/airflow/jobs/scheduler_job_runner.py

@@ -3062,6 +3062,7 @@ def _schedule_dag_run(
                     relevant_ti=last_unfinished_ti,
                     reason="timed_out",
                     execute=False,
+                    session=session,
                 )
 
             # Team name should be added before listeners are called in notify_dagrun_state_changed()

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

@@ -120,7 +120,10 @@
     from sqlalchemy.sql.elements import Case, ColumnElement
     from sqlalchemy.sql.selectable import Select
 
-    from airflow.api_fastapi.execution_api.datamodels.taskinstance import DagRun as DRDataModel
+    from airflow.api_fastapi.execution_api.datamodels.taskinstance import (
+        DagRun as DRDataModel,
+        TaskInstance as TIDataModel,
+    )
     from airflow.models.dag_version import DagVersion
     from airflow.models.taskinstancekey import TaskInstanceKey
     from airflow.sdk import DAG as SDKDAG
@@ -1333,6 +1336,7 @@ def recalculate(self) -> _UnfinishedStates:
                     relevant_ti=ti_causing_failure,
                     reason="task_failure",
                     execute=execute_callbacks,
+                    session=session,
                 )
 
             # Check if the max_consecutive_failed_dag_runs has been provided and not 0
@@ -1363,6 +1367,7 @@ def recalculate(self) -> _UnfinishedStates:
                     relevant_ti=last_succeeded_ti,
                     reason="success",
                     execute=execute_callbacks,
+                    session=session,
                 )
 
             if dag.deadline:
@@ -1400,6 +1405,7 @@ def recalculate(self) -> _UnfinishedStates:
                     relevant_ti=blocking_ti,
                     reason="all_tasks_deadlocked",
                     execute=execute_callbacks,
+                    session=session,
                 )
 
         # finally, if the leaves aren't done, the dag is still running
@@ -1508,30 +1514,79 @@ def notify_dagrun_state_changed(self, msg: str):
         # we can't get all the state changes on SchedulerJob,
         # or LocalTaskJob, so we don't want to "falsely advertise" we notify about that
 
+    def _build_callback_last_ti(self, relevant_ti: TI, *, session: Session) -> TIDataModel | None:
+        """
+        Build a callback context's ``last_ti``, standing in a Dag version if the record has none.
+
+        ``session`` is required, not defaulted: ``settings.Session`` is a ``scoped_session``,
+        so acquiring one here would hand back the caller's own and then commit and close it.
+
+        Unlike ``_ensure_ti_has_dag_version_id`` in the scheduler, the stand-in is reported
+        but never written back: this path only describes a task instance to a callback, and
+        healing the row belongs to the paths that own it. The consequence is that a purge
+        which heals the row to the latest version first wins, and the run's own version is
+        then never reported here.
+        """
+        from airflow.api_fastapi.execution_api.datamodels.taskinstance import TaskInstance as TIDataModel
+        from airflow.models.dag_version import DagVersion
+
+        if relevant_ti.dag_version_id is not None:
+            return TIDataModel.model_validate(relevant_ti, from_attributes=True)
+
+        dag_version_id = self.created_dag_version_id
+        if dag_version_id is not None:
+            self.log.info(
+                "Task instance %s has no dag_version_id; reporting it under the run's version %s "
+                "in the Dag callback context.",
+                relevant_ti,
+                dag_version_id,
+            )
+        else:
+            latest_dag_version = DagVersion.get_latest_version(self.dag_id, session=session)
+            if latest_dag_version is None:
+                self.log.warning(
+                    "Task instance %s has no dag_version_id and Dag %s has no version to stand in; "
+                    "omitting last_ti from the Dag callback context.",
+                    relevant_ti,
+                    self.dag_id,
+                )
+                return None
+            dag_version_id = latest_dag_version.id
+            self.log.info(
+                "Task instance %s and its run both have no dag_version_id (pre-versioning records); "
+                "reporting it under the latest version %s in the Dag callback context.",
+                relevant_ti,
+                dag_version_id,
+            )
+        # The datamodel rejects a null version, so the stand-in has to be spliced in
+        # before validation rather than copied onto a validated model.
+        values = {
+            name: getattr(relevant_ti, name)
+            for name in TIDataModel.model_fields
+            if hasattr(relevant_ti, name)
+        }
+        values["dag_version_id"] = dag_version_id
+        return TIDataModel.model_validate(values)
+
     def produce_dag_callback(
         self,
         dag: SerializedDAG,
         success: bool = True,
         relevant_ti: TI | None = None,
         reason: str = "success",
         execute: bool = False,
+        *,
+        session: Session,
     ) -> DagCallbackRequest | None:
         """Create a callback request for the DAG, or execute the callbacks directly if instructed, and return None."""
-        # Historical task instances created before the dag_version table existed (migration
-        # 0047_3_0_0_add_dag_versioning) have dag_version_id=None. The TaskInstance datamodel used to
-        # build the callback context requires a non-null UUID, so passing such a TI as last_ti would
-        # raise a ValidationError and crash the scheduler. Drop last_ti in that case; the callback still
-        # fires with a minimal context (dag, run_id, reason).
-        if relevant_ti is not None and relevant_ti.dag_version_id is None:
-            self.log.warning(
-                "Task instance %s has no dag_version_id (pre-versioning record); "
-                "omitting last_ti from the dag callback context.",
-                relevant_ti,
-            )
-            relevant_ti = None
         if not execute:
             from airflow.models.dag_version import _resolve_version_data
 
+            last_ti = (
+                self._build_callback_last_ti(relevant_ti, session=session)
+                if relevant_ti is not None
+                else None
+            )
             # Only carry version_data for pinned runs so the callback initializes the bundle
             # against the same version the run used.
             version_data = _resolve_version_data(self.created_dag_version, self.bundle_version)
@@ -1544,7 +1599,7 @@ def produce_dag_callback(
                 version_data=version_data,
                 context_from_server=DagRunContext(
                     dag_run=self,
-                    last_ti=relevant_ti,
+                    last_ti=last_ti,
                 ),
                 is_failure_callback=(not success),
                 msg=reason,
@@ -1554,22 +1609,25 @@ def produce_dag_callback(
             success=success,
             relevant_ti=relevant_ti,
             reason=reason,
+            session=session,
         )
         return None
 
     def execute_dag_callbacks(
-        self, dag: SDKDAG, success: bool = True, relevant_ti: TI | None = None, reason: str = "success"
+        self,
+        dag: SDKDAG,
+        success: bool = True,
+        relevant_ti: TI | None = None,
+        reason: str = "success",
+        *,
+        session: Session,
     ):
         """Only needed for `dag.test` where `execute_callbacks=True` is passed to `update_state`."""
-        from airflow.api_fastapi.execution_api.datamodels.taskinstance import (
-            TaskInstance as TIDataModel,
-            TIRunContext,
-        )
+        from airflow.api_fastapi.execution_api.datamodels.taskinstance import TIRunContext
         from airflow.models.dag import DagModel
         from airflow.sdk.execution_time.task_runner import RuntimeTaskInstance
 
-        if relevant_ti:
-            last_ti_model = TIDataModel.model_validate(relevant_ti, from_attributes=True)
+        if relevant_ti and (last_ti_model := self._build_callback_last_ti(relevant_ti, session=session)):
             task = dag.get_task(relevant_ti.task_id)
 
             runtime_ti = RuntimeTaskInstance.model_construct(

airflow-core/tests/unit/models/test_dag.py

@@ -1226,8 +1226,8 @@ def test_dag_handle_callback_crash(self, mock_incr, testing_dag_bundle):
             )
 
             # should not raise any exception
-        dag_run.execute_dag_callbacks(dag=dag, success=False)
-        dag_run.execute_dag_callbacks(dag=dag, success=True)
+            dag_run.execute_dag_callbacks(dag=dag, success=False, session=session)
+            dag_run.execute_dag_callbacks(dag=dag, success=True, session=session)
 
         mock_incr.assert_called_with(
             "dag.callback_exceptions",
@@ -1267,8 +1267,8 @@ def test_dag_handle_callback_with_removed_task(self, dag_maker, session, testing
             assert dag_run.get_task_instance(task_removed.task_id).state == TaskInstanceState.REMOVED
 
             # should not raise any exception
-            dag_run.execute_dag_callbacks(dag=dag, success=False)
-            dag_run.execute_dag_callbacks(dag=dag, success=True)
+            dag_run.execute_dag_callbacks(dag=dag, success=False, session=session)
+            dag_run.execute_dag_callbacks(dag=dag, success=True, session=session)
 
     @time_machine.travel(timezone.datetime(2025, 11, 11))
     @pytest.mark.parametrize(