fix/issue-66373: Adding back queued_dttm to TIRunContext model

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

@@ -429,6 +429,9 @@ class TIRunContext(BaseModel):
     always reflects when the task *first* started, not when it was rescheduled/resumed.
     """
 
+    queued_dttm: UtcDateTime | None = None
+    """When the task was queued. Used by listeners to measure queue wait time."""
+
 
 class PrevSuccessfulDagRunResponse(BaseModel):
     """Schema for response with previous successful DagRun information for Task Template Context."""

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

@@ -158,6 +158,7 @@ def ti_run(
             TI.try_number,
             TI.max_tries,
             TI.start_date,
+            TI.queued_dttm,
             TI.next_method,
             TI.hostname,
             TI.unixname,
@@ -314,6 +315,8 @@ def ti_run(
             context.next_method = ti.next_method
             context.next_kwargs = ti.next_kwargs
             context.start_date = ti.start_date
+        if ti.queued_dttm:
+            context.queued_dttm = ti.queued_dttm
     except SQLAlchemyError:
         log.exception("Error marking Task Instance state as running")
         raise HTTPException(

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

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

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

@@ -17,9 +17,12 @@
 
 from __future__ import annotations
 
-from cadwyn import VersionChange, endpoint, schema
+from cadwyn import ResponseInfo, VersionChange, convert_response_to_previous_version_for, endpoint, schema
 
-from airflow.api_fastapi.execution_api.datamodels.taskinstance import TIAwaitingInputStatePayload
+from airflow.api_fastapi.execution_api.datamodels.taskinstance import (
+    TIAwaitingInputStatePayload,
+    TIRunContext,
+)
 
 
 class AddVariableKeysEndpoint(VersionChange):
@@ -53,3 +56,16 @@ class AddAwaitingInputStatePayload(VersionChange):
         schema(TIAwaitingInputStatePayload).field("next_kwargs").didnt_exist,
         schema(TIAwaitingInputStatePayload).field("rendered_map_index").didnt_exist,
     )
+
+
+class AddQueuedDttmField(VersionChange):
+    """Add ``queued_dttm`` field to TIRunContext."""
+
+    description = __doc__
+
+    instructions_to_migrate_to_previous_version = (schema(TIRunContext).field("queued_dttm").didnt_exist,)
+
+    @convert_response_to_previous_version_for(TIRunContext)  # type: ignore[arg-type]
+    def remove_queued_dttm_field(response: ResponseInfo) -> None:  # type: ignore[misc]
+        """Remove queued_dttm field for older API versions."""
+        response.body.pop("queued_dttm", None)

airflow-core/tests/unit/api_fastapi/execution_api/versions/v2026_06_30/test_task_instances.py

@@ -0,0 +1,106 @@
+# 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.
+
+from __future__ import annotations
+
+import pytest
+
+from airflow._shared.timezones import timezone
+from airflow.utils.state import DagRunState, State
+
+from tests_common.test_utils.db import clear_db_runs
+
+pytestmark = pytest.mark.db_test
+
+TIMESTAMP_STR = "2026-01-01T00:00:00Z"
+TIMESTAMP = timezone.parse(TIMESTAMP_STR)
+
+RUN_PATCH_BODY = {
+    "state": "running",
+    "hostname": "h",
+    "unixname": "u",
+    "pid": 1,
+    "start_date": TIMESTAMP_STR,
+}
+
+
+@pytest.fixture
+def old_ver_client(client):
+    """Execution API version immediately before ``queued_dttm`` was added."""
+    client.headers["Airflow-API-Version"] = "2026-06-16"
+    return client
+
+
+class TestQueuedDttmFieldBackwardCompat:
+    @pytest.fixture(autouse=True)
+    def _freeze_time(self, time_machine):
+        time_machine.move_to(TIMESTAMP_STR, tick=False)
+
+    def setup_method(self):
+        clear_db_runs()
+
+    def teardown_method(self):
+        clear_db_runs()
+
+    def test_old_version_strips_queued_dttm(self, old_ver_client, session, create_task_instance):
+        ti = create_task_instance(
+            task_id="test_queued_dttm_downgrade",
+            state=State.QUEUED,
+            dagrun_state=DagRunState.RUNNING,
+            session=session,
+            start_date=TIMESTAMP,
+        )
+        ti.queued_dttm = TIMESTAMP
+        session.commit()
+
+        response = old_ver_client.patch(f"/execution/task-instances/{ti.id}/run", json=RUN_PATCH_BODY)
+
+        assert response.status_code == 200
+        assert "queued_dttm" not in response.json()
+
+    def test_head_version_includes_queued_dttm_when_set(self, client, session, create_task_instance):
+        queued_at = timezone.parse("2026-01-01T00:05:00Z")
+        ti = create_task_instance(
+            task_id="test_queued_dttm_head",
+            state=State.QUEUED,
+            dagrun_state=DagRunState.RUNNING,
+            session=session,
+            start_date=TIMESTAMP,
+        )
+        ti.queued_dttm = queued_at
+        session.commit()
+
+        response = client.patch(f"/execution/task-instances/{ti.id}/run", json=RUN_PATCH_BODY)
+
+        assert response.status_code == 200
+        assert response.json()["queued_dttm"] == "2026-01-01T00:05:00Z"
+
+    def test_head_version_omits_queued_dttm_when_not_set(self, client, session, create_task_instance):
+        ti = create_task_instance(
+            task_id="test_queued_dttm_head_null",
+            state=State.QUEUED,
+            dagrun_state=DagRunState.RUNNING,
+            session=session,
+            start_date=TIMESTAMP,
+        )
+        ti.queued_dttm = None
+        session.commit()
+
+        response = client.patch(f"/execution/task-instances/{ti.id}/run", json=RUN_PATCH_BODY)
+
+        assert response.status_code == 200
+        assert "queued_dttm" not in response.json()

providers/openlineage/tests/unit/openlineage/utils/test_utils.py

@@ -92,6 +92,7 @@
     AIRFLOW_V_3_0_3_PLUS,
     AIRFLOW_V_3_0_PLUS,
     AIRFLOW_V_3_2_PLUS,
+    AIRFLOW_V_3_3_PLUS,
 )
 
 BASH_OPERATOR_PATH = "airflow.providers.standard.operators.bash"
@@ -3065,7 +3066,7 @@ def test_taskinstance_info_af3():
     bundle_instance.name = "bundle_name"
     runtime_ti.bundle_instance = bundle_instance
 
-    assert dict(TaskInstanceInfo(runtime_ti)) == {
+    expected: dict = {
         "log_url": runtime_ti.log_url,
         "map_index": 2,
         "rendered_map_index": None,
@@ -3074,6 +3075,12 @@ def test_taskinstance_info_af3():
         "dag_bundle_name": "bundle_name",
     }
 
+    if AIRFLOW_V_3_3_PLUS:
+        # queued_dttm was added to RuntimeTaskInstance in 3.3.0
+        expected["queued_dttm"] = None
+
+    assert dict(TaskInstanceInfo(runtime_ti)) == expected
+
     runtime_ti.rendered_map_index = "country=PL"
     assert dict(TaskInstanceInfo(runtime_ti))["rendered_map_index"] == "country=PL"
 

task-sdk/src/airflow/sdk/api/datamodels/_generated.py

@@ -793,3 +793,4 @@ class TIRunContext(BaseModel):
     xcom_keys_to_clear: Annotated[list[str] | None, Field(title="Xcom Keys To Clear")] = None
     should_retry: Annotated[bool | None, Field(title="Should Retry")] = False
     start_date: Annotated[AwareDatetime | None, Field(title="Start Date")] = None
+    queued_dttm: Annotated[AwareDatetime | None, Field(title="Queued Dttm")] = None

task-sdk/src/airflow/sdk/execution_time/schema/schema.json

@@ -4921,6 +4921,19 @@
           ],
           "default": null,
           "title": "Start Date"
+        },
+        "queued_dttm": {
+          "anyOf": [
+            {
+              "format": "date-time",
+              "type": "string"
+            },
+            {
+              "type": "null"
+            }
+          ],
+          "default": null,
+          "title": "Queued Dttm"
         }
       },
       "required": [

task-sdk/src/airflow/sdk/execution_time/task_runner.py

@@ -240,6 +240,9 @@ class RuntimeTaskInstance(TaskInstance):
     start_date: AwareDatetime
     """Start date of the task instance."""
 
+    queued_dttm: AwareDatetime | None = None
+    """When the task was queued. Used to measure queue wait time."""
+
     end_date: AwareDatetime | None = None
 
     state: TaskInstanceState | None = None
@@ -969,6 +972,7 @@ def parse(what: StartupDetails, log: Logger) -> RuntimeTaskInstance:
         _ti_context_from_server=what.ti_context,
         max_tries=what.ti_context.max_tries,
         start_date=what.start_date,
+        queued_dttm=what.ti_context.queued_dttm,
         state=TaskInstanceState.RUNNING,
         sentry_integration=what.sentry_integration,
     )

task-sdk/src/airflow/sdk/types.py

@@ -140,6 +140,7 @@ class RuntimeTaskInstanceProtocol(Protocol):
     max_tries: int
     hostname: str | None = None
     start_date: AwareDatetime
+    queued_dttm: AwareDatetime | None = None
     end_date: AwareDatetime | None = None
     state: TaskInstanceState | None = None
     is_mapped: bool | None = None

task-sdk/tests/conftest.py

@@ -233,6 +233,7 @@ def __call__(
         should_retry: bool = ...,
         max_tries: int = ...,
         consumed_asset_events: Sequence[AssetEventDagRunReference] = ...,
+        queued_dttm: datetime | None = ...,
     ) -> TIRunContext: ...
 
 
@@ -275,6 +276,7 @@ def _make_context(
         should_retry: bool = False,
         max_tries: int = 0,
         consumed_asset_events: Sequence[AssetEventDagRunReference] = (),
+        queued_dttm: datetime | None = None,
     ) -> TIRunContext:
         return TIRunContext(
             dag_run=DagRun(
@@ -294,6 +296,7 @@ def _make_context(
             task_reschedule_count=task_reschedule_count,
             max_tries=max_tries,
             should_retry=should_retry,
+            queued_dttm=queued_dttm,
         )
 
     return _make_context

task-sdk/tests/task_sdk/execution_time/test_task_runner.py

@@ -246,6 +246,60 @@ def test_parse(test_dags_dir: Path, make_ti_context):
     assert ti.task.dag
 
 
+@mock.patch("airflow.sdk.execution_time.task_runner.DagBundlesManager")
+@mock.patch("airflow.dag_processing.dagbag.BundleDagBag")
+@pytest.mark.parametrize(
+    "queued_dttm",
+    [
+        pytest.param(timezone.datetime(2026, 1, 1, 0, 0), id="with_queued_dttm"),
+        pytest.param(None, id="without_queued_dttm"),
+    ],
+)
+def test_parse_propagates_queued_dttm(mock_dagbag, mock_bundle_manager, make_ti_context, queued_dttm):
+    """queued_dttm from TIRunContext must land on RuntimeTaskInstance."""
+    # Mock the bundle
+    mock_bundle = mock.Mock()
+    mock_bundle.path = Path("/tmp")
+    mock_bundle_manager.return_value.get_bundle.return_value = mock_bundle
+
+    # Mock the task to "assign" to the DAG
+    mock_task = mock.Mock(spec=BaseOperator)
+    mock_task.deserialization_allowed_class_fields = ()
+
+    # Mock the DAG
+    mock_dag = mock.Mock(spec=DAG)
+    mock_dag.task_dict = {"a": mock_task}
+    mock_dag.tasks = [mock_task]
+
+    # Mock the DagBag
+    mock_bag_instance = mock.Mock()
+    mock_bag_instance.dags = {"super_basic": mock_dag}
+    mock_dagbag.return_value = mock_bag_instance
+
+    what = StartupDetails(
+        ti=TaskInstanceDTO(
+            id=uuid7(),
+            task_id="a",
+            dag_id="super_basic",
+            run_id="c",
+            try_number=1,
+            dag_version_id=uuid7(),
+            pool_slots=1,
+            queue="default",
+            priority_weight=1,
+        ),
+        dag_rel_path="super_basic.py",
+        bundle_info=BundleInfo(name="my-bundle", version=None),
+        ti_context=make_ti_context(queued_dttm=queued_dttm),
+        start_date=timezone.utcnow(),
+        sentry_integration="",
+    )
+
+    ti = parse(what, mock.Mock())
+
+    assert ti.queued_dttm == queued_dttm
+
+
 @mock.patch("airflow.dag_processing.dagbag.BundleDagBag")
 def test_parse_dag_bag(mock_dagbag, test_dags_dir: Path, make_ti_context):
     """Test that checks that the BundleDagBag is constructed as expected during parsing"""