kaxil opened a new pull request, #75412: URL: https://github.com/apache/airflow/pull/75412
`AgentOperator(enable_hitl_review=True)` runs its review loop inside `execute()`: it pushes the output to XCom and polls for the reviewer's action every `hitl_poll_interval` seconds. The task stays `running` the whole time, so a review that takes an afternoon holds a worker slot and a pool slot for the afternoon. On Airflow 3.3+ the task now parks in `awaiting_input` between rounds and holds neither. The chat UI and the approve / reject / request-changes behaviour are unchanged; only the waiting changes. Airflow 3.1 and 3.2 keep the polling loop. ## How it works Each round, the worker opens a Human-in-the-loop request on the task instance (options *Approve*, *Request changes*, *Reject*, plus a `feedback` param) and raises `TaskAwaitingInput(method_name="resume_hitl_review")`, the same park that `LLMApprovalMixin` and `AgentOperator` tool approval already use. When the reviewer acts in the chat UI, the plugin's existing `/sessions/*` endpoint updates the session XComs, answers that request and resumes the task in one transaction, locking the task instance before the request row in the order core's response route and park transition use. The next round runs in a new worker process. The pydantic-ai conversation is saved in the task state store (with a sha256 in `next_kwargs`, so a changed or missing conversation fails the resume instead of regenerating from it), and the session and per-round outputs stay in XCom, which a resume of the same try keeps. `hitl_timeout` keeps its meaning: one wall-clock budget for all rounds, carried as an absolute deadline. Each park passes what is left as the `TaskAwaitingInput` timeout, so the scheduler's `awaiting_input` sweep resumes the task with a timeout event and it fails with `HITLTimeoutError` as before. A deadline that passes during regeneration fails the task instead of parking. ## Design rationale **Why a Human-in-the-loop request per round, when the plugin could resume the task itself.** The plugin could call `handle_event_submit` with no request behind it, but a click that lands after the worker pushes the round and before the park lands would never resume the task: the park transition only looks for an early answer on the request row. With a request per round, that answer is found at park time and the task resumes straight away. The timeout sweep, `responded_by`, and the Required Actions inbox come with it. **Why the plugin answers the request, rather than the chat UI calling core's response route.** Answering through core's route would leave the session XCom at `pending_review` until the worker restarts, so the chat would lose the immediate "feedback sent / regenerating" state it has today. Keeping the plugin's endpoints keeps that, keeps the 409 for a round already answered, and needs no frontend change. **Why the request text is the same every round.** On Airflow 3.3 a repeat request clears the earlier answer but keeps the first round's subject, body and params. So the request points the reviewer to the chat, which shows every round, and the worker ties an answer to its round through the session XCom rather than through the request. ## Gotchas - Each round is a new run of the task, so `execution_timeout`, `pre_execute` and `on_execute_callback` apply per round. `hitl_timeout` is what bounds the review as a whole. - The review also appears on the **Required Actions** page and can be answered there. An answer given there is written to the chat history by the worker on resume. *Request changes* with empty feedback from that form asks again for the same round instead of approving. - `hitl_poll_interval` is not used on 3.3+. It still applies on 3.1 and 3.2, so it is documented rather than deprecated. - The plugin resumes the task, so upgrade the provider on the API server no later than on the workers. - An approved output now crosses a process boundary. It comes back as the same object when it reads back from JSON (a registered model, `str`, plain types); otherwise the task returns the JSON form of what it returned before (a model under a list of output types or an output function's result, as a dict). A `datetime` inside such a dict comes back as an ISO string. - A `HITLReviewMixin` subclass that does not implement the new `dump_review_state` / `load_review_state` hooks keeps polling on 3.3+, so out-of-tree hosts do not break. Verified end to end on a local Airflow (postgres, scheduler, API server and the plugin's chat UI, with a 1-slot pool): request changes from the chat, request changes from Required Actions, then approve. The pool showed `occupied_slots: 0, open_slots: 1` at every park, the run succeeded on try 1, and each regenerated draft carried the full conversation. ## Follow-ups - On Airflow 3.4+, where a repeat request refreshes its text and `created_at`, the request could carry the round number and the current output, as tool approval does with its `request` param. - The saved-state plumbing (store, hash check, cleanup unless parked) now has a sibling in tool approval; the two could share a helper. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
