Add durable task instance launch records

Persist immutable executor launch tokens independently of mutable or
deleted TaskInstance rows so that expected stale executor launches
return a typed 409 stale_executor_launch instead of an opaque 404.

Refs #69760

Author: wolvery

airflow-core/docs/migrations-ref.rst

@@ -39,7 +39,10 @@ Here's the list of all the Database Migrations that are executed via when you ru
 +-------------------------+------------------+-------------------+--------------------------------------------------------------+
 | Revision ID             | Revises ID       | Airflow Version   | Description                                                  |
 +=========================+==================+===================+==============================================================+
-| ``c7f0a5d2e9b4`` (head) | ``76c46545c91e`` | ``3.4.0``         | Lower case team names.                                       |
+| ``3c5f8e9a1d2b`` (head) | ``c7f0a5d2e9b4`` | ``3.4.0``         | Add task_instance_launch table for durable executor launch   |
+|                         |                  |                   | records.                                                     |
++-------------------------+------------------+-------------------+--------------------------------------------------------------+
+| ``c7f0a5d2e9b4``        | ``76c46545c91e`` | ``3.4.0``         | Lower case team names.                                       |
 +-------------------------+------------------+-------------------+--------------------------------------------------------------+
 | ``76c46545c91e``        | ``3c525f44bea8`` | ``3.4.0``         | Add new index for trigger.                                   |
 +-------------------------+------------------+-------------------+--------------------------------------------------------------+

airflow-core/newsfragments/70931.significant.rst

@@ -0,0 +1,29 @@
+Durable executor launch records fence stale task executors
+
+Executor task launches are now recorded in a new ``task_instance_launch`` table that is
+independent of the mutable ``task_instance`` row. The scheduler pre-assigns an immutable
+launch token when a task instance moves from ``scheduled`` to ``queued`` and writes an
+``active`` launch record for it. When a worker calls the Execution API ``/run`` endpoint,
+the token it carries is validated against the durable record: a launch whose token has
+already reached a terminal state (``consumed`` or ``superseded``) now returns a typed
+``409 stale_executor_launch`` response instead of a generic ``404``, so a worker started
+from a stale/duplicated launch exits cleanly (return code 0) rather than corrupting the
+task instance's state. Launch records are superseded on reschedule of stuck-queued tasks,
+failed pod adoption, orphan reset, and task-instance clear/next-try; successful adoption
+preserves the token. The mechanism is executor-agnostic (CeleryExecutor, KubernetesExecutor,
+EdgeExecutor).
+
+The new ``409 stale_executor_launch`` behaviour is gated behind a new Execution API version
+(``v2026_06_30``); older Task SDK clients pinned to earlier Execution API versions continue
+to observe the previous ``404`` behaviour, so the change is backward compatible.
+
+* Types of change
+
+  * [ ] Dag changes
+  * [ ] Config changes
+  * [x] API changes
+  * [ ] CLI changes
+  * [x] Behaviour changes
+  * [ ] Plugin changes
+  * [ ] Dependency changes
+  * [ ] Code interface changes

airflow-core/src/airflow/api_fastapi/execution_api/datamodels/taskinstance.py

@@ -65,6 +65,8 @@ class TIEnterRunningPayload(StrictBaseModel):
     """Process Identifier on `hostname`"""
     start_date: UtcDateTime
     """When the task started executing"""
+    external_executor_id: str | None = None
+    """Executor token for durable launch validation"""
 
 
 # Create an enum to give a nice name in the generated datamodels
@@ -287,6 +289,8 @@ class TaskInstance(BaseModel):
     map_index: int = -1
     hostname: str | None = None
     context_carrier: dict | None = None
+    external_executor_id: str | None = None
+    """Executor token for durable launch validation"""
     # The supervisor routes tasks to a coordinator by queue. The default keeps
     # hand-built instances (tests, dry runs) valid; the executor workload
     # always sends the real value.

airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py

@@ -89,6 +89,7 @@
 from airflow.models.dagrun import DagRun as DR
 from airflow.models.hitl import HITLDetail
 from airflow.models.log import Log
+from airflow.models.task_instance_launch import TERMINAL_STATES, TaskInstanceLaunch
 from airflow.models.taskinstance import TaskInstance as TI, _stop_remaining_tasks
 from airflow.models.taskinstancehistory import TaskInstanceHistory as TIH
 from airflow.models.taskreschedule import TaskReschedule
