Skip to content

Commit 73c0b0f

Browse files
committed
fixup! Let the HITL review plugin address one loop pass or mapped slot
1 parent 3552782 commit 73c0b0f

5 files changed

Lines changed: 207 additions & 45 deletions

File tree

‎devel-common/src/tests_common/test_utils/taskinstance.py‎

Lines changed: 12 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -20,7 +20,7 @@
2020

2121
import asyncio
2222
import copy
23-
from typing import TYPE_CHECKING
23+
from typing import TYPE_CHECKING, Any
2424

2525
from airflow import settings
2626
from airflow.models.taskinstance import TaskInstance
@@ -146,23 +146,17 @@ def run_task_instance(
146146
from airflow.sdk.definitions.dag import _run_task
147147

148148
# Session handling is a mess in tests; use a fresh ti to run the task.
149-
if AIRFLOW_V_3_4_PLUS:
150-
new_ti = TaskInstance.get_task_instance(
151-
dag_id=ti.dag_id,
152-
run_id=ti.run_id,
153-
task_id=ti.task_id,
154-
map_index=ti.region_index,
155-
region_id=ti.region_id,
156-
**session_kwargs,
157-
)
158-
else:
159-
new_ti = TaskInstance.get_task_instance(
160-
dag_id=ti.dag_id,
161-
run_id=ti.run_id,
162-
task_id=ti.task_id,
163-
map_index=ti.map_index,
164-
**session_kwargs,
165-
)
149+
# Without a region, get_task_instance applies public_region_filter, which excludes loop regions, so
150+
# a loop pass would not be found and the caller's ti would run instead of this pass's row.
151+
region_kwargs: dict[str, Any] = {"region_id": ti.region_id} if AIRFLOW_V_3_4_PLUS else {}
152+
new_ti = TaskInstance.get_task_instance(
153+
dag_id=ti.dag_id,
154+
run_id=ti.run_id,
155+
task_id=ti.task_id,
156+
map_index=ti.map_index,
157+
**region_kwargs,
158+
**session_kwargs,
159+
)
166160
# Some tests don't even save the ti at all, in which case new_ti is None.
167161
taskrun_result = _run_task(ti=new_ti or ti, task=task)
168162
_dispose_in_process_async_connections()

‎providers/common/ai/docs/hitl_review.rst‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -164,6 +164,10 @@ The plugin exposes a FastAPI app at ``/hitl-review``. Base URL:
164164
- ``run_id``: Dag run ID.
165165
- ``task_id``: Task ID.
166166
- ``map_index``: Map index for mapped tasks. Use ``-1`` for non-mapped tasks or index for dynamic mapping.
167+
- ``region_id`` and ``region_index``: Select one pass of a loop. Send them together; ``region_index``
168+
without ``region_id`` is rejected. Only supported on Airflow 3.4 or later; earlier hosts return 400 when
169+
either is sent. A task inside a loop addressed without them returns 400, and a selection that matches
170+
more than one task instance returns 409.
167171

168172
Endpoints
169173
^^^^^^^^^

‎providers/common/ai/src/airflow/providers/common/ai/plugins/hitl_review.py‎

Lines changed: 11 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -91,6 +91,7 @@ def _get_bundle_url() -> str:
9191
if AIRFLOW_V_3_4_PLUS:
9292
from airflow.api_fastapi.core_api.services.public.task_coordinates import resolve_task_scope
9393
from airflow.models.dagbag import DBDagBag
94+
from airflow.models.task_coordinates import TaskCoordinateResolver
9495

9596
def _get_session():
9697
with create_session(scoped=False) as session:
@@ -105,11 +106,11 @@ def _read_xcom(
105106
run_id: str,
106107
task_id: str,
107108
map_index: int = -1,
108-
region_id: UUID | None = None,
109+
region_id: UUID | None,
109110
key: str,
110111
):
111112
"""Read a single XCom value from the database."""
112-
scope = {"region_id": region_id} if AIRFLOW_V_3_4_PLUS else {}
113+
scope = {"region_id": region_id, "include_node_regions": False} if AIRFLOW_V_3_4_PLUS else {}
113114
read = XComModel.get_many(
114115
run_id=run_id,
115116
key=key,
@@ -131,7 +132,7 @@ def _read_xcom_by_prefix(
131132
run_id: str,
132133
task_id: str,
133134
map_index: int = -1,
134-
region_id: UUID | None = None,
135+
region_id: UUID | None,
135136
prefix: str,
136137
) -> dict[int, Any]:
137138
"""Read all iteration-keyed XCom entries matching *prefix* (e.g. ``airflow_hitl_review_agent_output_``)."""
@@ -142,6 +143,7 @@ def _read_xcom_by_prefix(
142143
task_ids=task_id,
143144
map_indexes=map_index,
144145
region_id=region_id,
146+
include_node_regions=False,
145147
)
146148
entity = read.column_descriptions[0]["entity"]
147149
query = read.with_only_columns(entity.key, entity.value).where(entity.key.like(f"{prefix}%"))
@@ -167,7 +169,7 @@ def _write_xcom(
167169
run_id: str,
168170
task_id: str,
169171
map_index: int = -1,
170-
region_id: UUID | None = None,
172+
region_id: UUID | None,
171173
key: str,
172174
value,
173175
):
@@ -180,7 +182,7 @@ def _write_xcom(
180182
TI.run_id == run_id,
181183
TI.task_id == task_id,
182184
TI.region_index == map_index,
183-
TI.region_id == (region_id or UUID(int=0)),
185+
TI.region_id == region_id,
184186
)
185187
)
186188
if owner is None:
@@ -237,7 +239,7 @@ def _is_task_completed(
237239
run_id: str,
238240
task_id: str,
239241
map_index: int = -1,
240-
region_id: UUID | None = None,
242+
region_id: UUID | None,
241243
) -> bool:
242244
"""Return True if the task instance is no longer running."""
243245
query = select(TI.state).where(
@@ -249,7 +251,7 @@ def _is_task_completed(
249251
query = query.where(
250252
TI.working_set.is_(True),
251253
TI.region_index == map_index,
252-
TI.region_id == (region_id or UUID(int=0)),
254+
TI.region_id == region_id,
253255
)
254256
else:
255257
query = query.where(TI.map_index == map_index)
@@ -265,7 +267,7 @@ def _build_session_response(
265267
run_id: str,
266268
task_id: str,
267269
map_index: int = -1,
268-
region_id: UUID | None = None,
270+
region_id: UUID | None,
269271
) -> HITLReviewResponse | None:
270272
"""Build `HITLReviewResponse` from XCom entries."""
271273
raw = _read_xcom(
@@ -339,8 +341,7 @@ def _get_task_scope(
339341
dag_id=dag_id,
340342
run_id=run_id,
341343
task_id=task_id,
342-
session=db,
343-
dag_bag=DBDagBag(),
344+
resolver=TaskCoordinateResolver(DBDagBag(), db),
344345
map_index=map_index,
345346
region_id=region_id,
346347
region_index=region_index,

‎providers/common/ai/src/airflow/providers/common/ai/plugins/www/src/main.tsx‎

Lines changed: 22 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -59,10 +59,28 @@ const PluginComponent: FC<PluginComponentProps> = (props) => {
5959
? { region_id: taskInstance.region_id, region_index: taskInstance.region_index }
6060
: undefined;
6161

62-
return <ChatPage
63-
key={`${dagId}/${runId}/${taskId}/${mapIndex}/${taskInstance?.id ?? ""}/${region?.region_id ?? ""}/${region?.region_index ?? ""}`}
64-
dagId={dagId} runId={runId} taskId={taskId} mapIndex={mapIndex} region={region}
65-
/>;
62+
// Remounts ChatPage on any selection change: useSession keeps its first createApi(...) in a ref,
63+
// so without a new key it would keep sending the previous pass's region.
64+
const sessionKey = [
65+
dagId,
66+
runId,
67+
taskId,
68+
mapIndex,
69+
taskInstance?.id ?? "",
70+
region?.region_id ?? "",
71+
region?.region_index ?? "",
72+
].join("/");
73+
74+
return (
75+
<ChatPage
76+
key={sessionKey}
77+
dagId={dagId}
78+
runId={runId}
79+
taskId={taskId}
80+
mapIndex={mapIndex}
81+
region={region}
82+
/>
83+
);
6684
};
6785

6886
/**

0 commit comments

Comments
 (0)