Fix scheduler crash when resetting orphaned task instances

[akhilpratap1991]

adopt_or_reset_orphaned_tasks() reads deferred columns (the full row copied into TaskInstanceHistory by prepare_db_for_next_try, last_heartbeat_at) and the lazy dag_run relationship on TaskInstances that executors' try_adopt_task_instances() may hand back detached from the session. Those reads raise DetachedInstanceError, which is unhandled in _run_scheduler_loop — the scheduler exits, and since it dies before the orphan is reset, the same orphan crashes it again on every restart: an unrecoverable crash loop that halts all scheduling (observed in production with KubernetesExecutor as a scheduler CrashLoopBackOff). Where the attributes happen to be loaded, the writes to the detached instances are silently lost instead, so the orphans are never actually reset.

closes: #71272

This completes #67822, which fixed the first such read (repr(ti) on state/map_index) by widening load_only — but TaskInstanceHistory(ti) copies ~35 columns, so widening the column list further just moves the crash. Instead, re-select the rows to mutate by their already-loaded ids (with dag_run joined-loaded) so both the reset and adopt branches operate on fully loaded, session-bound instances, independent of what the executor did with the objects it was handed.

The regression test simulates the executor detaching the TaskInstances it was handed (as observed in production): without this change it fails with the production DetachedInstanceError (deferred load of try_number); with it, the orphan is reset with a TaskInstanceHistory audit row and the adopted TI gets its heartbeat/dag_run handling.

---

Was generative AI tooling used to co-author this PR?

• Yes — Claude Code (Opus 4.8)

Generated-by: Claude Code (Opus 4.8) following the guidelines

[boring-cyborg]

Congratulations on your first Pull Request and welcome to the Apache Airflow community! If you have any issues or are unsure about any anything please check our Contributors' Guide
Here are some useful points:

• Pay attention to the quality of your code (ruff, mypy and type annotations). Our prek-hooks will help you with that.
• In case of a new feature add useful documentation (in docstrings or in docs/ directory). Adding a new operator? Check this short guide Consider adding an example Dag that shows how users should use it.
• Consider using Breeze environment for testing locally, it's a heavy docker but it ships with a working Airflow and a lot of integrations.
• Be patient and persistent. It might take some time to get a review or get the final approval from Committers.
• Please follow ASF Code of Conduct for all communication including (but not limited to) comments on Pull Requests, Mailing list and Slack.
• Be sure to read the Airflow Coding style.
• Always keep your Pull Requests rebased, otherwise your build might fail due to changes not related to your commits.
  Apache Airflow is a community-driven project and together we are making it better 🚀.
  In case of doubts contact the developers at:
  Mailing List: dev@airflow.apache.org
  Slack: https://s.apache.org/airflow-slack

[uranusjr]

Static checks need fixing.

The description asserts this was "observed in production with KubernetesExecutor as a scheduler CrashLoopBackOff," but the test only reproduces it by manually calling session.expunge() in a mock. I think we should identify the actual, real offending code path that triggers the issue you described to decide if the fix is a good idea.

[akhilpratap1991]

Static checks are green on the current head — the initial run was sitting in the first-time-contributor approval gate (action_required), not failing; after approval, CI image checks / Static checks passed.

On the real code path — the concrete trigger is an executor opening and closing a scoped session while adopting:

1. KubernetesExecutor.try_adopt_task_instances() → completed-pod adoption scoping (added in KubernetesExecutor: scope periodic completed-pod adoption to dead schedulers #66400) runs with create_session(): — in released providers up to 10.17.x, which is what Airflow 3.2.2 constraints pin.
2. create_session() defaults to the thread-scoped registry (settings.Session), so it returns the very session adopt_or_reset_orphaned_tasks() is holding, and its finally: session.close() expunges everything the orphan query loaded.
3. Back in the scheduler loop, prepare_db_for_next_try() → TaskInstanceHistory.record_ti() reads columns deferred by the load_only (plus last_heartbeat_at / dag_run.conf in the adopt branch) on now-detached instances → DetachedInstanceError, unhandled in _run_scheduler_loop. This is exactly the production incident we hit (3.2.2 + cncf provider 10.17.1, scheduler CrashLoopBackOff): the scheduler dies before the reset flushes, so the same orphan kills every restart. Where the attributes happen to be loaded, the resets are silently lost instead — the loop "succeeds" without persisting anything.

#67850 already fixed that one call site provider-side (scoped=False, released in 10.18.0), and #67822 fixed the first read (repr). Both are point fixes: executors are separately released (and can be third-party), and any scoped-session close during adoption re-creates the condition — so this PR makes the scheduler loop itself robust by operating on re-selected, session-bound rows.

I've updated the test accordingly: the executor mock now performs the exact operation the released provider does (with create_session(): pass) instead of manually expunging. Without this PR's change it fails with the production DetachedInstanceError (deferred load of try_number); with it, it passes.

---

Drafted-by: Claude Code (Opus 4.8); reviewed by @akhilpratap1991 before posting