refactor(opencode,acp,orchestrator): unify session execution under SessionPool and modernize event routing architecture - #47
Conversation
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request integrates the OpenCode server with the SessionPool orchestration layer, replacing the legacy session management and protocol handler with a new integration layer, event adapter, and status bridge. While the transition to SessionPool-based routing is well-structured, several critical issues need to be addressed: a fire-and-forget task in the message routes should use state.create_background_task to avoid garbage collection; the SessionStatusBridge must be stopped in the finally block to prevent task and subscription leaks; the conversation history copy and agent state population during session forks must be restored; and asyncio.CancelledError should be excluded from broadcasting error events in the status bridge to prevent false error popups.
Important
The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.
2cabbae to
37c7e92
Compare
- Use state.create_background_task() for LSP warmup to prevent GC - Stop SessionStatusBridge in finally block to prevent task/subscription leak - Restore message copying during session fork (with session_id update) - Skip SessionErrorEvent broadcast for asyncio.CancelledError in status bridge - Add TODO for per-session agent conversation population during fork
Leoyzen
left a comment
Leoyzen
left a comment
There was a problem hiding this comment.
Review Fixes Applied
All 5 review comments have been addressed in commit 94ae0be81. Below is a point-by-point response:
1. _warmup_task GC issue (message_routes.py:104)
Fixed: Replaced asyncio.create_task(warmup()) with state.create_background_task(warmup(), name="warmup_lsp"). This ensures the task is tracked in state.background_tasks with a done callback, preventing GC mid-execution.
2. SessionStatusBridge leak (message_routes.py:622)
Fixed: Added await status_bridge.stop() in the finally block before unsubscribing from the EventBus. This cancels the bridge's background consumer task and releases the EventBus subscription.
3-4. Fork session message copy (session_routes.py:893, 925)
Partially fixed: Restored message copying logic:
- Retrieves
original_messagesfromstate.messages - Supports
request.message_idfiltering (copy up to specified message) - Copies messages with updated
session_idreferences tostate.messages[new_session_id]
Not applied: The suggestion to manually populate fork_agent.conversation.chat_messages was intentionally skipped because the current migration uses a shared agent instance (get_or_create_agent returns self.agent). Mutating the shared agent's conversation would pollute all sessions. Added a TODO comment to restore this once per-session agent instances are reintroduced.
5. CancelledError error broadcast (status_bridge.py:113)
Fixed: Added isinstance(exc, asyncio.CancelledError) guard before broadcasting SessionErrorEvent. Normal user cancellation/aborts no longer trigger false error popups.
All status_bridge tests pass (6/6). Ready for re-review.
|
/gemini review Note: All review comments from the previous review have been addressed in commit |
- Use state.create_background_task() for LSP warmup to prevent GC - Stop SessionStatusBridge in finally block to prevent task/subscription leak - Restore message copying during session fork (with session_id update) - Skip SessionErrorEvent broadcast for asyncio.CancelledError in status bridge - Add TODO for per-session agent conversation population during fork
94ae0be to
dd29aab
Compare
- Use state.create_background_task() for LSP warmup to prevent GC - Stop SessionStatusBridge in finally block to prevent task/subscription leak - Restore message copying during session fork (with session_id update) - Skip SessionErrorEvent broadcast for asyncio.CancelledError in status bridge - Add TODO for per-session agent conversation population during fork
dd29aab to
26fc20b
Compare
There was a problem hiding this comment.
Code Review
This pull request integrates the OpenCode server with the SessionPool orchestration layer, replacing the legacy protocol handler with a new OpenCodeSessionPoolIntegration class and a SessionStatusBridge to synchronize session status changes via the EventBus. It also introduces an OpenCodeEventAdapter to map AgentPool events to OpenCode SSE events and updates message routing and session lifecycle routes to delegate operations to the SessionPool. The review feedback recommends several improvements to error handling and state preservation: finalizing the assistant message with an aborted state upon request cancellation to prevent TUI lockups, populating the in-memory conversation history for forked agents, wrapping status bridge shutdown and database persistence calls in try-except blocks to allow graceful degradation to in-memory availability, and correctly converting monotonic timestamps to epoch timestamps during session state conversion.
Important
The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.
- Use state.create_background_task() for LSP warmup to prevent GC - Stop SessionStatusBridge in finally block to prevent task/subscription leak - Restore message copying during session fork (with session_id update) - Skip SessionErrorEvent broadcast for asyncio.CancelledError in status bridge - Add TODO for per-session agent conversation population during fork
26fc20b to
f10e0b3
Compare
🐛 Critical Issue: Auto-resume Events Lost Due to Per-Request Event Consumer LifecycleAfter reviewing the SessionPool integration in this PR, I found a critical race condition that causes auto-resume events to be lost. ProblemThe current implementation in # message_routes.py:_process_message_locked
status_bridge = SessionStatusBridge(...) # per-request
await status_bridge.start()
event_queue = await session_pool.event_bus.subscribe(session_id) # per-request
# ...
consumer_task = asyncio.create_task(_consume_events()) # per-requestBut # TurnRunner.run_loop
async with session.turn_lock:
await self._run_turn_unlocked(session_id, *initial_prompts, **kwargs)
await self._process_queued_work(session_id, session, **kwargs) # auto-resume hereThe
Impact
Root CauseThe fundamental issue is that event consumption lifecycle is tied to HTTP request lifecycle, but session execution lifecycle (including auto-resume) extends beyond a single request. Proposed SolutionSession-scoped event consumer instead of request-scoped:
This aligns with the architecture principle: session events are a session concern, not a request concern. Files to Modify
VerificationNeed to test:
Status: This issue exists in the current PR implementation and needs to be addressed before merging. Related: The |
- Use state.create_background_task() for LSP warmup to prevent GC - Stop SessionStatusBridge in finally block to prevent task/subscription leak - Restore message copying during session fork (with session_id update) - Skip SessionErrorEvent broadcast for asyncio.CancelledError in status bridge - Add TODO for per-session agent conversation population during fork
453040e to
717e0b9
Compare
|
/gemini review |
There was a problem hiding this comment.
Code Review
This pull request unifies OpenCode Server routes under the SessionPool orchestration layer, transitions to session-scoped EventBus consumers and status bridges, and introduces EventEnvelope wrapping for EventBus events. Feedback on these changes highlights several critical issues: first, RunExecutor is not updated to handle the new ToolCallCompleteEvent returned by process_tool_event(), which will cause lost events during graph-based team execution; second, a race condition in EventBus.subscribe() can lead to duplicate events being enqueued when a concurrent publish occurs during subscription; third, in fork_session, copying messages via session_pool.copy_messages and then calling append_message_to_session results in duplicate messages in persistent storage; fourth, in the MCP prompt command route, the assistant message is finalized prematurely because the handler does not wait for the background run to complete; and finally, set_messages_for_session is left as a no-op, which can cause stale in-memory state during session revert or compaction.
Important
The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.
Leoyzen
changed the title
- Thread 15: Fix fork_session duplicate messages by only appending to SessionPool in legacy mode (session_pool is None) - Thread 16: Fix execute_command premature completion by waiting for run_handle.complete_event with 30s timeout before finalizing - Thread 17: Fix set_messages_for_session no-op by updating in-memory state.messages cache
回复 Review Comments (Threads 13-17)以下是对未解决 review comment 的回复和修复说明： Thread 13 (CRITICAL): ToolCallCompleteEvent 丢失结论：当前代码已正确处理，无需修改。 经核查， combined = await process_tool_event(...)
if combined is not None:
await event_queue.put(combined)
Thread 14: EventBus.subscribe 竞态条件 ✅ 已修复Commit:
所有 34 个 EventBus 测试通过。 Thread 15: fork_session 消息重复 ✅ 已修复Commit:
Thread 16: execute_command 提前完成 ✅ 已修复Commit: 添加 Thread 17: set_messages_for_session 空实现 ✅ 已修复Commit: 将 以上修复已全部推送到 |
…onversion - Add OpenCodeEventAdapter for AgentPool->OpenCode event conversion - Add OpenCodeSessionPoolIntegration for SessionPool orchestration - Add SessionStatusBridge for RunHandle->SessionStatus sync - Route message handling through SessionPool.receive_request() - Delegate session CRUD (create/load/delete) to SessionPool - Delegate abort and fork to SessionPool - Wire input provider flow through SessionPool - Remove ServerState per-session agent creation, async queue - Remove orphaned OpenCodeProtocolHandler (handler.py) - Slim ServerState to SSE/connection-only (~309 lines) - Add comprehensive TDD tests for session integration and event conversion - Fix backward-compat references and lint issues
- Use state.create_background_task() for LSP warmup to prevent GC - Stop SessionStatusBridge in finally block to prevent task/subscription leak - Restore message copying during session fork (with session_id update) - Skip SessionErrorEvent broadcast for asyncio.CancelledError in status bridge - Add TODO for per-session agent conversation population during fork
When SpawnSessionStart is received, automatically subscribe to child session EventBus and forward events as SubAgentEvent to frontend. - Add _consume_child_events() to subscribe and forward child events - Detect SpawnSessionStart in parent event stream - Handle nested subagents recursively - Clean up subscriptions on StreamCompleteEvent/RunErrorEvent
… session-scoped event consumer refactor - Add OpenCodeSessionPoolIntegration to ServerState and create_app() - Use integration for session creation, message routing, and cleanup - Add close_session() to OpenCodeSessionPoolIntegration for proper cleanup - Update delete_session route to use integration.close_session() - Add backward compatibility when integration is not available - Fix CancelledError handling to not re-raise (matching old behavior) - Add conversation history preservation in CancelledError/ Exception handlers - Update test fixture to mock SessionPool for new architecture - Add session-scoped consumer tests Completes openspec change: session-scoped-event-consumer
…agents Per-session agents created by SessionController.get_or_create_session_agent() were missing pool-level resource providers (MCP aggregating provider and skills instruction provider), causing them to have zero tools when the agent config had no agent-level tools. This led to models outputting pseudo-tool-call text instead of executing actual function calls. The fix adds the same pool-level providers to per-session agents that shared agents already receive in AgentPool.__aenter__().
…esume Event consumer loop now creates a proper AssistantMessage (was UserMessage) and registers it in state.messages before processing the first event. Without this, PartUpdatedEvents were ignored by the TUI because the message store lacked the parent message entry. Red flag test: test_auto_resume_events_create_message_in_state
…l level SkillsTools provides list_skills and load_skill tools. Previously, these were only available to agents that explicitly configured in their agent config. Agents without this config (like librarian with all tools commented out) would receive skill metadata via SkillsInstructionProvider but had no actual tools to call, causing pseudo-tool-call output. This fix adds SkillsTools as a pool-level provider alongside MCP and skills_instruction providers, ensuring all agents (shared and per-session) have access to skill discovery and loading tools. Changes: - AgentPool.__init__(): create self.skills_tools_provider - AgentPool.__aenter__(): add skills_tools_provider to all shared agents - AgentPool.__aexit__(): remove skills_tools_provider during cleanup - SessionController.get_or_create_session_agent(): add skills_tools_provider to per-session agents
Unify all pydantic-ai stream events into a single FIFO path in SessionPool mode, eliminating the dual-path architecture that caused race conditions. Core changes: - process_tool_event(): Remove direct EventBus publish, always return combined ToolCallCompleteEvent. Caller decides routing. - _run_agentlet_core(): Add ToolCallStartEvent mapping for FunctionToolCallEvent and PartStartEvent(BaseToolCallPart), with deduplication by tool_call_id. Capture process_tool_event() return value and enqueue into local queue. - EventBusHooksAdapter: Disable before_tool_execute/after_tool_execute event publishing (now transparent passthroughs). Tool events come exclusively from the stream path. This fixes two observable bugs: 1. Missing ToolCallStartEvent in opencode TUI (start event wasn't mapped) 2. Race condition where ToolCallCompleteEvent arrived before start event and was silently dropped by event_processor (dual-path ordering hazard) Test coverage: - Updated red flag tests to verify fixed behavior - Added FIFO ordering test - Added duplicate suppression test - Added process_tool_event() no-direct-publish test - Added RunExecutor event_bus integration test - Added multiple tool calls ordering test Spec: Updated openspec/specs/unified-event-routing/spec.md with new scenarios for stream consumer flow, duplicate suppression, and PartStartEvent mapping.
Remove manual event routing from business layer tools: - TurnRunner now wraps child session events in SubAgentEvent via metadata - SessionPool.run_stream() supports scope parameter (session/descendants/subtree) - subagent_tools.py: remove manual EventBus subscription and SubAgentEvent wrapping - workers.py: use session_pool.run_stream(), remove manual event emission - agentpool_commands/pool.py: use session_pool.run_stream() for subagent spawning - create_child_session() now accepts **metadata for child session tracking Tests updated for unified event routing via EventBus descendants scope. Archives cleanup-business-layer-event-routing openspec change.
… legacy fallbacks
Recreate the thin-agentpool-core OpenSpec change after accidental deletion. Includes proposal, design, 3 delta specs, and tasks for thinning AgentPool core to native+acp agents only.
…to SessionPool - Remove 5 legacy fields from ServerState: messages, session_status, todos, input_providers, pending_questions - Migrate all route files to SessionPool/SessionController helpers: - message_routes.py: get_messages_for_session, append_message_to_session - session_routes.py: status + message helpers for all CRUD/ops - permission_routes.py: SessionController exclusively - question_routes.py: SessionController exclusively - Migrate non-route files: event_processor.py, status_bridge.py, session_pool_integration.py fallback removal - Update 33+ test files with backward-compat fixtures and helper usage - Centralize SessionStatusEvent handling in broadcast_event() Test results: 730 passed, 1 failed (pre-existing flaky timeout)
- Add EventEnvelope dataclass with source_session_id + transparent __getattr__ forwarding - Wrap all EventBus events in EventEnvelope at publish time - Remove session_id injection from producers (RunExecutor, helpers, event_emitter) - Adapt all consumers to unwrap EventEnvelope before type checks: - ACPProtocolHandler, ProtocolEventConsumerMixin - OpenCode server (status_bridge, session_pool_integration) - Claude Code Agent, ACP Agent, BaseAgent - Migrate tests to assert EventEnvelope wrapping behavior - Add dedicated EventEnvelope integration tests - Fix all ruff/mypy issues and e2e regressions All 112 key tests pass.
…umers - Override _get_subscription_scope to 'session' to prevent event interleaving - Add _on_spawn_session_start to create child consumers for sync subagents - Skip child consumer creation for background tasks (spawn_mechanism='task')
Use self.node._events.session_id instead of getattr fallback. Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent) Co-authored-by: Sisyphus <clio-agent@sisyphuslabs.ai>
Inherit MCP tool providers (kind==mcp) when creating child sessions. Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent) Co-authored-by: Sisyphus <clio-agent@sisyphuslabs.ai>
Extends exception handling in load_rules() to catch RequestError (and other exceptions) from remote filesystem calls via MCP/ACP. Previously only OSError/UnicodeDecodeError were caught, causing task failures when remote workspace-fs returned errors.
…e condition Moves replay buffer snapshot capture inside the same lock as subscriber registration in subscribe(), and moves replay buffer append inside the lock in publish(). This prevents concurrent publish() from delivering events to both the historical replay and the live queue.
- Thread 15: Fix fork_session duplicate messages by only appending to SessionPool in legacy mode (session_pool is None) - Thread 16: Fix execute_command premature completion by waiting for run_handle.complete_event with 30s timeout before finalizing - Thread 17: Fix set_messages_for_session no-op by updating in-memory state.messages cache
dc31055 to
5c6ae68
Compare
1 participant
Summary
This PR is a multi-phase architectural refactor that unifies OpenCode server execution under
SessionPool, modernizes the subagent event architecture, and introduces foundational infrastructure for immutable event routing. It replaces the original "SessionPool integration" scope with a comprehensive overhaul of session lifecycle, event propagation, and cross-protocol handler consistency.Architectural Overview
The refactor spans three major migrations and four architectural pillars:
Migrations (OpenCode Server → SessionPool)
SessionPoolintegration, message routing, event conversion, auto-subscriptionServerStatedict elimination, session CRUD, permission routes, legacy fallback removalResult:
ServerStateslimmed from 608 → ~309 lines; all session operations routed throughSessionPoolhelpers.Architectural Pillars
Subagent Event Architecture (Phase 2)
SubAgentEventwrapping inTurnRunnerEventBusToolPartregistration happens explicitly inEventProcessorContextEventBus Infrastructure (Phases 3, 8)
EventBuswith event IDs and deduplicationEventEnvelopefor immutable event routingProtocolEventConsumerMixin (Phase 7)
SpawnSessionStartplaceholder replacedMCP Provider Inheritance (Phase 8)
AgentContextsession_id retrieval for proper scopingNew Components
event_adapter.pyOpenCodeEventAdapter— convertsRichAgentStreamEvent→ OpenCodeEventsession_pool_integration.pyOpenCodeSessionPoolIntegration— bridges OpenCode routes withSessionPoolstatus_bridge.pySessionStatusBridge— syncsRunHandlestatus →SessionStatusEventevent_bridge.py/event_processor.pymixins.pyProtocolEventConsumerMixin+ConsumerShutdownModified Modules
OpenCode Server (heaviest impact):
SessionPoolhelpers (message_routes,session_routes,permission_routes,question_routes, etc.)state.py— legacy dict fields removed; backward-compat wrappers added during migrationhandler.py— orphanedOpenCodeProtocolHandlerdeletedOrchestrator & Agents:
TurnRunner— SubAgentEvent wrapping removedRunExecutor— unified tool event pathsAgentContext— session_id retrieval fixedpool.py— SkillsTools provider added at pool levelACP Server:
handler.py— adoptedProtocolEventConsumerMixinevent_converter.py— elicitation fixes, flat-format message supportRebase Context
✅ Rebased onto latest
develop/agentic(2026-06-09) — includes 6 ACP/MCP elicitation fixes:d41417e42fix(acp): send elicitation response back to MCP server instead of ClientSession63a21e72cfix(acp): handle MCP elicitation locally instead of forwarding to ACP client1a5c0f6c6fix(acp): handle flat-format mcp/message from ACP clients0a85a1da1test(acp): add fastmcp e2e integration tests for MCP-over-ACPc3149372bfix(mcp-over-acp): handle elicitation/create without inner id926a87f60fix(mcp-over-acp): add correlation registry for elicitation passthrough52 commits replayed cleanly with 0 conflicts
Previous HEAD:
2e547ed36→ New HEAD:b1b27f382Test Coverage
The 6 TDD RED failures require
EventProcessorhandlers forPartDeltaEvent.text/thinking,StreamCompleteEvent→SessionIdleEvent,RunErrorEvent,RunFailedEvent— to be implemented in a follow-up.Total: 70+ new/modified test files across
tests/orchestrator/,tests/servers/,tests/agents/,tests/toolsets/.Backward Compatibility
Thin backward-compat wrappers added to
ServerStateforget_or_create_agent(),_session_agents,remove_session_agent()to prevent breaking existing route references during the migration window. These will be removed in a future cleanup PR.Checklist
SessionPool.receive_request()Migration Path
This PR supersedes the original "SessionPool integration" scope. The changes are not backward-compatible for code depending on:
ServerStatedict mutations (e.g.,state.messages,state.todos)SubAgentEventwrapping behavior in customTurnRunnersubclassesOpenCodeProtocolHandler(deleted)Consumers should migrate to:
SessionPoolhelpers for session lifecycleEventBus+ProtocolEventConsumerMixinfor event consumption