Skip to content

Let the HITL review plugin address one loop pass or mapped slot #74351

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
ashb merged 1 commit into from
Oct 10, 2026
Merged

Let the HITL review plugin address one loop pass or mapped slot #74351

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 6 additions & 2 deletions devel-common/src/tests_common/test_utils/taskinstance.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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.
Expand Down
2 changes: 1 addition & 1 deletion providers/common/ai/.pre-commit-config.yaml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,7 +30,7 @@ repos:
language: node
files: |
(?x)
^src/airflow/providers/common/ai/plugins/www/.*\.(js|ts|tsx|yaml|css|json)$
^src/airflow/providers/common/ai/plugins/www/.*\.(m?js|ts|tsx|yaml|css|json)$
exclude: |
(?x)
^src/airflow/providers/common/ai/plugins/www/node-modules/.*|
Expand Down
5 changes: 5 additions & 0 deletions providers/common/ai/docs/hitl_review.rst
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -167,6 +167,11 @@ 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 loop pass or one mapped slot. Use the ``region_id`` and
``region_index`` of the task instance as returned by the public task instance endpoints. 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
^^^^^^^^^
Expand Down
Loading
Loading