diff --git a/airflow-core/docs/tutorial/hitl.rst b/airflow-core/docs/tutorial/hitl.rst index aea843486f404..a1dc91da68e95 100644 --- a/airflow-core/docs/tutorial/hitl.rst +++ b/airflow-core/docs/tutorial/hitl.rst @@ -221,6 +221,12 @@ calls involved (``~`` works as a wildcard for ``dag_id`` and ``dag_run_id``): PATCH /api/v2/dags/{dag_id}/dagRuns/{dag_run_id}/taskInstances/{task_id}/{map_index}/hitlDetails {"chosen_options": ["Approve"], "params_input": {}} +.. note:: + + Keys in ``params_input`` may not be Airflow's reserved serialization keys (``__classname__`` or + ``__id__``). A response containing one is rejected with ``400`` at submission time, because it + could not be serialized when the task resumes. + .. note:: ``response_timeout`` and timeout defaults are enforced by the scheduler, which does not run 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 301a3923dea8e..204db40180a92 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 @@ -16,6 +16,7 @@ # under the License. from __future__ import annotations +from collections.abc import Mapping from typing import Annotated import structlog @@ -23,6 +24,7 @@ from sqlalchemy import select from sqlalchemy.orm import joinedload +from airflow._shared.serialization import CLASSNAME, SCHEMA_ID from airflow._shared.timezones import timezone from airflow.api_fastapi.auth.managers.models.resource_details import DagAccessEntity from airflow.api_fastapi.common.db.common import SessionDep, paginated_select @@ -80,6 +82,27 @@ log = structlog.get_logger(__name__) +# Keys that ``serde.serialize`` refuses at any depth. A ``params_input`` carrying one could not be +# serialized into the resume event, so the response is rejected before it is stored. Extend this +# tuple if serde gains another reserved key; the write-side check and its tests read from here. +_SERDE_RESERVED_KEYS = (CLASSNAME, SCHEMA_ID) + + +def _find_serde_reserved_key(value: object) -> str | None: + """Return the first ``_SERDE_RESERVED_KEYS`` entry found at any depth in ``value``, else ``None``.""" + stack: list[object] = [value] + while stack: + current = stack.pop() + if isinstance(current, Mapping): + for key in _SERDE_RESERVED_KEYS: + if key in current: + return key + stack.extend(current.values()) + elif isinstance(current, (list, tuple)): + stack.extend(current) + return None + + def _get_task_instance_with_hitl_detail( dag_id: str, dag_run_id: str, @@ -221,6 +244,13 @@ def update_hitl_detail( status.HTTP_400_BAD_REQUEST, "Multiple options chosen but this Human-in-the-loop task accepts only a single option.", ) + reserved_key = _find_serde_reserved_key(update_hitl_detail_payload.params_input) + if reserved_key is not None: + raise HTTPException( + status.HTTP_400_BAD_REQUEST, + f"params_input may not contain the reserved key {reserved_key!r}, which cannot be " + "serialized when the task resumes.", + ) hitl_detail_model.responded_by = hitl_user hitl_detail_model.responded_at = timezone.utcnow() 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 ad7b069581ee3..171c167225bb1 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 @@ -29,6 +29,7 @@ from sqlalchemy.orm import Session from airflow._shared.timezones.timezone import utc, utcnow +from airflow.api_fastapi.core_api.routes.public.hitl import _SERDE_RESERVED_KEYS from airflow.models.hitl import HITLDetail from airflow.models.log import Log from airflow.models.taskinstance import TaskInstance as TIModel @@ -408,6 +409,75 @@ 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.parametrize("reserved_key", _SERDE_RESERVED_KEYS) + @pytest.mark.parametrize( + "make_params_input", + [ + pytest.param(lambda key: {key: "x"}, id="top-level"), + pytest.param(lambda key: {"nested": {key: "x"}}, id="nested-dict"), + pytest.param(lambda key: {"items": [{key: "x"}]}, id="inside-list"), + ], + ) + def test_should_respond_400_for_serde_reserved_key_in_params_input( + self, + test_client: TestClient, + sample_ti_url_identifier: str, + reserved_key: str, + make_params_input: Callable[[str], dict[str, Any]], + ) -> None: + """A params_input carrying a serde-reserved key at any depth is rejected (400) at submission time.""" + response = test_client.patch( + f"{sample_ti_url_identifier}/hitlDetails", + json={"chosen_options": ["Approve"], "params_input": make_params_input(reserved_key)}, + ) + assert response.status_code == 400 + assert reserved_key in response.json()["detail"] + + @time_machine.travel(datetime(2025, 7, 3, 0, 0, 0), tick=False) + @pytest.mark.usefixtures("sample_hitl_detail") + def test_rejected_reserved_key_leaves_task_resumable( + self, + test_client: TestClient, + sample_ti_url_identifier: str, + sample_ti: TaskInstance, + session: Session, + ) -> None: + """A rejected reserved-key response records nothing, so the parked task is still resumable by a corrected resubmission.""" + 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_id = None + session.commit() + + rejected = test_client.patch( + f"{sample_ti_url_identifier}/hitlDetails", + json={"chosen_options": ["Approve"], "params_input": {_SERDE_RESERVED_KEYS[0]: "x"}}, + ) + assert rejected.status_code == 400 + + session.expire_all() + parked = session.get(TIModel, sample_ti.id) + assert parked is not None + assert parked.state == TaskInstanceState.AWAITING_INPUT + detail = session.scalar(select(HITLDetail).where(HITLDetail.ti_id == sample_ti.id)) + assert detail is not None + assert detail.response_received is False + + accepted = test_client.patch( + f"{sample_ti_url_identifier}/hitlDetails", + json={"chosen_options": ["Approve"], "params_input": {"input_1": 2}}, + ) + assert accepted.status_code == 200 + + session.expire_all() + resumed = session.get(TIModel, sample_ti.id) + assert resumed is not None + assert resumed.state == TaskInstanceState.SCHEDULED + 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") def test_should_respond_200_to_assigned_users(