@@ -170,6 +171,7 @@ def ti_run(
             TI.unixname,
             TI.pid,
             TI.dag_version_id,
+            TI.external_executor_id,
             # This selects the raw JSON value, bypassing the deserialization -- we want that to happen on the
             # client
             column("next_kwargs", JSON),
@@ -186,6 +188,25 @@ def ti_run(
         ti = session.execute(old).one()
         log.debug("Retrieved task instance details", state=ti.state, dag_id=ti.dag_id, task_id=ti.task_id)
     except NoResultFound:
+        # TI not found - check if we have a token to validate against launch records
+        payload_token = ti_run_payload.external_executor_id
+        if payload_token:
+            # Check if this token has a known launch record
+            launch = TaskInstanceLaunch.get_by_token(payload_token, session=session)
+            if launch and launch.state in TERMINAL_STATES:
+                # Token is known but in a terminal state (consumed or superseded)
+                log.warning(
+                    "Stale executor token for missing task instance", token=payload_token, state=launch.state
+                )
+                raise HTTPException(
+                    status_code=status.HTTP_409_CONFLICT,
+                    detail={
+                        "reason": "stale_executor_launch",
+                        "message": "Executor token is stale or has been superseded",
+                    },
+                )
+            # Token is unknown/null - ambiguous, could be upgrade scenario or stale executor
+            # Let the 404 be returned
         log.error("Task Instance not found")
         raise HTTPException(
             status_code=status.HTTP_404_NOT_FOUND,
@@ -195,6 +216,39 @@ def ti_run(
             },
         )
 
+    # Validate launch token if provided (launch record presence drives enforcement)
+    # If TI has a DB token and that token has a launch record, enforce strict matching
+    payload_token = ti_run_payload.external_executor_id
+    db_token = ti.external_executor_id
+
+    if db_token:
+        # Check if this DB token has an active/terminal launch record
+        launch_record = TaskInstanceLaunch.get_by_token(db_token, session=session)
+        if launch_record:
+            # Launch record exists - enforce strict matching
+            if payload_token != db_token:
+                log.error(
+                    "External executor ID mismatch for known launch record",
+                    payload_token=payload_token,
+                    db_token=db_token,
+                    launch_state=launch_record.state,
+                )
+                raise HTTPException(
+                    status_code=status.HTTP_409_CONFLICT,
+                    detail={
+                        "reason": "stale_executor_launch",
+                        "message": "Executor token does not match known launch record",
+                    },
+                )
+    elif payload_token and payload_token != db_token:
+        # No launch record exists, preserve rolling compatibility
+        # Only enforce if both have tokens and they differ
+        log.debug(
+            "Token mismatch but no launch record (rolling upgrade scenario)",
+            payload_token=payload_token,
+            db_token=db_token,
+        )
+
     # We exclude_unset to avoid updating fields that are not set in the payload
     data = ti_run_payload.model_dump(exclude_unset=True)
 
@@ -265,6 +319,32 @@ def ti_run(
         result = session.execute(query)
         log.info("Task instance state updated", rows_affected=getattr(result, "rowcount", 0))
 
+        # Mark the launch record as consumed if we have a token and the transition to RUNNING succeeded.
+        # Only mark on successful transitions from QUEUED/RESTARTING (not on duplicate RUNNING requests).
+        if payload_token and previous_state in (TaskInstanceState.QUEUED, TaskInstanceState.RESTARTING):
+            marked = TaskInstanceLaunch.mark_consumed(
+                token=payload_token,
+                session=session,
+            )
+            if marked:
+                log.debug("Marked launch token as consumed", token=payload_token)
+            else:
+                # If mark_consumed returns False, the record was already consumed or missing
+                # Only allow on duplicate RUNNING request (same host/user/pid), otherwise it's a conflict
+                if previous_state == TaskInstanceState.RUNNING:
+                    log.debug("Launch token already consumed (duplicate request)", token=payload_token)
+                else:
+                    # Roll back the update since mark_consumed failed and this isn't a duplicate
+                    session.rollback()
+                    log.error("Failed to mark launch token as consumed", token=payload_token)
+                    raise HTTPException(
+                        status_code=status.HTTP_409_CONFLICT,
+                        detail={
+                            "reason": "stale_executor_launch",
+                            "message": "Launch record not found or already consumed",
+                        },
+                    )
+
         dr = (
             session.scalars(
                 select(DR)

airflow-core/src/airflow/api_fastapi/execution_api/versions/__init__.py

@@ -47,6 +47,7 @@
     AddPartitionDateField,
     AddRetryPolicyFields,
     AddTaskAndAssetStateStoreEndpoints,
+    AddTaskInstanceExternalExecutorIdField,
     AddTaskInstanceQueueField,
     AddTeamNameField,
     AddVariableKeysEndpoint,
@@ -61,6 +62,7 @@
         AddVariableKeysEndpoint,
         AddConnectionTestEndpoint,
         AddAwaitingInputStatePayload,
+        AddTaskInstanceExternalExecutorIdField,
         AddTaskInstanceQueueField,
         AddRetryPolicyFields,
         AddTeamNameField,

airflow-core/src/airflow/api_fastapi/execution_api/versions/v2026_06_30.py

@@ -29,6 +29,7 @@
     DagRun,
     TaskInstance,
     TIAwaitingInputStatePayload,
+    TIEnterRunningPayload,
     TIRetryStatePayload,
     TIRunContext,
 )
@@ -61,6 +62,17 @@ class AddTaskInstanceQueueField(VersionChange):
     instructions_to_migrate_to_previous_version = (schema(TaskInstance).field("queue").didnt_exist,)
 
 
+class AddTaskInstanceExternalExecutorIdField(VersionChange):
+    """Add the `external_executor_id` field to task instance launch payloads."""
+
+    description = __doc__
+
+    instructions_to_migrate_to_previous_version = (
+        schema(TaskInstance).field("external_executor_id").didnt_exist,
+        schema(TIEnterRunningPayload).field("external_executor_id").didnt_exist,
+    )
+
+
 class AddAwaitingInputStatePayload(VersionChange):
     """Add the awaiting_input task instance state transition payload (Human-in-the-loop, no trigger)."""
 

airflow-core/src/airflow/executors/workloads/task.py

@@ -45,7 +45,12 @@ class TaskInstanceDTO(TaskInstance):
     pool_slots: int
     priority_weight: int
 
-    external_executor_id: str | None = Field(default=None, exclude=True)
+    # NOT excluded: the durable launch token must reach the worker so it can echo
+    # it back on POST /run for stale-executor fencing. Executors that pre-assign
+    # external_executor_id (Celery, Kubernetes) rely on this surviving serialization
+    # into the workload JSON that is shipped to the worker/pod. executor_config stays
+    # excluded — it is a scheduler-only concern the worker never reads.
+    external_executor_id: str | None = Field(default=None)
     executor_config: dict | None = Field(default=None, exclude=True)
 
     # TODO: Task-SDK: Can we replace TaskInstanceKey with just the uuid across the codebase?

airflow-core/src/airflow/jobs/scheduler_job_runner.py

@@ -105,6 +105,7 @@
 from airflow.models.dagwarning import DagWarning, DagWarningType
 from airflow.models.pool import normalize_pool_name_for_stats
 from airflow.models.serialized_dag import SerializedDagModel
+from airflow.models.task_instance_launch import TaskInstanceLaunch, TaskInstanceLaunchState
 from airflow.models.taskinstance import TaskInstance
 from airflow.models.taskinstancekey import TaskInstanceKey
 from airflow.models.team import Team
@@ -1117,6 +1118,30 @@ def _executable_task_instances_to_queued(self, max_tis: int, session: Session) -
             else:
                 session.execute(queued_update)
 
+            # Bulk insert TaskInstanceLaunch records for active launches with non-null tokens.
+            # This happens after external_executor_ids have been synced into executable_tis in-memory.
+            # The DB-stored external_executor_id values will be read back in the next query if needed.
+            launch_records = []
+            for ti in executable_tis:
+                token = ti.external_executor_id
+                if token:
+                    launch_records.append(
+                        TaskInstanceLaunch(
+                            token=token,
+                            task_instance_id=str(ti.id),
+                            dag_id=ti.dag_id,
+                            task_id=ti.task_id,
+                            run_id=ti.run_id,
+                            map_index=ti.map_index if ti.map_index is not None else -1,
+                            try_number=ti.try_number,
+                            executor=ti.executor or "default",
+                            state=TaskInstanceLaunchState.ACTIVE,
+                        )
+                    )
+            if launch_records:
+                session.add_all(launch_records)
+                session.flush()
+
             for ti in executable_tis:
                 ti.emit_state_change_metric(TaskInstanceState.QUEUED)
 
@@ -3212,6 +3237,15 @@ def _reschedule_stuck_task(self, ti: TaskInstance, session: Session):
         filter_for_tis = TI.filter_for_tis([ti])
         if filter_for_tis is None:
             return
+
+        # Supersede any active launch record before rescheduling, so the old token
+        # cannot be reused if the old executor manages to come back.
+        if ti.external_executor_id:
+            TaskInstanceLaunch.mark_superseded(
+                token=ti.external_executor_id,
+                session=session,
+            )
+
         session.execute(
             update(TI)
             .where(filter_for_tis)
@@ -3220,6 +3254,7 @@ def _reschedule_stuck_task(self, ti: TaskInstance, session: Session):
                 queued_dttm=None,
                 queued_by_job_id=None,
                 scheduled_dttm=timezone.utcnow(),
+                external_executor_id=None,  # Clear so next queue cycle assigns fresh token
             )
             .execution_options(synchronize_session=False)
         )
@@ -3428,6 +3463,13 @@ def adopt_or_reset_orphaned_tasks(self, *, session: Session = NEW_SESSION) -> in
                     reset_tis_message = []
                     for ti in to_reset:
                         reset_tis_message.append(repr(ti))
+                        # Mark the current launch token as superseded before resetting the TI
+                        # so the old executor cannot re-adopt this task if it comes back
+                        if ti.external_executor_id:
+                            TaskInstanceLaunch.mark_superseded(
+                                token=ti.external_executor_id,
+                                session=session,
+                            )
                         # If we reset a TI, it will be eligible to be scheduled again.
                         # This can cause the scheduler to increase the try_number on the TI.
                         # Record the current try to TaskInstanceHistory first so users have an audit trail for

airflow-core/src/airflow/migrations/versions/0132_3_4_0_add_task_instance_launch_table.py

@@ -0,0 +1,85 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""
+Add task_instance_launch table for durable executor launch records.
+
+Revision ID: 3c5f8e9a1d2b
+Revises: c7f0a5d2e9b4
+Create Date: 2026-05-10 00:00:00.000000
+
+"""
+
+from __future__ import annotations
+
+import sqlalchemy as sa
+from alembic import op
+
+from airflow.utils.sqlalchemy import UtcDateTime
+
+# revision identifiers, used by Alembic.
+revision = "3c5f8e9a1d2b"
+down_revision = "c7f0a5d2e9b4"
+branch_labels = None
+depends_on = None
+airflow_version = "3.4.0"
+
+
+def upgrade():
+    """Create task_instance_launch table for durable executor token tracking."""
+    op.create_table(
+        "task_instance_launch",
+        sa.Column("token", sa.String(256), nullable=False),
+        sa.Column("task_instance_id", sa.String(250), nullable=False),
+        sa.Column("dag_id", sa.String(250), nullable=False),
+        sa.Column("task_id", sa.String(250), nullable=False),
+        sa.Column("run_id", sa.String(250), nullable=False),
+        sa.Column("map_index", sa.Integer(), nullable=False, server_default="-1"),
+        sa.Column("try_number", sa.Integer(), nullable=False),
+        sa.Column("executor", sa.String(256), nullable=False),
+        sa.Column("state", sa.String(20), nullable=False, server_default="active"),
+        sa.Column("created_at", UtcDateTime(timezone=True), nullable=False, server_default=sa.func.now()),
+        sa.Column("updated_at", UtcDateTime(timezone=True), nullable=False, server_default=sa.func.now()),
+        sa.Column("consumed_at", UtcDateTime(timezone=True), nullable=True),
+        sa.Column("superseded_at", UtcDateTime(timezone=True), nullable=True),
+        sa.PrimaryKeyConstraint("token", name="pk_task_instance_launch_token"),
+        sa.CheckConstraint(
+            "state IN ('active', 'consumed', 'superseded')",
+            name="state_enum",
+        ),
+    )
+    op.create_index(
+        "idx_task_instance_launch_task_instance_id",
+        "task_instance_launch",
+        ["task_instance_id"],
+    )
+    op.create_index(
+        "idx_task_instance_launch_state_updated",
+        "task_instance_launch",
+        ["state", "updated_at"],
+    )
+    op.create_index(
+        "idx_task_instance_launch_created_at",
+        "task_instance_launch",
+        ["created_at"],
+    )
+
+
+def downgrade():
+    """Drop task_instance_launch table."""
+    op.drop_table("task_instance_launch")