bbovenzi · Oct 6, 2026 · Jul 7, 2026 · Aug 24, 2026 · Sep 3, 2026 · Sep 17, 2026
@@ -57,6 +57,26 @@ class DAGRunStateCountsResponse(BaseModel):
     state_counts: dict[DagRunState, int]
 
 
+class DAGRecentTaskInstanceStateCountsResponse(BaseModel):
+    """
+    Task-instance state counts for a Dag's recent runs.
+
+    The counts cover every running Dag run, or the latest run when none is running;
+    ``run_ids`` lists those runs. ``state_counts`` only carries states present in them;
+    task instances without a state yet are keyed as ``no_status``.
+    """
+
+    dag_id: str
+    run_ids: list[str]
+    state_counts: dict[str, int]
+
+
+class DAGsRecentTaskInstanceStateCountsCollectionResponse(BaseModel):
+    """Collection of per-Dag recent task-instance state counts for the Dag list page."""
+
+    dags: list[DAGRecentTaskInstanceStateCountsResponse]
+
+
 class DAGsRunStateCountsCollectionResponse(BaseModel):
     """Collection of per-Dag DagRun-state counts for the Dag list page."""
 

@@ -994,6 +994,58 @@ paths:
             application/json:
               schema:
                 $ref: '#/components/schemas/HTTPValidationError'
+  /ui/dags/recent_task_instance_state_counts:
+    get:
+      tags:
+      - DAG
+      summary: Get Recent Task Instance State Counts
+      description: 'Return recent task-instance state counts for the Dags of the given
+        runs, for the Dag list page.
+
+
+        Like the Airflow 2 "Recent Tasks" column, a Dag''s counts cover all of its
+        running Dag
+
+        runs, or its latest run when none is running.
+
+
+        The Dag list response already carries the latest run of each Dag, so the caller
+        passes
+
+        those run ids straight in. Deriving the latest run again here would mean an
+
+        ``ORDER BY run_after DESC LIMIT 1`` per Dag, which has no supporting index
+        and degrades
+
+        badly once a Dag has many runs. Runs the caller may not read are dropped.'
+      operationId: get_recent_task_instance_state_counts_ui
+      security:
+      - OAuth2PasswordBearer: []
+      - HTTPBearer: []
+      parameters:
+      - name: dag_run_ids
+        in: query
+        required: true
+        schema:
+          type: array
+          items:
+            type: integer
+          minItems: 1
+          maxItems: 100
+          title: Dag Run Ids
+      responses:
+        '200':
+          description: Successful Response
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/DAGsRecentTaskInstanceStateCountsCollectionResponse'
+        '422':
+          description: Validation Error
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/HTTPValidationError'
   /ui/dependencies:
     get:
       tags:
@@ -3098,6 +3150,36 @@ components:
         that
 
         the API server/Web UI can use this data to render connection form UI.'
+    DAGRecentTaskInstanceStateCountsResponse:
+      properties:
+        dag_id:
+          type: string
+          title: Dag Id
+        run_ids:
+          items:
+            type: string
+          type: array
+          title: Run Ids
+        state_counts:
+          additionalProperties:
+            type: integer
+          type: object
+          title: State Counts
+      type: object
+      required:
+      - dag_id
+      - run_ids
+      - state_counts
+      title: DAGRecentTaskInstanceStateCountsResponse
+      description: 'Task-instance state counts for a Dag''s recent runs.
+
+
+        The counts cover every running Dag run, or the latest run when none is running;
+
+        ``run_ids`` lists those runs. ``state_counts`` only carries states present
+        in them;
+
+        task instances without a state yet are keyed as ``no_status``.'
     DAGRunLightResponse:
       properties:
         id:
@@ -3421,6 +3503,19 @@ components:
       - file_token
       title: DAGWithLatestDagRunsResponse
       description: DAG with latest dag runs response serializer.
