Let TriggerDagRunOperator own its execution via a new accessor

[1fanwang]

Opened for the [DISCUSS] Let TriggerDagRunOperator own its execution logic thread, so the accessor direction can be weighed as an actual diff rather than in the abstract — see Scope for what it does and doesn't cover.

Why

On Airflow 3, TriggerDagRunOperator.execute() raises DagRunTriggerException and the task runner does the trigger and the synchronous wait. So the trigger-and-wait contract (and any crash recovery) lives in two places that can drift. The poll half is already a first-class Task SDK accessor (ti.get_dagrun_state()); only the trigger half was missing, so it had to go through the exception side channel.

What

• Add ti.trigger_dag_run(), the counterpart to ti.get_dagrun_state() — same execution-API endpoint and scoped token the runner already uses, so no new authz surface.
• On Airflow 3.3+ non-deferrable runs, the operator does the submit and poll itself and subclasses ResumableJobMixin, so the durability contract lives in one place.
• Opt-in durable flag (default False): the triggered run id is persisted before polling, so a worker crash mid-wait reconnects to the in-flight run on retry instead of triggering a duplicate.
• The version-conditional ResumableJobMixin shim (real on 3.3+, a stub below) lives in common.compat.standard, so the operator body stays on the clean 3.3+ path and the older-version back-compat retires as a contained delete once the provider floor reaches 3.3.
• Deferrable, and Airflow < 3.3 / 2.11, keep the existing DagRunTriggerException path.

Scope

This is the 3.3+ end-state, not a drop-in for the full supported range: the operator owns execution on 3.3+, while 3.0–3.2 and 2.11 stay on today's path, so durability applies on 3.3+ only. Whether and when standard can require 3.3+ (and drop the fallbacks) is the provider min_version question under discussion on the thread. The lower-duplication interim that keeps a single runner path is #68952 / #68955.

Testing done

Standard provider suite: 41 passed, 41 skipped (the skips are Airflow-2-only, skipped on 3.4). The existing synchronous-path tests are re-homed to assert the accessor calls instead of the raised exception; deferrable / Airflow-2 tests are unchanged. New coverage: the accessor, the operator-owned path (trigger, skip/reset, sync wait success/failure, fire-and-forget), and the four durable states (persist-before-poll, reconnect, short-circuit an already-succeeded prior run, resubmit after a failed one). Task SDK accessor/runner tests pass.

Live E2E — Airflow 3.4 standalone (LocalExecutor + Postgres). Two DAGs: a parent TriggerDagRunOperator(wait_for_completion=True, durable=True, poke_interval=3) triggers a child whose only task sleeps 45s.

1. Clean path (no crash) — the operator owns execution via the accessors; the runner's DagRunTriggerException path is never taken.

t	parent task	child runs
8s	running (try 1)	1 (running)
48s	success (try 1)	1 (success)

Waiting for e2e_child on manual__…920653+00:00 to reach an allowed state [SUCCESS] ...
DagRunTriggerException occurrences in the task log: 0

2. Worker SIGKILL mid-wait (durable reconnect) — the parent's worker is kill -9ed while it polls; the scheduler retries the task.

t	parent task	child runs from this parent
6s	running (try 1), triggered child	1
—	SIGKILL the parent worker	1
9s	up_for_retry (try 1)	1
12s	running (try 2)	1 (no new trigger)
42s	success (try 2)	1

attempt=2: Reconnecting to existing job
attempt=2: Waiting for e2e_child on manual__…957584+00:00 ...   <- same run id as attempt=1
parent task: success (try_number=2)   child dag runs created: 1  (reconnected, not duplicated)

Raw state progression — crash run

child runs before crash test: 1
[t=3s]  parent=(none)       try=0          | child_total=1
[t=6s]  parent=running      try=1 pid=62403| child_total=2   <- triggered + polling
>>> SIGKILL parent worker pid=62403 (simulate crash mid-wait)
[t=9s]  parent=up_for_retry try=1          | child_total=2
[t=12s] parent=running      try=2 pid=62469| child_total=2   <- retry; NO new child
[t=15s..39s] parent=running try=2          | child_total=2   <- reconnect-polling the in-flight run
[t=42s] parent=success      try=2          | child_total=2
manual__2026-…T16:51:09.920653+00:00 = success   (clean run's child)
manual__2026-…T16:54:29.957584+00:00 = success   (crash run's child — single run despite the crash+retry)

Refs

The accessor-based option discussed on the [DISCUSS] thread, alongside the hand-rolled-in-the-runner durability in #68936 and the share-the-core refactor in #68952 / #68955.

---

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

• Yes (please specify the tool below)

[raphaelauv]

Hey, thanks for this big refacto.

airflow 2.11 is EOL , so maybe you could remove eveyrthing related to 2.11 ?