perf(event): iteration extension events and general-purpose status events#1889
Open
E2ern1ty wants to merge 3 commits into
Open
perf(event): iteration extension events and general-purpose status events#1889E2ern1ty wants to merge 3 commits into
E2ern1ty wants to merge 3 commits into
Conversation
… events Add iteration extension events and a general-purpose status event mechanism to the agent event system. Iteration extension events: - Add RequireIterationExtensionEvent, emitted when the reasoning-acting loop reaches its iteration limit (gated by the new ReActConfig.allow_iteration_extension, default False for backward compatibility). - Add IterationExtensionResultEvent to capture the user's decision and resume the paused loop. Approving extends the per-reply iteration budget (tracked via AgentState.iteration_extension); denying falls back to the existing ExceedMaxItersEvent termination path. - Track the paused state via AgentState.awaiting_iteration_extension so the pause/resume cycle survives persistence and process restarts. General status events: - Add a general-purpose, user-customizable StatusEvent (name + status + arbitrary data) plus Agent.emit_status for emitting custom status notifications into the active reply stream. - Wire context compaction to emit start/end status events as the canonical example of a time-consuming operation. Wiring: - Thread the new input event through reply/reply_stream and the chat service; convert the new events in the AG-UI protocol middleware; skip wake-up runs while awaiting an iteration extension decision. Tests: - Add tests for iteration extension (request/approve/deny, default budget, validation) and the status event mechanism (emit, drain, early-close cancellation, compaction), and extend the AG-UI tests.
Make the general-purpose status event mechanism work from any stage of a reply (reasoning, tool execution, context compaction, user middleware), not just during context compaction. Inspired by LangGraph's get_stream_writer, the active reply's status queue is now exposed through a ContextVar so emit_status can be called from anywhere in the call stack without threading a writer argument. - Run the reply pipeline (_reply) as a background task that pushes its output onto a single queue; a driver concurrently drains the queue and re-yields, so status events emitted during a long-running awaited operation surface promptly, interleaved in emission order with the main events. - Bind the status queue inside the producer task's own (copied) context to avoid the cross-context ContextVar reset pitfall on async-generator aclose(); the binding is discarded automatically when the task ends. - The driver propagates pipeline exceptions to the consumer and cancels the background task if the consumer abandons the stream early. - Remove the previous _drain_status_events helper and the per-instance _status_queue; the compaction call site reverts to a plain await. Rewrite the status event tests to exercise the mechanism through real reply_stream runs, including a tool that emits status events during the reply (demonstrating it is no longer tied to compaction), early-close cancellation, and exception propagation.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Closes #1696.
Improves the event system in two ways, as described in the issue.
1. Iteration extension events (HITL loop-limit handling)
RequireIterationExtensionEvent, emitted when the reasoning-acting loop reaches its iteration limit.IterationExtensionResultEventto capture the user's decision and resume the paused loop.ReActConfig.allow_iteration_extension(defaultFalse, fully backward compatible). When disabled, the loop terminates withExceedMaxItersEventas before.RequireIterationExtensionEventand pauses (mirroring the existing HITL tool-confirmation pause/resume). On resume:AgentState.iteration_extension) and continues. Approving without a positive count grants another fullmax_itersbudget so the loop always makes progress.ExceedMaxItersEventtermination path.AgentState.awaiting_iteration_extension/iteration_extension) lives onAgentState, so the pause/resume cycle survives persistence and process restarts.2. General-purpose status events
StatusEvent(name+status+ arbitrarydata) andAgent.emit_status(...)so the backend can notify the frontend about time-consuming operations without being tied to a specific operation.compressing_contextstart/end status events as the canonical example.ContextVar(inspired by LangGraph'sget_stream_writer), soemit_statusworks from any stage of a reply — reasoning, tool execution, compaction, or user middleware — and the events are interleaved promptly into the stream. Internally the reply pipeline runs as a background task whose output is drained by a driver, with exception propagation and early-close cancellation handled.Wiring
reply/reply_streamand the chat service; convert the new events in the AG-UI protocol middleware; skip wake-up runs while the agent is awaiting an iteration extension decision.Tests
tests/iteration_extension_test.py— request / approve / deny, default-budget grant, validation errors.tests/status_event_test.py— emit-outside-reply no-op, compaction status events in a real reply, a tool emitting status events during a reply (proves it is not tied to compaction), early-close cancellation, exception propagation.tests/agui_protocol_test.pyfor the three new events.black,flake8,mypy,pylintall clean; relevant test suites pass.Backward compatibility
allow_iteration_extension=Falseby default); existing behavior is unchanged.compress_context()keeps its-> Nonesignature.AgentState/ReActConfigfields have defaults, so existing serialized state/config deserialize unchanged.