+    DAGsRecentTaskInstanceStateCountsCollectionResponse:
+      properties:
+        dags:
+          items:
+            $ref: '#/components/schemas/DAGRecentTaskInstanceStateCountsResponse'
+          type: array
+          title: Dags
+      type: object
+      required:
+      - dags
+      title: DAGsRecentTaskInstanceStateCountsCollectionResponse
+      description: Collection of per-Dag recent task-instance state counts for the
+        Dag list page.
     DAGsRunStateCountsCollectionResponse:
       properties:
         dags:

@@ -22,7 +22,7 @@
 from typing import Annotated
 
 from fastapi import Depends, HTTPException, Query, status
-from sqlalchemy import false, func, literal, select, union_all
+from sqlalchemy import and_, false, func, literal, select, union_all
 from sqlalchemy.orm import defaultload
 
 from airflow.api_fastapi.auth.managers.models.resource_details import DagAccessEntity
@@ -66,7 +66,9 @@
 from airflow.api_fastapi.core_api.datamodels.ui.dags import (
     DagFolderCollectionResponse,
     DagFolderResponse,
+    DAGRecentTaskInstanceStateCountsResponse,
     DAGRunStateCountsResponse,
+    DAGsRecentTaskInstanceStateCountsCollectionResponse,
     DAGsRunStateCountsCollectionResponse,
     DagTimetableTypeCollectionResponse,
     DAGWithLatestDagRunsCollectionResponse,
@@ -481,3 +483,107 @@ def get_dag_run_state_counts(
         ],
         state_count_limit=STATE_COUNT_CAP,
     )
