This is an automated email from the ASF dual-hosted git repository.
shahar1 pushed a commit to branch airflow-ctl/v0-1-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/airflow-ctl/v0-1-test by this
push:
new 5617f7fadbb Accept the awaiting_input task instance state from Airflow
3.3+ servers (#76118)
5617f7fadbb is described below
commit 5617f7fadbba0455e21ecfe76146e0ae253f4113
Author: Shahar Epstein <[email protected]>
AuthorDate: Sat Oct 31 09:28:31 2026 +0200
Accept the awaiting_input task instance state from Airflow 3.3+ servers
(#76118)
Airflow 3.3.0+ servers emit the task-instance state awaiting_input for
Human-in-the-loop tasks (upstream #68028, b64a0e636d9). airflowctl's
TaskInstanceState lacked the value, so taskinstances list, tasks
states-for-dag-run and the tasks clear preview crashed with a pydantic
ValidationError whenever such a task instance existed on a 3.3.x or 3.4.0rc1
server. 3.2.x servers never emit the value, so the client is a strict superset
there.
Only the enum value and what the branch's generators derive from it are
backported, following 290f0226179: core utils/state.py (IntermediateTIState and
TaskInstanceState, needed so the generate-openapi-spec output and the
test_state union invariant stay consistent), both OpenAPI yamls, the UI
openapi-gen ts files and the task-sdk datamodels. Left out on purpose because
airflow-core, task-sdk and the UI are not released from this branch: the
State.AWAITING_INPUT alias, the state_color [...]
Widening IntermediateTIState also widens the execution API's
TITargetStatePayload.state and the task-sdk model without a Cadwyn version
bump, as upstream did; this has no runtime effect on this branch.
---
.../airflow/api_fastapi/core_api/openapi/_private_ui.yaml | 1 +
.../api_fastapi/core_api/openapi/v2-rest-api-generated.yaml | 1 +
.../src/airflow/ui/openapi-gen/requests/schemas.gen.ts | 2 +-
.../src/airflow/ui/openapi-gen/requests/types.gen.ts | 2 +-
airflow-core/src/airflow/utils/state.py | 2 ++
airflow-ctl/src/airflowctl/api/datamodels/generated.py | 1 +
airflow-ctl/tests/airflow_ctl/api/test_operations.py | 12 ++++++++++++
task-sdk/src/airflow/sdk/api/datamodels/_generated.py | 2 ++
8 files changed, 21 insertions(+), 2 deletions(-)
diff --git
a/airflow-core/src/airflow/api_fastapi/core_api/openapi/_private_ui.yaml
b/airflow-core/src/airflow/api_fastapi/core_api/openapi/_private_ui.yaml
index 598b5e94ffb..8d3bba67267 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/openapi/_private_ui.yaml
+++ b/airflow-core/src/airflow/api_fastapi/core_api/openapi/_private_ui.yaml
@@ -3183,6 +3183,7 @@ components:
- upstream_failed
- skipped
- deferred
+ - awaiting_input
title: TaskInstanceState
description: 'All possible states that a Task Instance can be in.
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 f0875f305bb..c4cd87f3a08 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
@@ -12918,6 +12918,7 @@ components:
- upstream_failed
- skipped
- deferred
+ - awaiting_input
title: TaskInstanceState
description: 'All possible states that a Task Instance can be in.
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 0c7ee2e5f1d..87243cf7363 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
@@ -5845,7 +5845,7 @@ export const $TaskInstanceResponse = {
export const $TaskInstanceState = {
type: 'string',
- enum: ['removed', 'scheduled', 'queued', 'running', 'success',
'restarting', 'failed', 'up_for_retry', 'up_for_reschedule', 'upstream_failed',
'skipped', 'deferred'],
+ enum: ['removed', 'scheduled', 'queued', 'running', 'success',
'restarting', 'failed', 'up_for_retry', 'up_for_reschedule', 'upstream_failed',
'skipped', 'deferred', 'awaiting_input'],
title: 'TaskInstanceState',
description: `All possible states that a Task Instance can be in.
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 2feb9ab4566..53cb0e415f8 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
@@ -1468,7 +1468,7 @@ export type TaskInstanceResponse = {
*
* Note that None is also allowed, so always use this in a type hint with
Optional.
*/
-export type TaskInstanceState = 'removed' | 'scheduled' | 'queued' | 'running'
| 'success' | 'restarting' | 'failed' | 'up_for_retry' | 'up_for_reschedule' |
'upstream_failed' | 'skipped' | 'deferred';
+export type TaskInstanceState = 'removed' | 'scheduled' | 'queued' | 'running'
| 'success' | 'restarting' | 'failed' | 'up_for_retry' | 'up_for_reschedule' |
'upstream_failed' | 'skipped' | 'deferred' | 'awaiting_input';
/**
* Task Instance body for get batch.
diff --git a/airflow-core/src/airflow/utils/state.py
b/airflow-core/src/airflow/utils/state.py
index 332efb10553..ace056c333f 100644
--- a/airflow-core/src/airflow/utils/state.py
+++ b/airflow-core/src/airflow/utils/state.py
@@ -56,6 +56,7 @@ class IntermediateTIState(str, Enum):
UP_FOR_RETRY = "up_for_retry"
UP_FOR_RESCHEDULE = "up_for_reschedule"
DEFERRED = "deferred"
+ AWAITING_INPUT = "awaiting_input"
def __str__(self) -> str:
return self.value
@@ -87,6 +88,7 @@ class TaskInstanceState(str, Enum):
UPSTREAM_FAILED = TerminalTIState.UPSTREAM_FAILED # One or more upstream
deps failed
SKIPPED = TerminalTIState.SKIPPED # Skipped by branching or some other
mechanism
DEFERRED = IntermediateTIState.DEFERRED # Deferrable operator waiting on
a trigger
+ AWAITING_INPUT = IntermediateTIState.AWAITING_INPUT # Parked waiting for
human input (HITL)
def __str__(self) -> str:
return self.value
diff --git a/airflow-ctl/src/airflowctl/api/datamodels/generated.py
b/airflow-ctl/src/airflowctl/api/datamodels/generated.py
index b8f1e473dad..e0ed5f45c03 100644
--- a/airflow-ctl/src/airflowctl/api/datamodels/generated.py
+++ b/airflow-ctl/src/airflowctl/api/datamodels/generated.py
@@ -809,6 +809,7 @@ class TaskInstanceState(str, Enum):
UPSTREAM_FAILED = "upstream_failed"
SKIPPED = "skipped"
DEFERRED = "deferred"
+ AWAITING_INPUT = "awaiting_input"
class TaskInstancesBatchBody(BaseModel):
diff --git a/airflow-ctl/tests/airflow_ctl/api/test_operations.py
b/airflow-ctl/tests/airflow_ctl/api/test_operations.py
index 9e510c55f63..ad701bec997 100644
--- a/airflow-ctl/tests/airflow_ctl/api/test_operations.py
+++ b/airflow-ctl/tests/airflow_ctl/api/test_operations.py
@@ -1809,6 +1809,18 @@ class TestTaskInstancesOperations:
response = client.task_instances.list(dag_id="dag_id",
dag_run_id="dag_run_id")
assert response == self.task_instance_collection_response
+ def test_list_accepts_awaiting_input_state(self):
+ """Servers on Airflow 3.3+ report Human-in-the-loop task instances as
``awaiting_input``."""
+ payload =
json.loads(self.task_instance_collection_response.model_dump_json())
+ payload["task_instances"][0]["state"] = "awaiting_input"
+
+ def handle_request(request: httpx.Request) -> httpx.Response:
+ return httpx.Response(200, json=payload)
+
+ client = make_api_client(transport=httpx.MockTransport(handle_request))
+ response = client.task_instances.list(dag_id="dag_id",
dag_run_id="dag_run_id")
+ assert response.task_instances[0].state ==
TaskInstanceState.AWAITING_INPUT
+
class TestTasksOperations:
dag_id = "dag_id"
diff --git a/task-sdk/src/airflow/sdk/api/datamodels/_generated.py
b/task-sdk/src/airflow/sdk/api/datamodels/_generated.py
index f0df33dda86..8573336e862 100644
--- a/task-sdk/src/airflow/sdk/api/datamodels/_generated.py
+++ b/task-sdk/src/airflow/sdk/api/datamodels/_generated.py
@@ -176,6 +176,7 @@ class IntermediateTIState(str, Enum):
UP_FOR_RETRY = "up_for_retry"
UP_FOR_RESCHEDULE = "up_for_reschedule"
DEFERRED = "deferred"
+ AWAITING_INPUT = "awaiting_input"
class PrevSuccessfulDagRunResponse(BaseModel):
@@ -341,6 +342,7 @@ class TaskInstanceState(str, Enum):
UPSTREAM_FAILED = "upstream_failed"
SKIPPED = "skipped"
DEFERRED = "deferred"
+ AWAITING_INPUT = "awaiting_input"
class TaskStatesResponse(BaseModel):
