feat(session-pool): implement SessionPool architecture for ACP/OpenCode

[Leoyzen]

Session Pool Architecture Implementation

This PR implements the session-pool-architecture as defined in the OpenSpec change. It introduces a centralized SessionPool for managing per-session agent lifecycle, event routing, and turn orchestration across ACP and OpenCode protocols.

Changes Overview

Group 1: BaseAgent Public API

• Add get_active_run_context() to safely access active run context
• Add is_turn_active() to check if a turn is running
• 16 comprehensive tests

Group 2: SessionPool Core Infrastructure

• SessionState: per-session state with turn locking
• EventBus: bounded queues with dropping, sentinel shutdown
• SessionController: per-session agent lifecycle, TTL cleanup
• TurnRunner: turn serialization, auto-resume, prompt injection
• SessionPool: high-level facade with feature flags
• 81 unit tests

Group 3: AgentPool Integration

• SessionPoolConfig Pydantic model
• AgentPool lifecycle integration (__aenter__/__aexit__)
• create_session() convenience method
• Per-protocol feature flags (acp.use_session_pool, opencode.use_session_pool)
• 27 integration tests

Group 4-5: Protocol Handlers

• ACPProtocolHandler: ACP session pool handler with event consumer
• OpenCodeProtocolHandler: OpenCode handler with SSE forwarding
• Both are canary-safe (fallback to legacy code when disabled)
• 30 handler tests

Group 6: Validation & Observability

• Prometheus metrics format (to_prometheus())
• Stress tests (1000 concurrent sessions)
• Benchmark tests (EventBus >600K events/sec)
• E2E integration tests (cross-protocol event passing)
• Ops playbooks for session-pool, ACP, and OpenCode handlers

Test Results

170 passed, 11 deselected

Design Highlights

• Canary-safe: All protocol handlers check per-agent feature flags
• Opt-in: SessionPool disabled by default (enable_session_pool=False)
• No breaking changes: Existing session management preserved
• Cross-protocol: EventBus enables ACP ↔ OpenCode event sharing

Related

• OpenSpec Change: session-pool-architecture
• Design Doc: ARCHITECTURE-ORCHESTRATOR.md

[gemini-code-assist]

Code Review

This pull request introduces a unified SessionPool orchestration layer under src/agentpool/orchestrator/ to consolidate session and agent lifecycle management, enforce turn serialization, and decouple event routing via a pub/sub EventBus. It integrates this layer into AgentPool with per-protocol feature flags for ACP and OpenCode, migrating both servers to use the new session-pool-backed handlers. The review feedback highlights critical improvements for robustness, including resolving a race condition in the auto-resume trigger, adding defensive validation for session IDs, gracefully handling stream closure exceptions in the event consumer, and wrapping resource cleanup loops in try-except blocks to guarantee complete teardown during shutdown and cleanup.

[Leoyzen]

修复完成：所有 Review Comments 已处理

感谢 @gemini-code-assist 的详细 review。以下是对 7 条 comment 的修复总结：

1. 🔴 Critical: _trigger_auto_resume Race Condition (core.py:826)

问题：注入在 _process_queued_work 排空队列后和 run_loop 释放锁之间到达，会永久卡住。

修复：移除了 if session.turn_lock.locked(): return 的 early return。现在 background task 会直接 async with session.turn_lock，让 late-arriving injection 安全地等待锁释放后被处理。

# Before
if session.turn_lock.locked():
    return
async with session.turn_lock:
    ...

# After
async with session.turn_lock:
    ...

2. 🟡 Medium: session_id 验证 (core.py:257)

修复：在 get_or_create_session 入口处添加验证：

if not session_id or not session_id.strip():
    raise ValueError("session_id cannot be empty or whitespace")

3. 🟡 Medium: ACP Event Consumer 连接异常 (acp/handler.py:140)

修复：区分处理连接关闭异常和真正的错误：

except (ConnectionResetError, BrokenPipeError) as e:
    logger.debug("Client connection closed gracefully", ...)
    break
except Exception as e:
    if isinstance(e, (anyio.ClosedResourceError, anyio.EndOfStream)):
        logger.debug("Stream closed gracefully", ...)
        break
    logger.exception("Failed to convert or send event", ...)

4. 🟡 Medium: OpenCode close_session 异常传播 (opencode/handler.py:277)

修复：将 contextlib.suppress 替换为显式的 try-except，确保非 CancelledError 异常被记录而不是传播：

try:
    await task
except asyncio.CancelledError:
    pass
except Exception:
    logger.exception("Unexpected exception during consumer task cancellation", ...)

5. 🟡 Medium: ACP close_session 异常传播 (acp/handler.py:239)

修复：在等待 consumer task 完成时，catch 所有意外异常并记录：

try:
    await asyncio.wait_for(task, timeout=5.0)
except TimeoutError:
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        pass
    except Exception:
        logger.exception("Unexpected exception during consumer task cancellation", ...)
except asyncio.CancelledError:
    pass
except Exception:
    logger.exception("Unexpected exception in consumer task during graceful shutdown", ...)

6. 🟡 Medium: SessionPool.shutdown 批量关闭容错 (core.py:922)

