kaxil · Jun 11, 2026 · Jun 11, 2026 · Jun 11, 2026 · Jun 11, 2026 · Jun 11, 2026
diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml
@@ -532,12 +532,6 @@ repos:
         language: python
         pass_filenames: false
         files: ^dev/registry/registry_tools/types\.py$|^registry/src/_data/types\.json$
-      - id: check-task-instance-dto-sync
-        name: Check BaseTaskInstanceDTO duplicate is in sync between core and task-sdk
-        entry: ./scripts/ci/prek/check_task_instance_dto_sync.py
-        language: python
-        pass_filenames: false
-        files: ^airflow-core/src/airflow/executors/workloads/task\.py$|^task-sdk/src/airflow/sdk/execution_time/workloads/task\.py$
       - id: ruff
         name: Run 'ruff' for extremely fast Python linting
         description: "Run 'ruff' for extremely fast Python linting"

diff --git a/airflow-core/docs/core-concepts/executor/index.rst b/airflow-core/docs/core-concepts/executor/index.rst
@@ -271,20 +271,18 @@ Example:
 
     ExecuteTask(
         token="mock",
-        ti=TaskInstance(
+        ti=TaskInstanceDTO(
             id=UUID("4d828a62-a417-4936-a7a6-2b3fabacecab"),
             task_id="mock",
             dag_id="mock",
             run_id="mock",
             try_number=1,
+            dag_version_id=UUID("4d828a62-a417-4936-a7a6-2b3fabacecab"),
             map_index=-1,
             pool_slots=1,
             queue="default",
             priority_weight=1,
             executor_config=None,
-            parent_context_carrier=None,
-            context_carrier=None,
-            queued_dttm=None,
         ),
         dag_rel_path=PurePosixPath("mock.py"),
         bundle_info=BundleInfo(name="n/a", version="no matter"),

@@ -286,6 +286,10 @@ class TaskInstance(BaseModel):
     map_index: int = -1
     hostname: str | None = None
     context_carrier: dict | None = None
+    # 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.
+    queue: str = "default"
 
 
 class AssetReferenceAssetEventDagRun(StrictBaseModel):

@@ -49,6 +49,7 @@
 from airflow.api_fastapi.execution_api.versions.v2026_06_30 import (
     AddAwaitingInputStatePayload,
     AddConnectionTestEndpoint,
+    AddTaskInstanceQueueField,
     AddVariableKeysEndpoint,
 )
 
@@ -59,6 +60,7 @@
         AddVariableKeysEndpoint,
         AddConnectionTestEndpoint,
         AddAwaitingInputStatePayload,
+        AddTaskInstanceQueueField,
     ),
     Version(
         "2026-06-16",

@@ -19,7 +19,10 @@
 
 from cadwyn import VersionChange, endpoint, schema
 
-from airflow.api_fastapi.execution_api.datamodels.taskinstance import TIAwaitingInputStatePayload
+from airflow.api_fastapi.execution_api.datamodels.taskinstance import (
+    TaskInstance,
+    TIAwaitingInputStatePayload,
+)
 
 
 class AddVariableKeysEndpoint(VersionChange):
@@ -41,6 +44,14 @@ class AddConnectionTestEndpoint(VersionChange):
     )
 
 
+class AddTaskInstanceQueueField(VersionChange):
+    """Add the `queue` field to the TaskInstance model."""
+
+    description = __doc__
+
+    instructions_to_migrate_to_previous_version = (schema(TaskInstance).field("queue").didnt_exist,)
+
+
 class AddAwaitingInputStatePayload(VersionChange):
     """Add the awaiting_input task instance state transition payload (Human-in-the-loop, no trigger)."""
 

@@ -18,12 +18,12 @@
 
 from __future__ import annotations
 
-import uuid
 from pathlib import Path
 from typing import TYPE_CHECKING, Literal
 
-from pydantic import BaseModel, Field
+from pydantic import Field
 
+from airflow.api_fastapi.execution_api.datamodels.taskinstance import TaskInstance
 from airflow.executors.workloads.base import BaseDagBundleWorkload, BundleInfo
 from airflow.utils.state import TaskInstanceState
 
@@ -33,36 +33,20 @@
     from airflow.models.taskinstancekey import TaskInstanceKey
 
 
-class BaseTaskInstanceDTO(BaseModel):
+class TaskInstanceDTO(TaskInstance):
     """
-    Base schema for TaskInstance with the minimal fields shared by Executors and the Task SDK.
+    The versioned execution API ``TaskInstance`` schema plus executor-only fields.
 
-    This definition is duplicated in :mod:`airflow.sdk.execution_time.workloads.task`
-    and the two are kept in sync by the ``check-task-instance-dto-sync`` prek
-    hook. Update both files together.
+    The base class is the single source of truth for the fields a worker needs;
+    the fields added here are executor concerns (queueing order and pool
+    accounting) the worker never reads.
     """
 
-    id: uuid.UUID
-    dag_version_id: uuid.UUID
-    task_id: str
-    dag_id: str
-    run_id: str
-    try_number: int
-    map_index: int = -1
-
     pool_slots: int
-    queue: str
     priority_weight: int
-    executor_config: dict | None = Field(default=None, exclude=True)
-
-    parent_context_carrier: dict | None = None
-    context_carrier: dict | None = None
-
-
-class TaskInstanceDTO(BaseTaskInstanceDTO):
-    """TaskInstanceDTO with executor-specific ``external_executor_id`` field and ``key`` property."""
 
     external_executor_id: str | None = Field(default=None, exclude=True)
+    executor_config: dict | None = Field(default=None, exclude=True)
 
     # TODO: Task-SDK: Can we replace TaskInstanceKey with just the uuid across the codebase?
     @property

diff --git a/airflow-core/tests/unit/executors/test_workloads.py b/airflow-core/tests/unit/executors/test_workloads.py
@@ -32,6 +32,7 @@
 from airflow.executors.workloads.task import ExecuteTask
 from airflow.executors.workloads.types import state_class_for_key
 from airflow.models.callback import CallbackKey
+from airflow.sdk.api.datamodels._generated import TaskInstance as GeneratedTaskInstance
 
 
 def test_task_instance_alias_keeps_backwards_compat():
@@ -134,3 +135,39 @@ def test_callback_dto_key_returns_callback_key_instance():
     assert isinstance(key, CallbackKey)
     assert key.id == cid
     assert str(key) == cid
+
+
+def test_workload_ti_round_trips_through_sdk_generated_model():
+    """
+    The executor-side DTO and the SDK's generated TaskInstance share the
+    execution API schema; the serialized workload must carry the routing
+    fields and exclude the executor-only ones.
+    """
+    ti = TaskInstanceDTO(
+        id=uuid4(),
+        dag_version_id=uuid4(),
+        task_id="test_task",
+        dag_id="test_dag",
+        run_id="test_run",
+        try_number=2,
+        map_index=3,
+        pool_slots=4,
+        queue="jdk-17",
+        priority_weight=5,
+        external_executor_id="celery-id",
+        executor_config={"KubernetesExecutor": {"image": "custom"}},
+    )
+
+    dumped = ti.model_dump(mode="json")
+    assert "external_executor_id" not in dumped
+    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
+    # part of the worker-facing schema.
+    assert dumped["pool_slots"] == 4
+    assert dumped["priority_weight"] == 5
+
+    received = GeneratedTaskInstance.model_validate(dumped)
+    assert received.queue == "jdk-17"
+    assert received.map_index == 3
+    assert not hasattr(received, "pool_slots")
diff --git a/devel-common/src/tests_common/pytest_plugin.py b/devel-common/src/tests_common/pytest_plugin.py
@@ -2584,16 +2584,12 @@ def _create_task_instance(
         should_retry: bool | None = None,
         max_tries: int | None = None,
     ) -> RuntimeTaskInstance:
-        from tests_common.test_utils.version_compat import AIRFLOW_V_3_3_PLUS
-
-        if AIRFLOW_V_3_3_PLUS:
-            from airflow.sdk.execution_time.workloads.task import TaskInstanceDTO
-        else:
-            from airflow.sdk.api.datamodels._generated import (  # type: ignore[no-redef,assignment]
-                TaskInstance as TaskInstanceDTO,
-            )
-
-        from airflow.sdk.api.datamodels._generated import DagRun, DagRunState, TIRunContext
+        from airflow.sdk.api.datamodels._generated import (
+            DagRun,
+            DagRunState,
+            TaskInstance as TaskInstanceDTO,
+            TIRunContext,
+        )
         from airflow.utils.types import DagRunType
 
         if isinstance(logical_date, str):
@@ -2678,9 +2674,7 @@ def _create_task_instance(
                 try_number=try_number,
                 map_index=map_index,  # type: ignore[arg-type]
                 dag_version_id=uuid7(),
-                pool_slots=1,
                 queue="default",
-                priority_weight=1,
             ),
             dag_rel_path="",
             bundle_info=BundleInfo(name="anything", version="any"),

@@ -29,7 +29,7 @@ import org.apache.airflow.sdk.execution.comm.DagRun
 import org.apache.airflow.sdk.execution.comm.StartupDetails
 import org.apache.airflow.sdk.execution.comm.SucceedTask
 import org.apache.airflow.sdk.execution.comm.TIRunContext
-import org.apache.airflow.sdk.execution.comm.TaskInstanceDTO
+import org.apache.airflow.sdk.execution.comm.TaskInstance
 import org.apache.airflow.sdk.execution.comm.TaskState
 import org.junit.jupiter.api.Assertions
 import org.junit.jupiter.api.DisplayName
@@ -76,7 +76,7 @@ class TaskTest {
   private fun startupDetails(taskId: String): StartupDetails =
     StartupDetails().also {
       it.ti =
-        TaskInstanceDTO().also { o ->
+        TaskInstance().also { o ->
           o.id = UUID.randomUUID()
           o.taskId = taskId
           o.dagId = "test_dag"

@@ -1303,6 +1303,7 @@ def test_try_adopt_task_instances(self, mock_executor):
             task.pool_slots = 1
             task.priority_weight = 1
             task.context_carrier = {}
+            task.hostname = None
             task.queued_dttm = dt.datetime.now()
             # Set up nested attributes for BundleInfo
             task.dag_model = mock.Mock()

@@ -1277,10 +1277,6 @@ components:
           type: string
           format: uuid
           title: Id
-        dag_version_id:
-          type: string
-          format: uuid
-          title: Dag Version Id
         task_id:
           type: string
           title: Task Id
@@ -1293,45 +1289,55 @@ components:
         try_number:
           type: integer
           title: Try Number
+        dag_version_id:
+          type: string
+          format: uuid
+          title: Dag Version Id
         map_index:
           type: integer
           title: Map Index
           default: -1
-        pool_slots:
-          type: integer
-          title: Pool Slots
-        queue:
-          type: string
-          title: Queue
-        priority_weight:
-          type: integer
-          title: Priority Weight
-        parent_context_carrier:
+        hostname:
           anyOf:
-          - additionalProperties: true
-            type: object
+          - type: string
           - type: 'null'
-          title: Parent Context Carrier
+          title: Hostname
         context_carrier:
           anyOf:
           - additionalProperties: true
             type: object
           - type: 'null'
           title: Context Carrier
+        queue:
+          type: string
+          title: Queue
+          default: default
+        pool_slots:
+          type: integer
+          title: Pool Slots
+        priority_weight:
+          type: integer
+          title: Priority Weight
       type: object
       required:
       - id
-      - dag_version_id
       - task_id
       - dag_id
       - run_id
       - try_number
+      - dag_version_id
       - pool_slots
-      - queue
       - priority_weight
       title: TaskInstanceDTO
-      description: TaskInstanceDTO with executor-specific ``external_executor_id``
-        field and ``key`` property.
+      description: 'The versioned execution API ``TaskInstance`` schema plus executor-only
+        fields.
+
+
+        The base class is the single source of truth for the fields a worker needs;
+
+        the fields added here are executor concerns (queueing order and pool
+
+        accounting) the worker never reads.'
     TaskInstanceState:
       type: string
       enum: