Check that external tasks and task groups exist on Airflow 3 in ExternalTaskSensor

[bingqin2]

On Airflow 3, check_existence=True on ExternalTaskSensor verified nothing: a worker has no database access, and the Airflow 2 lookup (DagModel plus a DagBag of the file) has no equivalent through the execution API. #72514 tracks this, and #72517 adds the Dag-level lookup (ti.get_dag, Airflow 3.2+). This adds the task and task-group half. It is independent of #72517: it only touches the Airflow 3 branch of the existence check and the trigger, so whichever merges second has a small rebase in _check_for_existence.

Depends on #73086, which makes GET /execution/task-instances/states answer 404 for a task that the Dag version of a named run does not define (and that has no instance in that run), and resolve a task group against the named run's version instead of the latest one. This PR is a draft until that lands and should merge after it.

How it decides

Per awaited run, on each poke until every awaited run has been seen:

• no run yet (get_dr_count is 0): keep waiting; nothing is known about a run before it exists (feat: implement DagTaskGroupsExistence and DagTasksExistence endpoints #67832)
• external_task_ids: get_task_states(task_ids=...) for that run; a 404 means the run's Dag version does not define one of the tasks: ExternalTaskNotFoundError, carrying the server's message that names them
• external_task_group_id: get_task_states(task_group_id=...) for that run; a 404 means the group is not defined for it: ExternalTaskGroupNotFoundError

Nothing is inferred from task-instance counts: a normal answer means the run's version defines the task, whether or not its instances exist yet. (The scheduler adds newly parsed tasks to an unfinished run of an unversioned bundle on a later pass, so a count taken in that window would be wrong, and the one-shot check would make it final.) On an API server without #73086 the task call never answers 404, so the sensor keeps waiting exactly as it does today; the group call answers 404 against the latest version there, which is what _poke_af3 already relies on.

Missing tasks and groups are configuration errors, so they are raised regardless of soft_fail, as on Airflow 2.

The deferrable path gets the same rule. WorkflowTrigger takes check_existence (serialized, default False, so already-serialized triggers are unaffected) and yields {"status": "not_found", "kind": "task" | "task_group", "message": ...}, which execute_complete maps to the matching exception. The Airflow 2 path is untouched.

Known limits

• Runs that are never created, for example an execution_date_fn pointing at dates that never run, still wait until timeout, as on Airflow 2.
• A sensor deployed before the awaited task exists in the external Dag at all is reported missing once the run exists, as on Airflow 2, which checked the current file once.
• Whether the external Dag itself is registered is not checked here; that is Stop ExternalTaskSensor waiting for missing Dags on Airflow 3.2+ #72517.

Changes

• utils/sensor_helper.py: _check_external_task_existence(api, ...), usable from the worker (ti) and from the triggerer (RuntimeTaskInstance); _not_found_message reads the server's 404 message out of the AirflowRuntimeError
• sensors/external_task.py: _check_for_existence_af3 runs the rule on each poke until every awaited run has been checked; the trigger receives check_existence; execute_complete handles not_found
• triggers/external_task.py: check_existence parameter, serialized; the existence loop before the state polling
• tests: 8 sensor tests (waits for the run then checks once, task unknown to the run's version, soft_fail ignored, group without instances keeps waiting, group unknown, other API errors kept, deferrable hands over to the trigger, execute_complete mapping) and 4 trigger tests (serialization, task unknown, group unknown, waits for the run)

Testing

• providers/standard: tests/unit/standard/sensors/test_external_task_sensor.py and tests/unit/standard/triggers/test_external_task.py (67 tests on Airflow 3)
• mypy on the three changed modules, prek hooks on the changed files

Part of #72514 (the task and task-group half; #72517 covers the Dag level). Depends on #73086.

---

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

• Yes (please specify the tool below)

Generated-by: Claude Code (Claude Fable 5.1) following the guidelines. I reviewed and understand all changes; the tests were run locally as listed above.

---

🤖 Generated with Claude Code

[Vamsi-klu]

​

[bingqin2]

You're right, and thanks for the pointer to _verify_integrity_if_dag_changed: for a run without a pinned bundle version the scheduler adds newly parsed tasks on a later pass, so a zero count in that window is not an answer, and the one-shot flag would have made it permanent. That inference is gone. The check now asks task-instances/states for the awaited task ids (or the task group) per existing run and treats only a 404 as "does not exist"; a normal answer means the run's version defines it, whether or not its instances exist yet. The 404 comes from #73086, which validates task_ids against the version each named run resolves to and also counts an existing instance as proof, so a task removed in a newer version is still found for a run that has it. An API server without that validation never 404s for a task, so with it the sensor simply keeps waiting, as it does today.

On mapped tasks: as far as I can see DagRun._create_tasks always creates a map_index=-1 placeholder while the length is unknown, so that case counted as present, but with the count inference gone it no longer matters here.

[bingqin2]

Agreed. The 404 is now the only signal on the group path as well (an empty result no longer raises), and the description states the dependency: with #73086 the group is resolved against the awaited run's version, so a group renamed after the run was created is still found for that run. Without it the 404 reflects the latest version, which is what _poke_af3 already relies on today: a renamed group makes the current sensor fail with an AirflowRuntimeError on main, and this PR only changes the exception type. I have converted this PR to a draft until #73086 lands.

[bingqin2]

You're right, and thanks for the pointer to _verify_integrity_if_dag_changed: for a run without a pinned bundle version the scheduler adds newly parsed tasks on a later pass, so a zero count in that window is not an answer, and the one-shot flag would have made it permanent. That inference is gone. The check now asks task-instances/states for the awaited task ids (or the task group) per existing run and treats only a 404 as "does not exist"; a normal answer means the run's version defines it, whether or not its instances exist yet. The 404 comes from #73086, which validates task_ids against the version each named run resolves to and also counts an existing instance as proof, so a task removed in a newer version is still found for a run that has it. An API server without that validation never 404s for a task, so with it the sensor simply keeps waiting, as it does today.

On mapped tasks: as far as I can see DagRun._create_tasks always creates a map_index=-1 placeholder while the length is unknown, so that case counted as present, but with the count inference gone it no longer matters here.

[bingqin2]

Agreed. The 404 is now the only signal on the group path as well (an empty result no longer raises), and the description states the dependency: with #73086 the group is resolved against the awaited run's version, so a group renamed after the run was created is still found for that run. Without it the 404 reflects the latest version, which is what _poke_af3 already relies on today: a renamed group makes the current sensor fail with an AirflowRuntimeError on main, and this PR only changes the exception type. I have converted this PR to a draft until #73086 lands.