修复：每个 session 单独 try-except，一个失败不阻止其他关闭：

for session_id in active_sessions:
    try:
        await self.close_session(session_id)
    except Exception:
        logger.exception("Failed to close session during shutdown", session_id=session_id)

7. 🟡 Medium: TTL Cleanup 批量关闭容错 (core.py:498)

修复：同上，每个过期 session 单独 try-except：

for session_id in expired_sessions:
    try:
        if self._cleanup_callback is not None:
            await self._cleanup_callback(session_id)
        else:
            await self.close_session(session_id)
    except Exception:
        logger.exception("Failed to close expired session during cleanup", session_id=session_id)

验证

所有 170 个测试通过：

170 passed, 11 deselected

Commit: 371e66e10

[gemini-code-assist]

感谢你的详细总结，Leoyzen。这些修复非常到位，特别是针对 _trigger_auto_resume 中 Race Condition 的处理，以及在 SessionPool 和 ACP/OpenCode 处理器中增加的异常捕获与容错机制，这极大地增强了系统的健壮性。

看到所有 170 个测试均已通过，这非常令人欣慰。这些针对性修复（包括对 session_id 的验证和优雅的连接关闭处理）对于生产环境的稳定性至关重要。目前看来，该 PR 已经具备了合并的条件。

[Leoyzen]

/gemini review

[gemini-code-assist]

Code Review

This pull request refactors AgentPool's session and turn management by unifying them under SessionPool and integrating pydantic-graph for team and workflow execution. It also adopts PydanticAI's native pending message queue for native agents, making BaseAgent a stateless execution engine. The review feedback highlights several critical bugs, including a missing contextlib import in claude_code_agent.py and missing null guards for optional hook callbacks in eventbus_hooks_adapter.py that could cause runtime crashes. Additionally, the reviewer recommended performance optimizations for stack walking in base_agent.py and defensive null guards for session_pool access in both base_agent.py and context.py to prevent potential crashes before the pool is fully initialized.

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]

第三轮 Review 修复完成

感谢 @gemini-code-assist 的持续 review。以下是对本轮 7 条 comment 的修复总结：

🔴 High Priority Fixes

1. eventbus_hooks_adapter.py — Hook Wrapper None 检查（4 处）

所有 4 个 wrapper 函数现在都包含 if original is not None: 防御检查：

• _wrap_before_run (line 106): 在 await original(ctx) 前检查
• _wrap_after_run (line 121): 在 await original(ctx, result=result) 前检查，否则返回原始 result
• _wrap_before_tool_execute (line 150): 在 await original(...) 前检查，否则返回原始 args
• _wrap_after_tool_execute (line 181): 在 await original(...) 前检查，否则返回原始 result

根因：Hooks() 默认构造时所有 hook 属性为 None。之前的代码假设 wrapper 只在 hook 已注册时被调用，但 EventBusHooksAdapter.as_capability() 无条件包装所有 4 个 hook，导致未注册时为 None 的 hook 被直接 await。

2. claude_code_agent.py — contextlib.suppress → 显式 try-except (line 1281)

# Before
with contextlib.suppress(Exception):
    await run_ctx.event_bus.unsubscribe(...)

# After
try:
    await run_ctx.event_bus.unsubscribe(...)
except Exception:
    self.log.warning("Failed to unsubscribe from event bus during cleanup", ...)

虽然 contextlib 已导入，但显式 try-except 可以记录异常信息，且避免了对 contextlib.suppress 的依赖（如果未来重构移除 import）。

🟡 Medium Priority Fixes

3. base_agent.py — inspect.getmodule(frame) → frame.f_globals.get("__name__") (line 135)

# Before: O(n×m) 的 sys.modules 查找
module = inspect.getmodule(frame)
if module is not None:
    module_name = module.__name__

# After: O(1) 的直接 frame 属性访问
module_name = frame.f_globals.get("__name__", "")

_should_bypass_session_pool() 在每次 agent run 时调用，是热路径。inspect.getmodule() 在深层 async 调用栈中开销显著。

4. base_agent.py — assert → 显式 if 检查 (line 604)

# Before: assert 在 Python -O 模式下被 strip
assert session_pool is not None

# After: 显式检查，-O 安全
if session_pool is None:
    return None

5. context.py — pool.session_pool 空值保护 (line 208)

# Before: 只检查 pool is not None，未检查 pool.session_pool
if pool is not None:
    child_session = await pool.session_pool.create_session(...)

# After: 双重检查，统一 fallback
if pool is not None and pool.session_pool is not None:
    ...
# 单一 fallback 路径
return generate_session_id()

验证

462 passed, 2 skipped, 2953 deselected

Commit: ddf876cd6

[gemini-code-assist]

感谢你的详细总结，Leoyzen。这些修复非常到位，特别是针对 eventbus_hooks_adapter.py 中 Hook Wrapper 的 None 检查，以及对 base_agent.py 中热路径代码的优化，这极大地增强了系统的健壮性和性能。看到所有 462 个测试均已通过，这非常令人欣慰。目前看来，该 PR 已经具备了合并的条件。