This is an automated email from the ASF dual-hosted git repository.
vatsrahul1001 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new d1491b82c90 Refuse Human-in-the-loop responses once the task stopped
waiting (#75645)
d1491b82c90 is described below
commit d1491b82c90d9c5178021885f16f8df1ca564b53
Author: Wei Lee <[email protected]>
AuthorDate: Tue Oct 27 01:37:35 2026 +0800
Refuse Human-in-the-loop responses once the task stopped waiting (#75645)
* Refuse Human-in-the-loop responses once the task stopped waiting
The hitlDetails PATCH recorded a response whenever none existed yet and
only resumed the task if it was still awaiting input or deferred. After
a response_timeout without defaults, an answer was stored and returned
200 although nothing resumed, so the UI reported success for a decision
that never took effect.
Return 409 before writing when the task instance is in any other state.
The check runs under the existing task instance lock, after the
authorization and option validation, and looks at the state only: a
request whose timeout has passed but that the scheduler has not yet
processed is still accepted.
* Show why the backend refused a Human-in-the-loop response
The error toast showed only the HTTP status text ("Conflict") when a
response was refused. Show the detail the backend returns, such as that
the task is no longer waiting for a response, and fall back to the
previous message when there is none.
* Cover the refusal toast for validation, mapping, missing and forbidden
details
---
.../api_fastapi/core_api/routes/public/hitl.py | 32 ++---
.../HITLTaskInstances/HITLResponseForm.test.tsx | 104 +++++++++++++++-
.../airflow/ui/src/queries/useUpdateHITLDetail.ts | 16 ++-
.../core_api/routes/public/test_hitl.py | 135 +++++++++++++++++++--
4 files changed, 260 insertions(+), 27 deletions(-)
diff --git
a/airflow-core/src/airflow/api_fastapi/core_api/routes/public/hitl.py
b/airflow-core/src/airflow/api_fastapi/core_api/routes/public/hitl.py
index 85965610353..63fa36abeff 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/routes/public/hitl.py
+++ b/airflow-core/src/airflow/api_fastapi/core_api/routes/public/hitl.py
@@ -267,27 +267,29 @@ def update_hitl_detail(
"Multiple options chosen but this Human-in-the-loop task accepts
only a single option.",
)
+ if locked_ti.state not in (TaskInstanceState.AWAITING_INPUT,
TaskInstanceState.DEFERRED):
+ raise HTTPException(
+ status.HTTP_409_CONFLICT,
+ "The Task Instance is no longer waiting for a response, so this
Human-in-the-loop response "
+ "was not recorded.",
+ )
+
hitl_detail_model.responded_by = dict(hitl_user)
hitl_detail_model.responded_at = timezone.utcnow()
hitl_detail_model.chosen_options =
update_hitl_detail_payload.chosen_options
hitl_detail_model.params_input =
dict(update_hitl_detail_payload.params_input)
session.add(hitl_detail_model)
- # Event-driven resume: if the task is parked waiting for this input,
transition it directly,
- # without a trigger. handle_event_submit packs the response into
next_kwargs["event"], sets
- # state=SCHEDULED + scheduled_dttm; the scheduler then re-queues
execute_complete. Gated on the
- # parked states so a finished/cleared TI is never resurrected. `locked_ti`
was locked at the top
- # (TI-before-HITLDetail order), so a concurrent clear cannot interleave
between this state check
- # and the resume and have its reset silently overwritten.
- if locked_ti is not None and locked_ti.state in (
- TaskInstanceState.AWAITING_INPUT,
- TaskInstanceState.DEFERRED,
- ):
- handle_event_submit(
- TriggerEvent(hitl_detail_model.as_resume_event_payload()),
- task_instance=locked_ti,
- session=session,
- )
+ # Event-driven resume: transition the parked task directly, without a
trigger.
+ # handle_event_submit packs the response into next_kwargs["event"], sets
state=SCHEDULED +
+ # scheduled_dttm; the scheduler then re-queues execute_complete.
`locked_ti` was locked at the top
+ # (TI-before-HITLDetail order), so a concurrent clear cannot interleave
between the state check
+ # above and the resume and have its reset silently overwritten.
+ handle_event_submit(
+ TriggerEvent(hitl_detail_model.as_resume_event_payload()),
+ task_instance=locked_ti,
+ session=session,
+ )
return HITLDetailResponse.model_validate(hitl_detail_model)
diff --git
a/airflow-core/src/airflow/ui/src/pages/HITLTaskInstances/HITLResponseForm.test.tsx
b/airflow-core/src/airflow/ui/src/pages/HITLTaskInstances/HITLResponseForm.test.tsx
index 9c35067370b..d220d4e7577 100644
---
a/airflow-core/src/airflow/ui/src/pages/HITLTaskInstances/HITLResponseForm.test.tsx
+++
b/airflow-core/src/airflow/ui/src/pages/HITLTaskInstances/HITLResponseForm.test.tsx
@@ -17,12 +17,17 @@
* under the License.
*/
/// <reference types="@testing-library/jest-dom" />
+import { QueryClient } from "@tanstack/react-query";
import "@testing-library/jest-dom/vitest";
-import { fireEvent, render, screen } from "@testing-library/react";
-import { describe, expect, it, vi } from "vitest";
+import { fireEvent, render, renderHook, screen, waitFor } from
"@testing-library/react";
+import { afterEach, describe, expect, it, vi } from "vitest";
+import { TaskInstanceService } from "openapi/requests/services.gen";
import type { HITLDetailHistory, TaskInstanceHistoryResponse } from
"openapi/requests/types.gen";
+import { toaster } from "src/system-components";
+
+import type { useUpdateHITLDetail as UseUpdateHITLDetail } from
"src/queries/useUpdateHITLDetail";
import { Wrapper } from "src/utils/Wrapper";
import { HITLResponseForm } from "./HITLResponseForm";
@@ -297,3 +302,98 @@ describe("HITLResponseForm – multiple requires a
selection", () => {
expect(screen.queryByText("response.selectAtLeastOne")).not.toBeInTheDocument();
});
});
+
+const submitRejectedAnswer = async (rejection: unknown) => {
+ const { useUpdateHITLDetail: useRealUpdateHITLDetail } = await
vi.importActual<{
+ useUpdateHITLDetail: typeof UseUpdateHITLDetail;
+ }>("src/queries/useUpdateHITLDetail");
+
+ vi.spyOn(TaskInstanceService,
"updateHitlDetail").mockRejectedValue(rejection);
+ const createToast = vi.spyOn(toaster, "create").mockImplementation(() =>
"toast-id");
+ const invalidateQueries = vi.spyOn(QueryClient.prototype,
"invalidateQueries");
+ const { result } = renderHook(
+ () =>
+ useRealUpdateHITLDetail({ dagId: "test_dag", dagRunId: "run_1",
mapIndex: -1, taskId: "test_task" }),
+ { wrapper: Wrapper },
+ );
+
+ result.current.updateHITLResponse({ chosen_options: ["Approve"],
params_input: {} });
+
+ return { createToast, invalidateQueries };
+};
+
+describe("useUpdateHITLDetail – a refused answer", () => {
+ const refusalDetail =
+ "The Task Instance is no longer waiting for a response, so this
Human-in-the-loop response was not recorded.";
+
+ afterEach(() => {
+ vi.restoreAllMocks();
+ });
+
+ it("shows the reason the backend gave instead of the bare status text",
async () => {
+ const { createToast } = await submitRejectedAnswer({
+ body: { detail: refusalDetail },
+ message: "Conflict",
+ status: 409,
+ });
+
+ await waitFor(() =>
+ expect(createToast).toHaveBeenCalledWith(
+ expect.objectContaining({ description: refusalDetail, type: "error" }),
+ ),
+ );
+ });
+
+ it.each([
+ {
+ detail: [
+ { loc: ["body", "chosen_options"], msg: "Field required", type:
"missing" },
+ { loc: ["body", "params_input", "name"], msg: "Input should be a
string", type: "string_type" },
+ ],
+ expected: "body.chosen_options Field required, body.params_input.name
Input should be a string",
+ name: "validation entries",
+ },
+ {
+ detail: { reason: "stale request" },
+ expected: "reason: stale request",
+ name: "a mapping",
+ },
+ ])("joins $name detail into one description", async ({ detail, expected })
=> {
+ const { createToast } = await submitRejectedAnswer({
+ body: { detail },
+ message: "Unprocessable Entity",
+ status: 422,
+ });
+
+ await waitFor(() =>
+ expect(createToast).toHaveBeenCalledWith({
+ description: expected,
+ title: "response.error",
+ type: "error",
+ }),
+ );
+ });
+
+ it("falls back to the error message when the backend gave no detail", async
() => {
+ const { createToast } = await submitRejectedAnswer({ message: "Internal
Server Error", status: 500 });
+
+ await waitFor(() =>
+ expect(createToast).toHaveBeenCalledWith({
+ description: "Internal Server Error",
+ title: "hitl:response.error",
+ type: "error",
+ }),
+ );
+ });
+
+ it("leaves a 403 to the MutationCache handler but still reloads the stale
request", async () => {
+ const { createToast, invalidateQueries } = await submitRejectedAnswer({
+ body: { detail: "Forbidden" },
+ message: "Forbidden",
+ status: 403,
+ });
+
+ await waitFor(() => expect(invalidateQueries).toHaveBeenCalledTimes(3));
+ expect(createToast).not.toHaveBeenCalled();
+ });
+});
diff --git a/airflow-core/src/airflow/ui/src/queries/useUpdateHITLDetail.ts
b/airflow-core/src/airflow/ui/src/queries/useUpdateHITLDetail.ts
index b351a714278..a2646dac946 100644
--- a/airflow-core/src/airflow/ui/src/queries/useUpdateHITLDetail.ts
+++ b/airflow-core/src/airflow/ui/src/queries/useUpdateHITLDetail.ts
@@ -36,7 +36,7 @@ import {
import { toaster } from "src/system-components";
-import { createErrorToaster } from "src/utils";
+import { createErrorToaster, getApiErrorDetail, getErrorStatus } from
"src/utils";
import type { HITLResponseParams } from "src/utils/hitl";
import { gridQueryKeys, tiPerAttemptQueryKeys } from "./gridViewQueryKeys";
@@ -88,7 +88,19 @@ export const useUpdateHITLDetail = ({
};
const onError = async (apiError: unknown) => {
- createErrorToaster(apiError, { titleKey: "hitl:response.error" },
translate);
+ const detail = getApiErrorDetail(apiError);
+
+ // The backend says why it refused the answer (e.g. the task stopped
waiting); show that rather
+ // than the bare status text. 403 stays with the MutationCache handler, as
in createErrorToaster.
+ if (detail !== undefined && getErrorStatus(apiError) !== 403) {
+ toaster.create({
+ description: Array.isArray(detail) ? detail.join(", ") : detail,
+ title: translate("response.error"),
+ type: "error",
+ });
+ } else {
+ createErrorToaster(apiError, { titleKey: "hitl:response.error" },
translate);
+ }
// A refused answer (the request was replaced or already answered) leaves
a stale request on
// screen; reload it so the next answer goes to what is waiting now.
await Promise.all(
diff --git
a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_hitl.py
b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_hitl.py
index 07b164fdae2..221cf9f95f9 100644
--- a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_hitl.py
+++ b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_hitl.py
@@ -463,6 +463,14 @@ def _assert_sample_audit_log(audit_log: Log) -> None:
assert json.loads(audit_log.extra) == expected_extra
[email protected]
+def park_sample_ti(sample_ti: TaskInstance, session: Session) -> None:
+ ti = session.get(TIModel, sample_ti.id)
+ assert ti is not None
+ ti.state = TaskInstanceState.AWAITING_INPUT
+ session.commit()
+
+
@pytest.fixture
def sample_update_payload() -> dict[str, Any]:
return {"chosen_options": ["Approve"], "params_input": {"input_1": 2}}
@@ -514,7 +522,7 @@ class TestUpdateHITLDetailEndpoint:
assert verify_session.get(HITLDetail, current.id) is None
@time_machine.travel(datetime(2025, 7, 3, 0, 0, 0), tick=False)
- @pytest.mark.usefixtures("sample_hitl_detail")
+ @pytest.mark.usefixtures("sample_hitl_detail", "park_sample_ti")
@mock.patch(
"airflow.api_fastapi.auth.managers.simple.user.SimpleAuthManagerUser.get_display_name",
return_value="Jane Doe",
@@ -534,7 +542,7 @@ class TestUpdateHITLDetailEndpoint:
assert response.json()["responded_by"] == {"id": "test", "name": "Jane
Doe"}
@time_machine.travel(datetime(2025, 7, 3, 0, 0, 0), tick=False)
- @pytest.mark.usefixtures("sample_hitl_detail")
+ @pytest.mark.usefixtures("sample_hitl_detail", "park_sample_ti")
def test_should_respond_200_with_existing_response(
self,
test_client: TestClient,
@@ -612,7 +620,7 @@ class TestUpdateHITLDetailEndpoint:
assert response.status_code == 400
assert "Invalid options" in response.json()["detail"]
- @pytest.mark.usefixtures("sample_hitl_detail")
+ @pytest.mark.usefixtures("sample_hitl_detail", "park_sample_ti")
def test_should_validate_against_row_refreshed_under_lock(
self,
test_client: TestClient,
@@ -666,7 +674,7 @@ class TestUpdateHITLDetailEndpoint:
assert "reserved serialization keys" in detail
assert reserved_key in detail
- @pytest.mark.usefixtures("sample_hitl_detail")
+ @pytest.mark.usefixtures("sample_hitl_detail", "park_sample_ti")
def test_should_respond_200_for_json_string_holding_a_reserved_key(
self,
test_client: TestClient,
@@ -725,7 +733,7 @@ class TestUpdateHITLDetailEndpoint:
assert "event" in (resumed.next_kwargs or {})
@time_machine.travel(datetime(2025, 7, 3, 0, 0, 0), tick=False)
- @pytest.mark.usefixtures("sample_hitl_detail_respondent")
+ @pytest.mark.usefixtures("sample_hitl_detail_respondent", "park_sample_ti")
def test_should_respond_200_to_assigned_users(
self,
test_client: TestClient,
@@ -818,7 +826,7 @@ class TestUpdateHITLDetailEndpoint:
assert response.json() == {"detail":
expected_hitl_detail_not_found_error_msg}
@time_machine.travel(datetime(2025, 7, 3, 0, 0, 0), tick=False)
- @pytest.mark.usefixtures("sample_hitl_detail")
+ @pytest.mark.usefixtures("sample_hitl_detail", "park_sample_ti")
def test_should_respond_409(
self,
test_client: TestClient,
@@ -852,6 +860,117 @@ class TestUpdateHITLDetailEndpoint:
)
}
+ @pytest.mark.usefixtures("sample_hitl_detail")
+ @pytest.mark.parametrize(
+ "state",
+ [
+ None,
+ TaskInstanceState.SCHEDULED,
+ TaskInstanceState.QUEUED,
+ TaskInstanceState.RUNNING,
+ TaskInstanceState.FAILED,
+ ],
+ )
+ def test_a_response_to_a_task_no_longer_waiting_is_refused(
+ self,
+ test_client: TestClient,
+ sample_ti_url_identifier: str,
+ sample_update_payload: dict[str, Any],
+ sample_ti: TaskInstance,
+ session: Session,
+ state: TaskInstanceState | None,
+ ) -> None:
+ ti = session.get(TIModel, sample_ti.id)
+ assert ti is not None
+ ti.state = state
+ ti.next_method = "execute_complete"
+ ti.next_kwargs = {}
+ session.commit()
+
+ response =
test_client.patch(f"{sample_ti_url_identifier}/hitlDetails",
json=sample_update_payload)
+
+ assert response.status_code == 409
+ assert response.json() == {
+ "detail": (
+ "The Task Instance is no longer waiting for a response, so
this Human-in-the-loop "
+ "response was not recorded."
+ )
+ }
+ session.expire_all()
+ detail = session.get(HITLDetail, sample_ti.id)
+ assert detail is not None
+ assert not detail.response_received
+ assert detail.responded_at is None
+ assert detail.responded_by is None
+ assert detail.chosen_options is None
+ refreshed = session.get(TIModel, sample_ti.id)
+ assert refreshed is not None
+ assert refreshed.state == state
+ assert refreshed.next_kwargs == {}
+
+ @pytest.mark.usefixtures("sample_hitl_detail")
+ @pytest.mark.parametrize("state", [TaskInstanceState.AWAITING_INPUT,
TaskInstanceState.DEFERRED])
+ def test_a_response_to_a_waiting_task_is_recorded_and_resumes_it(
+ self,
+ test_client: TestClient,
+ sample_ti_url_identifier: str,
+ sample_update_payload: dict[str, Any],
+ sample_ti: TaskInstance,
+ session: Session,
+ state: TaskInstanceState,
+ ) -> None:
+ ti = session.get(TIModel, sample_ti.id)
+ assert ti is not None
+ ti.state = state
+ ti.next_method = "execute_complete"
+ ti.next_kwargs = {}
+ session.commit()
+
+ response =
test_client.patch(f"{sample_ti_url_identifier}/hitlDetails",
json=sample_update_payload)
+
+ assert response.status_code == 200
+ session.expire_all()
+ detail = session.get(HITLDetail, sample_ti.id)
+ assert detail is not None
+ assert detail.response_received
+ assert detail.chosen_options == ["Approve"]
+ refreshed = session.get(TIModel, sample_ti.id)
+ assert refreshed is not None
+ assert refreshed.state == TaskInstanceState.SCHEDULED
+ assert "event" in (refreshed.next_kwargs or {})
+
+ @time_machine.travel(datetime(2025, 7, 3, 0, 0, 0, tzinfo=utc), tick=False)
+ @pytest.mark.usefixtures("sample_hitl_detail")
+ def
test_a_response_after_trigger_timeout_is_accepted_while_still_awaiting_input(
+ self,
+ test_client: TestClient,
+ sample_ti_url_identifier: str,
+ sample_update_payload: dict[str, Any],
+ sample_ti: TaskInstance,
+ session: Session,
+ ) -> None:
+ """The gate looks at state only: the scheduler has not timed the task
out yet, so it is still waiting."""
+ ti = session.get(TIModel, sample_ti.id)
+ assert ti is not None
+ ti.state = TaskInstanceState.AWAITING_INPUT
+ ti.next_method = "execute_complete"
+ ti.next_kwargs = {}
+ ti.trigger_timeout = datetime(2025, 7, 2, 0, 0, 0, tzinfo=utc)
+ session.commit()
+
+ response =
test_client.patch(f"{sample_ti_url_identifier}/hitlDetails",
json=sample_update_payload)
+
+ assert response.status_code == 200
+ session.expire_all()
+ detail = session.get(HITLDetail, sample_ti.id)
+ assert detail is not None
+ assert detail.response_received
+ assert detail.chosen_options == ["Approve"]
+ refreshed = session.get(TIModel, sample_ti.id)
+ assert refreshed is not None
+ assert refreshed.state == TaskInstanceState.SCHEDULED
+ assert "event" in (refreshed.next_kwargs or {})
+
@pytest.mark.usefixtures("sample_hitl_detail")
def test_a_response_for_a_replaced_request_is_refused(
self,
@@ -880,7 +999,7 @@ class TestUpdateHITLDetailEndpoint:
assert unanswered is not None
assert not unanswered.response_received
- @pytest.mark.usefixtures("sample_hitl_detail")
+ @pytest.mark.usefixtures("sample_hitl_detail", "park_sample_ti")
def test_a_stale_response_to_an_answered_request_reports_it_answered(
self,
test_client: TestClient,
@@ -921,7 +1040,7 @@ class TestUpdateHITLDetailEndpoint:
assert response.status_code == 422
- @pytest.mark.usefixtures("sample_hitl_detail")
+ @pytest.mark.usefixtures("sample_hitl_detail", "park_sample_ti")
def test_a_response_naming_the_waiting_request_is_applied(
self,
test_client: TestClient,
