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 1f4bc05aa0a Allow Human-in-the-loop tasks to be answered while still
running (#75413)
1f4bc05aa0a is described below
commit 1f4bc05aa0a13072b91984ff46d5c5d5153a2278
Author: Kaxil Naik <[email protected]>
AuthorDate: Thu Nov 5 20:11:48 2026 +0000
Allow Human-in-the-loop tasks to be answered while still running (#75413)
A running task could write its Human-in-the-loop request but not read the
answer: the task supervisor served CreateHITLDetailPayload and rejected
GetHITLDetailResponse as an unhandled request, since only the triggerer
needed it. Serve it from the task supervisor too, with a
HITLDetailResponseResult
reply type in the Task SDK comms. The Execution API route is already scoped
to ti:self, so a task can only read its own response.
The UI and the pending-actions queries treated a HITL task as waiting for a
response only while it was deferred or awaiting_input. A task that checks
for its answer before parking is running at that point, so count it as
waiting too, but only while the attempt that wrote the request runs it for
the first time: an attempt that started after the request was written has
not asked yet, and a resumed run (for example a timeout without defaults)
has stopped waiting. The response endpoint refuses answers to a running
task in those two cases, which it used to accept and then lose.
* Decide whether a running Human-in-the-loop task waits without the
worker's clock
A running task instance's unanswered request is always its own, since
every attempt gets a new task instance id. So a running task waits on
its first run, or on a resumed run that asked again after the server
resumed it: compare created_at with scheduled_dttm, both stamped by the
server, instead of with the worker's start_date. That keeps refusing a
run resumed after a timeout, and accepts an operator that asks again
from a resumed run. The response endpoint resumes only a parked task;
a running one reads the answer itself.
The supervisor and the triggerer share one handler and one
HITLDetailResponseResult, since core requires the Task SDK that ships
it. List the new supervisor body under 2027-02-28, the in-progress
version, since 2026-10-30 ships in 3.4.
The Task SDK client retries POST /hitlDetails/{ti_id}, and the endpoint
rewrites the request and clears any stored response. Now that a running
task can be answered, a reviewer could answer the request the first
attempt stored, get a 200, and then lose the answer to the client's
retry. The client now sends an Idempotency-Key per request, reused by
its retries; the endpoint stores it and returns the stored request
unchanged when a POST repeats it.
---
airflow-core/docs/migrations-ref.rst | 4 +-
.../airflow/api_fastapi/common/parameters/misc.py | 10 +-
.../api_fastapi/core_api/routes/public/hitl.py | 6 +-
.../airflow/api_fastapi/core_api/routes/ui/dags.py | 9 +-
.../api_fastapi/execution_api/routes/hitl.py | 16 +-
.../src/airflow/jobs/triggerer_job_runner.py | 23 +--
...154_3_5_0_add_idempotency_key_to_hitl_detail.py | 53 ++++++
airflow-core/src/airflow/models/hitl.py | 43 +++++
.../ui/src/components/NeedsReviewButton.tsx | 3 +-
.../airflow/ui/src/hooks/useRequiredActionTabs.ts | 6 +-
.../HITLTaskInstances/HITLTaskInstances.test.tsx | 11 +-
.../pages/HITLTaskInstances/HITLTaskInstances.tsx | 4 +-
airflow-core/src/airflow/ui/src/utils/hitl.test.ts | 48 +++---
airflow-core/src/airflow/ui/src/utils/hitl.ts | 12 +-
airflow-core/src/airflow/utils/db.py | 2 +-
.../core_api/routes/public/test_hitl.py | 46 ++++++
.../api_fastapi/core_api/routes/ui/test_dags.py | 30 ++++
.../execution_api/versions/head/test_hitl.py | 45 ++++++
.../tests/unit/dag_processing/test_processor.py | 1 +
generated/known_sdk_imports_in_core.txt | 2 +-
.../pkg/execution/genmodels/discriminators.gen.go | 4 +
go-sdk/pkg/execution/genmodels/models.gen.go | 83 ++++++----
go-sdk/schema/supervisor-schema.json | 72 +++++++++
task-sdk/src/airflow/sdk/api/client.py | 3 +
task-sdk/src/airflow/sdk/execution_time/comms.py | 13 ++
.../airflow/sdk/execution_time/request_handlers.py | 10 ++
.../airflow/sdk/execution_time/schema/schema.json | 72 +++++++++
.../execution_time/schema/versions/v2027_02_28.py | 9 +-
.../src/airflow/sdk/execution_time/supervisor.py | 4 +
task-sdk/tests/task_sdk/api/test_client.py | 21 +++
.../task_sdk/execution_time/test_supervisor.py | 40 +++--
ts-sdk/schema/supervisor-schema.json | 72 +++++++++
ts-sdk/src/generated/supervisor.ts | 177 ++++++++++++---------
33 files changed, 758 insertions(+), 196 deletions(-)
diff --git a/airflow-core/docs/migrations-ref.rst
b/airflow-core/docs/migrations-ref.rst
index 324dda5a79f..a6ee6caa085 100644
--- a/airflow-core/docs/migrations-ref.rst
+++ b/airflow-core/docs/migrations-ref.rst
@@ -39,7 +39,9 @@ Here's the list of all the Database Migrations that are
executed via when you ru
+-------------------------+------------------+-------------------+--------------------------------------------------------------+
| Revision ID | Revises ID | Airflow Version | Description
|
+=========================+==================+===================+==============================================================+
-| ``16d7301f124b`` (head) | ``5fbfa42a2d01`` | ``3.5.0`` | Add index
on callback trigger ID. |
+| ``a8b48633fa60`` (head) | ``16d7301f124b`` | ``3.5.0`` | Add
idempotency_key to hitl_detail. |
++-------------------------+------------------+-------------------+--------------------------------------------------------------+
+| ``16d7301f124b`` | ``5fbfa42a2d01`` | ``3.5.0`` | Add index
on callback trigger ID. |
+-------------------------+------------------+-------------------+--------------------------------------------------------------+
| ``5fbfa42a2d01`` | ``38c8df3fcb9c`` | ``3.5.0`` | Add
recorded state and owner to trigger. |
+-------------------------+------------------+-------------------+--------------------------------------------------------------+
diff --git a/airflow-core/src/airflow/api_fastapi/common/parameters/misc.py
b/airflow-core/src/airflow/api_fastapi/common/parameters/misc.py
index bdf53a993cd..e25cd99cbf3 100644
--- a/airflow-core/src/airflow/api_fastapi/common/parameters/misc.py
+++ b/airflow-core/src/airflow/api_fastapi/common/parameters/misc.py
@@ -37,7 +37,7 @@ from airflow.api_fastapi.common.parameters.search import (
from airflow.models.connection import Connection
from airflow.models.dag import DagModel
from airflow.models.errors import ParseImportError
-from airflow.models.hitl import HITLDetail
+from airflow.models.hitl import HITLDetail, awaiting_response_clause
from airflow.models.pool import Pool
from airflow.models.task_coordinates import public_map_index_expression
from airflow.models.taskinstance import TaskInstance
@@ -55,17 +55,11 @@ class _PendingActionsFilter(BaseParam[bool]):
if self.value is None and self.skip_none:
return select
- from airflow.models.hitl import HITLDetail
- from airflow.models.taskinstance import TaskInstance
-
# Join with HITLDetail and TaskInstance to find Dags
pending_actions_count_subquery = (
sql_select(func.count(HITLDetail.ti_id))
.join(TaskInstance, HITLDetail.ti_id == TaskInstance.id)
- .where(
- HITLDetail.responded_at.is_(None),
- TaskInstance.state.in_((TaskInstanceState.DEFERRED,
TaskInstanceState.AWAITING_INPUT)),
- )
+ .where(HITLDetail.responded_at.is_(None),
awaiting_response_clause())
.where(TaskInstance.dag_id == DagModel.dag_id)
.scalar_subquery()
)
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 63fa36abeff..a4a96550d08 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,7 +267,7 @@ 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):
+ if not hitl_detail_model.is_awaited_by(locked_ti):
raise HTTPException(
status.HTTP_409_CONFLICT,
"The Task Instance is no longer waiting for a response, so this
Human-in-the-loop response "
@@ -280,6 +280,10 @@ def update_hitl_detail(
hitl_detail_model.params_input =
dict(update_hitl_detail_payload.params_input)
session.add(hitl_detail_model)
+ if locked_ti.state == TaskInstanceState.RUNNING:
+ # The running task reads the answer itself; if it parks first, the
park transition resumes it.
+ return HITLDetailResponse.model_validate(hitl_detail_model)
+
# 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
diff --git a/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/dags.py
b/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/dags.py
index c0c747befcd..e018a21f649 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/dags.py
+++ b/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/dags.py
@@ -88,11 +88,11 @@ from
airflow.api_fastapi.core_api.services.public.task_coordinates import TaskCo
from airflow.configuration import conf
from airflow.models import DagModel, DagRun
from airflow.models.dag_favorite import DagFavorite
-from airflow.models.hitl import HITLDetail
+from airflow.models.hitl import HITLDetail, awaiting_response_clause
from airflow.models.renderedtifields import load_legacy_rendered_fields
from airflow.models.task_coordinates import TaskCoordinateResolver
from airflow.models.taskinstance import TaskInstance
-from airflow.utils.state import DagRunState, State, TaskInstanceState
+from airflow.utils.state import DagRunState, State
dags_router = AirflowRouter(prefix="/dags", tags=["DAG"])
@@ -285,10 +285,7 @@ def get_dags(
.options(
defaultload(HITLDetail.task_instance).joinedload(TaskInstance.rendered_task_instance_fields)
)
- .where(
- HITLDetail.responded_at.is_(None),
- TaskInstance.state.in_((TaskInstanceState.DEFERRED,
TaskInstanceState.AWAITING_INPUT)),
- )
+ .where(HITLDetail.responded_at.is_(None),
awaiting_response_clause())
.where(TaskInstance.dag_id.in_([dag.dag_id for dag in dags]))
.order_by(TaskInstance.dag_id)
)
diff --git a/airflow-core/src/airflow/api_fastapi/execution_api/routes/hitl.py
b/airflow-core/src/airflow/api_fastapi/execution_api/routes/hitl.py
index e27349684e3..d86c499a20a 100644
--- a/airflow-core/src/airflow/api_fastapi/execution_api/routes/hitl.py
+++ b/airflow-core/src/airflow/api_fastapi/execution_api/routes/hitl.py
@@ -16,11 +16,12 @@
# under the License.
from __future__ import annotations
+from typing import Annotated
from uuid import UUID
import structlog
from cadwyn import VersionedAPIRouter
-from fastapi import HTTPException, Security, status
+from fastapi import Header, HTTPException, Security, status
from sqlalchemy import select
from airflow._shared.timezones import timezone
@@ -54,11 +55,13 @@ def upsert_hitl_detail(
task_instance_id: UUID,
payload: HITLDetailRequest,
session: SessionDep,
+ idempotency_key: Annotated[UUID | None, Header(alias="Idempotency-Key")] =
None,
) -> HITLDetailRequest:
"""
Create a Human-in-the-loop detail for a specific Task Instance.
- There're 3 cases handled here.
+ A POST that repeats the ``Idempotency-Key`` of the stored request is a
retry of it, so the request and
+ any response it already has are kept. Otherwise there're 3 cases handled
here.
1. If a HITLOperator task instance does not have a HITLDetail,
a new HITLDetail is created without a response section.
@@ -79,6 +82,7 @@ def upsert_hitl_detail(
"params": payload.params,
"assignees": [user.model_dump() for user in payload.assigned_users],
"created_at": timezone.utcnow(),
+ "idempotency_key": idempotency_key,
}
# Same lock order as the park transition and the Core API response path
(TaskInstance, then
# hitl_detail), so a response landing between the read and the flush
cannot survive the rewrite.
@@ -86,6 +90,14 @@ def upsert_hitl_detail(
hitl_detail_model = session.scalar(
select(HITLDetail).where(HITLDetail.ti_id ==
task_instance_id).with_for_update(of=HITLDetail)
)
+ if (
+ hitl_detail_model is not None
+ and idempotency_key is not None
+ and hitl_detail_model.idempotency_key == idempotency_key
+ ):
+ # The client retried a POST whose response it lost. A running task can
be answered by then,
+ # so rewriting the request would discard that answer.
+ return HITLDetailRequest.model_validate(hitl_detail_model)
if not hitl_detail_model:
hitl_detail_model = HITLDetail(ti_id=task_instance_id, **request_part)
else:
diff --git a/airflow-core/src/airflow/jobs/triggerer_job_runner.py
b/airflow-core/src/airflow/jobs/triggerer_job_runner.py
index ea862dc9025..649b8ca8276 100644
--- a/airflow-core/src/airflow/jobs/triggerer_job_runner.py
+++ b/airflow-core/src/airflow/jobs/triggerer_job_runner.py
@@ -64,7 +64,6 @@ from airflow.models.task_coordinates import
TaskCoordinateResolver
from airflow.models.trigger import RecordedFailure, Trigger, TriggerKind
from airflow.observability.metrics import stats_utils
from airflow.sdk._shared.secrets_masker import redact # noqa: SDK001
-from airflow.sdk.api.datamodels._generated import HITLDetailResponse
from airflow.sdk.bases.xcom import _current_trigger_ti
from airflow.sdk.definitions.asset import Asset
from airflow.sdk.execution_time import supervisor
@@ -93,6 +92,7 @@ from airflow.sdk.execution_time.comms import (
GetVariable,
GetVariableKeys,
GetXCom,
+ HITLDetailResponseResult,
MaskSecret,
OKResponse,
PutVariable,
@@ -121,6 +121,7 @@ from airflow.sdk.execution_time.request_handlers import (
handle_get_connection,
handle_get_dag_run_state,
handle_get_dr_count,
+ handle_get_hitl_detail_response,
handle_get_previous_ti,
handle_get_task_states,
handle_get_ti_count,
@@ -460,23 +461,6 @@ class messages:
events_persisted: list[int] | None = None
-class HITLDetailResponseResult(HITLDetailResponse):
- """Response to GetHITLDetailResponse request."""
-
- type: Literal["HITLDetailResponseResult"] = "HITLDetailResponseResult"
-
- @classmethod
- def from_api_response(cls, response: HITLDetailResponse) ->
HITLDetailResponseResult:
- """
- Create result class from API Response.
-
- API Response is autogenerated from the API schema, so we need to
convert it to Result
- for communication between the Supervisor and the task process since it
needs a
- discriminator field.
- """
- return cls(**response.model_dump(exclude_defaults=True),
type="HITLDetailResponseResult")
-
-
ToTriggerRunner = Annotated[
messages.StartTriggerer
| messages.TriggerStateSync
@@ -804,8 +788,7 @@ class TriggerRunnerSupervisor(WatchedSubprocess):
)
resp =
HITLDetailResponseResult.from_api_response(response=api_resp)
elif isinstance(msg, GetHITLDetailResponse):
- api_resp = self.client.hitl.get_detail_response(ti_id=msg.ti_id)
- resp =
HITLDetailResponseResult.from_api_response(response=api_resp)
+ resp, dump_opts = handle_get_hitl_detail_response(self.client, msg)
elif isinstance(msg, MaskSecret):
handle_mask_secret(msg)
elif isinstance(msg, ClearAssetStateStoreByName):
diff --git
a/airflow-core/src/airflow/migrations/versions/0154_3_5_0_add_idempotency_key_to_hitl_detail.py
b/airflow-core/src/airflow/migrations/versions/0154_3_5_0_add_idempotency_key_to_hitl_detail.py
new file mode 100644
index 00000000000..097ef7168a3
--- /dev/null
+++
b/airflow-core/src/airflow/migrations/versions/0154_3_5_0_add_idempotency_key_to_hitl_detail.py
@@ -0,0 +1,53 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""
+Add idempotency_key to hitl_detail.
+
+Revision ID: a8b48633fa60
+Revises: 16d7301f124b
+Create Date: 2026-11-04 17:48:09.000000
+
+"""
+
+from __future__ import annotations
+
+import sqlalchemy as sa
+from alembic import op
+
+from airflow.migrations.utils import disable_sqlite_fkeys
+
+revision = "a8b48633fa60"
+down_revision = "16d7301f124b"
+branch_labels = None
+depends_on = None
+airflow_version = "3.5.0"
+
+
+def upgrade():
+ """Add the idempotency_key column to hitl_detail."""
+ with disable_sqlite_fkeys(op):
+ with op.batch_alter_table("hitl_detail", schema=None) as batch_op:
+ batch_op.add_column(sa.Column("idempotency_key", sa.Uuid(),
nullable=True))
+
+
+def downgrade():
+ """Remove the idempotency_key column from hitl_detail."""
+ with disable_sqlite_fkeys(op):
+ with op.batch_alter_table("hitl_detail", schema=None) as batch_op:
+ batch_op.drop_column("idempotency_key")
diff --git a/airflow-core/src/airflow/models/hitl.py
b/airflow-core/src/airflow/models/hitl.py
index ef7c1e4aa21..30b985f891a 100644
--- a/airflow-core/src/airflow/models/hitl.py
+++ b/airflow-core/src/airflow/models/hitl.py
@@ -30,11 +30,17 @@ from sqlalchemy.sql.functions import FunctionElement
from airflow._shared.timezones import timezone
from airflow.models.base import Base
from airflow.utils.sqlalchemy import UtcDateTime
+from airflow.utils.state import TaskInstanceState
if TYPE_CHECKING:
from sqlalchemy.sql import ColumnElement
from sqlalchemy.sql.compiler import SQLCompiler
+ from airflow.models.taskinstance import TaskInstance
+
+# A task parks for its answer in ``deferred`` before Airflow 3.3 and in
``awaiting_input`` from 3.3.
+HITL_PARKED_STATES = (TaskInstanceState.DEFERRED,
TaskInstanceState.AWAITING_INPUT)
+
class JSONExtract(FunctionElement):
"""
@@ -163,6 +169,8 @@ class HITLDetail(Base, HITLDetailPropertyMixin):
params: Mapped[dict] = mapped_column(sa.JSON(), nullable=False, default={})
assignees: Mapped[list[dict[str, str]] | None] = mapped_column(sa.JSON(),
nullable=True)
created_at: Mapped[datetime] = mapped_column(UtcDateTime,
default=timezone.utcnow, nullable=False)
+ # Sent by the Task SDK client with the request, so a retried POST can be
told from a new request.
+ idempotency_key: Mapped[UUID | None] = mapped_column(Uuid(), nullable=True)
# Response Content Detail
responded_at: Mapped[datetime | None] = mapped_column(UtcDateTime,
nullable=True)
@@ -206,3 +214,38 @@ class HITLDetail(Base, HITLDetailPropertyMixin):
"responded_by_user": self.responded_by_user,
"timedout": timedout,
}
+
+ def is_awaited_by(self, ti: TaskInstance) -> bool:
+ """
+ Whether ``ti`` is waiting for the answer to this request.
+
+ A parked task is. A running task is while it checks for the answer
before parking: on its
+ first run (``next_method`` unset), or on a resumed run that asked
again after the server
+ resumed it (``scheduled_dttm``). A run resumed after a timeout without
an answer is not.
+ Every attempt gets a new task instance id, so the request is always
``ti``'s own.
+ :func:`awaiting_response_clause` is the SQL form.
+ """
+ if ti.state in HITL_PARKED_STATES:
+ return True
+ if ti.state != TaskInstanceState.RUNNING:
+ return False
+ return ti.next_method is None or (
+ ti.scheduled_dttm is not None and self.created_at >=
ti.scheduled_dttm
+ )
+
+
+def awaiting_response_clause() -> ColumnElement[bool]:
+ """Return the SQL form of :meth:`HITLDetail.is_awaited_by`; join
``task_instance`` first."""
+ # Circular import: airflow.models.taskinstance imports this module at load
time.
+ from airflow.models.taskinstance import TaskInstance
+
+ return sa.or_(
+ TaskInstance.state.in_(HITL_PARKED_STATES),
+ sa.and_(
+ TaskInstance.state == TaskInstanceState.RUNNING,
+ sa.or_(
+ TaskInstance.next_method.is_(None),
+ HITLDetail.created_at >= TaskInstance.scheduled_dttm,
+ ),
+ ),
+ )
diff --git a/airflow-core/src/airflow/ui/src/components/NeedsReviewButton.tsx
b/airflow-core/src/airflow/ui/src/components/NeedsReviewButton.tsx
index 2bfc93d521a..5fe645bd6e4 100644
--- a/airflow-core/src/airflow/ui/src/components/NeedsReviewButton.tsx
+++ b/airflow-core/src/airflow/ui/src/components/NeedsReviewButton.tsx
@@ -26,6 +26,7 @@ import { useTaskInstanceServiceGetHitlDetails } from
"openapi/queries";
import { HITLReviewModal } from
"src/components/HITLReview/HITLReviewModal.tsx";
import { useHITLReviewModalRouteSync } from
"src/components/HITLReview/useHITLReviewModalRouteSync.ts";
+import { HITL_PENDING_STATES } from "src/utils/hitl";
import { useAutoRefresh } from "src/utils/query";
import { NeedsReviewIndicator } from "./NeedsReviewIndicator";
@@ -52,7 +53,7 @@ const usePendingHitl = ({
dagRunId: runId ?? "~",
orderBy: ["dag_id", "run_after", "created_at", "task_display_name"],
responseReceived: false,
- state: ["deferred", "awaiting_input"],
+ state: HITL_PENDING_STATES,
taskId,
},
undefined,
diff --git a/airflow-core/src/airflow/ui/src/hooks/useRequiredActionTabs.ts
b/airflow-core/src/airflow/ui/src/hooks/useRequiredActionTabs.ts
index 55e926d127b..03ffca15ad5 100644
--- a/airflow-core/src/airflow/ui/src/hooks/useRequiredActionTabs.ts
+++ b/airflow-core/src/airflow/ui/src/hooks/useRequiredActionTabs.ts
@@ -23,6 +23,8 @@ import { useLocation, useNavigate } from "react-router-dom";
import { useTaskInstanceServiceGetHitlDetails } from "openapi/queries";
+import { isHITLPending } from "src/utils/hitl";
+
export type HITLQueryParams = {
dagId: string;
dagRunId?: string;
@@ -97,9 +99,7 @@ export const useRequiredActionTabs = (
const hasHitlData = enabled && (hitlData?.total_entries ?? 0) > 0;
const pendingActionsCount =
hitlData?.hitl_details.filter(
- (hitl) =>
- (hitl.task_instance.state === "deferred" || hitl.task_instance.state
=== "awaiting_input") &&
- !hitl.response_received,
+ (hitl) => isHITLPending(hitl.task_instance.state) &&
!hitl.response_received,
).length ?? 0;
const processedTabs = tabs
diff --git
a/airflow-core/src/airflow/ui/src/pages/HITLTaskInstances/HITLTaskInstances.test.tsx
b/airflow-core/src/airflow/ui/src/pages/HITLTaskInstances/HITLTaskInstances.test.tsx
index 1440662aa07..a43491a4ac4 100644
---
a/airflow-core/src/airflow/ui/src/pages/HITLTaskInstances/HITLTaskInstances.test.tsx
+++
b/airflow-core/src/airflow/ui/src/pages/HITLTaskInstances/HITLTaskInstances.test.tsx
@@ -165,19 +165,20 @@ describe("HITLTaskInstances – mapIndex URL param handling
(#66428)", () => {
});
// ---------------------------------------------------------------------------
-// State filter: pending required-action rows park in either "deferred"
(pre-3.3)
-// or "awaiting_input" (from 3.3). Filtering on unreceived responses must
request
-// both states, otherwise awaiting_input rows are hidden from the listing.
+// State filter: pending required-action rows park in "deferred" (pre-3.3) or
+// "awaiting_input" (from 3.3), or are still "running" while the task checks
for the
+// response before parking. Filtering on unreceived responses must request all
three,
+// otherwise those rows are hidden from the listing.
// ---------------------------------------------------------------------------
describe("HITLTaskInstances – pending state filter", () => {
- it("requests both deferred and awaiting_input when filtering for unreceived
responses", () => {
+ it("requests deferred, awaiting_input and running when filtering for
unreceived responses", () => {
mockSearchParams = new URLSearchParams("response_received=false");
render(<HITLTaskInstances />, { wrapper: Wrapper });
const args = lastListingCall()?.[0] as { state?: Array<string> } |
undefined;
- expect(args?.state).toEqual(["deferred", "awaiting_input"]);
+ expect(args?.state).toEqual(["deferred", "awaiting_input", "running"]);
});
it("does not send a state filter when not filtering for unreceived
responses", () => {
diff --git
a/airflow-core/src/airflow/ui/src/pages/HITLTaskInstances/HITLTaskInstances.tsx
b/airflow-core/src/airflow/ui/src/pages/HITLTaskInstances/HITLTaskInstances.tsx
index dcf3c3cb910..882ed4e8566 100644
---
a/airflow-core/src/airflow/ui/src/pages/HITLTaskInstances/HITLTaskInstances.tsx
+++
b/airflow-core/src/airflow/ui/src/pages/HITLTaskInstances/HITLTaskInstances.tsx
@@ -45,7 +45,7 @@ import { SearchParamsKeys, type SearchParamsKeysType } from
"src/constants/searc
import { useAdvancedSearchArg } from "src/hooks/useAdvancedSearch";
import { useConfig } from "src/queries/useConfig";
import { useAutoRefresh, useDocumentTitle } from "src/utils";
-import { getHITLState, isHITLPending } from "src/utils/hitl";
+import { getHITLState, HITL_PENDING_STATES, isHITLPending } from
"src/utils/hitl";
import { getTaskInstanceLink } from "src/utils/links";
import { HITLFilters } from "./HITLFilters";
@@ -303,7 +303,7 @@ export const HITLTaskInstances = ({
Boolean(effectiveResponseReceived) && effectiveResponseReceived !==
"all"
? effectiveResponseReceived === "true"
: undefined,
- state: effectiveResponseReceived === "false" ? ["deferred",
"awaiting_input"] : undefined,
+ state: effectiveResponseReceived === "false" ? HITL_PENDING_STATES :
undefined,
subjectSearch,
taskId,
...taskIdArg,
diff --git a/airflow-core/src/airflow/ui/src/utils/hitl.test.ts
b/airflow-core/src/airflow/ui/src/utils/hitl.test.ts
index 309de054b97..a4fc6ffa64a 100644
--- a/airflow-core/src/airflow/ui/src/utils/hitl.test.ts
+++ b/airflow-core/src/airflow/ui/src/utils/hitl.test.ts
@@ -19,7 +19,7 @@
import type { TFunction } from "i18next";
import { describe, it, expect, vi } from "vitest";
-import type { HITLDetail } from "openapi/requests/types.gen";
+import type { HITLDetail, TaskInstanceState } from
"openapi/requests/types.gen";
import { getHITLFormData, getHITLParamsDict, getHITLState, isHITLPending }
from "./hitl";
@@ -51,13 +51,12 @@ const createMockHITLDetail = (overrides?:
Partial<HITLDetail>): HITLDetail =>
}) as HITLDetail;
describe("isHITLPending", () => {
- it("treats a deferred task as pending", () => {
- expect(isHITLPending("deferred")).toBe(true);
- });
-
- it("treats an awaiting_input task as pending", () => {
- expect(isHITLPending("awaiting_input")).toBe(true);
- });
+ it.each<TaskInstanceState>(["deferred", "awaiting_input", "running"])(
+ "treats a %s task as pending",
+ (state) => {
+ expect(isHITLPending(state)).toBe(true);
+ },
+ );
it("treats a finished/cleared task as not pending", () => {
expect(isHITLPending("success")).toBe(false);
@@ -67,22 +66,25 @@ describe("isHITLPending", () => {
});
describe("getHITLState", () => {
- it("reports a 'required' state for an awaiting_input task without a
response", () => {
- const hitlDetail = createMockHITLDetail({
- response_received: false,
- task_instance: {
- dag_id: "test_dag",
- dag_run_id: "test_run",
- map_index: -1,
- state: "awaiting_input",
- task_id: "test_task",
- },
- } as Partial<HITLDetail>);
+ it.each<TaskInstanceState>(["awaiting_input", "running"])(
+ "reports a 'required' state for a %s task without a response",
+ (state) => {
+ const hitlDetail = createMockHITLDetail({
+ response_received: false,
+ task_instance: {
+ dag_id: "test_dag",
+ dag_run_id: "test_run",
+ map_index: -1,
+ state,
+ task_id: "test_task",
+ },
+ } as Partial<HITLDetail>);
- // Empty params + non-approval options -> choice task; the point is that a
parked
- // awaiting_input task is "required", not "noResponseReceived".
- expect(getHITLState(mockTranslate,
hitlDetail)).toBe("state.choiceRequired");
- });
+ // Empty params + non-approval options -> choice task; the point is that
a task still
+ // waiting for its answer is "required", not "noResponseReceived".
+ expect(getHITLState(mockTranslate,
hitlDetail)).toBe("state.choiceRequired");
+ },
+ );
it("reports no response received for a finished task without a response", ()
=> {
const hitlDetail = createMockHITLDetail({
diff --git a/airflow-core/src/airflow/ui/src/utils/hitl.ts
b/airflow-core/src/airflow/ui/src/utils/hitl.ts
index ca03b8f03f3..4fcd47ed5d1 100644
--- a/airflow-core/src/airflow/ui/src/utils/hitl.ts
+++ b/airflow-core/src/airflow/ui/src/utils/hitl.ts
@@ -29,11 +29,15 @@ export type HITLResponseParams = {
request_created_at?: string;
};
-// A HITL task is "pending a response" while parked: pre-3.3 it parks in
"deferred", from 3.3 it
-// parks in "awaiting_input". Either way an unanswered parked task is
"response required", not
-// "no response received" (which is reserved for tasks no longer parked, e.g.
cleared/finished).
+// A HITL task waits for a response in one of these states: pre-3.3 it parks
in "deferred", from 3.3
+// it parks in "awaiting_input", and it is still "running" while it checks for
the response before
+// parking. An unanswered task in one of them is "response required", not "no
response received"
+// (which is reserved for tasks no longer waiting, e.g. cleared/finished). A
run resumed after a
+// timeout is also "running" until it fails; the server refuses answers to it.
+export const HITL_PENDING_STATES: Array<TaskInstanceState> = ["deferred",
"awaiting_input", "running"];
+
export const isHITLPending = (state?: TaskInstanceState | null): boolean =>
- state === "deferred" || state === "awaiting_input";
+ state !== undefined && state !== null && HITL_PENDING_STATES.includes(state);
const getChosenOptionsValue = (hitlDetail: HITLDetailHistory) => {
// if response_received is true, display the chosen_options, otherwise
display the defaults
diff --git a/airflow-core/src/airflow/utils/db.py
b/airflow-core/src/airflow/utils/db.py
index 5aa8350c822..345c35c8746 100644
--- a/airflow-core/src/airflow/utils/db.py
+++ b/airflow-core/src/airflow/utils/db.py
@@ -112,7 +112,7 @@ _REVISION_HEADS_MAP: dict[str, str] = {
"3.2.0": "1d6611b6ab7c",
"3.3.0": "d2f4e1b3c5a7",
"3.4.0": "b4f1a7d2c903",
- "3.5.0": "16d7301f124b",
+ "3.5.0": "a8b48633fa60",
}
# Prefix used to identify tables holding data moved during migration.
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 bcc836f491b..e8eb36ce51c 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
@@ -607,6 +607,52 @@ class TestUpdateHITLDetailEndpoint:
assert refreshed.next_method == "execute_complete"
assert "event" in (refreshed.next_kwargs or {})
+ @pytest.mark.parametrize(
+ ("next_method", "resumed_offset", "expected_status"),
+ [
+ pytest.param(None, None, 200, id="first-run"),
+ pytest.param(
+ "resume_after_tool_approval", timedelta(minutes=-5), 200,
id="resumed-run-asked-again"
+ ),
+ pytest.param("execute_complete", timedelta(minutes=5), 409,
id="resumed-after-a-timeout"),
+ ],
+ )
+ @pytest.mark.usefixtures("sample_hitl_detail")
+ def test_response_to_a_running_task_is_stored_while_it_waits(
+ self,
+ test_client: TestClient,
+ sample_ti_url_identifier: str,
+ sample_update_payload: dict[str, Any],
+ sample_ti: TaskInstance,
+ session: Session,
+ next_method: str | None,
+ resumed_offset: timedelta | None,
+ expected_status: int,
+ ) -> None:
+ detail = session.scalar(select(HITLDetail).where(HITLDetail.ti_id ==
sample_ti.id))
+ ti = session.get(TIModel, sample_ti.id)
+ assert detail is not None
+ assert ti is not None
+ ti.state = TaskInstanceState.RUNNING
+ ti.next_method = next_method
+ # start_date comes from the worker's clock, which can run ahead of the
API server's.
+ ti.start_date = detail.created_at + timedelta(seconds=2)
+ # The server stamps scheduled_dttm when it resumes a task.
+ ti.scheduled_dttm = None if resumed_offset is None else
detail.created_at + resumed_offset
+ session.commit()
+
+ response =
test_client.patch(f"{sample_ti_url_identifier}/hitlDetails",
json=sample_update_payload)
+
+ assert response.status_code == expected_status
+ session.expire_all()
+ refreshed_detail =
session.scalar(select(HITLDetail).where(HITLDetail.ti_id == sample_ti.id))
+ refreshed_ti = session.get(TIModel, sample_ti.id)
+ assert refreshed_detail is not None
+ assert refreshed_ti is not None
+ assert refreshed_detail.response_received is (expected_status == 200)
+ # The running task picks the answer up itself, so the response never
resumes it.
+ assert refreshed_ti.state == TaskInstanceState.RUNNING
+
@pytest.mark.usefixtures("sample_hitl_detail")
def test_should_respond_400_for_invalid_option(
self,
diff --git
a/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_dags.py
b/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_dags.py
index cc33d43f540..cc968f0b680 100644
--- a/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_dags.py
+++ b/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_dags.py
@@ -16,6 +16,7 @@
# under the License.
from __future__ import annotations
+from datetime import timedelta
from typing import TYPE_CHECKING
from unittest import mock
@@ -475,6 +476,35 @@ class TestGetDagRuns(TestPublicDagEndpoint):
assert without_pending.status_code == 200
assert without_pending.json()["total_entries"] == 3
+ def test_pending_actions_count_a_running_task_only_while_it_waits(
+ self, test_client: TestClient, setup_hitl_data, session: Session
+ ):
+ details =
session.scalars(select(HITLDetail).order_by(HITLDetail.subject)).all()
+ first_run, resumed_after_timeout, resumed_and_asked_again, finished, _
= (
+ detail.task_instance for detail in details
+ )
+ details[3].responded_at = None
+ finished.state = TaskInstanceState.SUCCESS
+ first_run.state = TaskInstanceState.RUNNING
+ # The server stamps scheduled_dttm when it resumes a task.
+ resumed_after_timeout.state = TaskInstanceState.RUNNING
+ resumed_after_timeout.next_method = "execute_complete"
+ resumed_after_timeout.scheduled_dttm = utcnow() + timedelta(minutes=5)
+ resumed_and_asked_again.state = TaskInstanceState.RUNNING
+ resumed_and_asked_again.next_method = "resume_after_tool_approval"
+ resumed_and_asked_again.scheduled_dttm = utcnow() -
timedelta(minutes=5)
+ session.commit()
+
+ response = test_client.get("/dags", params={"has_pending_actions":
True})
+
+ assert response.status_code == 200
+ assert response.json()["total_entries"] == 1
+ pending_actions = [action for dag in response.json()["dags"] for
action in dag["pending_actions"]]
+ assert sorted(action["subject"] for action in pending_actions) == [
+ "This is subject 0",
+ "This is subject 2",
+ ]
+
def test_should_response_401(self, unauthenticated_test_client):
response = unauthenticated_test_client.get("/dags", params={})
assert response.status_code == 401
diff --git
a/airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_hitl.py
b/airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_hitl.py
index c05fd5808f1..67d681df068 100644
---
a/airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_hitl.py
+++
b/airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_hitl.py
@@ -161,6 +161,51 @@ def test_upsert_hitl_detail(
assert hitl_detail.response_received is False
[email protected](
+ ("same_key", "expected_chosen_options"),
+ [
+ pytest.param(True, ["Approve"], id="retry-of-the-stored-request"),
+ pytest.param(False, None, id="new-request"),
+ ],
+)
+def test_upsert_hitl_detail_keeps_the_answer_to_a_retried_request(
+ client: TestClient,
+ create_task_instance: CreateTaskInstance,
+ session: Session,
+ same_key: bool,
+ expected_chosen_options: list[str] | None,
+) -> None:
+ ti = create_task_instance()
+ session.commit()
+ request_json = {"ti_id": str(ti.id), "subject": "Ship it?", "options":
["Approve", "Reject"]}
+ first_key = uuid7()
+ first = client.post(
+ f"/execution/hitlDetails/{ti.id}", json=request_json,
headers={"Idempotency-Key": str(first_key)}
+ )
+ assert first.status_code == 201
+ # A reviewer answers the running task before the client's retry arrives.
+ stored = session.scalars(select(HITLDetail).where(HITLDetail.ti_id ==
ti.id)).one()
+ first_created_at = stored.created_at
+ stored.chosen_options = ["Approve"]
+ stored.responded_at = convert_to_utc(datetime(2025, 7, 3, 0, 0, 0))
+ stored.responded_by = {"id": "1", "name": "reviewer"}
+ session.commit()
+ second_key = first_key if same_key else uuid7()
+
+ second = client.post(
+ f"/execution/hitlDetails/{ti.id}", json=request_json,
headers={"Idempotency-Key": str(second_key)}
+ )
+
+ assert second.status_code == 201
+ session.expire_all()
+ hitl_detail = session.scalars(select(HITLDetail).where(HITLDetail.ti_id ==
ti.id)).one()
+ assert hitl_detail.chosen_options == expected_chosen_options
+ assert hitl_detail.response_received is same_key
+ assert hitl_detail.idempotency_key == second_key
+ # created_at decides whether a resumed run is waiting, so a retry must not
move it.
+ assert (hitl_detail.created_at == first_created_at) is same_key
+
+
def test_upsert_hitl_detail_with_empty_option(
client: TestClient,
create_task_instance: CreateTaskInstance,
diff --git a/airflow-core/tests/unit/dag_processing/test_processor.py
b/airflow-core/tests/unit/dag_processing/test_processor.py
index f56de3f3141..6e6707057bb 100644
--- a/airflow-core/tests/unit/dag_processing/test_processor.py
+++ b/airflow-core/tests/unit/dag_processing/test_processor.py
@@ -2414,6 +2414,7 @@ class TestDagProcessingMessageTypes:
"InactiveAssetsResult",
"CreateHITLDetailPayload",
"HITLDetailRequestResult",
+ "HITLDetailResponseResult",
# AIP-103 task/asset store results — worker-only responses to the
above messages.
"TaskStateStoreResult",
"AssetStateStoreResult",
diff --git a/generated/known_sdk_imports_in_core.txt
b/generated/known_sdk_imports_in_core.txt
index b0019965223..920b73e4478 100644
--- a/generated/known_sdk_imports_in_core.txt
+++ b/generated/known_sdk_imports_in_core.txt
@@ -11,7 +11,7 @@ airflow-core/src/airflow/dag_processing/manager.py::5
airflow-core/src/airflow/dag_processing/processor.py::14
airflow-core/src/airflow/exceptions.py::1
airflow-core/src/airflow/executors/base_executor.py::3
-airflow-core/src/airflow/jobs/triggerer_job_runner.py::19
+airflow-core/src/airflow/jobs/triggerer_job_runner.py::18
airflow-core/src/airflow/models/__init__.py::5
airflow-core/src/airflow/models/asset.py::2
airflow-core/src/airflow/models/connection.py::5
diff --git a/go-sdk/pkg/execution/genmodels/discriminators.gen.go
b/go-sdk/pkg/execution/genmodels/discriminators.gen.go
index 8736a6e2c97..e17b9156241 100644
--- a/go-sdk/pkg/execution/genmodels/discriminators.gen.go
+++ b/go-sdk/pkg/execution/genmodels/discriminators.gen.go
@@ -77,6 +77,7 @@ const (
TypeGetXComSequenceItem = "GetXComSequenceItem"
TypeGetXComSequenceSlice = "GetXComSequenceSlice"
TypeHITLDetailRequestResult = "HITLDetailRequestResult"
+ TypeHITLDetailResponseResult = "HITLDetailResponseResult"
TypeInactiveAssetsResult = "InactiveAssetsResult"
TypeMaskSecret = "MaskSecret"
TypeOKResponse = "OKResponse"
@@ -282,6 +283,9 @@ func EnsureType(m any) any {
case HITLDetailRequestResult:
b.Type = TypeHITLDetailRequestResult
return b
+ case HITLDetailResponseResult:
+ b.Type = TypeHITLDetailResponseResult
+ return b
case InactiveAssetsResult:
b.Type = TypeInactiveAssetsResult
return b
diff --git a/go-sdk/pkg/execution/genmodels/models.gen.go
b/go-sdk/pkg/execution/genmodels/models.gen.go
index 3cefba3e3f9..066e7ebce5c 100644
--- a/go-sdk/pkg/execution/genmodels/models.gen.go
+++ b/go-sdk/pkg/execution/genmodels/models.gen.go
@@ -223,6 +223,8 @@ type BundleInfo struct {
VersionData *VersionData `msgpack:"version_data,omitempty"`
}
+type ChosenOptions []string
+
type ClearAssetStateStoreByName struct {
// Name corresponds to the JSON schema field "name".
Name string `msgpack:"name"`
@@ -1261,6 +1263,27 @@ type HITLDetailRequestResult struct {
Type string `msgpack:"type,omitempty"`
}
+// Response to GetHITLDetailResponse request.
+type HITLDetailResponseResult struct {
+ // ChosenOptions corresponds to the JSON schema field "chosen_options".
+ ChosenOptions *ChosenOptions `msgpack:"chosen_options"`
+
+ // ParamsInput corresponds to the JSON schema field "params_input".
+ ParamsInput *ParamsInput `msgpack:"params_input,omitempty"`
+
+ // RespondedAt corresponds to the JSON schema field "responded_at".
+ RespondedAt interface{} `msgpack:"responded_at"`
+
+ // RespondedByUser corresponds to the JSON schema field
"responded_by_user".
+ RespondedByUser *HITLUser `msgpack:"responded_by_user,omitempty"`
+
+ // ResponseReceived corresponds to the JSON schema field
"response_received".
+ ResponseReceived bool `msgpack:"response_received"`
+
+ // Type corresponds to the JSON schema field "type".
+ Type string `msgpack:"type,omitempty"`
+}
+
// Schema for a Human-in-the-loop users.
type HITLUser struct {
// ID corresponds to the JSON schema field "id".
@@ -1842,24 +1865,41 @@ type TaskInstance struct {
TryNumber int `msgpack:"try_number"`
}
+type Warnings []interface{}
+
+type TriggerKwargs map[string]interface{}
+
type TaskInstanceState string
+type VersionData map[string]interface{}
+
const TaskInstanceStateAwaitingInput TaskInstanceState = "awaiting_input"
const TaskInstanceStateDeferred TaskInstanceState = "deferred"
-const TaskInstanceStateFailed TaskInstanceState = "failed"
-const TaskInstanceStateQueued TaskInstanceState = "queued"
-const TaskInstanceStateRemoved TaskInstanceState = "removed"
-const TaskInstanceStateRestarting TaskInstanceState = "restarting"
-const TaskInstanceStateRunning TaskInstanceState = "running"
-const TaskInstanceStateScheduled TaskInstanceState = "scheduled"
const TaskInstanceStateSkipped TaskInstanceState = "skipped"
-const TaskInstanceStateSuccess TaskInstanceState = "success"
+const TaskInstanceStateUpstreamFailed TaskInstanceState = "upstream_failed"
const TaskInstanceStateUpForReschedule TaskInstanceState = "up_for_reschedule"
const TaskInstanceStateUpForRetry TaskInstanceState = "up_for_retry"
-const TaskInstanceStateUpstreamFailed TaskInstanceState = "upstream_failed"
+const TaskInstanceStateFailed TaskInstanceState = "failed"
+
+const TaskInstanceStateRestarting TaskInstanceState = "restarting"
+const TaskInstanceStateSuccess TaskInstanceState = "success"
+const TaskInstanceStateRunning TaskInstanceState = "running"
+const TaskInstanceStateQueued TaskInstanceState = "queued"
type TaskOutlets []AssetProfile
+const TaskInstanceStateScheduled TaskInstanceState = "scheduled"
+const TaskInstanceStateRemoved TaskInstanceState = "removed"
+
+// Variable schema for responses with fields that are needed for Runtime.
+type VariableResponse struct {
+ // Key corresponds to the JSON schema field "key".
+ Key string `msgpack:"key"`
+
+ // Value corresponds to the JSON schema field "value".
+ Value interface{} `msgpack:"value"`
+}
+
// Response containing the first reschedule date for a task instance.
type TaskRescheduleStartDate struct {
// StartDate corresponds to the JSON schema field "start_date".
@@ -1869,6 +1909,12 @@ type TaskRescheduleStartDate struct {
Type string `msgpack:"type,omitempty"`
}
+type TaskStateState string
+
+const TaskStateStateFailed TaskStateState = "failed"
+const TaskStateStateSkipped TaskStateState = "skipped"
+const TaskStateStateRemoved TaskStateState = "removed"
+
// Update a task's state.
//
// If a process exits without sending one of these the state will be derived
from
@@ -1895,12 +1941,6 @@ type TaskState struct {
Type string `msgpack:"type,omitempty"`
}
-type TaskStateState string
-
-const TaskStateStateFailed TaskStateState = "failed"
-const TaskStateStateRemoved TaskStateState = "removed"
-const TaskStateStateSkipped TaskStateState = "skipped"
-
// Response to GetTaskStateStore; wraps the generated API response for
supervisor
// to worker comms.
type TaskStateStoreResult struct {
@@ -1950,19 +1990,6 @@ type TriggerDagRun struct {
Type string `msgpack:"type,omitempty"`
}
-type TriggerKwargs map[string]interface{}
-
-type Warnings []interface{}
-
-// Variable schema for responses with fields that are needed for Runtime.
-type VariableResponse struct {
- // Key corresponds to the JSON schema field "key".
- Key string `msgpack:"key"`
-
- // Value corresponds to the JSON schema field "value".
- Value interface{} `msgpack:"value"`
-}
-
type UpdateDagRunNote struct {
// Note corresponds to the JSON schema field "note".
Note interface{} `msgpack:"note"`
@@ -1974,8 +2001,6 @@ type UpdateDagRunNote struct {
Type string `msgpack:"type,omitempty"`
}
-type VersionData map[string]interface{}
-
// Update the response content part of an existing Human-in-the-loop response.
type UpdateHITLDetail struct {
// ChosenOptions corresponds to the JSON schema field "chosen_options".
diff --git a/go-sdk/schema/supervisor-schema.json
b/go-sdk/schema/supervisor-schema.json
index 1db287cb17d..eb5e7ad8e8d 100644
--- a/go-sdk/schema/supervisor-schema.json
+++ b/go-sdk/schema/supervisor-schema.json
@@ -3156,6 +3156,78 @@
"title": "HITLDetailRequestResult",
"type": "object"
},
+ "HITLDetailResponseResult": {
+ "description": "Response to GetHITLDetailResponse request.",
+ "properties": {
+ "response_received": {
+ "title": "Response Received",
+ "type": "boolean"
+ },
+ "responded_by_user": {
+ "anyOf": [
+ {
+ "$ref": "#/$defs/HITLUser"
+ },
+ {
+ "type": "null"
+ }
+ ],
+ "default": null
+ },
+ "responded_at": {
+ "anyOf": [
+ {
+ "format": "date-time",
+ "type": "string"
+ },
+ {
+ "type": "null"
+ }
+ ],
+ "title": "Responded At"
+ },
+ "chosen_options": {
+ "anyOf": [
+ {
+ "items": {
+ "type": "string"
+ },
+ "type": "array"
+ },
+ {
+ "type": "null"
+ }
+ ],
+ "title": "Chosen Options"
+ },
+ "params_input": {
+ "anyOf": [
+ {
+ "additionalProperties": true,
+ "type": "object"
+ },
+ {
+ "type": "null"
+ }
+ ],
+ "default": null,
+ "title": "Params Input"
+ },
+ "type": {
+ "const": "HITLDetailResponseResult",
+ "default": "HITLDetailResponseResult",
+ "title": "Type",
+ "type": "string"
+ }
+ },
+ "required": [
+ "response_received",
+ "responded_at",
+ "chosen_options"
+ ],
+ "title": "HITLDetailResponseResult",
+ "type": "object"
+ },
"HITLUser": {
"description": "Schema for a Human-in-the-loop users.",
"properties": {
diff --git a/task-sdk/src/airflow/sdk/api/client.py
b/task-sdk/src/airflow/sdk/api/client.py
index 2638a2a697c..5bc78e64f2c 100644
--- a/task-sdk/src/airflow/sdk/api/client.py
+++ b/task-sdk/src/airflow/sdk/api/client.py
@@ -1119,6 +1119,9 @@ class HITLOperations:
resp = self.client.post(
f"/hitlDetails/{ti_id}",
content=payload.model_dump_json(),
+ # One key per request, reused by the client's retries, so a
retried POST cannot replace
+ # the request (and discard an answer) after the first attempt was
stored.
+ headers={"content-type": "application/json", "Idempotency-Key":
str(uuid.uuid4())},
)
return HITLDetailRequest.model_validate_json(resp.read())
diff --git a/task-sdk/src/airflow/sdk/execution_time/comms.py
b/task-sdk/src/airflow/sdk/execution_time/comms.py
index f0e8336426c..d6ce0112563 100644
--- a/task-sdk/src/airflow/sdk/execution_time/comms.py
+++ b/task-sdk/src/airflow/sdk/execution_time/comms.py
@@ -78,6 +78,7 @@ from airflow.sdk.api.datamodels._generated import (
DagRun,
DagRunStateResponse,
HITLDetailRequest,
+ HITLDetailResponse,
InactiveAssetsResponse,
LoopDecisionPayload,
PreviousTIResponse,
@@ -840,6 +841,17 @@ class HITLDetailRequestResult(HITLDetailRequest):
return cls(**hitl_request.model_dump(exclude_defaults=True),
type="HITLDetailRequestResult")
+class HITLDetailResponseResult(HITLDetailResponse):
+ """Response to GetHITLDetailResponse request."""
+
+ type: Literal["HITLDetailResponseResult"] = "HITLDetailResponseResult"
+
+ @classmethod
+ def from_api_response(cls, response: HITLDetailResponse) ->
HITLDetailResponseResult:
+ """Add the discriminator field the tagged union needs to the
HITLDetailResponse API response."""
+ return cls(**response.model_dump(exclude_defaults=True),
type="HITLDetailResponseResult")
+
+
class DagResult(DagResponse):
type: Literal["DagResult"] = "DagResult"
@@ -884,6 +896,7 @@ ToTask = Annotated[
| InactiveAssetsResult
| CreateHITLDetailPayload
| HITLDetailRequestResult
+ | HITLDetailResponseResult
| OKResponse
| PreviousDagRunResult,
Field(discriminator="type"),
diff --git a/task-sdk/src/airflow/sdk/execution_time/request_handlers.py
b/task-sdk/src/airflow/sdk/execution_time/request_handlers.py
index 2a2d16833df..7141b3b0807 100644
--- a/task-sdk/src/airflow/sdk/execution_time/request_handlers.py
+++ b/task-sdk/src/airflow/sdk/execution_time/request_handlers.py
@@ -57,6 +57,7 @@ from airflow.sdk.execution_time.comms import (
GetConnection,
GetDagRunState,
GetDRCount,
+ GetHITLDetailResponse,
GetPreviousDagRun,
GetPreviousTI,
GetTaskStates,
@@ -67,6 +68,7 @@ from airflow.sdk.execution_time.comms import (
GetXComCount,
GetXComSequenceItem,
GetXComSequenceSlice,
+ HITLDetailResponseResult,
MaskSecret,
PrevSuccessfulDagRunResult,
PutVariable,
@@ -168,6 +170,14 @@ def handle_get_task_states(client: Client, msg:
GetTaskStates) -> tuple[BaseMode
return task_states_map, {}
+def handle_get_hitl_detail_response(
+ client: Client, msg: GetHITLDetailResponse
+) -> tuple[BaseModel | None, dict[str, bool]]:
+ """Fetch the response to a task instance's Human-in-the-loop request."""
+ resp = client.hitl.get_detail_response(ti_id=msg.ti_id)
+ return HITLDetailResponseResult.from_api_response(resp), {}
+
+
def handle_get_previous_ti(client: Client, msg: GetPreviousTI) ->
tuple[BaseModel | None, dict[str, bool]]:
"""Fetch the previous task instance."""
resp = client.task_instances.get_previous(
diff --git a/task-sdk/src/airflow/sdk/execution_time/schema/schema.json
b/task-sdk/src/airflow/sdk/execution_time/schema/schema.json
index 1db287cb17d..eb5e7ad8e8d 100644
--- a/task-sdk/src/airflow/sdk/execution_time/schema/schema.json
+++ b/task-sdk/src/airflow/sdk/execution_time/schema/schema.json
@@ -3156,6 +3156,78 @@
"title": "HITLDetailRequestResult",
"type": "object"
},
+ "HITLDetailResponseResult": {
+ "description": "Response to GetHITLDetailResponse request.",
+ "properties": {
+ "response_received": {
+ "title": "Response Received",
+ "type": "boolean"
+ },
+ "responded_by_user": {
+ "anyOf": [
+ {
+ "$ref": "#/$defs/HITLUser"
+ },
+ {
+ "type": "null"
+ }
+ ],
+ "default": null
+ },
+ "responded_at": {
+ "anyOf": [
+ {
+ "format": "date-time",
+ "type": "string"
+ },
+ {
+ "type": "null"
+ }
+ ],
+ "title": "Responded At"
+ },
+ "chosen_options": {
+ "anyOf": [
+ {
+ "items": {
+ "type": "string"
+ },
+ "type": "array"
+ },
+ {
+ "type": "null"
+ }
+ ],
+ "title": "Chosen Options"
+ },
+ "params_input": {
+ "anyOf": [
+ {
+ "additionalProperties": true,
+ "type": "object"
+ },
+ {
+ "type": "null"
+ }
+ ],
+ "default": null,
+ "title": "Params Input"
+ },
+ "type": {
+ "const": "HITLDetailResponseResult",
+ "default": "HITLDetailResponseResult",
+ "title": "Type",
+ "type": "string"
+ }
+ },
+ "required": [
+ "response_received",
+ "responded_at",
+ "chosen_options"
+ ],
+ "title": "HITLDetailResponseResult",
+ "type": "object"
+ },
"HITLUser": {
"description": "Schema for a Human-in-the-loop users.",
"properties": {
diff --git
a/task-sdk/src/airflow/sdk/execution_time/schema/versions/v2027_02_28.py
b/task-sdk/src/airflow/sdk/execution_time/schema/versions/v2027_02_28.py
index 8a0e320a1e8..b75e46f5aef 100644
--- a/task-sdk/src/airflow/sdk/execution_time/schema/versions/v2027_02_28.py
+++ b/task-sdk/src/airflow/sdk/execution_time/schema/versions/v2027_02_28.py
@@ -15,7 +15,14 @@
# specific language governing permissions and limitations
# under the License.
-"""Supervisor schema version 2027-02-28."""
+"""
+Supervisor schema version 2027-02-28.
+
+New bodies in this version:
+
+- ``HITLDetailResponseResult``: the supervisor's reply when a running task
reads its own
+ Human-in-the-loop response.
+"""
from __future__ import annotations
diff --git a/task-sdk/src/airflow/sdk/execution_time/supervisor.py
b/task-sdk/src/airflow/sdk/execution_time/supervisor.py
index c2ba367faef..38c14782351 100644
--- a/task-sdk/src/airflow/sdk/execution_time/supervisor.py
+++ b/task-sdk/src/airflow/sdk/execution_time/supervisor.py
@@ -105,6 +105,7 @@ from airflow.sdk.execution_time.comms import (
GetDagRun,
GetDagRunState,
GetDRCount,
+ GetHITLDetailResponse,
GetPreviousDagRun,
GetPreviousTI,
GetPrevSuccessfulDagRun,
@@ -155,6 +156,7 @@ from airflow.sdk.execution_time.request_handlers import (
handle_get_connection,
handle_get_dag_run_state,
handle_get_dr_count,
+ handle_get_hitl_detail_response,
handle_get_prev_successful_dag_run,
handle_get_previous_dag_run,
handle_get_previous_ti,
@@ -827,6 +829,7 @@ class WatchedSubprocess:
_register_client_handler(GetConnection, handle_get_connection),
_register_client_handler(GetDagRunState, handle_get_dag_run_state),
_register_client_handler(GetDRCount, handle_get_dr_count),
+ _register_client_handler(GetHITLDetailResponse,
handle_get_hitl_detail_response),
_register_client_handler(GetPreviousDagRun,
handle_get_previous_dag_run),
_register_client_handler(GetPreviousTI, handle_get_previous_ti),
_register_client_handler(GetTaskStates, handle_get_task_states),
@@ -2431,6 +2434,7 @@ class ActivitySubprocess(WatchedSubprocess):
GetConnection,
GetDRCount,
GetDagRunState,
+ GetHITLDetailResponse,
GetPrevSuccessfulDagRun,
GetPreviousDagRun,
GetPreviousTI,
diff --git a/task-sdk/tests/task_sdk/api/test_client.py
b/task-sdk/tests/task_sdk/api/test_client.py
index a0bd43dffbb..c9b862fb336 100644
--- a/task-sdk/tests/task_sdk/api/test_client.py
+++ b/task-sdk/tests/task_sdk/api/test_client.py
@@ -2011,6 +2011,27 @@ class TestHITLOperations:
assert result.multiple is False
assert result.assigned_users is None
+ def test_add_response_retries_with_the_same_idempotency_key(self) -> None:
+ ti_id = uuid7()
+ keys: list[str] = []
+
+ def handle_request(request: httpx2.Request) -> httpx2.Response:
+ keys.append(request.headers["Idempotency-Key"])
+ if len(keys) == 1:
+ return httpx2.Response(status_code=500, json={"detail":
"Internal Server Error"})
+ return httpx2.Response(
+ status_code=201, json={"ti_id": str(ti_id), "options":
["Approve"], "subject": "Ship it?"}
+ )
+
+ client = make_client(transport=httpx2.MockTransport(handle_request))
+ with time_machine.travel("2023-01-01T00:00:00Z", tick=False):
+ client.hitl.add_response(ti_id=ti_id, options=["Approve"],
subject="Ship it?")
+ client.hitl.add_response(ti_id=ti_id, options=["Approve"],
subject="Ship it?")
+
+ first_attempt, retry, next_request = keys
+ assert retry == first_attempt
+ assert next_request != first_attempt
+
def test_update_response(self, time_machine: TimeMachineFixture) -> None:
time_machine.move_to(datetime(2025, 7, 3, 0, 0, 0))
ti_id = uuid7()
diff --git a/task-sdk/tests/task_sdk/execution_time/test_supervisor.py
b/task-sdk/tests/task_sdk/execution_time/test_supervisor.py
index bf20ba7e111..d2381dc6046 100644
--- a/task-sdk/tests/task_sdk/execution_time/test_supervisor.py
+++ b/task-sdk/tests/task_sdk/execution_time/test_supervisor.py
@@ -68,6 +68,7 @@ from airflow.sdk.api.datamodels._generated import (
DagRun,
DagRunState,
DagRunType,
+ HITLUser,
LoopDecisionPayload,
PreviousTIResponse,
TaskInstance,
@@ -128,6 +129,7 @@ from airflow.sdk.execution_time.comms import (
GetXComSequenceItem,
GetXComSequenceSlice,
HITLDetailRequestResult,
+ HITLDetailResponseResult,
InactiveAssetsResult,
MaskSecret,
OKResponse,
@@ -3818,6 +3820,29 @@ REQUEST_TEST_CASES = [
),
test_id="create_hitl_detail_payload",
),
+ RequestTestCase(
+ message=GetHITLDetailResponse(ti_id=TI_ID),
+ expected_body={
+ "response_received": True,
+ "responded_by_user": {"id": "alice", "name": "Alice"},
+ "responded_at": timezone.parse("2024-10-31T12:00:00Z"),
+ "chosen_options": ["Approve"],
+ "params_input": {"reason": "ship it"},
+ "type": "HITLDetailResponseResult",
+ },
+ client_mock=ClientMock(
+ method_path="hitl.get_detail_response",
+ kwargs={"ti_id": TI_ID},
+ response=HITLDetailResponseResult(
+ response_received=True,
+ responded_by_user=HITLUser(id="alice", name="Alice"),
+ responded_at=timezone.parse("2024-10-31T12:00:00Z"),
+ chosen_options=["Approve"],
+ params_input={"reason": "ship it"},
+ ),
+ ),
+ test_id="get_hitl_detail_response",
+ ),
RequestTestCase(
message=MaskSecret(value=["iter1", "iter2", {"key": "value"}],
name="test_secret"),
mask_secret_args=(["iter1", "iter2", {"key": "value"}], "test_secret"),
@@ -4403,16 +4428,11 @@ class TestHandleRequest:
)
@patch.object(ActivitySubprocess, "send_msg", autospec=True)
- @pytest.mark.parametrize(
- "message",
- [
- GetHITLDetailResponse(ti_id=TI_ID),
- UpdateHITLDetail(ti_id=TI_ID, chosen_options=["approved"]),
- ],
- )
- def test_rejects_unsupported_task_messages(self, send_msg,
watched_subprocess, message):
+ def test_rejects_unsupported_task_messages(self, send_msg,
watched_subprocess):
process, _ = watched_subprocess
- process._handle_request(message, structlog.get_logger(), req_id=42)
+ process._handle_request(
+ UpdateHITLDetail(ti_id=TI_ID, chosen_options=["approved"]),
structlog.get_logger(), req_id=42
+ )
send_msg.assert_called_once_with(
process,
@@ -4536,7 +4556,7 @@ class TestHandleRequest:
assert decoder.validate_python(frame.body) == client_mock.response
def test_registered_message_types(self):
- expected = set(get_args(get_args(ToSupervisor)[0])) -
{GetHITLDetailResponse, UpdateHITLDetail}
+ expected = set(get_args(get_args(ToSupervisor)[0])) -
{UpdateHITLDetail}
assert set(ActivitySubprocess._request_handlers) == expected
assert {type(case.message) for case in REQUEST_TEST_CASES} == expected
diff --git a/ts-sdk/schema/supervisor-schema.json
b/ts-sdk/schema/supervisor-schema.json
index 1db287cb17d..eb5e7ad8e8d 100644
--- a/ts-sdk/schema/supervisor-schema.json
+++ b/ts-sdk/schema/supervisor-schema.json
@@ -3156,6 +3156,78 @@
"title": "HITLDetailRequestResult",
"type": "object"
},
+ "HITLDetailResponseResult": {
+ "description": "Response to GetHITLDetailResponse request.",
+ "properties": {
+ "response_received": {
+ "title": "Response Received",
+ "type": "boolean"
+ },
+ "responded_by_user": {
+ "anyOf": [
+ {
+ "$ref": "#/$defs/HITLUser"
+ },
+ {
+ "type": "null"
+ }
+ ],
+ "default": null
+ },
+ "responded_at": {
+ "anyOf": [
+ {
+ "format": "date-time",
+ "type": "string"
+ },
+ {
+ "type": "null"
+ }
+ ],
+ "title": "Responded At"
+ },
+ "chosen_options": {
+ "anyOf": [
+ {
+ "items": {
+ "type": "string"
+ },
+ "type": "array"
+ },
+ {
+ "type": "null"
+ }
+ ],
+ "title": "Chosen Options"
+ },
+ "params_input": {
+ "anyOf": [
+ {
+ "additionalProperties": true,
+ "type": "object"
+ },
+ {
+ "type": "null"
+ }
+ ],
+ "default": null,
+ "title": "Params Input"
+ },
+ "type": {
+ "const": "HITLDetailResponseResult",
+ "default": "HITLDetailResponseResult",
+ "title": "Type",
+ "type": "string"
+ }
+ },
+ "required": [
+ "response_received",
+ "responded_at",
+ "chosen_options"
+ ],
+ "title": "HITLDetailResponseResult",
+ "type": "object"
+ },
"HITLUser": {
"description": "Schema for a Human-in-the-loop users.",
"properties": {
diff --git a/ts-sdk/src/generated/supervisor.ts
b/ts-sdk/src/generated/supervisor.ts
index 8cc93eb7bed..0c9da591eaf 100644
--- a/ts-sdk/src/generated/supervisor.ts
+++ b/ts-sdk/src/generated/supervisor.ts
@@ -533,20 +533,27 @@ export type Params1 = {
} | null;
export type AssignedUsers1 = HITLUser[] | null;
export type Type54 = "HITLDetailRequestResult";
+export type ResponseReceived = boolean;
+export type RespondedAt = string | null;
+export type ChosenOptions = string[] | null;
+export type ParamsInput = {
+ [k: string]: unknown;
+} | null;
+export type Type55 = "HITLDetailResponseResult";
export type InactiveAssets = AssetProfile[] | null;
-export type Type55 = "InactiveAssetsResult";
+export type Type56 = "InactiveAssetsResult";
export type Stop1 = boolean;
export type Reason = string | null;
export type Name14 = string | null;
-export type Type56 = "MaskSecret";
+export type Type57 = "MaskSecret";
export type Ok = boolean;
-export type Type57 = "OKResponse";
+export type Type58 = "OKResponse";
export type DataIntervalStart3 = string | null;
export type DataIntervalEnd3 = string | null;
export type StartDate4 = string | null;
export type EndDate3 = string | null;
-export type Type58 = "PrevSuccessfulDagRunResult";
-export type Type59 = "PreviousDagRunResult";
+export type Type59 = "PrevSuccessfulDagRunResult";
+export type Type60 = "PreviousDagRunResult";
export type TaskId8 = string;
export type DagId20 = string;
export type RunId13 = string;
@@ -559,39 +566,39 @@ export type MapIndex6 = number | null;
export type RegionId2 = string | null;
export type RegionIndex1 = number | null;
export type Duration = number | null;
-export type Type60 = "PreviousTIResult";
+export type Type61 = "PreviousTIResult";
export type Key14 = string;
export type Value1 = string | null;
export type Description = string | null;
-export type Type61 = "PutVariable";
+export type Type62 = "PutVariable";
export type State5 = "up_for_reschedule";
export type RescheduleDate = string;
export type EndDate5 = string;
-export type Type62 = "RescheduleTask";
-export type Type63 = "ResendLoggingFD";
+export type Type63 = "RescheduleTask";
+export type Type64 = "ResendLoggingFD";
export type State6 = "up_for_retry";
export type EndDate6 = string;
export type RenderedMapIndex2 = string | null;
export type RetryDelaySeconds = number | null;
export type RetryReason = string | null;
-export type Type64 = "RetryTask";
-export type Type65 = "SentFDs";
+export type Type65 = "RetryTask";
+export type Type66 = "SentFDs";
export type Fds = number[];
export type Name15 = string;
export type Key15 = string;
-export type Type66 = "SetAssetStateStoreByName";
+export type Type67 = "SetAssetStateStoreByName";
export type Uri9 = string;
export type Key16 = string;
-export type Type67 = "SetAssetStateStoreByUri";
+export type Type68 = "SetAssetStateStoreByUri";
export type TimeoutSeconds = number;
-export type Type68 = "SetExecutionTimeout";
-export type Type69 = "SetRenderedFields";
+export type Type69 = "SetExecutionTimeout";
+export type Type70 = "SetRenderedFields";
export type RenderedMapIndex3 = string;
-export type Type70 = "SetRenderedMapIndex";
+export type Type71 = "SetRenderedMapIndex";
export type TiId8 = string;
export type Key17 = string;
export type ExpiresAt = string | null;
-export type Type71 = "SetTaskStateStore";
+export type Type72 = "SetTaskStateStore";
export type Key18 = string;
export type DagId21 = string;
export type RunId14 = string;
@@ -599,13 +606,13 @@ export type TaskId9 = string;
export type MapIndex7 = number | null;
export type DagResult1 = boolean;
export type MappedLength = number | null;
-export type Type72 = "SetXCom";
+export type Type73 = "SetXCom";
export type Tasks = (string | [unknown, unknown])[];
-export type Type73 = "SkipDownstreamTasks";
+export type Type74 = "SkipDownstreamTasks";
export type DagRelPath = string;
export type StartDate6 = string;
export type SentryIntegration = string;
-export type Type74 = "StartupDetails";
+export type Type75 = "StartupDetails";
export type State7 = "success";
export type EndDate7 = string;
export type TaskOutlets = AssetProfile[] | null;
@@ -615,22 +622,22 @@ export type OutletEvents =
}[]
| null;
export type RenderedMapIndex4 = string | null;
-export type Type75 = "SucceedTask";
+export type Type76 = "SucceedTask";
export type Count1 = number;
-export type Type76 = "TICount";
+export type Type77 = "TICount";
export type Breadcrumbs = {
[k: string]: unknown;
}[];
-export type Type77 = "TaskBreadcrumbsResult";
+export type Type78 = "TaskBreadcrumbsResult";
export type StartDate7 = string | null;
-export type Type78 = "TaskRescheduleStartDate";
+export type Type79 = "TaskRescheduleStartDate";
export type State8 = "failed" | "skipped" | "removed";
export type EndDate8 = string | null;
-export type Type79 = "TaskState";
+export type Type80 = "TaskState";
export type RenderedMapIndex5 = string | null;
export type RetryReason1 = string | null;
-export type Type80 = "TaskStateStoreResult";
-export type Type81 = "TaskStatesResult";
+export type Type81 = "TaskStateStoreResult";
+export type Type82 = "TaskStatesResult";
export type LogicalDate6 = string | null;
export type RunAfter2 = string | null;
export type Conf2 = {
@@ -641,34 +648,34 @@ export type PartitionKey7 = string | null;
export type Note2 = string | null;
export type DagId22 = string;
export type DagRunId = string;
-export type Type82 = "TriggerDagRun";
+export type Type83 = "TriggerDagRun";
export type TiId9 = string;
export type Note3 = string | null;
-export type Type83 = "UpdateDagRunNote";
+export type Type84 = "UpdateDagRunNote";
export type TiId10 = string;
/**
* @minItems 1
*/
-export type ChosenOptions = [string, ...string[]];
-export type ParamsInput = {
+export type ChosenOptions1 = [string, ...string[]];
+export type ParamsInput1 = {
[k: string]: unknown;
} | null;
-export type Type84 = "UpdateHITLDetail";
+export type Type85 = "UpdateHITLDetail";
export type TiId11 = string;
-export type Type85 = "ValidateInletsAndOutlets";
+export type Type86 = "ValidateInletsAndOutlets";
export type Keys = string[];
export type TotalEntries = number;
-export type Type86 = "VariableKeysResult";
+export type Type87 = "VariableKeysResult";
export type Key19 = string;
export type Value2 = string | null;
-export type Type87 = "VariableResult";
+export type Type88 = "VariableResult";
export type Len = number;
-export type Type88 = "XComCountResponse";
+export type Type89 = "XComCountResponse";
export type Key20 = string;
-export type Type89 = "XComResult";
-export type Type90 = "XComSequenceIndexResult";
+export type Type90 = "XComResult";
+export type Type91 = "XComSequenceIndexResult";
export type Root = JsonValue[];
-export type Type91 = "XComSequenceSliceResult";
+export type Type92 = "XComSequenceSliceResult";
export interface SupervisorWireSchema {}
/**
@@ -1625,6 +1632,20 @@ export interface HITLDetailRequestResult {
assigned_users?: AssignedUsers1;
type?: Type54;
}
+/**
+ * Response to GetHITLDetailResponse request.
+ *
+ * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema
+ * via the `definition` "HITLDetailResponseResult".
+ */
+export interface HITLDetailResponseResult {
+ response_received: ResponseReceived;
+ responded_by_user?: HITLUser | null;
+ responded_at: RespondedAt;
+ chosen_options: ChosenOptions;
+ params_input?: ParamsInput;
+ type?: Type55;
+}
/**
* Response of InactiveAssets requests.
*
@@ -1633,7 +1654,7 @@ export interface HITLDetailRequestResult {
*/
export interface InactiveAssetsResult {
inactive_assets?: InactiveAssets;
- type?: Type55;
+ type?: Type56;
}
/**
* A loop gate's evaluation of its ``until`` callable.
@@ -1654,7 +1675,7 @@ export interface LoopDecisionPayload {
export interface MaskSecret {
value: JsonValue;
name?: Name14;
- type?: Type56;
+ type?: Type57;
}
/**
* This interface was referenced by `SupervisorWireSchema`'s JSON-Schema
@@ -1662,7 +1683,7 @@ export interface MaskSecret {
*/
export interface OKResponse {
ok: Ok;
- type?: Type57;
+ type?: Type58;
}
/**
* This interface was referenced by `SupervisorWireSchema`'s JSON-Schema
@@ -1673,7 +1694,7 @@ export interface PrevSuccessfulDagRunResult {
data_interval_end?: DataIntervalEnd3;
start_date?: StartDate4;
end_date?: EndDate3;
- type?: Type58;
+ type?: Type59;
}
/**
* Response containing previous Dag run information.
@@ -1683,7 +1704,7 @@ export interface PrevSuccessfulDagRunResult {
*/
export interface PreviousDagRunResult {
dag_run?: DagRun | null;
- type?: Type59;
+ type?: Type60;
}
/**
* Schema for response with previous TaskInstance information.
@@ -1713,7 +1734,7 @@ export interface PreviousTIResponse {
*/
export interface PreviousTIResult {
task_instance?: PreviousTIResponse | null;
- type?: Type60;
+ type?: Type61;
}
/**
* This interface was referenced by `SupervisorWireSchema`'s JSON-Schema
@@ -1723,7 +1744,7 @@ export interface PutVariable {
key: Key14;
value: Value1;
description: Description;
- type?: Type61;
+ type?: Type62;
}
/**
* Update a task instance state to reschedule/up_for_reschedule.
@@ -1735,14 +1756,14 @@ export interface RescheduleTask {
state?: State5;
reschedule_date: RescheduleDate;
end_date: EndDate5;
- type?: Type62;
+ type?: Type63;
}
/**
* This interface was referenced by `SupervisorWireSchema`'s JSON-Schema
* via the `definition` "ResendLoggingFD".
*/
export interface ResendLoggingFD {
- type?: Type63;
+ type?: Type64;
}
/**
* Update a task instance state to up_for_retry.
@@ -1756,14 +1777,14 @@ export interface RetryTask {
rendered_map_index?: RenderedMapIndex2;
retry_delay_seconds?: RetryDelaySeconds;
retry_reason?: RetryReason;
- type?: Type64;
+ type?: Type65;
}
/**
* This interface was referenced by `SupervisorWireSchema`'s JSON-Schema
* via the `definition` "SentFDs".
*/
export interface SentFDs {
- type?: Type65;
+ type?: Type66;
fds: Fds;
}
/**
@@ -1774,7 +1795,7 @@ export interface SetAssetStateStoreByName {
name: Name15;
key: Key15;
value: JsonValue;
- type?: Type66;
+ type?: Type67;
}
/**
* This interface was referenced by `SupervisorWireSchema`'s JSON-Schema
@@ -1784,7 +1805,7 @@ export interface SetAssetStateStoreByUri {
uri: Uri9;
key: Key16;
value: JsonValue;
- type?: Type67;
+ type?: Type68;
}
/**
* Tell the supervisor the task's ``execution_timeout`` so it can enforce it
from outside the task process.
@@ -1794,7 +1815,7 @@ export interface SetAssetStateStoreByUri {
*/
export interface SetExecutionTimeout {
timeout_seconds: TimeoutSeconds;
- type?: Type68;
+ type?: Type69;
}
/**
* Payload for setting RTIF for a task instance.
@@ -1804,7 +1825,7 @@ export interface SetExecutionTimeout {
*/
export interface SetRenderedFields {
rendered_fields: RenderedFields;
- type?: Type69;
+ type?: Type70;
}
export interface RenderedFields {
[k: string]: JsonValue;
@@ -1817,7 +1838,7 @@ export interface RenderedFields {
*/
export interface SetRenderedMapIndex {
rendered_map_index: RenderedMapIndex3;
- type?: Type70;
+ type?: Type71;
}
/**
* This interface was referenced by `SupervisorWireSchema`'s JSON-Schema
@@ -1828,7 +1849,7 @@ export interface SetTaskStateStore {
key: Key17;
value: JsonValue;
expires_at: ExpiresAt;
- type?: Type71;
+ type?: Type72;
}
/**
* This interface was referenced by `SupervisorWireSchema`'s JSON-Schema
@@ -1843,7 +1864,7 @@ export interface SetXCom {
map_index?: MapIndex7;
dag_result?: DagResult1;
mapped_length?: MappedLength;
- type?: Type72;
+ type?: Type73;
}
/**
* Update state of downstream tasks within a task instance to 'skipped', while
updating current task to success state.
@@ -1853,7 +1874,7 @@ export interface SetXCom {
*/
export interface SkipDownstreamTasks {
tasks: Tasks;
- type?: Type73;
+ type?: Type74;
}
/**
* This interface was referenced by `SupervisorWireSchema`'s JSON-Schema
@@ -1866,7 +1887,7 @@ export interface StartupDetails {
start_date: StartDate6;
ti_context: TIRunContext;
sentry_integration: SentryIntegration;
- type?: Type74;
+ type?: Type75;
}
/**
* Update a task's state to success. Includes task_outlets and outlet_events
for registering asset events.
@@ -1881,7 +1902,7 @@ export interface SucceedTask {
outlet_events?: OutletEvents;
rendered_map_index?: RenderedMapIndex4;
loop_decision?: LoopDecisionPayload | null;
- type?: Type75;
+ type?: Type76;
}
/**
* Response containing count of Task Instances matching certain filters.
@@ -1891,7 +1912,7 @@ export interface SucceedTask {
*/
export interface TICount {
count: Count1;
- type?: Type76;
+ type?: Type77;
}
/**
* This interface was referenced by `SupervisorWireSchema`'s JSON-Schema
@@ -1899,7 +1920,7 @@ export interface TICount {
*/
export interface TaskBreadcrumbsResult {
breadcrumbs: Breadcrumbs;
- type?: Type77;
+ type?: Type78;
}
/**
* Response containing the first reschedule date for a task instance.
@@ -1909,7 +1930,7 @@ export interface TaskBreadcrumbsResult {
*/
export interface TaskRescheduleStartDate {
start_date: StartDate7;
- type?: Type78;
+ type?: Type79;
}
/**
* Update a task's state.
@@ -1924,7 +1945,7 @@ export interface TaskRescheduleStartDate {
export interface TaskState {
state: State8;
end_date?: EndDate8;
- type?: Type79;
+ type?: Type80;
rendered_map_index?: RenderedMapIndex5;
retry_reason?: RetryReason1;
loop_decision?: LoopDecisionPayload | null;
@@ -1937,7 +1958,7 @@ export interface TaskState {
*/
export interface TaskStateStoreResult {
value: JsonValue | null;
- type?: Type80;
+ type?: Type81;
}
/**
* This interface was referenced by `SupervisorWireSchema`'s JSON-Schema
@@ -1945,7 +1966,7 @@ export interface TaskStateStoreResult {
*/
export interface TaskStatesResult {
task_states: TaskStates;
- type?: Type81;
+ type?: Type82;
}
export interface TaskStates {
[k: string]: unknown;
@@ -1963,7 +1984,7 @@ export interface TriggerDagRun {
note?: Note2;
dag_id: DagId22;
run_id: DagRunId;
- type?: Type82;
+ type?: Type83;
}
/**
* This interface was referenced by `SupervisorWireSchema`'s JSON-Schema
@@ -1972,7 +1993,7 @@ export interface TriggerDagRun {
export interface UpdateDagRunNote {
ti_id: TiId9;
note: Note3;
- type?: Type83;
+ type?: Type84;
}
/**
* Update the response content part of an existing Human-in-the-loop response.
@@ -1982,9 +2003,9 @@ export interface UpdateDagRunNote {
*/
export interface UpdateHITLDetail {
ti_id: TiId10;
- chosen_options: ChosenOptions;
- params_input?: ParamsInput;
- type?: Type84;
+ chosen_options: ChosenOptions1;
+ params_input?: ParamsInput1;
+ type?: Type85;
}
/**
* This interface was referenced by `SupervisorWireSchema`'s JSON-Schema
@@ -1992,7 +2013,7 @@ export interface UpdateHITLDetail {
*/
export interface ValidateInletsAndOutlets {
ti_id: TiId11;
- type?: Type85;
+ type?: Type86;
}
/**
* This interface was referenced by `SupervisorWireSchema`'s JSON-Schema
@@ -2001,7 +2022,7 @@ export interface ValidateInletsAndOutlets {
export interface VariableKeysResult {
keys: Keys;
total_entries: TotalEntries;
- type?: Type86;
+ type?: Type87;
}
/**
* This interface was referenced by `SupervisorWireSchema`'s JSON-Schema
@@ -2010,7 +2031,7 @@ export interface VariableKeysResult {
export interface VariableResult {
key: Key19;
value: Value2;
- type?: Type87;
+ type?: Type88;
}
/**
* This interface was referenced by `SupervisorWireSchema`'s JSON-Schema
@@ -2018,7 +2039,7 @@ export interface VariableResult {
*/
export interface XComCountResponse {
len: Len;
- type?: Type88;
+ type?: Type89;
}
/**
* Response to ReadXCom request.
@@ -2029,7 +2050,7 @@ export interface XComCountResponse {
export interface XComResult {
key: Key20;
value: JsonValue | null;
- type?: Type89;
+ type?: Type90;
}
/**
* This interface was referenced by `SupervisorWireSchema`'s JSON-Schema
@@ -2037,7 +2058,7 @@ export interface XComResult {
*/
export interface XComSequenceIndexResult {
root: JsonValue;
- type?: Type90;
+ type?: Type91;
}
/**
* This interface was referenced by `SupervisorWireSchema`'s JSON-Schema
@@ -2045,7 +2066,7 @@ export interface XComSequenceIndexResult {
*/
export interface XComSequenceSliceResult {
root: Root;
- type?: Type91;
+ type?: Type92;
}
/** Cadwyn schema version this SDK was generated against.
