Show current task executions in the UI

A loop pass or a mapped expansion reuses a task id and map index, so the UI
cannot identify an execution by (task_id, map_index): the logs, XCom, state
or HITL page opened for one pass could show another's. Pages and mutations
therefore carry the region id and index, and row selection is keyed by attempt
UUID so two executions that share public coordinates stay distinct even when
they are selected together.

The run-level Execution tab lists live task instances grouped by loop
iteration and mapped expansion, because that is how a user thinks about what
is running. Clearing a selection goes by UUID through a dialog that offers the
later-iteration and whole-expansion choices the clear API takes, so a user can
decide how far a clear reaches. Gantt segments link to the exact region and
try, and a retired try opens only when both are given, since a try number alone
is ambiguous.

Only the current view is covered. Selecting history by clear boundary is
deliberately absent, so no trajectory parameter is read or written.

Author: ashb

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

@@ -60,6 +60,7 @@ class TaskInstanceResponse(OmitsMissingRegion, BaseModel):
     map_index: int
     region_id: RegionId = None
     region_index: RegionIndex = None
+    in_loop: bool
     logical_date: datetime | None
     run_after: datetime
     start_date: datetime | None

airflow-core/src/airflow/api_fastapi/core_api/openapi/_private_ui.yaml

@@ -5250,6 +5250,9 @@ components:
         region_index:
           type: integer
           title: Region Index
+        in_loop:
+          type: boolean
+          title: In Loop
         logical_date:
           anyOf:
           - type: string
@@ -5401,6 +5404,7 @@ components:
       - dag_id
       - dag_run_id
       - map_index
+      - in_loop
       - logical_date
       - run_after
       - start_date

airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml

@@ -17896,6 +17896,9 @@ components:
         region_index:
           type: integer
           title: Region Index
+        in_loop:
+          type: boolean
+          title: In Loop
         logical_date:
           anyOf:
           - type: string
@@ -18047,6 +18050,7 @@ components:
       - dag_id
       - dag_run_id
       - map_index
+      - in_loop
       - logical_date
       - run_after
       - start_date

airflow-core/src/airflow/api_fastapi/core_api/services/public/task_coordinates.py

@@ -53,6 +53,8 @@ def __getattr__(self, name: str) -> Any:
             if self.projected_map_index is not None:
                 return self.projected_map_index
             return self.resolver.public_map_index(self.value)
+        if name == "in_loop":
+            return self.value.region_id != SENTINEL_REGION_ID and self.resolver._is_in_loop(self.value)
         return getattr(self.value, name)
 
 

airflow-core/src/airflow/models/task_coordinates.py

@@ -164,6 +164,7 @@ class TaskCoordinateResolver:
     _dags: dict[UUID, SerializedDAG] = attrs.field(factory=dict, init=False)
     _region_nodes: dict[UUID, str | None] = attrs.field(factory=dict, init=False)
     _regional_tasks: dict[tuple[str, str | None, str], bool] = attrs.field(factory=dict, init=False)
+    _looped_tasks: dict[tuple[UUID, str], bool] = attrs.field(factory=dict, init=False)
 
     @classmethod
     def for_dag(cls, dag: SerializedDAG | None, session: Session) -> TaskCoordinateResolver:
