Read current task executions by exact region and try

A loop pass or a mapped expansion reuses a task id and map index, so reading
a task instance, its XCom, logs, links, HITL detail or state by (task_id,
map_index) no longer names one execution. The public readers take region_id and
region_index and resolve them against the live producers. An index without a
region is rejected because an index means nothing without the region it indexes.
Responses return the region coordinates and a map_index derived from the pinned
definition, so mapped tasks keep presenting their slot as before while a task
inside a loop presents -1. Filters and sorts use the same derived value so
paging and cursors agree with what the responses show.

Retired attempts stay in task_instance under their own UUID, so try details,
logs and external log links select one by try_number plus the same coordinates.
The external log URL used to ignore the try number, which became ambiguous once
several rows share a coordinate. Operator extra links read their XCom from the
selected region and pinned definition, since the coordinates alone no longer say
which execution produced it.

The run-level execution listing returns the region ancestry beside each live
task instance because its coordinates cannot be interpreted without it. The
Gantt payload carries the attempt id and region coordinates so a bar identifies
one attempt.

Author: ashb

airflow-core/src/airflow/api_fastapi/common/parameters/misc.py

@@ -39,6 +39,7 @@
 from airflow.models.errors import ParseImportError
 from airflow.models.hitl import HITLDetail
 from airflow.models.pool import Pool
+from airflow.models.task_coordinates import public_map_index_expression
 from airflow.models.taskinstance import TaskInstance
 from airflow.models.variable import Variable
 from airflow.utils.state import TaskInstanceState
@@ -245,7 +246,7 @@ def _optional_boolean(value: bool | None) -> bool | None:
     FilterParam[int | None],
     Depends(
         filter_param_factory(
-            TaskInstance.region_index,
+            public_map_index_expression(TaskInstance),
             int | None,
             filter_name="map_index",
         )

airflow-core/src/airflow/api_fastapi/common/parameters/sort.py

@@ -44,7 +44,7 @@ def __init__(
         self,
         allowed_attrs: list[str],
         model: Base,
-        to_replace: dict[str, str | Column | list[Column]] | None = None,
+        to_replace: dict[str, str | ColumnElement | list[ColumnElement]] | None = None,
     ) -> None:
         super().__init__()
         self.allowed_attrs = allowed_attrs
@@ -78,7 +78,7 @@ def _resolve(self) -> list[tuple[str, ColumnElement, bool]]:
             # it back to the actual row accessor via ``to_replace`` when reading values
             # for cursor encoding.
             attr_name = lstriped_orderby
-            column: Column | None = None
+            column: ColumnElement | None = None
             if self.to_replace:
                 replacement = self.to_replace.get(lstriped_orderby, lstriped_orderby)
                 if isinstance(replacement, str):
@@ -91,6 +91,8 @@ def _resolve(self) -> list[tuple[str, ColumnElement, bool]]:
                     is_desc = order_by_value.startswith("-")
                     for col in replacement:
                         col_attr_name = col.key
+                        if col_attr_name is None:
+                            raise ValueError("Compound sort expressions must have labels")
                         resolved.append((col_attr_name, col, is_desc))
                     continue
                 else:

airflow-core/src/airflow/api_fastapi/common/parameters/task_instance.py

@@ -37,6 +37,7 @@
 )
 from airflow.api_fastapi.compat import HTTP_422_UNPROCESSABLE_CONTENT
 from airflow.models.dag_version import DagVersion
+from airflow.models.task_coordinates import public_map_index_expression
 from airflow.models.taskinstance import TaskInstance
 from airflow.utils.state import TaskInstanceState
 
