This is an automated email from the ASF dual-hosted git repository.
kaxil 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 3cf3f4b2a9d Refuse a HITL response for a request the task has since
replaced (#74688)
3cf3f4b2a9d is described below
commit 3cf3f4b2a9d1aa216ac209f15e9796a322ceb727
Author: Kaxil Naik <[email protected]>
AuthorDate: Wed Oct 14 08:24:47 2026 +0100
Refuse a HITL response for a request the task has since replaced (#74688)
A task can ask for a human response again in one Dag run: after a
retry or clear, or an operator that pauses more than once, such as
common.ai's per-tool approval. Since #73235 the repeat request replaces
the subject, body and options and clears the earlier answer, but the
response endpoint still accepts any answer for the task instance. A
reviewer with the earlier request's form still open could approve the
new request while reading the old one.
The response payload takes an optional request_created_at, the
created_at of the request the client loaded. When it no longer matches
the waiting request, the endpoint returns 409. The Required Actions
form sends it; API callers that leave it out answer whatever request
is waiting, as before.
---
airflow-core/docs/tutorial/hitl.rst | 6 +-
.../api_fastapi/core_api/datamodels/hitl.py | 11 ++-
.../core_api/openapi/v2-rest-api-generated.yaml | 11 +++
.../api_fastapi/core_api/routes/public/hitl.py | 11 +++
.../airflow/ui/openapi-gen/requests/schemas.gen.ts | 13 ++++
.../airflow/ui/openapi-gen/requests/types.gen.ts | 4 +
.../src/components/HITLReview/HITLReviewDetail.tsx | 8 +-
.../HITLTaskInstances/HITLResponseForm.test.tsx | 36 ++++++++-
.../pages/HITLTaskInstances/HITLResponseForm.tsx | 6 +-
.../ui/src/pages/TaskInstance/HITLResponse.tsx | 2 +-
.../airflow/ui/src/queries/useUpdateHITLDetail.ts | 12 ++-
airflow-core/src/airflow/ui/src/utils/hitl.ts | 2 +
.../core_api/routes/public/test_hitl.py | 89 ++++++++++++++++++++++
.../src/airflowctl/api/datamodels/generated.py | 7 ++
14 files changed, 209 insertions(+), 9 deletions(-)
diff --git a/airflow-core/docs/tutorial/hitl.rst
b/airflow-core/docs/tutorial/hitl.rst
index 56b26235b59..62b94baa228 100644
--- a/airflow-core/docs/tutorial/hitl.rst
+++ b/airflow-core/docs/tutorial/hitl.rst
@@ -219,7 +219,11 @@ calls involved (``~`` works as a wildcard for ``dag_id``
and ``dag_run_id``):
# Submit the response; the test run resumes the task on its next poll.
# map_index is -1 for non-mapped tasks.
PATCH
/api/v2/dags/{dag_id}/dagRuns/{dag_run_id}/taskInstances/{task_id}/{map_index}/hitlDetails
- {"chosen_options": ["Approve"], "params_input": {}}
+ {"chosen_options": ["Approve"], "params_input": {}, "request_created_at":
"<created_at from the GET>"}
+
+``request_created_at`` is optional. Copy it unchanged from the request you
read: if the task asked
+again in the meantime (a retry, a clear, or an operator that pauses more than
once in one run), the
+answer is refused with ``409`` instead of being applied to a request you did
not read.
.. note::
diff --git a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/hitl.py
b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/hitl.py
index 1f502aa4d9d..31a5d9a07d4 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/hitl.py
+++ b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/hitl.py
@@ -20,7 +20,7 @@ from collections.abc import Iterable, Mapping
from datetime import datetime
from typing import Any
-from pydantic import Field, field_validator
+from pydantic import AwareDatetime, Field, field_validator
from airflow.api_fastapi.core_api.base import BaseModel
from airflow.api_fastapi.core_api.datamodels.common import find_reserved_keys
@@ -33,6 +33,15 @@ class UpdateHITLDetailPayload(BaseModel):
chosen_options: list[str] = Field(min_length=1)
params_input: Mapping = Field(default_factory=dict)
+ request_created_at: AwareDatetime | None = Field(
+ default=None,
+ description=(
+ "The ``created_at`` of the request this response answers, exactly
as the GET endpoint "
+ "returned it. When the task has since replaced its request (it
asked again after a "
+ "resume, retry or clear), the response is refused with 409 instead
of answering a "
+ "request the user did not see. Leave it out to answer whatever
request is waiting."
+ ),
+ )
@field_validator("params_input")
@classmethod
diff --git
a/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml
b/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml
index 28b9241bc06..9fba36f2126 100644
---
a/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml
+++
b/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml
@@ -18847,6 +18847,17 @@ components:
additionalProperties: true
type: object
title: Params Input
+ request_created_at:
+ anyOf:
+ - type: string
+ format: date-time
+ - type: 'null'
+ title: Request Created At
+ description: The ``created_at`` of the request this response
answers, exactly
+ as the GET endpoint returned it. When the task has since replaced
its
+ request (it asked again after a resume, retry or clear), the
response
+ is refused with 409 instead of answering a request the user did
not see.
+ Leave it out to answer whatever request is waiting.
type: object
required:
- chosen_options
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 91887a64cb4..85965610353 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
@@ -224,6 +224,17 @@ def update_hitl_detail(
"and is not allowed to write again."
),
)
+ if (
+ update_hitl_detail_payload.request_created_at is not None
+ and update_hitl_detail_payload.request_created_at !=
hitl_detail_model.created_at
+ ):
+ # The task asked again since the client loaded the request, so this
answer was given
+ # while reading other content (an operator that pauses more than once
in one run).
+ raise HTTPException(
+ status.HTTP_409_CONFLICT,
+ "The Human-in-the-loop request was replaced after it was loaded.
Reload it and answer "
+ "the request that is waiting now.",
+ )
user_id = user.get_id()
user_name = user.get_name()
diff --git a/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts
b/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts
index 68ccfe0903b..236c239ff9f 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts
@@ -9549,6 +9549,19 @@ export const $UpdateHITLDetailPayload = {
additionalProperties: true,
type: 'object',
title: 'Params Input'
+ },
+ request_created_at: {
+ anyOf: [
+ {
+ type: 'string',
+ format: 'date-time'
+ },
+ {
+ type: 'null'
+ }
+ ],
+ title: 'Request Created At',
+ description: 'The ``created_at`` of the request this response
answers, exactly as the GET endpoint returned it. When the task has since
replaced its request (it asked again after a resume, retry or clear), the
response is refused with 409 instead of answering a request the user did not
see. Leave it out to answer whatever request is waiting.'
}
},
type: 'object',
diff --git a/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts
b/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts
index 69a3ad485c6..06ac2f54295 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts
@@ -2462,6 +2462,10 @@ export type UpdateHITLDetailPayload = {
params_input?: {
[key: string]: unknown;
};
+ /**
+ * The ``created_at`` of the request this response answers, exactly as the
GET endpoint returned it. When the task has since replaced its request (it
asked again after a resume, retry or clear), the response is refused with 409
instead of answering a request the user did not see. Leave it out to answer
whatever request is waiting.
+ */
+ request_created_at?: string | null;
};
export type ValidationError = {
diff --git
a/airflow-core/src/airflow/ui/src/components/HITLReview/HITLReviewDetail.tsx
b/airflow-core/src/airflow/ui/src/components/HITLReview/HITLReviewDetail.tsx
index bf7d3afc025..aebe1c78252 100644
--- a/airflow-core/src/airflow/ui/src/components/HITLReview/HITLReviewDetail.tsx
+++ b/airflow-core/src/airflow/ui/src/components/HITLReview/HITLReviewDetail.tsx
@@ -50,7 +50,13 @@ export const HITLReviewDetail = ({
<VStack alignItems="stretch" gap={4}>
<HITLReviewDetailSummary detail={detail} onOpenTask={onOpenTask} />
- <HITLResponseForm hitlDetail={detail} key={ti.id} namespace={ti.id}
onResponded={onResponded} />
+ <HITLResponseForm
+ hitlDetail={detail}
+ // A new request from the same task (it asked again) gets a fresh form.
+ key={`${ti.id}-${detail.created_at}`}
+ namespace={ti.id}
+ onResponded={onResponded}
+ />
</VStack>
);
};
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 220ecb081ba..b1be04925d2 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
@@ -18,7 +18,7 @@
*/
/// <reference types="@testing-library/jest-dom" />
import "@testing-library/jest-dom/vitest";
-import { render, screen } from "@testing-library/react";
+import { fireEvent, render, screen } from "@testing-library/react";
import { describe, expect, it, vi } from "vitest";
import type { HITLDetailHistory, TaskInstanceHistoryResponse } from
"openapi/requests/types.gen";
@@ -38,8 +38,10 @@ vi.mock("react-i18next", () => ({
}),
}));
+const { updateHITLResponse } = vi.hoisted(() => ({ updateHITLResponse: vi.fn()
}));
+
vi.mock("src/queries/useUpdateHITLDetail", () => ({
- useUpdateHITLDetail: () => ({ updateHITLResponse: vi.fn() }),
+ useUpdateHITLDetail: () => ({ updateHITLResponse }),
}));
// ---------------------------------------------------------------------------
@@ -176,3 +178,33 @@ describe("HITLResponseForm – pending-state enablement", ()
=> {
expect(screen.getByTestId("hitl-option-No")).toBeDisabled();
});
});
+
+describe("HITLResponseForm – the answer names the request it was given for",
() => {
+ it("sends the created_at of the request the form was loaded for", () => {
+ updateHITLResponse.mockClear();
+ const loaded = "2026-10-14T01:00:00.123456Z";
+
+ renderForm(["Approve", "Reject"], { created_at: loaded });
+ fireEvent.click(screen.getByTestId("hitl-option-Approve"));
+
+ expect(updateHITLResponse).toHaveBeenCalledWith(
+ expect.objectContaining({ chosen_options: ["Approve"],
request_created_at: loaded }),
+ );
+ });
+
+ it("keeps the loaded created_at when the request under an open form is
replaced", () => {
+ updateHITLResponse.mockClear();
+ const loaded = "2026-10-14T01:00:00.123456Z";
+ const { rerender } = renderForm(["Approve", "Reject"], { created_at:
loaded });
+
+ // The task asked again while the form stayed open; a polled refetch hands
the form the new request.
+ rerender(
+ <HITLResponseForm
+ hitlDetail={makeHITLDetail(["Approve", "Reject"], { created_at:
"2026-10-14T01:05:00.000000Z" })}
+ />,
+ );
+ fireEvent.click(screen.getByTestId("hitl-option-Approve"));
+
+ expect(updateHITLResponse).toHaveBeenCalledWith(expect.objectContaining({
request_created_at: loaded }));
+ });
+});
diff --git
a/airflow-core/src/airflow/ui/src/pages/HITLTaskInstances/HITLResponseForm.tsx
b/airflow-core/src/airflow/ui/src/pages/HITLTaskInstances/HITLResponseForm.tsx
index 3e5fdb87bf6..9f192753a63 100644
---
a/airflow-core/src/airflow/ui/src/pages/HITLTaskInstances/HITLResponseForm.tsx
+++
b/airflow-core/src/airflow/ui/src/pages/HITLTaskInstances/HITLResponseForm.tsx
@@ -16,7 +16,7 @@
* specific language governing permissions and limitations
* under the License.
*/
-import { useState } from "react";
+import { useRef, useState } from "react";
import { Button, Box, Spacer, HStack, Text } from "@chakra-ui/react";
import { useTranslation } from "react-i18next";
@@ -64,6 +64,8 @@ export const HITLResponseForm = ({ hitlDetail, namespace =
"hitl", onResponded }
const [errors, setErrors] = useState<boolean>(false);
const [isSubmitting, setIsSubmitting] = useState<boolean>(false);
const { paramsDict } = useParamStore(namespace);
+ // The request this form was loaded for. The server refuses the answer if
the task has asked again since.
+ const loadedCreatedAt = useRef(hitlDetail.created_at);
const [searchParams] = useSearchParams();
const { preloadedHITLOptions } = getPreloadHITLFormData(searchParams,
hitlDetail);
@@ -97,7 +99,7 @@ export const HITLResponseForm = ({ hitlDetail, namespace =
"hitl", onResponded }
try {
const formData = getHITLFormData(paramsDict, option);
- updateHITLResponse(formData);
+ updateHITLResponse({ ...formData, request_created_at:
loadedCreatedAt.current });
} catch {
setErrors(true);
}
diff --git
a/airflow-core/src/airflow/ui/src/pages/TaskInstance/HITLResponse.tsx
b/airflow-core/src/airflow/ui/src/pages/TaskInstance/HITLResponse.tsx
index 24c7032ae2b..eff5c2866cd 100644
--- a/airflow-core/src/airflow/ui/src/pages/TaskInstance/HITLResponse.tsx
+++ b/airflow-core/src/airflow/ui/src/pages/TaskInstance/HITLResponse.tsx
@@ -99,7 +99,7 @@ export const HITLResponse = () => {
taskInstance={taskInstance}
/>
)}
- <HITLResponseForm hitlDetail={hitlDetail} />
+ <HITLResponseForm hitlDetail={hitlDetail} key={hitlDetail.created_at} />
</Box>
);
};
diff --git a/airflow-core/src/airflow/ui/src/queries/useUpdateHITLDetail.ts
b/airflow-core/src/airflow/ui/src/queries/useUpdateHITLDetail.ts
index c08455d4c62..b351a714278 100644
--- a/airflow-core/src/airflow/ui/src/queries/useUpdateHITLDetail.ts
+++ b/airflow-core/src/airflow/ui/src/queries/useUpdateHITLDetail.ts
@@ -87,8 +87,17 @@ export const useUpdateHITLDetail = ({
onSuccess?.();
};
- const onError = (apiError: unknown) => {
+ const onError = async (apiError: unknown) => {
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(
+ [
+ [useTaskInstanceServiceGetHitlDetailsKey],
+ [useTaskInstanceServiceGetHitlDetailKey, { dagId, dagRunId }],
+ [useTaskInstanceServiceGetHitlDetailTryDetailKey, { dagId, dagRunId }],
+ ].map((key) => queryClient.invalidateQueries({ queryKey: key })),
+ );
};
const { isPending, mutate } = useTaskInstanceServiceUpdateHitlDetail({
@@ -102,6 +111,7 @@ export const useUpdateHITLDetail = ({
const requestBody = {
chosen_options: updateHITLResponseRequestBody.chosen_options ?? [],
params_input: updateHITLResponseRequestBody.params_input ?? {},
+ request_created_at: updateHITLResponseRequestBody.request_created_at,
};
try {
diff --git a/airflow-core/src/airflow/ui/src/utils/hitl.ts
b/airflow-core/src/airflow/ui/src/utils/hitl.ts
index 573dab8c9d8..64af365f145 100644
--- a/airflow-core/src/airflow/ui/src/utils/hitl.ts
+++ b/airflow-core/src/airflow/ui/src/utils/hitl.ts
@@ -25,6 +25,8 @@ import type { ParamSchema, ParamSpec, ParamsSpec } from
"src/queries/useDagParam
export type HITLResponseParams = {
chosen_options?: Array<string>;
params_input?: Record<string, unknown>;
+ // The created_at of the request the form was loaded for; the server refuses
a stale answer.
+ request_created_at?: string;
};
// A HITL task is "pending a response" while parked: pre-3.3 it parks in
"deferred", from 3.3 it
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 0960ac451ed..07b164fdae2 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
@@ -852,6 +852,95 @@ class TestUpdateHITLDetailEndpoint:
)
}
+ @pytest.mark.usefixtures("sample_hitl_detail")
+ def test_a_response_for_a_replaced_request_is_refused(
+ self,
+ test_client: TestClient,
+ sample_ti_url_identifier: str,
+ sample_ti: TaskInstance,
+ session: Session,
+ ) -> None:
+ """A form loaded before the task asked again names the old request; it
must not answer the new one."""
+ loaded =
test_client.get(f"{sample_ti_url_identifier}/hitlDetails").json()["created_at"]
+ # The task asks again: core replaces the request, and with it
created_at.
+ detail = session.get(HITLDetail, sample_ti.id)
+ assert detail is not None
+ detail.created_at = detail.created_at + timedelta(minutes=5)
+ session.commit()
+
+ response = test_client.patch(
+ f"{sample_ti_url_identifier}/hitlDetails",
+ json={"chosen_options": ["Approve"], "params_input": {},
"request_created_at": loaded},
+ )
+
+ assert response.status_code == 409
+ assert "replaced after it was loaded" in response.json()["detail"]
+ session.expire_all()
+ unanswered = session.get(HITLDetail, sample_ti.id)
+ assert unanswered is not None
+ assert not unanswered.response_received
+
+ @pytest.mark.usefixtures("sample_hitl_detail")
+ def test_a_stale_response_to_an_answered_request_reports_it_answered(
+ self,
+ test_client: TestClient,
+ sample_ti_url_identifier: str,
+ sample_ti: TaskInstance,
+ session: Session,
+ ) -> None:
+ """Nothing is waiting, so the error must not tell the user to answer
the waiting request."""
+ loaded =
test_client.get(f"{sample_ti_url_identifier}/hitlDetails").json()["created_at"]
+ detail = session.get(HITLDetail, sample_ti.id)
+ assert detail is not None
+ detail.created_at = detail.created_at + timedelta(minutes=5)
+ session.commit()
+ answered = test_client.patch(
+ f"{sample_ti_url_identifier}/hitlDetails",
+ json={"chosen_options": ["Approve"], "params_input": {"input_1":
2}},
+ )
+ assert answered.status_code == 200
+
+ response = test_client.patch(
+ f"{sample_ti_url_identifier}/hitlDetails",
+ json={"chosen_options": ["Approve"], "params_input": {},
"request_created_at": loaded},
+ )
+
+ assert response.status_code == 409
+ assert "already been updated" in response.json()["detail"]
+
+ @pytest.mark.usefixtures("sample_hitl_detail")
+ def test_a_naive_request_created_at_is_rejected(
+ self,
+ test_client: TestClient,
+ sample_ti_url_identifier: str,
+ ) -> None:
+ response = test_client.patch(
+ f"{sample_ti_url_identifier}/hitlDetails",
+ json={"chosen_options": ["Approve"], "request_created_at":
"2025-07-03T00:00:00"},
+ )
+
+ assert response.status_code == 422
+
+ @pytest.mark.usefixtures("sample_hitl_detail")
+ def test_a_response_naming_the_waiting_request_is_applied(
+ self,
+ test_client: TestClient,
+ sample_ti_url_identifier: str,
+ ) -> None:
+ loaded =
test_client.get(f"{sample_ti_url_identifier}/hitlDetails").json()["created_at"]
+
+ response = test_client.patch(
+ f"{sample_ti_url_identifier}/hitlDetails",
+ json={
+ "chosen_options": ["Approve"],
+ "params_input": {"input_1": 2},
+ "request_created_at": loaded,
+ },
+ )
+
+ assert response.status_code == 200
+ assert response.json()["chosen_options"] == ["Approve"]
+
@pytest.mark.usefixtures("sample_hitl_detail")
def test_should_respond_422_with_empty_option(
self,
diff --git a/airflow-ctl/src/airflowctl/api/datamodels/generated.py
b/airflow-ctl/src/airflowctl/api/datamodels/generated.py
index 337a83c76cc..bfa82fcc1dc 100644
--- a/airflow-ctl/src/airflowctl/api/datamodels/generated.py
+++ b/airflow-ctl/src/airflowctl/api/datamodels/generated.py
@@ -1515,6 +1515,13 @@ class UpdateHITLDetailPayload(BaseModel):
chosen_options: Annotated[list[str], Field(min_length=1, title="Chosen
Options")]
params_input: Annotated[dict[str, Any] | None, Field(title="Params
Input")] = None
+ request_created_at: Annotated[
+ datetime | None,
+ Field(
+ description="The ``created_at`` of the request this response
answers, exactly as the GET endpoint returned it. When the task has since
replaced its request (it asked again after a resume, retry or clear), the
response is refused with 409 instead of answering a request the user did not
see. Leave it out to answer whatever request is waiting.",
+ title="Request Created At",
+ ),
+ ] = None
class ValidationError(BaseModel):
