feat(acp): batch SessionUpdate delivery during replay

[Million-mo]

Summary

Optimize ACP session/load replay by batching SessionUpdate delivery instead of sending each update as a separate session/update JSON-RPC notification.

Problem

replay() sends 300-600 individual session/update notifications for a 100-message session, each with a sequential await on TCP flush. Estimated load time: 150-300 seconds.

Solution

Collect all SessionUpdate objects first (pure CPU conversion, no I/O), then deliver in batches of 20 via _batch_session_updates ext_notification. Clients without batch support automatically fall back to sequential delivery.

Changes

• src/acp/agent/notifications.py:

  • Refactor _replay_request/_replay_response into pure _collect_request_updates/_collect_response_updates (return list[SessionUpdate], no I/O)
  • Add send_batch_update() — uses ext_notification("_batch_session_updates", ...) when supported, falls back to sequential session/update
  • Add notification_batch_size (default 20) and notification_flush_interval (default 0.0) to __init__
  • Add set_batch_support(bool) for capability-based opt-in
  • Rewrite replay() as collect-then-batch-send
  • Keep _replay_request/_replay_response as thin wrappers for backward compatibility

• src/agentpool_server/acp_server/session.py:

  • Detect batch support from client_capabilities.field_meta["_batch_session_updates"] in ACPSession.__post_init__

• tests/acp/test_notifications_replay.py: 6 new tests (batch mode, fallback, ordering, custom size, pure collector, empty)

• tests/acp/benchmark_replay.py: Benchmark script comparing batch vs sequential

Design

• Uses ACP ext_notification extension mechanism — no protocol spec changes
• Client advertises support via ClientCapabilities.field_meta["_batch_session_updates"]
• Default _batch_supported=False — sequential fallback when client doesn't opt in
• Ordering preserved: collect phase builds list in message order, batch chunking doesn't reorder

OpenSpec

Change: openspec/changes/acp-notification-batching/ (proposal + design + specs + tasks, all complete)

Test Results

pytest:  29 passed (20 replay + 9 acp_load)
ruff:    All checks passed
mypy:    Success: no issues found

Note

This PR is based on develop/agentic and is independent of PR #65 (eliminate-pool-level-agents). The only overlap is session.py (3 lines added in __post_init__), which should rebase cleanly.

Co-authored-by: Sisyphus clio-agent@sisyphuslabs.ai

[gemini-code-assist]

Code Review

This pull request implements batching for SessionUpdate delivery during session/load replay to optimize performance. It refactors replay logic into side-effect-free collectors and introduces the _batch_session_updates extension notification with a graceful sequential fallback. Feedback recommends validating the batch size and flush interval parameters during initialization to prevent runtime errors, and updating the proposal document to explicitly mark the rejected default flush interval option as 'Rejected' for historical context.

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.

[Million-mo]

/gemini review

[gemini-code-assist]

Code Review

This pull request introduces batching for SessionUpdate notifications during session replay (session/load) to reduce wire roundtrips and improve load times. It refactors the replay mechanism to collect updates in memory first, then chunks and delivers them via a new _batch_session_updates extension notification if supported by the client, with a graceful fallback to sequential delivery. The reviewer noted a potential serialization issue in send_batch_update where calling model_dump() without mode="json" in Pydantic v2 could leave raw Python objects (like datetime) in the payload, potentially causing JSON-RPC serialization failures.

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]

PR Review & Analysis

Overview

PR quality is solid — OpenSpec artifacts are complete, pure function separation (CPU vs I/O) is clean, backward compatibility is well handled, and test coverage is comprehensive. Below are some suggestions and analysis points for discussion.

---

1. Code Quality Notes

Done well:

• Pure function refactoring (_collect_request_updates / _collect_response_updates) makes conversion testable without a client connection — good separation of concerns
• Graceful fallback: default _batch_supported=False ensures clients without batch support are unaffected
• Constructor parameter validation (batch_size > 0, flush_interval >= 0)

Could improve:

1. notification_flush_interval is effectively dead code — default is 0.0, and the design doc explicitly states no artificial delay is needed (the await on each send_batch_update provides natural flow control). This is YAGNI — a config knob introduced without a real use case. Consider removing it until there's an actual need.

2. _tool_call_inputs cache state during collection — _collect_response_updates populates the cache, _collect_request_updates pops it. If message ordering is incorrect (e.g., a ModelRequest with ToolReturnPart appears before the corresponding ModelResponse with ToolCallPart), the cache won't match. The design doc mentions this risk but no defensive handling exists. Consider adding a guard or at least a debug log when pop() misses.

3. tests/acp/benchmark_replay.py is not a pytest test — it's a standalone script with if __name__ == "__main__". Placing it under tests/ means pytest's test discovery will import it. Consider moving it to scripts/ or benchmarks/.

4. test_from_config_capabilities_not_duplicated change — The monkeypatch.setenv("OPENAI_API_KEY", ...) fix appears unrelated to notification batching. Should this be a separate PR?

---

2. Why not asyncio.gather for concurrent sends?

Considered alternative: fire all session/update notifications as concurrent tasks, await once via asyncio.gather.

Doesn't work because the underlying transport is a single connection (stdio pipe or TCP socket). All tasks write to the same connection, which has a write lock / internal queue — writes are serialized regardless. asyncio.gather only eliminates Python-level await scheduling overhead (microseconds), not TCP flush cost (milliseconds). Plus, asyncio.gather doesn't guarantee completion order, which breaks ToolCallStart → ToolCallProgress ordering.

The batch approach is fundamentally better: it reduces wire-level message count (600 → 30), reducing JSON serialization, JSON-RPC framing, and client-side dispatch overhead — not just Python scheduling.

---

3. drain_and_merge reuse consideration

The codebase has a drain_and_merge() component in src/agentpool/orchestrator/event_bus.py that coalesces streaming events (text deltas, tool call deltas, progress) for protocol subscribers. Considered whether this could be reused in replay().

Conclusion: not directly reusable. They operate at different layers:

• drain_and_merge coalesces live streaming events (EventEnvelope / RichAgentStreamEvent) during active runs
• replay() converts historical ModelMessages to SessionUpdate objects for one-shot session loading

Different data types, different lifecycle (continuous vs one-shot). The PR's design doc explicitly considered and rejected a streaming pipe approach (Decision 2), choosing collect-then-send for simplicity.

However, if future work extends batching to live streaming notifications (currently a Non-Goal), the drain_and_merge pattern — particularly _merge_key() + _merge_envelopes() — would be worth referencing for coalescing before batch delivery.

---

4. Spec context: when is replay needed?

Per ACP spec (checked agent-client-protocol schema):

v1:

• session/load: "Stream the entire conversation history back to the client via notifications" — LoadSessionResponse only carries session metadata (modes, configOptions), no message history in response payload
• session/resume: Resume without replaying history

v2:

• Merged into session/resume with optional replayFrom parameter:
  • replayFrom: null → resume without replay (v1's session/resume)
  • replayFrom: { "type": "start" } → replay full history (v1's session/load)

So replay via notifications is required by spec — the response payload doesn't carry conversation history. The batching optimization in this PR is the right approach: it reduces notification count while staying spec-compliant.

A more fundamental fix (returning history in the response payload) would require a protocol spec change, which is out of scope.

---

Summary

PR is approvable with minor cleanup suggestions. The batching approach is spec-compliant and architecturally sound. Main actionable items:

• Consider removing notification_flush_interval (YAGNI)
• Move benchmark script out of tests/
• Split unrelated CI fix into separate PR

[Leoyzen]

@Million-mo ci 失败了