@@ -237,7 +238,7 @@ def _transform_ti_states(states: list[str] | None) -> list[TaskInstanceState | N
     FilterParam[list[int]],
     Depends(
         filter_param_factory(
-            TaskInstance.region_index,
+            public_map_index_expression(TaskInstance),
             list[int],
             FilterOptionEnum.ANY_EQUAL,
             filter_name="map_index",

airflow-core/src/airflow/api_fastapi/core_api/datamodels/task_instance_history.py

@@ -18,6 +18,7 @@
 
 from datetime import datetime
 from typing import Annotated, cast
+from uuid import UUID
 
 from pydantic import (
     AliasPath,
@@ -27,21 +28,25 @@
 )
 
 from airflow._shared.secrets_masker import redact
+from airflow.api_fastapi.common.region import OmitsMissingRegion, RegionId, RegionIndex
 from airflow.api_fastapi.core_api.base import BaseModel
 from airflow.api_fastapi.core_api.datamodels.dag_versions import DagVersionResponse
 from airflow.utils.state import TaskInstanceState
 
 
-class TaskInstanceHistoryResponse(BaseModel):
+class TaskInstanceHistoryResponse(OmitsMissingRegion, BaseModel):
     """TaskInstanceHistory serializer for responses."""
 
+    id: UUID
     task_id: str
     dag_id: str
 
     # todo: this should not be aliased; it's ambiguous with dag run's "id" - airflow 3.0
     run_id: str = Field(alias="dag_run_id")
 
     map_index: int
+    region_id: RegionId = None
+    region_index: RegionIndex = None
     start_date: datetime | None
     end_date: datetime | None
     duration: float | None
@@ -64,6 +69,7 @@ class TaskInstanceHistoryResponse(BaseModel):
     executor: str | None
     executor_config: Annotated[str, BeforeValidator(str)]
     dag_version: DagVersionResponse | None
+    note: str | None = None
     state_reason: str | None = Field(
         default=None,
         validation_alias="retry_reason",

airflow-core/src/airflow/api_fastapi/core_api/datamodels/ui/execution.py

@@ -0,0 +1,66 @@
+# 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
+
+from datetime import datetime
+from uuid import UUID
+
+from pydantic import Field
+
+from airflow.api_fastapi.common.region import OmitsMissingRegion, RegionId, RegionIndex
+from airflow.api_fastapi.core_api.base import BaseModel
+from airflow.utils.state import TaskInstanceState
+
+
+class ExecutionTaskResponse(OmitsMissingRegion, BaseModel):
+    """A task try together with the coordinates that address it exactly."""
+
+    id: UUID
+    dag_id: str
+    run_id: str = Field(alias="dag_run_id")
+    task_id: str
+    task_display_name: str
+    region_id: RegionId = None
+    region_index: RegionIndex = None
+    map_index: int
+    try_number: int
+    state: TaskInstanceState | None
+    start_date: datetime | None
+    end_date: datetime | None
+    duration: float | None
+    dag_version_id: UUID | None
+    operator: str | None
+    note: str | None = None
+
+
+class ExecutionRegionResponse(BaseModel):
+    """Immutable region structure for interpreting task coordinates."""
+
+    id: UUID
+    node_id: str
+    parent_region_id: UUID | None
+    parent_region_index: int | None
+    forked_from_region_id: UUID | None
+    resumes_from_index: int
+
+
+class ExecutionCollectionResponse(BaseModel):
+    """A page of task executions with their region ancestry."""
+
+    task_instances: list[ExecutionTaskResponse]
+    regions: list[ExecutionRegionResponse]
+    total_entries: int

airflow-core/src/airflow/api_fastapi/core_api/datamodels/ui/gantt.py

@@ -18,14 +18,20 @@
 from __future__ import annotations
 
 from datetime import datetime
+from uuid import UUID
 
+from airflow.api_fastapi.common.region import OmitsMissingRegion, RegionId, RegionIndex
 from airflow.api_fastapi.core_api.base import BaseModel
 from airflow.utils.state import TaskInstanceState
 
 
-class GanttTaskInstance(BaseModel):
+class GanttTaskInstance(OmitsMissingRegion, BaseModel):
     """Task instance data for Gantt chart."""
 
+    id: UUID
+    region_id: RegionId = None
+    region_index: RegionIndex = None
+    map_index: int
     task_id: str
     task_display_name: str
     try_number: int

airflow-core/src/airflow/api_fastapi/core_api/datamodels/xcom.py

@@ -19,20 +19,24 @@
 from collections.abc import Iterable
 from datetime import datetime
 from typing import Any
+from uuid import UUID
 
 from pydantic import AliasPath, Field, field_validator
 
+from airflow.api_fastapi.common.region import OmitsMissingRegion, RegionId, RegionIndex
 from airflow.api_fastapi.core_api.base import BaseModel, StrictBaseModel
 from airflow.api_fastapi.core_api.datamodels.common import find_reserved_keys
 
 
-class XComResponse(BaseModel):
+class XComResponse(OmitsMissingRegion, BaseModel):
     """Serializer for a xcom item."""
 
     key: str
     timestamp: datetime
     logical_date: datetime | None
     map_index: int
+    region_id: RegionId = None
+    region_index: RegionIndex = None
     task_id: str
     dag_id: str
     run_id: str
@@ -105,6 +109,8 @@ class XComCreateBody(StrictBaseModel):
     key: str
     value: Any
     map_index: int = -1
+    region_id: UUID | None = None
+    region_index: int | None = Field(default=None, ge=-1)
 
     @field_validator("value")
     @classmethod
@@ -117,6 +123,8 @@ class XComUpdateBody(StrictBaseModel):
 
     value: Any
     map_index: int = -1
+    region_id: UUID | None = None
+    region_index: int | None = Field(default=None, ge=-1)
 
     @field_validator("value")
     @classmethod