[kaxil]

Stacked on #75413, which lets a running task read its own Human-in-the-loop response. Until that merges, this PR's diff includes its commit.

A Human-in-the-loop task stops running as soon as it writes its request and waits in awaiting_input. When the answer comes seconds later, from someone who clicked the link in the notification or from a program answering through the REST API, the task still goes back through the scheduler, an executor slot and a worker, which parses the Dag file again before execute_complete runs.

response_poll_window keeps the task running and checking for the answer for up to that long first:

ApprovalOperator(
    task_id="approve_deploy",
    subject="Deploy this build?",
    response_poll_window=timedelta(minutes=2),
    response_timeout=timedelta(hours=4),
)

An answer in the first two minutes is used in the same task run. Otherwise the task parks as it does today for the rest of the four hours. It is off by default, so existing tasks do not change.

Here an ApprovalOperator with a two-minute window and a 10-minute response_timeout is approved from the Required Actions page while the task is still running. The response is recorded at 18:34:53 and the task finishes at 18:34:54 in its first attempt, without parking:

hitl-poll-window-demo.mp4

Design rationale

An answer inside the window goes through execute_complete. The operator builds the same event a resumed task receives, so option and params validation and the subclass logic (ApprovalOperator skipping downstream tasks or failing on reject, HITLBranchOperator picking branches) run exactly as they do on resume.

The window is part of response_timeout, not added to it. The deadline still counts from when the request is written. The window is capped at response_timeout, and the task parks with whatever is left; if nothing is left, the scheduler's timeout sweep applies the defaults on its next pass, so the timeout logic stays in one place.

It counts against execution_timeout. The window runs inside execute(), and __init__ rejects a window that is not shorter than execution_timeout, which would otherwise kill the task before it could park.

Capability check, not a version gate. The task can only read its response if its Task SDK serves that request (#75413). The supervisor and the task always run from the same Task SDK install, so the operator checks for the SDK's HITLDetailResponseResult type. Without it, the window is ignored with a warning and the task parks straight away, rather than failing on the first read.

A failed read parks instead of failing the task. Checking early is optional, so if a read fails after the SDK client's retries, the task logs a warning and parks as usual.

Gotchas

• The window holds a worker slot and a pool slot, which a parked task does not. Keep it to the few minutes in which a quick answer is likely.
• On a local setup with the LocalExecutor, the scheduler round trip after parking took about a second, so the saving there is small. It grows with the cost of starting a task (a pod per task, a busy scheduler); I have not measured those.
• Answering from the Required Actions page during the window needs the UI change in Allow Human-in-the-loop tasks to be answered while still running #75413. Before that, only REST API answers are picked up in the window.
• The task checks once a second, through the supervisor and the Execution API, for as long as the window lasts.

---

• Read the Pull Request Guidelines for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
• For fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
• When adding dependency, check compliance with the ASF 3rd Party License Policy.
• For significant user-facing changes create newsfragment: {pr_number}.significant.rst, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.