feat: Durable Elicitation Bridge — checkpoint/resume for elicitation requests - #170
Closed
Million-mo wants to merge 38 commits into
Closed
feat: Durable Elicitation Bridge — checkpoint/resume for elicitation requests#170Million-mo wants to merge 38 commits into
Million-mo wants to merge 38 commits into
Conversation
…tationResumePayload and ElicitationDeferredEvent - Extend deferred_kind with 'elicitation' variant - Add optional fields: elicitation_message, elicitation_schema, elicitation_mode, mcp_server_id - Add ElicitationResumePayload Schema for accept/decline/cancel responses - Add ElicitationDeferredEvent dataclass to RichAgentStreamEvent union - Serialization backward-compatible (optional fields default to None) Refs: wolf1069b/wolfharness#107
…kpoint and protocol placeholders - Define ElicitationResolutionStrategy Protocol with async resolve() method - Implement CheckpointResolutionStrategy delegating to CheckpointManager.checkpoint() - Add ProtocolResolutionStrategy placeholder for future MRTR/SEP-2663 support Refs: wolf1069b/wolfharness#107
…and implementations - Add supports_durable_elicitation property to InputProvider base class (defaults to False; StdlibInputProvider and MockInputProvider inherit this default) - Add dynamic property to ACPInputProvider checking self.session.checkpoint_enabled at runtime - Add dynamic property to OpenCodeInputProvider checking session checkpoint config via session controller at runtime - Add checkpoint_enabled field to ACPSession and SessionState (default False) as foundation for runtime checks Refs: wolf1069b/wolfharness#107
…el and side-channel Add _pending_elicitation_deferral side-channel to AgentContext for durable elicitation. When provider.supports_durable_elicitation is True, handle_elicitation() stores params and returns a sentinel decline result instead of calling get_elicitation() directly. MCPClient.call_tool() checks the side-channel after the MCP call returns and raises CallDeferred with elicitation metadata, enabling checkpoint-based deferral. Key design decisions: - Sentinel return (decline) prevents FastMCP from blocking on the handler - Side-channel dict on AgentContext avoids exception-based control flow inside the elicitation callback (FastMCP catches exceptions) - except CallDeferred: raise placed BEFORE broad except Exception to prevent CallDeferred from being swallowed into RuntimeError - Exhaustive match on ElicitRequestParams union (FormParams | URLParams) with no getattr/hasattr Refs: wolf1069b/wolfharness#107
…nFutureRegistry, wire into capability chain - Create elicitation_bridge.py with ElicitationDeferredBridge as a HandleDeferredToolCalls capability following the deferred_bridge/approval_bridge factory function pattern - Implement ElicitationFutureRegistry: per-session registry of asyncio.Future instances with register/resolve/reject_all lifecycle methods - Bridge handler inspects DeferredToolRequests.calls for entries with metadata['deferred_kind'] == 'elicitation', checkpoints via CheckpointManager, emits ElicitationDeferredEvent, registers future, and returns None (block). Non-matching calls pass through. - Wire elicitation bridge into get_agentlet() capability chain AFTER deferred_bridge and BEFORE approval_bridge Refs: wolf1069b/wolfharness#107
…ode converters - Mark ElicitationDeferredEvent as immediate in _is_immediate() in core.py - ACP event_converter: emit ToolCallStart with elicitation params in _meta - OpenCode event_processor: create ToolPart with elicitation metadata Refs: wolf1069b/wolfharness#107
…th in-process and crash recovery paths - Add elicitation_payloads parameter to resume_session() (optional, default None) - Add _try_in_process_elicitation_resume() helper: resolves futures in ElicitationFutureRegistry when agent run is still alive (in-process) - Extend _resume_native_agent() with crash recovery: pre-populates cached_elicitation_responses on AgentRunContext so handle_elicitation() returns cached response during MCP tool re-execution - Add elicitation_registry and cached_elicitation_responses fields to AgentRunContext - Update handle_elicitation() to check cached responses before deferring - Store ElicitationFutureRegistry on run_ctx in get_agentlet() - Add __contains__ to ElicitationFutureRegistry for membership checking - Add SessionClosedError exception - Call registry.reject_all(SessionClosedError()) on session close - Validate elicitation payloads cover all elicitation deferred calls - Exclude elicitation calls from deferred_tool_results validation Refs: wolf1069b/wolfharness#107
Covers Tasks 9.1-9.7, 9.14-9.16, 9.18-9.19, 9.21-9.22: - PendingDeferredCall serialization roundtrip (with/without elicitation fields) - handle_elicitation durable=True (side-channel + decline sentinel) - handle_elicitation durable=False (direct get_elicitation call) - MCPClient.call_tool raises CallDeferred when side-channel is set - MCPClient.call_tool normal path (no deferral) - Elicitation bridge handles elicitation + passthrough non-elicitation calls - ElicitationFutureRegistry register/resolve/reject_all lifecycle - ACPInputProvider.supports_durable_elicitation dynamic property - OpenCodeInputProvider.supports_durable_elicitation dynamic property - Side-channel cleanup on error (finally block clears _current_elicitation_handler) - Bridge positioning: elicitation handled before approval - Backward compatibility: old-format PendingDeferredCall deserialization - ElicitationResumePayload decline/cancel/accept actions + cached response - Strategy classes: CheckpointResolutionStrategy, ProtocolResolutionStrategy - Elicitation timeout field serialization/deserialization Refs: wolf1069b/wolfharness#107
…exts AsyncMock returns truthy mocks for any attribute, which incorrectly triggered the CallDeferred side-channel check in MCPClient.call_tool(). Refs: wolf1069b/wolfharness#107
Task 7 changed _resume_native_agent() to call agent.run_stream() instead of agent.run(). Updated test mocks to use async generators instead of AsyncMock to properly match the async iterator interface. Refs: wolf1069b/wolfharness#107
Design correction: - handle_elicitation() now raises CallDeferred directly (not side-channel) - MCP elicitation_handler catches CallDeferred, converts to side-channel (FastMCP workaround isolated to MCP client only) - Local tools like question_for_user need zero adaptation PR review fixes: - #1: Accumulate all pending_calls, single checkpoint after loop (prevents overwrite when multiple parallel elicitation calls) - #2: Bypass SessionBusyError for in-process elicitation resume (add allow_active_run parameter to _with_resume_lock) CI fixes: - ruff: D202 blank line after docstring, D104 missing package docstring - ruff format: 3 files reformatted - snapshot: update ACP event converter snapshot (pre-existing drift) Refs: wolf1069b/wolfharness#107
- test_approval_bridge: expect 3 HandleDeferredToolCalls (deferred + elicitation + approval) - test_capability_chain: patch elicitation bridge, verify deferred < elicitation < approval order Refs: wolf1069b/wolfharness#107
_ACPSessionProxy is used by ACPProtocolHandler when creating ACPInputProvider for elicitation. Without checkpoint_enabled, ACPInputProvider.supports_durable_elicitation raises AttributeError: '_ACPSessionProxy' object has no attribute 'checkpoint_enabled'. Refs: wolf1069b/wolfharness#107
…nd configured SessionState.checkpoint_enabled now defaults to True when SessionController has a store (session persistence configured). ACP _ACPSessionProxy also passes checkpoint_enabled based on store availability. This means durable elicitation is automatically active for ACP and OpenCode servers when storage is configured, without requiring explicit configuration. Refs: wolf1069b/wolfharness#107
…s present Without DeferredToolRequests in the agent's output_type, pydantic-ai raises 'A deferred tool call was present, but DeferredToolRequests is not among output types' when handle_elicitation() raises CallDeferred. This was the root cause of the production error: ❌ Error: [engineer] A deferred tool call was present... Refs: wolf1069b/wolfharness#107
Covers: DeferredToolRequests in output_type, checkpoint_enabled auto-set, _ACPSessionProxy durable support, bridge handler with real DeferredToolRequests, ACP event converter with real EventBus, handle_elicitation CallDeferred raise. Refs: wolf1069b/wolfharness#107
…tation The ACP event converter was emitting a ToolCallStart NOTIFICATION for ElicitationDeferredEvent, but the ACP client expects an interactive elicitation/create REQUEST (request-response, not fire-and-forget). Now when _handle_event receives ElicitationDeferredEvent: 1. Spawns a background task (non-blocking for event consumer loop) 2. Sends elicitation/create request to ACP client via ACPRequests 3. Waits for user response 4. Builds ElicitationResumePayload from response 5. Calls session_pool.resume_session() with the payload This closes the loop: handle_elicitation() → CallDeferred → bridge checkpoint → ElicitationDeferredEvent → ACP elicitation/create → user responds → resume_session → future resolved (in-process) or crash recovery. Refs: wolf1069b/wolfharness#107
The in-process resume path was fundamentally broken: handle_elicitation() raises CallDeferred, which ends the agent run with DeferredToolRequests as output. Nobody awaits the ElicitationFutureRegistry futures, so resolving them does nothing — the agent run doesn't continue and the stream is dead. Fix: always fall through to crash recovery path (_resume_native_agent), which re-executes the agent with cached_elicitation_responses. This correctly re-runs the tool, handle_elicitation() returns the cached response, the tool completes, and the agent produces events normally. The in-process futures are still resolved (for registry cleanup), but the early return is removed. The has_in_process_elicitation bypass of SessionBusyError is kept (old run is completed but still registered). Refs: wolf1069b/wolfharness#107
…tive_run=True The elicitation bridge checkpoint saves checkpoint data but doesn't update the session store status from 'active' to 'checkpointed'. When resume_session() is called with allow_active_run=True (in-process elicitation resume), the status check was still rejecting 'active' status, causing SessionBusyError. Now when allow_active_run=True, both 'checkpointed' and 'active' statuses are allowed in the persisted session data check. Refs: #107
…llDeferred Major design correction: handle_elicitation() now has three paths: 1. Crash recovery: returns cached response (unchanged) 2. MCP tools (in_mcp_callback=True): raises CallDeferred (unchanged — FastMCP callbacks can't await for long periods) 3. Local tools (in_mcp_callback=False): checkpoints, emits event, registers future, and **awaits the future** — agent run suspends without ending. When user responds, future resolves, tool completes naturally, agent run continues. No re-execution. Changes: - context.py: Added checkpoint_manager to AgentRunContext, in_mcp_callback to AgentContext, rewrote handle_elicitation() with three-path logic - agent.py: Store checkpoint_mgr on run_ctx.checkpoint_manager - client.py: Set agent_ctx.in_mcp_callback=True before MCP call, clear in finally block - core.py: resume_session() now returns early when in-process futures are resolved (agent run continues naturally, no crash recovery needed) - Tests: Split test 9.2 into MCP path (raises CallDeferred) and local path (awaits future), updated protocol integration test Refs: wolf1069b/wolfharness#107
…ssion data SessionData.pending_deferred_calls is not updated when local tools suspend on await future (only checkpoint storage is updated). The in-process detection now checks the ElicitationFutureRegistry directly instead of relying on session data. Also added __len__ to ElicitationFutureRegistry for the non-empty check. Fixes: SessionBusyError when resuming after local tool elicitation Refs: wolf1069b/wolfharness#107
… resume Two-level test covering the full in-process elicitation resume lifecycle: - Level 1: Direct NativeTurn.execute() — core turn generator - Level 2: RunHandle.start() → EventBus → consumer — integration layer Both levels currently pass, confirming the core agent run and RunHandle integration correctly yield StreamCompleteEvent after future resolution. If the bug reappears, these tests will pinpoint which layer fails. Refs: wolf1069b/wolfharness#107
…efaulting to zeros NativeTurn.execute() was constructing ChatMessage without setting the usage field, causing it to default to RequestUsage() (all zeros). The ACP event converter reads message.usage (not message.cost_info) for UsageUpdate notifications, so clients always saw 0 tokens. Fix: extract RequestUsage from the last ModelResponse in new_messages and pass it to ChatMessage constructor. Refs: wolf1069b/wolfharness#107
1. context.py: Add finally block to remove future from registry on timeout/cancellation, preventing ValueError on retry. 2. core.py: Guard against concurrent run if in-process elicitation resolution fails — raise SessionBusyError if run is still active. 3. handler.py: Re-raise CancelledError instead of swallowing it. User cancel comes as response.action='cancel', not CancelledError. 4. core.py: Fix misleading comment — no filtering needed because API contract separates elicitation payloads from deferred results. Refs: #107
Replace ACP AgentMessageChunk ("🔄 Session resumed...") with a
backend-only log. The frontend doesn't need this notification —
streaming events from the resuming agent run are sufficient.
Refs: #107
Re-applied elicitation bridge changes to new module structure: - session_controller.py: SessionState.checkpoint_enabled, SessionClosedError - session_pool.py: resume_session with elicitation_payloads, _try_in_process_elicitation_resume, _resume_native_agent with cached_elicitation, close_session with reject_all, _with_resume_lock allow_active_run - event_bus.py: ElicitationDeferredEvent in _is_immediate - core.py: re-export SessionClosedError - Tests: fix EventBus API (Queue.get instead of receive_stream.receive) Refs: #107
Add field to BaseAgentConfig (defaults to 300s).
Supports string (5m, 300s), int (seconds), timedelta, or null
(infinite wait). The value flows through AgentRunContext to
handle_elicitation(), replacing the hardcoded 300s.
Also add to ElicitationDeferredEvent so frontends
can display countdown timers.
YAML example:
agents:
my_agent:
elicitation_timeout: 600s # or null for infinite
Refs: #107
P0: Pass real message_history to checkpoint in handle_elicitation() - Add current_messages field to AgentRunContext - Set it in tool_wrapping.py from ctx.messages before each tool call - Use run_ctx.current_messages instead of [] in checkpoint call - Fixes crash recovery re-executing all prior tool calls (duplicate side effects) P1a: Checkpoint failure no longer silently swallowed - Use logger.warning with explicit message about degraded durability - Don't set run_ctx.checkpointed=True on failure P1b: Cancel ACP elicitation tasks on session close - Cancel pending _elicitation_tasks in _after_consumer_loop - Cancel and clear tasks in close_session P2: Update session store status to 'checkpointed' after elicitation bridge checkpoint - Both handle_elicitation() (local tools) and elicitation_bridge.py (MCP tools) now update session store status from 'active' to 'checkpointed' after saving checkpoint - Eliminates the 'liminal' status that required allow_active_run workaround in _with_resume_lock - Wrapped in try/except so failures don't break the checkpoint path Refs: #107
- P0: verify handle_elicitation passes current_messages to checkpoint (not empty list) — prevents crash recovery re-executing tools - P1a: verify checkpoint failure doesn't set checkpointed=True — the in-process future await still works but crash recovery is degraded - P2: verify session store status is updated to 'checkpointed' after elicitation checkpoint, and skipped if already checkpointed Refs: #107
… violation agentpool_config.nodes was importing agentpool.utils.parse_time, breaking the 'Config must not import from core' contract. Inlined a simple regex-based parser in the field_validator to maintain layer separation. Refs: #107
…gentlet() from_config() pre-built capabilities and stored them in _extra_capabilities, while get_agentlet() also iterated self.config.capabilities and built them again — producing duplicate FunctionToolset instances with conflicting tool names (e.g. two 'task' tools from BackgroundTaskCapability). Fix: skip pre-building in from_config(); let get_agentlet() handle all capability construction lazily from self.config.capabilities.
AbstractCapability base class uses KW_ONLY for its fields (id, description, defer_loading), all with defaults. The subclass field hook_manager (no default) was not keyword-only, causing a dataclass TypeError when running against pydantic-ai v1.x where the base class is @DataClass(init=False).
cancel_session() only cancelled the parent session's RunHandle, leaving child (subagent) sessions running independently. Added _cancel_subagent_runs() which recursively walks the _parent_of tree and calls cancel_run_for_session() for each child before cancelling the parent. Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent) Co-authored-by: Sisyphus <clio-agent@sisyphuslabs.ai>
…rable-elicitation-bridge # Conflicts: # tests/agents/native_agent/test_get_agentlet_capabilities.py # tests/servers/acp_server/test_acp_protocol_handler_cancel.py
Owner
Author
|
Owner
Author
|
Owner
Author
|
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
2 participants
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
Implements durable elicitation: elicitation requests (form-based "ask the user" prompts) from both MCP tools and local tools survive process crashes via checkpoint/resume infrastructure.
Closes #107
Architecture
Two Resolution Paths
Local tools (e.g.
question_for_user): Inline future resolution —handle_elicitation()checkpoints, emitsElicitationDeferredEvent, registers anasyncio.Future, and awaits it. The agent run suspends (not ends) at theawaitpoint. When the user responds,resume_session()resolves the future,handle_elicitation()returns the response, the tool function continues naturally. Zero re-execution.MCP tools:
handle_elicitation()raisesCallDeferred(FastMCP's callback wrapper catches exceptions, so the MCPelicitation_handlerinclient.pyconverts to side-channel + sentinel).call_tool()re-raisesCallDeferredafter the MCP call returns. The run ends withDeferredToolRequestsand resumes via crash recovery withcached_elicitation_responses.Key Components
handle_elicitation()incontext.py: Three-path dispatch — crash recovery (cached response), MCP tools (raiseCallDeferred), local tools (await future)ElicitationDeferredBridge:HandleDeferredToolCallscapability wired afterdeferred_bridge, beforeapproval_bridge. Checkpoints session, emits event, registers futureElicitationFutureRegistry: Per-session future management withregister(),resolve(),remove(),reject_all()ElicitationDeferredEvent, sendselicitation/createrequest to client (two-way), collects response, callsresume_session()resume_session()→_resume_native_agent()re-creates agent, setscached_elicitation_responses, callsagent.run_stream(). During re-execution,handle_elicitation()returns cached responsecheckpoint_enabledset toTruewhenSessionController.store is not NoneElicitationResolutionStrategy: Abstraction withCheckpointResolutionStrategy(current) +ProtocolResolutionStrategy(MRTR placeholder)Session Status Fix
_with_resume_locknow allows"active"session status whenallow_active_run=True, because the elicitation bridge saves checkpoint data without updating the session store status from"active"to"checkpointed".Resume Notification
SessionResumeEventis logged backend-only — no ACP notification sent to frontend. Streaming events from the resuming agent run are sufficient.Files Changed (34 files, +4000/-92 lines)
New files:
src/agentpool/agents/native_agent/elicitation_strategy.py— Strategy protocol + implementationssrc/agentpool/agents/native_agent/elicitation_bridge.py— Bridge capability + future registrytests/elicitation/test_unit_elicitation.py— 19 unit teststests/elicitation/test_integration_elicitation.py— 10 integration teststests/elicitation/test_protocol_integration.py— 10 protocol integration teststests/elicitation/test_resume_lifecycle.py— Resume lifecycle testsopenspec/changes/durable-elicitation-bridge/— OpenSpec change artifactsModified files:
src/agentpool/sessions/models.py—PendingDeferredCallextension,ElicitationResumePayloadsrc/agentpool/agents/events/events.py—ElicitationDeferredEventsrc/agentpool/agents/context.py— Three-pathhandle_elicitation(),CallDeferred, cached responses, future awaitsrc/agentpool/mcp_server/client.py— Side-channel +CallDeferredre-raisesrc/agentpool/agents/native_agent/agent.py— Bridge wiring,DeferredToolRequestsin output_typesrc/agentpool/orchestrator/core.py—resume_session(), in-process + crash recovery,SessionClosedError, auto-enable, concurrent-run guardsrc/agentpool/ui/base.py—supports_durable_elicitationpropertysrc/agentpool_server/acp_server/—checkpoint_enabled, dynamic property, event converter, elicitation handlersrc/agentpool_server/opencode_server/— Dynamic property, event processordocs/architecture/user_interaction.md— +103 lines documenting the flowTest Results
uv run ruff check src/— All checks passeduv run ruff format --check src/— All files formatteduv run mypy src/— 0 errorsPR Review Fixes
finally: registry.remove(handle)SessionBusyErrorguard before crash recoveryCancelledErrorswallowedaction="cancel"response)agent_ctxcould be Noneif agent_ctx:guard already prevents handler definitionGuardrails Verified
MCPClient.call_tool()wrapper changed)ProtocolResolutionStrategyraisesNotImplementedError)supports_durable_elicitation=False)getattr/hasattr— full type safetyRefs: wolf1069b/wolfharness#107