Skip to content

Commit 3d8e2a3

Browse files
committed
Let the HITL review plugin address one loop pass or mapped slot
The plugin located a task instance by (task_id, map_index) and stored its review state as XCom under that address. With loop passes sharing a task id and map index that picks an arbitrary pass, so approving or rejecting could resolve another pass's session and write the human action to it. The plugin endpoints now take region_id and region_index and resolve them by the same live-producer rules as the public API, so an ambiguous selection is rejected instead of guessed. Reading, writing and checking completion all use that exact attempt. The review UI forwards the selected task instance's region coordinates when the host provides them. The plugin also runs on hosts that predate regions, where an unknown region_id would break the request, so a host without task instance context keeps the map-only request shape. A modern host that has not resolved a selection renders no session rather than opening another task's. Tests that re-fetch a task instance to run it need the region too, otherwise they run another pass's row.
1 parent 6adc196 commit 3d8e2a3

13 files changed

Lines changed: 745 additions & 43 deletions

File tree

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

Lines changed: 6 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -20,15 +20,15 @@
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
2727
from airflow.utils.session import NEW_SESSION
2828

2929
from tests_common.test_utils.compat import SerializedBaseOperator, SerializedMappedOperator
3030
from tests_common.test_utils.dag import create_scheduler_dag
31-
from tests_common.test_utils.version_compat import AIRFLOW_V_3_0_PLUS, AIRFLOW_V_3_2_PLUS
31+
from tests_common.test_utils.version_compat import AIRFLOW_V_3_0_PLUS, AIRFLOW_V_3_2_PLUS, AIRFLOW_V_3_4_PLUS
3232

3333
try:
3434
from airflow.serialization.serialized_objects import create_scheduler_operator
@@ -146,11 +146,15 @@ 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+
# 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 {}
149152
new_ti = TaskInstance.get_task_instance(
150153
dag_id=ti.dag_id,
151154
run_id=ti.run_id,
152155
task_id=ti.task_id,
153156
map_index=ti.map_index,
157+
**region_kwargs,
154158
**session_kwargs,
155159
)
156160
# Some tests don't even save the ti at all, in which case new_ti is None.

‎providers/common/ai/.pre-commit-config.yaml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -30,7 +30,7 @@ repos:
3030
language: node
3131
files: |
3232
(?x)
33-
^src/airflow/providers/common/ai/plugins/www/.*\.(js|ts|tsx|yaml|css|json)$
33+
^src/airflow/providers/common/ai/plugins/www/.*\.(m?js|ts|tsx|yaml|css|json)$
3434
exclude: |
3535
(?x)
3636
^src/airflow/providers/common/ai/plugins/www/node-modules/.*|

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

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -167,6 +167,11 @@ The plugin exposes a FastAPI app at ``/hitl-review``. Base URL:
167167
- ``run_id``: Dag run ID.
168168
- ``task_id``: Task ID.
169169
- ``map_index``: Map index for mapped tasks. Use ``-1`` for non-mapped tasks or index for dynamic mapping.
170+
- ``region_id`` and ``region_index``: Select one loop pass or one mapped slot. Use the ``region_id`` and
171+
``region_index`` of the task instance as returned by the public task instance endpoints. Send them
172+
together; ``region_index`` without ``region_id`` is rejected. Only supported on Airflow 3.4 or later;
173+
earlier hosts return 400 when either is sent. A task inside a loop addressed without them returns 400,
174+
and a selection that matches more than one task instance returns 409.
170175

171176
Endpoints
172177
^^^^^^^^^

0 commit comments

Comments
 (0)