+
+
+@dags_router.get(
+    "/recent_task_instance_state_counts",
+    dependencies=[
+        Depends(requires_access_dag(method="GET")),
+        Depends(requires_access_dag(method="GET", access_entity=DagAccessEntity.TASK_INSTANCE)),
+    ],
+    operation_id="get_recent_task_instance_state_counts_ui",
+)
+def get_recent_task_instance_state_counts(
+    session: SessionDep,
+    readable_dags_filter: ReadableDagsFilterDep,
+    dag_run_ids: Annotated[
+        list[int], Query(min_length=1, max_length=conf.getint("api", "maximum_page_limit"))
+    ],
+) -> DAGsRecentTaskInstanceStateCountsCollectionResponse:
+    """
+    Return recent task-instance state counts for the Dags of the given runs, for the Dag list page.
+
+    Like the Airflow 2 "Recent Tasks" column, a Dag's counts cover all of its running Dag
+    runs, or its latest run when none is running.
+
+    The Dag list response already carries the latest run of each Dag, so the caller passes
+    those run ids straight in. Deriving the latest run again here would mean an
+    ``ORDER BY run_after DESC LIMIT 1`` per Dag, which has no supporting index and degrades
+    badly once a Dag has many runs. Runs the caller may not read are dropped.
+    """
+    permitted_dag_ids = readable_dags_filter.value or set()
+
+    dags: list[DAGRecentTaskInstanceStateCountsResponse] = []
+    if not permitted_dag_ids:
+        return DAGsRecentTaskInstanceStateCountsCollectionResponse(dags=dags)
+
+    # Ascending run_after: if two runs of one Dag are passed, the newer one wins below.
+    requested_runs = session.execute(
+        select(DagRun.id, DagRun.dag_id, DagRun.run_id)
+        .where(DagRun.id.in_(set(dag_run_ids)), DagRun.dag_id.in_(permitted_dag_ids))
+        .order_by(DagRun.run_after)
+    ).all()
+    latest_run_id_by_dag: dict[str, str] = {row.dag_id: row.run_id for row in requested_runs}
+    latest_run_pk_by_dag: dict[str, int] = {row.dag_id: row.id for row in requested_runs}
+
+    if latest_run_id_by_dag:
+        # One statement picks the counted runs and counts their task instances: every running
+        # run of the requested Dags (found through the partial idx_dag_run_running_dags index),
+        # plus the requested latest run of each Dag with none running. Counting one UNION ALL
+        # branch per run instead grows with the running runs and stops scaling at a few dozen
+        # per Dag. The task instances counted are bounded by the Dag's task structure times its
+        # running runs, so the per-state counts are exact (no cap needed here, unlike the
+        # cross-run counts in get_dag_run_state_counts).
+        running_runs = (
+            select(DagRun.dag_id, DagRun.run_id, DagRun.run_after)
+            .where(DagRun.state == DagRunState.RUNNING, DagRun.dag_id.in_(latest_run_id_by_dag))
+            .cte("running_runs")
+        )
+        counted_runs = union_all(
+            select(running_runs.c.dag_id, running_runs.c.run_id, running_runs.c.run_after),
+            select(DagRun.dag_id, DagRun.run_id, DagRun.run_after).where(
+                DagRun.id.in_(latest_run_pk_by_dag.values()),
+                DagRun.dag_id.not_in(select(running_runs.c.dag_id)),
+            ),
+        ).subquery("counted_runs")
+        # Outer join so a counted run without task instances still lists its run id.
+        counts = session.execute(
+            select(
+                counted_runs.c.dag_id,
+                counted_runs.c.run_id,
+                TaskInstance.state,
+                func.count(TaskInstance.id).label("cnt"),
+            )
+            .select_from(counted_runs)
+            .outerjoin(
+                TaskInstance,
+                and_(
+                    TaskInstance.dag_id == counted_runs.c.dag_id,
+                    TaskInstance.run_id == counted_runs.c.run_id,
+                ),
+            )
+            .group_by(
+                counted_runs.c.dag_id, counted_runs.c.run_after, counted_runs.c.run_id, TaskInstance.state
+            )
+            .order_by(counted_runs.c.run_after, counted_runs.c.run_id)
+        )
+        counts_by_dag: dict[str, dict[str, int]] = {dag_id: {} for dag_id in latest_run_id_by_dag}
+        run_ids_by_dag: dict[str, list[str]] = {}
+        for row in counts:
+            run_ids = run_ids_by_dag.setdefault(row.dag_id, [])
+            if not run_ids or run_ids[-1] != row.run_id:
+                run_ids.append(row.run_id)
+            if row.cnt:
+                state_key = row.state if row.state is not None else "no_status"
+                counts_by_dag[row.dag_id][state_key] = counts_by_dag[row.dag_id].get(state_key, 0) + row.cnt
+
+        dags = [
+            DAGRecentTaskInstanceStateCountsResponse(
+                dag_id=dag_id,
+                run_ids=run_ids,
+                state_counts=counts_by_dag[dag_id],
+            )
+            for dag_id, run_ids in sorted(run_ids_by_dag.items())
+        ]
+
+    return DAGsRecentTaskInstanceStateCountsCollectionResponse(dags=dags)
@@ -473,6 +473,12 @@ export const useDagServiceGetDagRunStateCountsUiKey = "DagServiceGetDagRunStateC
 export const UseDagServiceGetDagRunStateCountsUiKeyFn = ({ dagIds }: {
   dagIds: string[];
 }, queryKey?: Array<unknown>) => [useDagServiceGetDagRunStateCountsUiKey, ...(queryKey ?? [{ dagIds }])];
+export type DagServiceGetRecentTaskInstanceStateCountsUiDefaultResponse = Awaited<ReturnType<typeof DagService.getRecentTaskInstanceStateCountsUi>>;
+export type DagServiceGetRecentTaskInstanceStateCountsUiQueryResult<TData = DagServiceGetRecentTaskInstanceStateCountsUiDefaultResponse, TError = unknown> = UseQueryResult<TData, TError>;
+export const useDagServiceGetRecentTaskInstanceStateCountsUiKey = "DagServiceGetRecentTaskInstanceStateCountsUi";
+export const UseDagServiceGetRecentTaskInstanceStateCountsUiKeyFn = ({ dagRunIds }: {
+  dagRunIds: number[];
+}, queryKey?: Array<unknown>) => [useDagServiceGetRecentTaskInstanceStateCountsUiKey, ...(queryKey ?? [{ dagRunIds }])];
 export type EventLogServiceGetEventLogDefaultResponse = Awaited<ReturnType<typeof EventLogService.getEventLog>>;
 export type EventLogServiceGetEventLogQueryResult<TData = EventLogServiceGetEventLogDefaultResponse, TError = unknown> = UseQueryResult<TData, TError>;
 export const useEventLogServiceGetEventLogKey = "EventLogServiceGetEventLog";

