Prevent DetachedInstanceError when processing executor event (#54334)

Author: kaxil

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

@@ -39,7 +39,7 @@
 
 from airflow import settings
 from airflow._shared.timezones import timezone
-from airflow.api_fastapi.execution_api.datamodels.taskinstance import TIRunContext
+from airflow.api_fastapi.execution_api.datamodels.taskinstance import DagRun as DRDataModel, TIRunContext
 from airflow.callbacks.callback_requests import DagCallbackRequest, DagRunContext, TaskCallbackRequest
 from airflow.configuration import conf
 from airflow.dag_processing.bundles.base import BundleUsageTrackingManager
@@ -768,6 +768,7 @@ def process_executor_events(
             select(TI)
             .where(filter_for_tis)
             .options(selectinload(TI.dag_model))
+            .options(selectinload(TI.dag_run).selectinload(DagRun.consumed_asset_events))
             .options(joinedload(TI.dag_version))
         )
         # row lock this entire set of taskinstances to make sure the scheduler doesn't fail when we have
@@ -888,7 +889,7 @@ def process_executor_events(
                         ti=ti,
                         msg=msg,
                         context_from_server=TIRunContext(
-                            dag_run=ti.dag_run,
+                            dag_run=DRDataModel.model_validate(ti.dag_run, from_attributes=True),
                             max_tries=ti.max_tries,
                             variables=[],
                             connections=[],
@@ -2266,7 +2267,7 @@ def _purge_task_instances_without_heartbeats(
                 ti=ti,
                 msg=str(task_instance_heartbeat_timeout_message_details),
                 context_from_server=TIRunContext(
-                    dag_run=ti.dag_run,
+                    dag_run=DRDataModel.model_validate(ti.dag_run, from_attributes=True),
                     max_tries=ti.max_tries,
                     variables=[],
                     connections=[],