@@ -201,7 +202,16 @@ def get_task(
             raise ValueError(f"Pinned Dag for {dag_id}/{run_id} not found")
         return dag.get_task(task_id)
 
-    def find_task(self, ti: TaskInstance) -> SerializedOperator | None:
+    def _is_in_loop(self, ti: TaskCoordinate) -> bool:
+        if ti.dag_version_id is None:
+            return False
+        key = (ti.dag_version_id, ti.task_id)
+        if key not in self._looped_tasks:
+            task = self.find_task(ti)
+            self._looped_tasks[key] = task is not None and enclosing_loop(task) is not None
+        return self._looped_tasks[key]
+
+    def find_task(self, ti: TaskCoordinate) -> SerializedOperator | None:
         """Return the task of ``ti`` from its pinned Dag, or None when that Dag or the task is gone."""
         if ti.dag_version_id is None or (dag := self.get_dag(ti.dag_version_id)) is None:
             return None

airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts

@@ -8067,6 +8067,10 @@ export const $TaskInstanceResponse = {
             type: 'integer',
             title: 'Region Index'
         },
+        in_loop: {
+            type: 'boolean',
+            title: 'In Loop'
+        },
         logical_date: {
             anyOf: [
                 {
@@ -8351,7 +8355,7 @@ export const $TaskInstanceResponse = {
         }
     },
     type: 'object',
-    required: ['id', 'task_id', 'dag_id', 'dag_run_id', 'map_index', 'logical_date', 'run_after', 'start_date', 'end_date', 'duration', 'state', 'try_number', 'max_tries', 'task_display_name', 'dag_display_name', 'hostname', 'unixname', 'pool', 'pool_slots', 'queue', 'priority_weight', 'operator', 'operator_name', 'queued_when', 'scheduled_when', 'pid', 'executor', 'executor_config', 'note', 'rendered_map_index', 'trigger', 'triggerer_job', 'dag_version'],
+    required: ['id', 'task_id', 'dag_id', 'dag_run_id', 'map_index', 'in_loop', 'logical_date', 'run_after', 'start_date', 'end_date', 'duration', 'state', 'try_number', 'max_tries', 'task_display_name', 'dag_display_name', 'hostname', 'unixname', 'pool', 'pool_slots', 'queue', 'priority_weight', 'operator', 'operator_name', 'queued_when', 'scheduled_when', 'pid', 'executor', 'executor_config', 'note', 'rendered_map_index', 'trigger', 'triggerer_job', 'dag_version'],
     title: 'TaskInstanceResponse',
     description: 'TaskInstance serializer for responses.'
 } as const;

airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts

@@ -2199,6 +2199,7 @@ export type TaskInstanceResponse = {
     map_index: number;
     region_id?: string;
     region_index?: number;
+    in_loop: boolean;
     logical_date: string | null;
     run_after: string;
     start_date: string | null;

airflow-core/src/airflow/ui/public/i18n/locales/en/common.json

@@ -466,6 +466,7 @@
     "executorConfig": "Executor Config",
     "hostname": "Hostname",
     "id": "ID",
+    "iteration": "Iteration",
     "maxTries": "Max Tries",
     "pid": "PID",
     "pool": "Pool",

airflow-core/src/airflow/ui/public/i18n/locales/en/dag.json

@@ -68,6 +68,17 @@
     "upcomingCount_one": "{{count}} Upcoming Deadline",
     "upcomingCount_other": "{{count}} Upcoming Deadlines"
   },
+  "execution": {
+    "clearSelected": "Clear selected executions",
+    "empty": "No executions in this run.",
+    "iteration": "{{node}} · Iteration {{index}}",
+    "mapped_one": "{{task}} ({{count}} mapped task on this page)",
+    "mapped_other": "{{task}} ({{count}} mapped tasks on this page)",
+    "page": "Showing {{start}}–{{end}} of {{total}} executions",
+    "runTasks": "Run tasks",
+    "select": "Select {{task}}",
+    "title": "Execution"
+  },
   "extraLinks": "Extra Links",
   "grid": {
     "buttons": {

airflow-core/src/airflow/ui/public/i18n/locales/en/dags.json

@@ -92,13 +92,15 @@
       "existingTasks": "Clear existing tasks",
       "future": "Future",
       "keepTaskState": "Keep task state and resume",
+      "laterIterations": "Later iterations",
       "onlyFailed": "Clear only failed tasks",
       "past": "Past",
       "preventRunningTasks": "Prevent rerun if task is running",
       "queueNew": "Queue up new tasks",
       "runOnLatestVersion": "Run with latest bundle version",
       "runOnLatestVersionForced": "Always uses the latest — there's no earlier version to go back to",
-      "upstream": "Upstream"
+      "upstream": "Upstream",
+      "wholeExpansion": "Whole expansion"
     }
   },
   "runStateCounts": {