Stop ExternalTaskSensor waiting for missing Dags on Airflow 3.2+

[ignacioparicio]

ExternalTaskSensor(check_existence=True) is supposed to fail immediately when the external Dag does not exist. On Airflow 3 it does not check at all. When external_dag_id refers to a Dag that does not exist, the sensor waits until its timeout instead of reporting the problem immediately.

The same thing happens when waiting is handed off to the triggerer (deferrable=True): the wait starts without a check.

Note:

• Airflow 2 is not affected.
• This only changes behavior for sensors with check_existence=True.

Where it happens

In providers/standard/src/airflow/providers/standard/sensors/external_task.py:

def _poke_af3(self, context: Context, dttm_filter: Sequence[datetime.datetime]) -> bool:
    from airflow.providers.standard.utils.sensor_helper import _get_count_by_matched_states

    self._has_checked_existence = True  # set here, then never read again
    ti = context["ti"]

_has_checked_existence is meant to stop the same check from being repeated. In the Airflow 3 code it is set before any check has happened.

The check itself is behind this condition:

if self.check_existence and not self._has_checked_existence:
    self._check_for_existence(session=session)

Both copies of the condition are in Airflow 2-only code, so Airflow 3 never calls _check_for_existence.

How it got here

The existing check reads the metadata database and the Dag file. An Airflow 3 worker has access to neither, so it cannot use that implementation.

• Make ExternalTaskSensor work with Task SDK  #48651 (6775bf7bae) split the sensor into Airflow 2 and Airflow 3 implementations. The status checks were moved to the new Airflow 3 implementation, but the existence check was not.
• fix: ExternalTaskSensor check_existence ignored in deferrable mode (Airflow < 3.0) #64394 (ce88001bb2) later fixed check_existence when deferrable=True, but only for Airflow 2. The Airflow 3 code still starts waiting without checking.

The missing piece was a way for a worker to ask whether a Dag exists. #56955 added that to the execution API in Airflow 3.2.

#67832 explored a broader check for individual tasks and task groups. It was closed because the answer depends on the Dag version used by a particular run. This PR does not add a new API. It uses the existing Dag lookup only to answer whether the external Dag itself exists.

What this PR does:

• external_task.py: on Airflow 3.2 and later, asks the execution API whether the external Dag exists before the sensor starts waiting. A missing Dag raises ExternalDagNotFoundError; other API errors are left unchanged.
• external_task.py: applies the check whether the sensor waits itself or hands the wait to the triggerer, and documents the limits below.
• test_external_task_sensor.py: covers both ways of waiting, repeated checks, unexpected API errors, and older Airflow 3 versions.

What this PR does not fix:

• On Airflow 3, external_task_ids and external_task_group_id are still not checked. The external Dag is checked, but the execution API cannot say whether a task exists in the Dag version used by a particular run. When an external task or task group is configured, the sensor now warns that only the external Dag was checked. The full task and task-group check is not part of this PR.
• Airflow 3.0 and 3.1 cannot perform the Dag check because the required execution API was added in Airflow 3.2. They also log a warning and keep the existing behavior.

related: #72514

---

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

• Yes, Codex (GPT-5)

Used to trace the regression through the history above, check what an Airflow 3 worker can ask the execution API, and review the implementation and tests.

[jroachgolf84]

closes: #72514