Validate KubernetesExecutor launches with external executor IDs

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 launch token assigned when the task was queued"""
 
 
 # Create an enum to give a nice name in the generated datamodels
@@ -291,6 +293,7 @@ class TaskInstance(BaseModel):
     # hand-built instances (tests, dry runs) valid; the executor workload
     # always sends the real value.
     queue: str = "default"
+    external_executor_id: str | None = None
 
 
 class AssetReferenceAssetEventDagRun(StrictBaseModel):

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

@@ -170,6 +170,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),
@@ -197,6 +198,13 @@ def ti_run(
 
     # We exclude_unset to avoid updating fields that are not set in the payload
     data = ti_run_payload.model_dump(exclude_unset=True)
+    # Only fence on the launch token when the worker actually presented a non-null token. Older
+    # Task SDK clients may omit the field entirely, and newer clients can serialize it as null
+    # before scheduler-side preassignment is available everywhere. Treating either case as a stale
+    # launch would 409 every task start for pre-assigning executors (KubernetesExecutor,
+    # CeleryExecutor) during rolling upgrades. When it is absent/null we fall back to state-based
+    # validation, matching the pre-token behavior.
+    payload_external_executor_id = data.pop("external_executor_id", None)
 
     # don't update start date when resuming from deferral
     if ti.next_kwargs:
@@ -215,6 +223,23 @@ def ti_run(
         ti_run_payload.pid,
     ):
         log.info("Duplicate start request received", hostname=ti_run_payload.hostname)
+    elif (
+        payload_external_executor_id is not None
+        and ti.external_executor_id is not None
+        and ti.external_executor_id != payload_external_executor_id
+    ):
+        log.warning(
+            "Cannot start Task Instance with stale executor launch token",
+            expected_external_executor_id=ti.external_executor_id,
+            provided_external_executor_id=payload_external_executor_id,
+        )
+        raise HTTPException(
+            status_code=status.HTTP_409_CONFLICT,
+            detail={
+                "reason": "stale_executor_launch",
+                "message": "TI executor launch token does not match the current queued task instance",
+            },
+        )
     elif previous_state not in (TaskInstanceState.QUEUED, TaskInstanceState.RESTARTING):
         log.warning(
             "Cannot start Task Instance in invalid state",

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,7 @@ class TaskInstanceDTO(TaskInstance):
     pool_slots: int
     priority_weight: int
 
-    external_executor_id: str | None = Field(default=None, exclude=True)
+    external_executor_id: str | None = 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/tests/unit/api_fastapi/execution_api/versions/head/test_task_instances.py

@@ -333,6 +333,125 @@ def test_ti_run_state_to_running(
         )
         assert response.status_code == 409
 
+    def test_ti_run_rejects_stale_external_executor_id(self, client, session, create_task_instance):
+        """A worker can only start the launch token currently assigned to the queued TI."""
+        ti = create_task_instance(
+            task_id="test_ti_run_rejects_stale_external_executor_id",
+            state=State.QUEUED,
+            dagrun_state=DagRunState.RUNNING,
+            session=session,
+            dag_id=str(uuid4()),
+        )
+        ti.external_executor_id = "current-launch-token"
+        session.commit()
+
+        response = client.patch(
+            f"/execution/task-instances/{ti.id}/run",
+            json={
+                "state": "running",
+                "hostname": "random-hostname",
+                "unixname": "random-unixname",
+                "pid": 100,
+                "start_date": "2024-09-30T12:00:00Z",
+                "external_executor_id": "stale-launch-token",
+            },
+        )
+
+        assert response.status_code == 409
+        assert response.json()["detail"]["reason"] == "stale_executor_launch"
+        session.refresh(ti)
+        assert ti.state == State.QUEUED
+
+    def test_ti_run_accepts_matching_external_executor_id(self, client, session, create_task_instance):
+        """A worker presenting the launch token currently assigned to the TI can start it."""
+        ti = create_task_instance(
+            task_id="test_ti_run_accepts_matching_external_executor_id",
+            state=State.QUEUED,
+            dagrun_state=DagRunState.RUNNING,
+            session=session,
+            dag_id=str(uuid4()),
+        )
+        ti.external_executor_id = "current-launch-token"
+        session.commit()
+
+        response = client.patch(
+            f"/execution/task-instances/{ti.id}/run",
+            json={
+                "state": "running",
+                "hostname": "random-hostname",
+                "unixname": "random-unixname",
+                "pid": 100,
+                "start_date": "2024-09-30T12:00:00Z",
+                "external_executor_id": "current-launch-token",
+            },
+        )
+
+        assert response.status_code == 200
+        session.refresh(ti)
+        assert ti.state == State.RUNNING
+        # The launch token is preserved; matching it must not clear or overwrite the field.
+        assert ti.external_executor_id == "current-launch-token"
+
+    def test_ti_run_allows_missing_external_executor_id_for_old_clients(
+        self, client, session, create_task_instance
+    ):
+        """An older worker that omits the launch token must not be fenced out (rolling upgrade)."""
+        ti = create_task_instance(
+            task_id="test_ti_run_allows_missing_external_executor_id_for_old_clients",
+            state=State.QUEUED,
+            dagrun_state=DagRunState.RUNNING,
+            session=session,
+            dag_id=str(uuid4()),
+        )
+        ti.external_executor_id = "current-launch-token"
+        session.commit()
+
+        # Payload deliberately omits ``external_executor_id`` entirely, as a pre-token Task SDK would.
+        response = client.patch(
+            f"/execution/task-instances/{ti.id}/run",
+            json={
+                "state": "running",
+                "hostname": "random-hostname",
+                "unixname": "random-unixname",
+                "pid": 100,
+                "start_date": "2024-09-30T12:00:00Z",
+            },
+        )
+
+        assert response.status_code == 200
+        session.refresh(ti)
+        assert ti.state == State.RUNNING
+
+    def test_ti_run_allows_null_external_executor_id_for_rolling_upgrade(
+        self, client, session, create_task_instance
+    ):
+        """A worker sending a null launch token must not be fenced out during rolling upgrades."""
+        ti = create_task_instance(
+            task_id="test_ti_run_allows_null_external_executor_id_for_rolling_upgrade",
+            state=State.QUEUED,
+            dagrun_state=DagRunState.RUNNING,
+            session=session,
+            dag_id=str(uuid4()),
+        )
+        ti.external_executor_id = "current-launch-token"
+        session.commit()
+
+        response = client.patch(
+            f"/execution/task-instances/{ti.id}/run",
+            json={
+                "state": "running",
+                "hostname": "random-hostname",
+                "unixname": "random-unixname",
+                "pid": 100,
+                "start_date": "2024-09-30T12:00:00Z",
+                "external_executor_id": None,
+            },
+        )
+
+        assert response.status_code == 200
+        session.refresh(ti)
+        assert ti.state == State.RUNNING
+
     def test_ti_run_returns_execution_token(
         self, client, exec_app, session, create_task_instance, time_machine
     ):

airflow-core/tests/unit/executors/test_workloads.py

@@ -159,7 +159,7 @@ def test_workload_ti_round_trips_through_sdk_generated_model():
     )
 
     dumped = ti.model_dump(mode="json")
-    assert "external_executor_id" not in dumped
+    assert dumped["external_executor_id"] == "celery-id"
     assert "executor_config" not in dumped
     # Executor-side scheduling fields stay on the workload wire (older workers
     # deserialize the workload with a model that requires them) but are not
@@ -169,6 +169,7 @@ def test_workload_ti_round_trips_through_sdk_generated_model():
 
     received = GeneratedTaskInstance.model_validate(dumped)
     assert received.queue == "jdk-17"
+    assert received.external_executor_id == "celery-id"
     assert received.map_index == 3
     assert not hasattr(received, "pool_slots")