Use min(start_date) for the task-reschedules start_date route

ti_run now returns the earliest reschedule start date via min(start_date).
Use the same definition in GET /task-reschedules/{ti_id}/start_date instead
of picking the first row by id, so the two cannot drift apart.

Co-Authored-By: Claude <noreply@anthropic.com>

Author: hkc-8010

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/tests/unit/api_fastapi/execution_api/versions/head/test_task_instances.py

@@ -19,7 +19,7 @@
 
 import re
 from contextlib import nullcontext
-from datetime import datetime
+from datetime import datetime, timedelta
 from types import SimpleNamespace
 from typing import TYPE_CHECKING
 from unittest import mock
@@ -3932,6 +3932,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")