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.

Author: ashb

airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_extra_links.py

@@ -143,8 +143,8 @@ def body():
         loop = create_loop(body, max_iterations=4)
     run = dag_maker.create_dagrun()
     sibling = next(ti for ti in run.task_instances if ti.task_id == "body.work" and ti.region_index == 0)
-    selected_region = DynamicRegion.get_or_create(
-        dag_id=run.dag_id, run_id=run.run_id, node_id=loop.group_id, session=session
+    selected_region = DynamicRegion(
+        dag_id=run.dag_id, run_id=run.run_id, node_id=loop.group_id, forked_from_region_id=sibling.region_id
     )
     session.add(selected_region)
     session.flush()

airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_hitl.py

@@ -71,8 +71,8 @@ def body():
         loop = create_loop(body, max_iterations=4)
     run = dag_maker.create_dagrun()
     region = session.scalar(select(DynamicRegion).where(DynamicRegion.node_id == loop.group_id))
-    sibling_region = DynamicRegion.get_or_create(
-        dag_id=run.dag_id, run_id=run.run_id, node_id=loop.group_id, session=session
+    sibling_region = DynamicRegion(
+        dag_id=run.dag_id, run_id=run.run_id, node_id=loop.group_id, forked_from_region_id=region.id
     )
     session.add(sibling_region)
     session.flush()

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

@@ -20,15 +20,15 @@
 
 import asyncio
 import copy
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Any
 
 from airflow import settings
 from airflow.models.taskinstance import TaskInstance
 from airflow.utils.session import NEW_SESSION
 
 from tests_common.test_utils.compat import SerializedBaseOperator, SerializedMappedOperator
 from tests_common.test_utils.dag import create_scheduler_dag
-from tests_common.test_utils.version_compat import AIRFLOW_V_3_0_PLUS, AIRFLOW_V_3_2_PLUS
+from tests_common.test_utils.version_compat import AIRFLOW_V_3_0_PLUS, AIRFLOW_V_3_2_PLUS, AIRFLOW_V_3_4_PLUS
 
 try:
     from airflow.serialization.serialized_objects import create_scheduler_operator
@@ -146,11 +146,15 @@ def run_task_instance(
     from airflow.sdk.definitions.dag import _run_task
 
     # Session handling is a mess in tests; use a fresh ti to run the task.
+    # Without a region, get_task_instance applies public_region_filter, which excludes loop regions, so
+    # a loop pass would not be found and the caller's ti would run instead of this pass's row.
+    region_kwargs: dict[str, Any] = {"region_id": ti.region_id} if AIRFLOW_V_3_4_PLUS else {}
     new_ti = TaskInstance.get_task_instance(
         dag_id=ti.dag_id,
         run_id=ti.run_id,
         task_id=ti.task_id,
         map_index=ti.map_index,
+        **region_kwargs,
         **session_kwargs,
     )
     # Some tests don't even save the ti at all, in which case new_ti is None.

providers/common/ai/docs/hitl_review.rst

@@ -164,6 +164,10 @@ The plugin exposes a FastAPI app at ``/hitl-review``. Base URL:
 - ``run_id``: Dag run ID.
 - ``task_id``: Task ID.
 - ``map_index``: Map index for mapped tasks. Use ``-1`` for non-mapped tasks or index for dynamic mapping.
+- ``region_id`` and ``region_index``: Select one pass of a loop. Send them together; ``region_index``
+  without ``region_id`` is rejected. Only supported on Airflow 3.4 or later; earlier hosts return 400 when
+  either is sent. A task inside a loop addressed without them returns 400, and a selection that matches
+  more than one task instance returns 409.
 
 Endpoints
 ^^^^^^^^^