Refuse Human-in-the-loop responses once the task stopped waiting (#75645)

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)
 
 

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";
@@ -208,3 +213,98 @@ describe("HITLResponseForm – the answer names the request it was given for", (
     expect(updateHITLResponse).toHaveBeenCalledWith(expect.objectContaining({ request_created_at: loaded }));
   });
 });
+
+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();
+  });
+});

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(

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
 
 
+@pytest.fixture
+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 @@ def archive_after_lookup(*args, **kwargs):
             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 @@ def test_response_records_responder_display_name(
         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 @@ def test_should_respond_400_for_invalid_option(
         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 @@ def test_should_respond_422_for_reserved_key_in_params_input(
         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 @@ def test_rejected_reserved_key_leaves_task_resumable(
         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 @@ def test_should_respond_404_without_hitl_detail(
         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 @@ def test_should_respond_409(
             )
         }
 
+    @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 @@ def test_a_response_for_a_replaced_request_is_refused(
         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 @@ def test_a_naive_request_created_at_is_rejected(
 
         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,