Reproduce executor-side session detachment via scoped session in test

Review asked for the concrete code path that detaches the TaskInstances
rather than simulating it with session.expunge(). The mechanism is the
executor opening and closing a scoped session while adopting:
create_session() returns the scheduler's own thread-scoped session and
closes it on exit, expunging everything the orphan query loaded.
KubernetesExecutor's completed-pod adoption did exactly this from #66400
until #67850, and released providers up to 10.17.x still ship it (the
combination Airflow 3.2.2 constraints pin). The mock now performs that
exact operation, so the test exercises the production mechanism
end-to-end instead of hand-detaching instances.

Author: akhilpratap1991

airflow-core/tests/unit/jobs/test_scheduler_job.py

@@ -3659,10 +3659,14 @@ def refuse_adoption(tis):
     def test_adopt_or_reset_orphaned_tasks_handles_detached_tis(self, dag_maker, session, mock_executor):
         """
         Reset and adoption must not depend on the session state of the TaskInstances the
-        executor hands back: the reset path copies the full row into TaskInstanceHistory and
-        the adopt path reads last_heartbeat_at / dag_run.conf, none of which are loaded by the
-        orphan query — on a detached instance those reads raise DetachedInstanceError and
-        crash the scheduler loop.
+        executor hands back. An executor can open-and-close a scoped session while adopting
+        (KubernetesExecutor's completed-pod adoption did exactly that until #67850, and
+        released providers <= 10.17.x still do); closing the shared thread-scoped session
+        detaches every TaskInstance the orphan query loaded. The reset path then copies the
+        full row into TaskInstanceHistory and the adopt path reads last_heartbeat_at /
+        dag_run.conf — reads of unloaded attributes that raise DetachedInstanceError on a
+        detached instance and crash the scheduler loop; where attributes happen to be
+        loaded, the writes are silently lost instead.
         """
         from airflow.models.taskinstancehistory import TaskInstanceHistory
 
@@ -3689,8 +3693,12 @@ def test_adopt_or_reset_orphaned_tasks_handles_detached_tis(self, dag_maker, ses
         session.expunge_all()
 
         def adopt_second_reset_first(tis):
-            for ti in tis:
-                session.expunge(ti)
+            # What KubernetesExecutor's completed-pod adoption does mid-adopt in released
+            # providers (<= 10.17.x; fixed on its side by #67850): create_session() returns
+            # the scheduler's own thread-scoped session, and closing it detaches every
+            # TaskInstance the orphan query loaded.
+            with create_session():
+                pass
             return [ti for ti in tis if ti.task_id == "op1"]
 
         mock_executor.try_adopt_task_instances.side_effect = adopt_second_reset_first