fix(orchestrator): eliminate duplicate events in session_pool.run_stream - #338
Open
Leoyzen wants to merge 12 commits into
Open
fix(orchestrator): eliminate duplicate events in session_pool.run_stream#338Leoyzen wants to merge 12 commits into
Leoyzen wants to merge 12 commits into
Conversation
SessionPool._run_stream_run_turn() previously drained events from BOTH run_handle.start() (which publishes every event to the EventBus via ProtocolChannel.publish) AND a direct EventBus subscription, delivering each streamed event twice to the consumer. This surfaced in the opencode client as duplicated model output after slash/skill commands, which are the only production path consuming session_pool.run_stream(). Fix: drive start() as a background side-effect (discarding yields, exactly like SessionController._consume_run) and consume events from the EventBus subscription only — a single delivery path, matching the _consume_run/ACP architecture. Tool-published mid-turn events (SpawnSessionStart) arrive on the same EventBus path, preserving order. Adds an integration regression test asserting concatenated text deltas through session_pool.run_stream() match the model output exactly once.
4 tasks
Previous approach used a background asyncio task to drive start() while consuming from the EventBus subscription. This caused two regressions: 1. test_cross_provider_session_lifecycle (scope=descendants): the background task's CancelledError from our own cancel surfaced as a spurious run error. 2. test_subsequent_run_after_break: anyio cancel scopes cannot be exited from a different task, so cancelling the background driver deadlocked during GeneratorExit cleanup (180s hang). Fix: drive start() in the CURRENT task (like the original dual-drain), but yield events ONLY from the EventBus subscription (bus_queue), never from gen directly. start() publishes every event to the EventBus before yielding it (ProtocolChannel.publish runs before in _execute_turn), so each event is already in bus_queue when we drain it with get_nowait(). This gives a single delivery path with no background task, no cross-task cancel scope issues, and exactly-once events.
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.
Problem
The opencode client showed duplicated model output (every text token appearing twice), visible immediately after a slash/skill command (e.g.
/lodestone). Access-log evidence showed the same SSE delta delivered twice per event, and two concurrent assistant messages with identical content.Root Cause
SessionPool._run_stream_run_turn()(insrc/agentpool/orchestrator/session_pool_runs.py) drained events from two sources simultaneously:run_handle.start()directly — which already publishes every event to the EventBus (ProtocolChannel.publish→event_bus.publishinsideRunHandle._execute_turn)bus_queue)Each streamed event was therefore yielded twice to the consumer. The slash-command executor is the only production path consuming
session_pool.run_stream(), which is why normal prompts (which go through_consume_run→ EventBus-only) never showed the bug.The double-path was introduced in
a56b10419(session-debt-cleanup #171) as a patch with the intent of capturing tool-published mid-turn events (SpawnSessionStart), but it incorrectly assumedstart()does not publish to the same EventBus — a wrong-premise patch on top of the single-publish architecture.Fix
Consume events from the EventBus subscription only — a single delivery path:
start()as a background side-effect, discarding its yields — exactly likeSessionController._consume_run(async for _ in gen: pass) and the ACP path.bus_queueuntilStreamCompleteEvent/RunErrorEvent.finally: cancel the driver task,gen.aclose(), unsubscribe, clearcurrent_run_id, pop the run handle.This unifies
run_streamwith the_consume_run/ ACP architecture: one source of truth (EventBus), each event delivered exactly once, ordering preserved, tool-published events still delivered.Verification
tests/team_mode/test_session_pool_run_stream_no_dup.py(@pytest.mark.integration): drives a real AgentPool + FunctionModel streaming known chunks throughsession_pool.run_stream()and asserts the concatenated text matches the source exactly once. Failed on the old code (each chunk doubled), passes with the fix.tests/team_mode/suite: 318 passed.ruff check/ruff formatclean;mypyclean on the changed file.