Avoid a supervisor round trip before poke for rescheduled sensors

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

@@ -448,6 +448,9 @@ class TIRunContext(BaseModel):
     task_reschedule_count: int = 0
     """How many times the task has been rescheduled."""
 
+    first_task_reschedule_start_date: UtcDateTime | None = None
+    """The first reschedule start date for the task instance, if it has been rescheduled."""
+
     max_tries: int
     """Maximum number of tries for the task instance (from DB)."""
 

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

@@ -321,12 +321,12 @@ def ti_run(
             read = build_xcom_read_query(producer_ids=select(TI.id).where(TI.id == task_instance_id))
             entity = xcom_entity(read)
             xcom_keys = list(session.scalars(read.with_only_columns(entity.key)))
-        task_reschedule_count = (
-            session.scalar(
-                select(func.count(TaskReschedule.id)).where(TaskReschedule.ti_id == task_instance_id)
-            )
-            or 0
-        )
+        task_reschedule_count, first_task_reschedule_start_date = session.execute(
+            select(
+                func.count(TaskReschedule.id),
+                func.min(TaskReschedule.start_date),
+            ).where(TaskReschedule.ti_id == task_instance_id)
+        ).one()
 
         context = TIRunContext(
             dag_run=dr,
@@ -352,6 +352,8 @@ def ti_run(
                 terminal_task_id=group.terminal_task_id,
                 terminal_is_mapped=terminal.get_needs_expansion(),
             )
+        if first_task_reschedule_start_date is not None:
+            context.first_task_reschedule_start_date = first_task_reschedule_start_date
 
         # Only set for lang-SDK (foreign-runtime) tasks with a captured TaskFlow arg
         # spec; the route excludes unset fields, keeping regular responses lean.

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

@@ -20,7 +20,7 @@
 from uuid import UUID
 
 from fastapi import APIRouter, Security, status
-from sqlalchemy import select
+from sqlalchemy import func, select
 
 from airflow.api_fastapi.common.db.common import AsyncSessionDep
 from airflow.api_fastapi.common.types import UtcDateTime
@@ -42,11 +42,6 @@
 @router.get("/{task_instance_id}/start_date")
 async def get_start_date(task_instance_id: UUID, session: AsyncSessionDep) -> UtcDateTime | None:
     """Get the first reschedule date if found, None if no records exist."""
-    start_date = await session.scalar(
-        select(TaskReschedule.start_date)
-        .where(TaskReschedule.ti_id == task_instance_id)
-        .order_by(TaskReschedule.id.asc())
-        .limit(1)
+    return await session.scalar(
+        select(func.min(TaskReschedule.start_date)).where(TaskReschedule.ti_id == task_instance_id)
     )
-
-    return start_date

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

@@ -56,6 +56,7 @@
     AddArgBindingsToTIRunContext,
     AddCallbackRunEndpoint,
     AddDagRunNoteUpdateEndpoint,
+    AddFirstTaskRescheduleStartDateField,
     AddLoopContext,
     AddMultiTeamToTIRunContext,
     AddStoppedTaskReport,
@@ -81,6 +82,7 @@
         AddStoppedTaskReport,
         AddTaskInstanceRegionCoordinates,
         IdentifyArchivedTaskStateUpdates,
+        AddFirstTaskRescheduleStartDateField,
     ),
     Version(
         "2026-06-30",

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

@@ -133,3 +133,13 @@ class AddMultiTeamToTIRunContext(VersionChange):
     def remove_multi_team_field(response: ResponseInfo) -> None:  # type: ignore[misc]
         """Strip ``multi_team`` from the run context for older clients."""
         response.body.pop("multi_team", None)
+
+
+class AddFirstTaskRescheduleStartDateField(VersionChange):
+    """Add the ``first_task_reschedule_start_date`` field to TIRunContext."""
+
+    description = __doc__
+
+    instructions_to_migrate_to_previous_version = (
+        schema(TIRunContext).field("first_task_reschedule_start_date").didnt_exist,
+    )

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

@@ -21,7 +21,7 @@
 import time
 from concurrent.futures import ThreadPoolExecutor
 from contextlib import nullcontext
-from datetime import datetime
+from datetime import datetime, timedelta
 from threading import Barrier, Event, local
 from types import SimpleNamespace
 from typing import TYPE_CHECKING
@@ -877,6 +877,60 @@ def test_ti_run_state_to_running(
         )
         assert response.status_code == 409
 
+    def test_ti_run_state_includes_first_task_reschedule_start_date(
+        self,
+        client,
+        session,
+        create_task_instance,
+    ):
+        """Test that running a rescheduled Task Instance includes its first reschedule start date."""
+        instant_str = "2024-09-30T12:00:00Z"
+        instant = timezone.parse(instant_str)
+        first_reschedule_start_date = timezone.datetime(2024, 9, 30, 10)
+        second_reschedule_start_date = timezone.datetime(2024, 9, 30, 11)
+
+        ti = create_task_instance(
+            task_id="test_ti_run_state_includes_first_task_reschedule_start_date",
+            state=State.QUEUED,
+            dagrun_state=DagRunState.RUNNING,
+            session=session,
+            start_date=instant,
+            dag_id=str(uuid4()),
+        )
+        session.add_all(
+            [
+                TaskReschedule(
+                    ti_id=ti.id,
+                    start_date=second_reschedule_start_date,
+                    end_date=timezone.datetime(2024, 9, 30, 11, 1),
+                    reschedule_date=timezone.datetime(2024, 9, 30, 11, 2),
+                ),
+                TaskReschedule(
+                    ti_id=ti.id,
+                    start_date=first_reschedule_start_date,
+                    end_date=timezone.datetime(2024, 9, 30, 10, 1),
+                    reschedule_date=timezone.datetime(2024, 9, 30, 10, 2),
+                ),
+            ]
+        )
+        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": instant_str,
+            },
+        )
+
+        assert response.status_code == 200
+        result = response.json()
+        assert result["task_reschedule_count"] == 2
+        assert result["first_task_reschedule_start_date"] == "2024-09-30T10:00:00Z"
+
     def test_ti_run_returns_execution_token(
         self, client, exec_app, session, create_task_instance, time_machine
     ):
@@ -5254,6 +5308,30 @@ def test_get_start_date(self, client, session, create_task_instance):
         assert response.status_code == 200
         assert response.json() == "2024-01-01T00:00:00Z"
 
+    def test_get_start_date_returns_earliest_not_first_inserted(self, client, session, create_task_instance):
+        ti = create_task_instance(
+            task_id="test_get_start_date_returns_earliest_not_first_inserted",
+            state=State.RUNNING,
+            start_date=timezone.datetime(2024, 1, 1),
+            session=session,
+        )
+        # Insert the later reschedule first so that ordering by id and by start_date disagree
+        for start_date in (timezone.datetime(2024, 1, 2), timezone.datetime(2024, 1, 1)):
+            session.add(
+                TaskReschedule(
+                    ti_id=ti.id,
+                    start_date=start_date,
+                    end_date=start_date + timedelta(hours=1),
+                    reschedule_date=start_date + timedelta(hours=2),
+                )
+            )
+            session.flush()
+        session.commit()
+
+        response = client.get(f"/execution/task-reschedules/{ti.id}/start_date")
+        assert response.status_code == 200
+        assert response.json() == "2024-01-01T00:00:00Z"
+
     def test_get_start_date_not_found(self, client):
         ti_id = "0182e924-0f1e-77e6-ab50-e977118bc139"
         response = client.get(f"/execution/task-reschedules/{ti_id}/start_date")

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

@@ -17,6 +17,8 @@
 
 from __future__ import annotations
 
+from uuid import uuid4
+
 import pytest
 from sqlalchemy import select
 
@@ -28,7 +30,7 @@
 from airflow.models.trigger import Trigger
 from airflow.models.xcom import XComModelV2
 from airflow.sdk import task
-from airflow.utils.state import State
+from airflow.utils.state import DagRunState, State
 
 from tests_common.test_utils.db import clear_db_runs
 
@@ -224,7 +226,7 @@ def test_legacy_worker_heartbeat_rejects_restarting_task(client, session, create
 
 @pytest.fixture
 def old_ver_client(client):
-    """Execution API version immediately before ``arg_bindings`` was added."""
+    """Execution API version immediately before the ``2026-10-30`` fields were added."""
     client.headers["Airflow-API-Version"] = "2026-06-30"
     return client
 
@@ -329,3 +331,42 @@ def test_mutations_after_retry_response_by_version(
     stored = XComModelV2.get_for_attempt(ti.id, "extra_link", session=session)
     assert (stored is not None) is xcom_kept
     assert XComModelV2.get_for_attempt(successor.id, "extra_link", session=session) is None
+
+
+class TestFirstTaskRescheduleStartDateBackwardCompat:
+    def setup_method(self):
+        clear_db_runs()
+
+    def teardown_method(self):
+        clear_db_runs()
+
+    def test_old_version_strips_first_task_reschedule_start_date(
+        self,
+        old_ver_client,
+        session,
+        create_task_instance,
+    ):
+        ti = create_task_instance(
+            task_id="test_first_task_reschedule_start_date_compat",
+            state=State.QUEUED,
+            dagrun_state=DagRunState.RUNNING,
+            session=session,
+            start_date=timezone.datetime(2024, 9, 30, 12),
+            dag_id=str(uuid4()),
+        )
+        session.add(
+            TaskReschedule(
+                ti_id=ti.id,
+                start_date=timezone.datetime(2024, 9, 30, 10),
+                end_date=timezone.datetime(2024, 9, 30, 10, 1),
+                reschedule_date=timezone.datetime(2024, 9, 30, 10, 2),
+            )
+        )
+        session.commit()
+
+        response = old_ver_client.patch(f"/execution/task-instances/{ti.id}/run", json=RUN_PATCH_BODY)
+
+        assert response.status_code == 200
+        result = response.json()
+        assert result["task_reschedule_count"] == 1
+        assert "first_task_reschedule_start_date" not in result