@@ -944,6 +944,25 @@ export const ensureUseDagServiceGetDagRunStateCountsUiData = (queryClient: Query
   dagIds: string[];
 }) => queryClient.ensureQueryData({ queryKey: Common.UseDagServiceGetDagRunStateCountsUiKeyFn({ dagIds }), queryFn: () => DagService.getDagRunStateCountsUi({ dagIds }) });
 /**
+* Get Recent Task Instance State Counts
+* Return recent task-instance state counts for the Dags of the given runs, for the Dag list page.
+*
+* Like the Airflow 2 "Recent Tasks" column, a Dag's counts cover all of its running Dag
+* runs, or its latest run when none is running.
+*
+* The Dag list response already carries the latest run of each Dag, so the caller passes
+* those run ids straight in. Deriving the latest run again here would mean an
+* ``ORDER BY run_after DESC LIMIT 1`` per Dag, which has no supporting index and degrades
+* badly once a Dag has many runs. Runs the caller may not read are dropped.
+* @param data The data for the request.
+* @param data.dagRunIds
+* @returns DAGsRecentTaskInstanceStateCountsCollectionResponse Successful Response
+* @throws ApiError
+*/
+export const ensureUseDagServiceGetRecentTaskInstanceStateCountsUiData = (queryClient: QueryClient, { dagRunIds }: {
+  dagRunIds: number[];
+}) => queryClient.ensureQueryData({ queryKey: Common.UseDagServiceGetRecentTaskInstanceStateCountsUiKeyFn({ dagRunIds }), queryFn: () => DagService.getRecentTaskInstanceStateCountsUi({ dagRunIds }) });
+/**
 * Get Event Log
 * @param data The data for the request.
 * @param data.eventLogId

@@ -944,6 +944,25 @@ export const prefetchUseDagServiceGetDagRunStateCountsUi = (queryClient: QueryCl
   dagIds: string[];
 }) => queryClient.prefetchQuery({ queryKey: Common.UseDagServiceGetDagRunStateCountsUiKeyFn({ dagIds }), queryFn: () => DagService.getDagRunStateCountsUi({ dagIds }) });
 /**
+* Get Recent Task Instance State Counts
+* Return recent task-instance state counts for the Dags of the given runs, for the Dag list page.
+*
+* Like the Airflow 2 "Recent Tasks" column, a Dag's counts cover all of its running Dag
+* runs, or its latest run when none is running.
+*
+* The Dag list response already carries the latest run of each Dag, so the caller passes
+* those run ids straight in. Deriving the latest run again here would mean an
+* ``ORDER BY run_after DESC LIMIT 1`` per Dag, which has no supporting index and degrades
+* badly once a Dag has many runs. Runs the caller may not read are dropped.
+* @param data The data for the request.
+* @param data.dagRunIds
+* @returns DAGsRecentTaskInstanceStateCountsCollectionResponse Successful Response
+* @throws ApiError
+*/
+export const prefetchUseDagServiceGetRecentTaskInstanceStateCountsUi = (queryClient: QueryClient, { dagRunIds }: {
+  dagRunIds: number[];
+}) => queryClient.prefetchQuery({ queryKey: Common.UseDagServiceGetRecentTaskInstanceStateCountsUiKeyFn({ dagRunIds }), queryFn: () => DagService.getRecentTaskInstanceStateCountsUi({ dagRunIds }) });
+/**
 * Get Event Log
 * @param data The data for the request.
 * @param data.eventLogId

@@ -944,6 +944,25 @@ export const useDagServiceGetDagRunStateCountsUi = <TData = Common.DagServiceGet
   dagIds: string[];
 }, queryKey?: TQueryKey, options?: Omit<UseQueryOptions<TData, TError>, "queryKey" | "queryFn">) => useQuery<TData, TError>({ queryKey: Common.UseDagServiceGetDagRunStateCountsUiKeyFn({ dagIds }, queryKey), queryFn: () => DagService.getDagRunStateCountsUi({ dagIds }) as TData, ...options });
 /**
+* Get Recent Task Instance State Counts
+* Return recent task-instance state counts for the Dags of the given runs, for the Dag list page.
+*
+* Like the Airflow 2 "Recent Tasks" column, a Dag's counts cover all of its running Dag
+* runs, or its latest run when none is running.
+*
+* The Dag list response already carries the latest run of each Dag, so the caller passes
+* those run ids straight in. Deriving the latest run again here would mean an
+* ``ORDER BY run_after DESC LIMIT 1`` per Dag, which has no supporting index and degrades
+* badly once a Dag has many runs. Runs the caller may not read are dropped.
+* @param data The data for the request.
+* @param data.dagRunIds
+* @returns DAGsRecentTaskInstanceStateCountsCollectionResponse Successful Response
+* @throws ApiError
+*/
+export const useDagServiceGetRecentTaskInstanceStateCountsUi = <TData = Common.DagServiceGetRecentTaskInstanceStateCountsUiDefaultResponse, TError = unknown, TQueryKey extends Array<unknown> = unknown[]>({ dagRunIds }: {
+  dagRunIds: number[];
+}, queryKey?: TQueryKey, options?: Omit<UseQueryOptions<TData, TError>, "queryKey" | "queryFn">) => useQuery<TData, TError>({ queryKey: Common.UseDagServiceGetRecentTaskInstanceStateCountsUiKeyFn({ dagRunIds }, queryKey), queryFn: () => DagService.getRecentTaskInstanceStateCountsUi({ dagRunIds }) as TData, ...options });
+/**
 * Get Event Log
 * @param data The data for the request.
 * @param data.eventLogId

@@ -944,6 +944,25 @@ export const useDagServiceGetDagRunStateCountsUiSuspense = <TData = Common.DagSe
   dagIds: string[];
 }, queryKey?: TQueryKey, options?: Omit<UseQueryOptions<TData, TError>, "queryKey" | "queryFn">) => useSuspenseQuery<TData, TError>({ queryKey: Common.UseDagServiceGetDagRunStateCountsUiKeyFn({ dagIds }, queryKey), queryFn: () => DagService.getDagRunStateCountsUi({ dagIds }) as TData, ...options });
 /**
+* Get Recent Task Instance State Counts
+* Return recent task-instance state counts for the Dags of the given runs, for the Dag list page.
+*
+* Like the Airflow 2 "Recent Tasks" column, a Dag's counts cover all of its running Dag
+* runs, or its latest run when none is running.
+*
+* The Dag list response already carries the latest run of each Dag, so the caller passes
+* those run ids straight in. Deriving the latest run again here would mean an
+* ``ORDER BY run_after DESC LIMIT 1`` per Dag, which has no supporting index and degrades
+* badly once a Dag has many runs. Runs the caller may not read are dropped.
+* @param data The data for the request.
+* @param data.dagRunIds
+* @returns DAGsRecentTaskInstanceStateCountsCollectionResponse Successful Response
+* @throws ApiError
+*/
+export const useDagServiceGetRecentTaskInstanceStateCountsUiSuspense = <TData = Common.DagServiceGetRecentTaskInstanceStateCountsUiDefaultResponse, TError = unknown, TQueryKey extends Array<unknown> = unknown[]>({ dagRunIds }: {
+  dagRunIds: number[];
+}, queryKey?: TQueryKey, options?: Omit<UseQueryOptions<TData, TError>, "queryKey" | "queryFn">) => useSuspenseQuery<TData, TError>({ queryKey: Common.UseDagServiceGetRecentTaskInstanceStateCountsUiKeyFn({ dagRunIds }, queryKey), queryFn: () => DagService.getRecentTaskInstanceStateCountsUi({ dagRunIds }) as TData, ...options });
+/**
 * Get Event Log
 * @param data The data for the request.
 * @param data.eventLogId