From d4074d67047394cdf2b3cea3100b12ab82753d31 Mon Sep 17 00:00:00 2001 From: Million <57204720+Million-mo@users.noreply.github.com> Date: Wed, 26 Aug 2026 16:41:24 +0800 Subject: [PATCH] fix(opencode): eliminate EventBus loopback causing duplicate SSE projections (#391) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(opencode): eliminate EventBus loopback causing duplicate SSE projections The OpenCodeEventBridge republished protocol projections (MessageUpdatedEvent, PartUpdatedEvent, etc.) back into the same EventBus that carries native agent events. This created a feedback loop: native events → event bridge → broadcast → EventBus republish → SSE delivery, causing duplicate renders in attached OpenCode TUI clients (issue #380). Architecture change — direct-wire SSE: - state.broadcast_event() now fans projections directly to per-connection SSE subscriber queues instead of republishing to EventBus - global_routes._event_generator reads from state.event_subscribers queues (no EventBus subscription, no CustomEvent unwrapping) - Reconnect replay via state.replay_projections() using Last-Event-ID - Deleted event_bridge.py (the loopback republisher) EventBus source isolation (defense-in-depth): - EventEnvelope gains source_hint field; publish() accepts source_hint - subscribe() accepts exclude_source param (filters both live fanout and replay) - ProtocolEventConsumerMixin hooks: _get_subscription_replay() and _get_subscription_exclude_source() (defaults: replay=True, exclude=None) - OpenCode overrides: replay=False, exclude_source={"opencode_event_bridge"} Session consumer replay alignment: - OpenCode session-level consumers now use replay=False (matching the global SSE endpoint's first-connect policy), preventing stale events from being redelivered on consumer startup Testing: - 7082 tests passed (full suite), ruff/mypy clean - New e2e test: test_attach_existing_session_first_prompt_renders_once - New unit tests: EventBus source_hint/exclude_source (4 tests) - Rewritten integration tests for direct-wire SSE model Note: A residual TUI-side duplication may still be visible in opencode attach mode due to the TUI's local echo (createUserMessage) not matching the server-generated message ID in the SSE event. This is tracked as an opencode TUI bug (sst/opencode#14372, #24773, #29478) with upstream fix PR #31945 still unmerged. The server-side fix in this commit eliminates the EventBus loopback path; the remaining duplication is purely client-side. * fix(opencode): address review — QueueFull policy, typed session-id extraction, mock alignment Review-driven fixes on PR #391 (direct-wire SSE loopback elimination): BLOCKER: broadcast_event now handles asyncio.QueueFull per-subscriber with the same drop-oldest policy as EventBus._enqueue, so a stalled SSE client can no longer abort fanout to every other subscriber. Adds structured warning logging on overflow plus a debug fanout log (telemetry on the delivery critical path). New regression test: test_broadcast_event_drop_oldest_on_queue_full. MAJOR: ServerState.extract_session_id now delegates to the typed global_routes._extract_session_id (match-based, no getattr) instead of the getattr probe that read info.id for MessageUpdatedEvent — buffering message.updated under per-message keys and leaking memory. Typed variant reads props.info.session_id as documented. Test mocks in test_global_event.py / test_sse_compliance.py aligned with production: deque(maxlen=100) buffers, drop-oldest overflow, typed extractor — the queue-full behavior is no longer suppressed out of the green suite. MINOR: replay_projections replays merged buffers in global event_id order (monotonic SSE ids for reconnecting clients) and stops with a warning on QueueFull instead of silently dropping a suffix. MINOR: _get_subscription_exclude_source documented as currently inert (its only producer was deleted with the loopback bridge; kept as defense-in-depth, exercised by unit tests). Nits: stale event_bridge docstrings/comment/names updated across test_event_pipeline_e2e.py and conftest.py; corrected the pre-existing dedup-set claim in src/wolfharness/AGENTS.md (the set is a private ACPEventConverter field, not on SessionController); changelog trailing newline; ADR eventbus-replay.md annotated as superseded by the direct-wire design (PR #391). Verified: ruff clean, mypy strict clean (686 files), 326 affected tests pass. --- ...26-08-25-fix-opencode-eventbus-loopback.md | 33 ++ docs/adr/eventbus-replay.md | 12 +- src/wolfharness/AGENTS.md | 2 +- src/wolfharness/orchestrator/event_bus.py | 64 +++- src/wolfharness_server/AGENTS.md | 2 +- src/wolfharness_server/mixins.py | 41 ++- .../opencode_server/event_bridge.py | 146 -------- .../opencode_server/opencode_event_bridge.py | 37 +- .../opencode_server/routes/global_routes.py | 122 +++---- .../opencode_server/state.py | 158 +++++++-- .../e2e/test_user_message_no_duplicate_e2e.py | 105 ++++++ tests/orchestrator/test_event_bus.py | 73 ++++ tests/servers/opencode_server/conftest.py | 20 +- .../opencode_server/test_event_bridge.py | 319 ++++++------------ .../test_event_pipeline_e2e.py | 87 ++--- .../opencode_server/test_global_event.py | 121 ++++--- .../test_message_id_alignment.py | 95 +++--- .../opencode_server/test_sse_compliance.py | 65 +++- tests/servers/test_subagent_event_mixin.py | 54 ++- 19 files changed, 918 insertions(+), 638 deletions(-) create mode 100644 changelog/unreleased/2026-08-25-fix-opencode-eventbus-loopback.md delete mode 100644 src/wolfharness_server/opencode_server/event_bridge.py diff --git a/changelog/unreleased/2026-08-25-fix-opencode-eventbus-loopback.md b/changelog/unreleased/2026-08-25-fix-opencode-eventbus-loopback.md new file mode 100644 index 000000000..6585ae3f1 --- /dev/null +++ b/changelog/unreleased/2026-08-25-fix-opencode-eventbus-loopback.md @@ -0,0 +1,33 @@ +# Fix EventBus loopback: duplicate first user message on serve + attach + +`wolfharness serve-opencode` + `opencode attach` rendered the first user +message twice in the TUI (issue #380). Root cause was a design flaw: the +OpenCode server republished its own protocol projections +(`MessageUpdatedEvent` / `PartUpdatedEvent`) back into the same EventBus +that carries native agent events, creating a feedback loop. Prior fixes had +patched each crossing of the loop (`60d6fda50` sync clears replay buffer, +`cde629b6d` first-connect `replay=False`, `47baa747b6` C4 consumer skip) +but never removed the loop itself. + +## What changed + +- **SSE direct-wire projections**: `ServerState.broadcast_event()` now + delivers OpenCode projections straight to per-connection SSE subscriber + queues. The `OpenCodeEventBridge` (which republished into the EventBus) is + deleted; the EventBus carries only native agent events. +- **Loopback isolation**: `EventBus.publish()` accepts a `source_hint` and + `EventBus.subscribe()` accepts `exclude_source`, so a producer can never + re-consume its own output. The OpenCode session consumer subscribes with + `exclude_source={"opencode_event_bridge"}` as belt-and-suspenders. +- **Replay alignment**: the OpenCode session consumer subscribes with + `replay=False` (matching the SSE endpoint's first-connect policy), so + replayed native events can no longer be re-converted into duplicate + projections. Recovery paths opt in to `replay=True` explicitly via a new + `_get_subscription_replay()` hook. +- **Reconnect replay preserved**: per-session projection buffers in + `ServerState` keep `Last-Event-ID` conditional replay working on the + direct-wire path. +- **ACP `_displayed_message_ids` retained**: it guards a separate + duplication vector (the same native event published twice with one + `message_id`), which this change does not eliminate; fixing that at the + publish source is a follow-up. diff --git a/docs/adr/eventbus-replay.md b/docs/adr/eventbus-replay.md index b2e2fd53f..e5587f730 100644 --- a/docs/adr/eventbus-replay.md +++ b/docs/adr/eventbus-replay.md @@ -1,7 +1,15 @@ # EventBus Replay Buffer Design -> **Status**: Accepted — this design has been implemented in the AgentPool codebase. -> See `docs/explanation/` for the current architecture documentation. +> **Status**: **Superseded** — the planned "Migration B" (SSE endpoints migrate +> from `state.event_subscribers` to EventBus-only subscription, with `ServerState` +> as a pure EventBus consumer) was **reversed** by PR #391 (issue #380): the +> OpenCode server now delivers projections **direct-wire** to per-connection SSE +> subscriber queues via `ServerState.broadcast_event`, and the EventBus replay +> buffer is no longer the SSE replay path. Reconnect replay is served by +> `ServerState.replay_projections()` (`Last-Event-ID`). This ADR is retained as +> historical record of the original replay-buffer design for the EventBus. +> +> Kept as historical record; see `docs/explanation/` for current architecture. ## Overview diff --git a/src/wolfharness/AGENTS.md b/src/wolfharness/AGENTS.md index 84d5236de..cd6192c81 100644 --- a/src/wolfharness/AGENTS.md +++ b/src/wolfharness/AGENTS.md @@ -155,7 +155,7 @@ A system-level event that carries inserted user message content through the even Internal paths (`"background_task"`, `"internal"`) are always displayed since they have no prior protocol emission. Protocol paths (`"protocol"`) may be deduplicated against the protocol handler's own ad-hoc emission. -**Dedup Mechanism**: The event carries `message_id`, which is shared with the protocol handler's ad-hoc emission path. Protocol handlers generate the ID first, register it in a per-session dedup set, emit the message to the client, then pass the same ID to `send_message()` -> `_route_message()`. The `ACPEventConverter` and `EventProcessor` check the dedup set and skip if the `message_id` is already displayed. The dedup set lives as `dict[str, set[str]]` on `SessionController` (keyed by `session_id`) and is passed to converters as `displayed_message_ids: set[str]`. +**Dedup Mechanism**: The event carries `message_id`, which is shared with the protocol handler's ad-hoc emission path. Protocol handlers generate the ID first, register it in a per-session dedup set, emit the message to the client, then pass the same ID to `send_message()` -> `_route_message()`. The `ACPEventConverter` checks the set and skips if the `message_id` is already displayed (guards dual-publish of the same native event — steer/followup fire-and-forget + `EnqueuedMessagesEvent`). The dedup set is a private per-converter field `_displayed_message_ids` on `ACPEventConverter` (keyed by `session_id`), populated by the ACP protocol handler's `_emit_user_message_chunks()`. The OpenCode `EventProcessor` has NO dedup set: it renders exactly once because OpenCode projections are delivered direct-wire to SSE subscriber queues (never republished into the EventBus — issue #380). **Publication Points**: The event is published from: - `SessionController._route_message()` — for all routing paths (initial, steer, followup), if EventBus is available diff --git a/src/wolfharness/orchestrator/event_bus.py b/src/wolfharness/orchestrator/event_bus.py index 06ba4d9aa..32d3081c9 100644 --- a/src/wolfharness/orchestrator/event_bus.py +++ b/src/wolfharness/orchestrator/event_bus.py @@ -77,6 +77,13 @@ class EventEnvelope: """The original event payload (unmodified).""" event_id: int = 0 """Monotonic event ID assigned at publish time (0 for legacy/test envelopes).""" + source_hint: str | None = None + """Provenance tag set at publish time (e.g. ``"opencode_event_bridge"``). + + Lets subscribers exclude events published by a specific source via + ``subscribe(exclude_source=...)``, preventing a producer from + re-consuming its own output (loopback isolation). + """ def __getattr__(self, name: str) -> Any: """Forward attribute access to the wrapped event.""" @@ -216,7 +223,10 @@ def _merge_progress_events(events: list[ToolCallProgressEvent]) -> ToolCallProgr def _rebind(template: EventEnvelope, new_event: Any) -> EventEnvelope: """Create new EventEnvelope with merged event, preserving source_session_id and event_id.""" return EventEnvelope( - source_session_id=template.source_session_id, event=new_event, event_id=template.event_id + source_session_id=template.source_session_id, + event=new_event, + event_id=template.event_id, + source_hint=template.source_hint, ) @@ -312,6 +322,9 @@ async def drain_and_merge( yield env +_SubscriberEntry = tuple[asyncio.Queue[EventEnvelope], str, frozenset[str]] + + class EventBus: """PubSub event bus for cross-turn event streaming. @@ -362,7 +375,7 @@ def __init__( f"Must be one of: {sorted(_VALID_OVERFLOW_POLICIES)}" ) - self._subscribers: dict[str, list[tuple[asyncio.Queue[EventEnvelope], str]]] = {} + self._subscribers: dict[str, list[_SubscriberEntry]] = {} self._session_tree: dict[str, list[str]] = {} self._lock = asyncio.Lock() self._max_queue_size = max_queue_size @@ -379,6 +392,7 @@ async def subscribe( *, replay: bool = True, last_event_id: int | None = None, + exclude_source: frozenset[str] | None = None, ) -> asyncio.Queue[EventEnvelope]: """Subscribe to events for a session. @@ -402,6 +416,11 @@ async def subscribe( For ``scope="all"``, the filtering applies per-buffer after collecting from all buffers. + ``exclude_source`` filters at publish fanout time: events published + with a matching ``source_hint`` are never queued to this subscriber + (live or replayed). This lets a producer subscribe to a bus without + re-consuming its own output (loopback isolation). + Args: session_id: The session to subscribe to. scope: Subscription scope - "session" (exact match), @@ -419,6 +438,8 @@ async def subscribe( events with ``event_id > last_event_id``. Gap detection falls back to full replay when the buffer is missing contiguous events. + exclude_source: Provenance tags whose published events this + subscriber must not receive. Returns: An ``asyncio.Queue`` to consume events from. @@ -426,7 +447,11 @@ async def subscribe( queue: asyncio.Queue[EventEnvelope] = asyncio.Queue(maxsize=self._max_queue_size) async with self._lock: - self._subscribers.setdefault(session_id, []).append((queue, scope)) + self._subscribers.setdefault(session_id, []).append(( + queue, + scope, + exclude_source or frozenset(), + )) if scope == "all": historical_events: list[EventEnvelope] = [] for buffer in self._replay_buffers.values(): @@ -452,6 +477,16 @@ async def subscribe( env for env in historical_events if env.event_id > last_event_id ] + # Apply source exclusion to replayed events too (same rule as live + # fanout in _send), so a subscriber never sees its own output even + # through the replay buffer. + if exclude_source and historical_events: + historical_events = [ + env + for env in historical_events + if env.source_hint is None or env.source_hint not in exclude_source + ] + for envelope in historical_events: try: queue.put_nowait(envelope) @@ -492,7 +527,7 @@ async def unsubscribe( async with self._lock: if session_id in self._subscribers: self._subscribers[session_id] = [ - (q, sc) for q, sc in self._subscribers[session_id] if q is not queue + (q, sc, ex) for q, sc, ex in self._subscribers[session_id] if q is not queue ] if not self._subscribers[session_id]: del self._subscribers[session_id] @@ -590,7 +625,9 @@ async def _send(self, session_id: str, envelope: EventEnvelope) -> None: targets: list[tuple[asyncio.Queue[EventEnvelope], str]] = [] for subscriber_sid, subscribers in self._subscribers.items(): - for queue, scope in subscribers: + for queue, scope, excluded in subscribers: + if envelope.source_hint is not None and envelope.source_hint in excluded: + continue if self._should_receive(session_id, subscriber_sid, scope): targets.append((queue, scope)) @@ -615,7 +652,13 @@ async def _send(self, session_id: str, envelope: EventEnvelope) -> None: with contextlib.suppress(Exception): queue.shutdown() - async def publish(self, session_id: str, event: Any) -> None: + async def publish( + self, + session_id: str, + event: Any, + *, + source_hint: str | None = None, + ) -> None: """Publish an event to all subscribers for a session. Wraps the event in an EventEnvelope and sends it directly via _send(). @@ -628,12 +671,17 @@ async def publish(self, session_id: str, event: Any) -> None: Args: session_id: The session that produced the event. event: The event to broadcast. + source_hint: Optional provenance tag so subscribers can exclude + this publisher's events (see ``subscribe(exclude_source=...)``). """ if isinstance(event, PartDeltaEvent) and event.delta is None: return self._event_counter += 1 envelope = EventEnvelope( - source_session_id=session_id, event=event, event_id=self._event_counter + source_session_id=session_id, + event=event, + event_id=self._event_counter, + source_hint=source_hint, ) await self._send(session_id, envelope) @@ -650,7 +698,7 @@ async def close_session(self, session_id: str) -> None: async with self._lock: subscribers = self._subscribers.pop(session_id, []) - queues = [queue for queue, _scope in subscribers] + queues = [queue for queue, _scope, _ex in subscribers] for queue in queues: with contextlib.suppress(Exception): diff --git a/src/wolfharness_server/AGENTS.md b/src/wolfharness_server/AGENTS.md index a398f055e..eb858b9ae 100644 --- a/src/wolfharness_server/AGENTS.md +++ b/src/wolfharness_server/AGENTS.md @@ -20,7 +20,7 @@ | ACP: syntax detection | `acp_server/syntax_detection.py` — maps file extensions/dotfiles to language identifiers | | ACP: slash commands | `acp_server/commands/skill_commands.py` (`ACPSkillBridge`) + `debug_commands.py` + `docs_commands/` | | ACP: session lifecycle | `acp_server/session.py` (`ACPSession`) + `session_manager.py` (`ACPSessionManager`) | -| OpenCode: event processing pipeline | `opencode_server/event_processor.py` → `stream_adapter.py` → `event_adapter.py` → `event_bridge.py` | +| OpenCode: event processing pipeline | `opencode_server/event_processor.py` → `stream_adapter.py` → `event_adapter.py` → `state.broadcast_event()` (direct-wire to SSE queues; no EventBus republish) | | OpenCode: session integration | `opencode_server/session_pool_integration.py` — `OpenCodeSessionPoolIntegration`, main bridge to SessionPool | | OpenCode: models (20 files) | `opencode_server/models/` — Pydantic models matching OpenCode API types (`parts.py`, `session.py`, `message.py`, `events.py`, etc.) | | OpenCode: route handlers (14 files) | `opencode_server/routes/` — `session_routes.py`, `message_routes.py`, `file_routes.py`, `config_routes.py`, `agent_routes.py`, etc. | diff --git a/src/wolfharness_server/mixins.py b/src/wolfharness_server/mixins.py index 826c20431..c599c408b 100644 --- a/src/wolfharness_server/mixins.py +++ b/src/wolfharness_server/mixins.py @@ -83,6 +83,42 @@ def _get_subscription_scope(self) -> str: """ return "descendants" + def _get_subscription_replay(self) -> bool: + """Return whether the session consumer replays buffered events on subscribe. + + Defaults to ``True`` (historical behavior). Protocol servers whose + client reloads full state via an explicit sync (e.g. OpenCode) may + override to ``False`` so the consumer aligns with the SSE endpoint's + first-connect policy — replaying buffered events would re-convert + native events into duplicate projections for messages the client has + already loaded. Crash-recovery paths that genuinely need historical + events must override this back to ``True``. + + Returns: + Whether the consumer should replay the EventBus replay buffer. + """ + return True + + def _get_subscription_exclude_source(self) -> frozenset[str] | None: + """Return sources whose published events this consumer must not receive. + + Defaults to ``None``. Protocol servers that republish their own + projections into the EventBus (loopback) should override to exclude + that source so a producer can never re-consume its own output. + + !!! note + Currently inert in production: the OpenCode server delivers + projections direct-wire to SSE queues and no producer calls + ``publish(..., source_hint=...)`` anymore (the loopback bridge + that did was removed). The hook is kept as defense-in-depth for + any future producer that re-enters the EventBus; unit tests + exercise both the hook and the ``exclude_source`` filter. + + Returns: + A frozenset of ``source_hint`` values to exclude, or ``None``. + """ + return None + async def _before_consumer_loop(self, session_id: str) -> None: # noqa: B027 """Hook called before the consumer loop starts reading from the stream. @@ -158,7 +194,10 @@ async def start_event_consumer(self, session_id: str) -> None: return receive_stream = await self.event_bus.subscribe( - session_id, scope=self._get_subscription_scope() + session_id, + scope=self._get_subscription_scope(), + replay=self._get_subscription_replay(), + exclude_source=self._get_subscription_exclude_source(), ) self._consumer_streams[session_id] = receive_stream diff --git a/src/wolfharness_server/opencode_server/event_bridge.py b/src/wolfharness_server/opencode_server/event_bridge.py deleted file mode 100644 index bf30c93af..000000000 --- a/src/wolfharness_server/opencode_server/event_bridge.py +++ /dev/null @@ -1,146 +0,0 @@ -"""Event bridge that publishes OpenCode events to the SessionPool EventBus. - -Provides :class:`OpenCodeEventBridge` which republishes events destined for -OpenCode clients to the SessionPool's :class:`EventBus` for EventBus-based -routing to all consumers. - -**Event flow** - -1. Caller invokes ``state.broadcast_event(event)`` (or ``bridge.publish(event)``). -2. Bridge extracts ``session_id`` from the event's ``properties``. -3. If a session_id is present, the event is wrapped in a - :class:`CustomEvent` and published to the EventBus for that session. -4. EventBus subscribers (status bridges, protocol adapters, SSE clients) - receive the wrapped event. -""" - -from __future__ import annotations - -from typing import TYPE_CHECKING, Any - -from wolfharness.agents.events.events import CustomEvent -from wolfharness.log import get_logger - - -if TYPE_CHECKING: - from wolfharness.orchestrator.core import EventBus - from wolfharness_server.opencode_server.models.events import Event - from wolfharness_server.opencode_server.state import ServerState - - -logger = get_logger(__name__) - - -class OpenCodeEventBridge: - """Bridge that publishes OpenCode events to the SessionPool EventBus. - - Every event published through this bridge is made available on the - SessionPool EventBus so that all consumers (SSE clients, status bridges, - protocol adapters) receive events via EventBus subscriptions. - - Args: - state: The OpenCode server state. - event_bus: The SessionPool EventBus to republish events into. - """ - - def __init__(self, state: ServerState, event_bus: EventBus) -> None: - """Initialize the bridge.""" - self._state = state - self._event_bus = event_bus - - async def publish(self, event: Event) -> None: - """Publish an event to the EventBus. - - Extracts ``session_id`` from the event properties. If a session_id - is present, the event is wrapped in a :class:`CustomEvent` and - published to the EventBus for that session. EventBus subscribers - (SSE clients, status bridges, protocol adapters) receive the - wrapped event. - - Args: - event: An OpenCode protocol event (e.g. ``SessionStatusEvent``, - ``PartUpdatedEvent``, ``MessageUpdatedEvent``). - """ - # Step 1: extract session_id - session_id = self._extract_session_id(event) - if session_id is None: - return - - # Step 2: wrap and republish to EventBus - wrapped = self._wrap_event(event) - try: - await self._event_bus.publish(session_id, wrapped) - except Exception: - logger.exception( - "Failed to republish event to EventBus", - session_id=session_id, - event_type=getattr(event, "type", "unknown"), - ) - - @staticmethod - def _extract_session_id(event: Event) -> str | None: - """Extract session_id from an OpenCode event's properties. - - Most session-scoped events inherit from ``SessionIdProperties`` and - expose ``properties.session_id`` directly. However, some event types - nest session_id deeper: - - - ``PartUpdatedEvent`` → ``properties.part.session_id`` - - ``SessionCreatedEvent`` / ``SessionUpdatedEvent`` → ``properties.info.id`` - - ``MessageUpdatedEvent`` → ``properties.info.session_id`` - - This method tries each known path in order and returns the first - non-None ``str`` result. - - Args: - event: The OpenCode event to inspect. - - Returns: - The session ID string, or ``None`` if the event is global. - """ - properties = getattr(event, "properties", None) - if properties is None: - return None - - # Fast path: direct session_id (SessionIdProperties subclasses) - session_id = getattr(properties, "session_id", None) - if isinstance(session_id, str): - return session_id - - # PartUpdatedEvent: session_id is at properties.part.session_id - part = getattr(properties, "part", None) - if part is not None: - sid = getattr(part, "session_id", None) - if isinstance(sid, str): - return sid - - # SessionCreated / SessionUpdated: session_id is at properties.info.id - info = getattr(properties, "info", None) - if info is not None: - sid = getattr(info, "id", None) - if isinstance(sid, str): - return sid - - return None - - @staticmethod - def _wrap_event(event: Event) -> CustomEvent[Any]: - """Wrap an OpenCode event in a :class:`CustomEvent`. - - The wrapped event preserves the original event as ``event_data`` and - uses the OpenCode event type (prefixed with ``opencode:``) as the - custom event type. This makes it easy for EventBus consumers to - distinguish OpenCode protocol events from native agent events. - - Args: - event: The OpenCode event to wrap. - - Returns: - A :class:`CustomEvent` carrying the original OpenCode event. - """ - event_type = getattr(event, "type", "opencode:unknown") - return CustomEvent( - event_data=event, - event_type=f"opencode:{event_type}", - source="opencode_event_bridge", - ) diff --git a/src/wolfharness_server/opencode_server/opencode_event_bridge.py b/src/wolfharness_server/opencode_server/opencode_event_bridge.py index 578f30e30..d8ff70c52 100644 --- a/src/wolfharness_server/opencode_server/opencode_event_bridge.py +++ b/src/wolfharness_server/opencode_server/opencode_event_bridge.py @@ -13,7 +13,6 @@ from typing import TYPE_CHECKING, Any, cast from wolfharness.agents.events.events import ( - CustomEvent, RunErrorEvent, RunFailedEvent, RunStartedEvent, @@ -150,6 +149,32 @@ def _get_subscription_scope(self) -> str: """ return "session" + def _get_subscription_replay(self) -> bool: + """Return whether the session consumer replays buffered events on subscribe. + + Overridden to ``False`` so the session consumer matches the SSE + endpoint's first-connect policy (``global_routes`` subscribes with + ``replay=False`` when no ``Last-Event-ID`` is present). The client + loads full session state via ``sync()``; replaying buffered native + events would re-convert them into duplicate projections for messages + the client has already loaded — the duplication race documented in + issue #380 and ``test_duplication_reproduction.py``. + + Crash-recovery paths that genuinely need historical events MUST opt + in by overriding this back to ``True``. + """ + return False + + def _get_subscription_exclude_source(self) -> frozenset[str] | None: + """Return sources whose published events this consumer must not receive. + + Overridden to exclude the OpenCode projection bridge source so this + consumer can never re-consume its own loopback output. Belt-and- + suspenders on top of removing the loopback republish (see the + ``fix-opencode-eventbus-loopback`` change). + """ + return frozenset({"opencode_event_bridge"}) + def _create_assistant_message(self, session_id: str) -> tuple[str, MessageWithParts]: """Create a fresh assistant message for a new turn. @@ -680,16 +705,6 @@ async def _handle_event( # noqa: PLR0915 case _: pass - # C4: CustomEvent wraps SSE broadcast events (e.g. - # SessionCreatedEvent) republished from the OpenCodeEventBridge. - # These are not real agent events and must NOT trigger assistant - # message registration. Only skip bridge-wrapped CustomEvents - # (source="opencode_event_bridge"); tool-emitted CustomEvents - # (source=None or tool name) may carry meaningful payload and - # should fall through to adapter processing. - if isinstance(event, CustomEvent) and event.source == "opencode_event_bridge": - return - ctx = self._contexts.get(session_id) if ctx is None: return diff --git a/src/wolfharness_server/opencode_server/routes/global_routes.py b/src/wolfharness_server/opencode_server/routes/global_routes.py index e09d2ab91..18bf30526 100644 --- a/src/wolfharness_server/opencode_server/routes/global_routes.py +++ b/src/wolfharness_server/opencode_server/routes/global_routes.py @@ -5,14 +5,13 @@ import asyncio import contextlib import json -from typing import TYPE_CHECKING, Any, cast +from typing import TYPE_CHECKING, Any import anyio from fastapi import APIRouter, Query from sse_starlette.sse import EventSourceResponse from wolfharness import log -from wolfharness.agents.events.events import CustomEvent from wolfharness_server.opencode_server.dependencies import StateDep from wolfharness_server.opencode_server.models import Event, GlobalEvent, HealthResponse from wolfharness_server.opencode_server.models.app import ( @@ -210,20 +209,21 @@ async def _event_generator( # noqa: PLR0915 ) -> AsyncGenerator[dict[str, Any]]: """Generate SSE events for connected clients. - Events are received from the SessionPool EventBus via a global - subscription (``"__global_sse__"``) and streamed to SSE clients. + Events are OpenCode protocol projections delivered directly from + ``ServerState``: ``broadcast_event`` fans projections out to each + connection's subscriber queue (no EventBus round-trip). Reconnecting + clients replay missed projections via ``Last-Event-ID``. Args: - state: The server state holding subscribers and event factory + state: The server state holding subscriber queues and projection buffers wrap_payload: Whether to wrap events in GlobalEvent envelopes last_event_id: The last event ID received by the client, for replay """ factory = state.get_event_factory() if wrap_payload else None - # Track connection for diagnostic subscriber counting. - # The queue is maintained only for the event_subscribers counter; - # actual event delivery now flows exclusively through the EventBus. - _sentinel_queue: asyncio.Queue[Event] = asyncio.Queue() - state.event_subscribers.append(_sentinel_queue) + # Each SSE connection owns a queue registered with ServerState. Live + # projections are delivered here by broadcast_event. + queue: asyncio.Queue[tuple[int, Event]] = asyncio.Queue(maxsize=1000) + state.event_subscribers.append(queue) subscriber_count = len(state.event_subscribers) logger.info("SSE: New client connected (total subscribers: %s)", subscriber_count) @@ -236,23 +236,13 @@ async def _event_generator( # noqa: PLR0915 state._first_subscriber_triggered = True state.create_background_task(state.on_first_subscriber(), name="on_first_subscriber") - # Subscribe to EventBus for global SSE events. - # When the client provides a Last-Event-ID, use conditional replay: only - # events with event_id > last_event_id are replayed. When no Last-Event-ID - # is provided, replay all buffered events (default behavior). + # Reconnect replay: when the client provides a Last-Event-ID, replay the + # buffered projections with event_id > last_event_id before live events. + # First connections (no Last-Event-ID) get live events only — the client + # loads full state via sync() (matches the pre-loopback SSE policy). parsed_last_event_id: int | None = int(last_event_id) if last_event_id is not None else None - event_bus_stream: asyncio.Queue[Any] | None = None - session_controller = getattr(state, "session_controller", None) - if session_controller is not None: - session_pool = getattr(state.pool, "session_pool", None) - if session_pool is not None: - event_bus = session_pool.event_bus - event_bus_stream = await event_bus.subscribe( - "__global_sse__", - scope="all", - replay=parsed_last_event_id is not None, - last_event_id=parsed_last_event_id, - ) + if parsed_last_event_id is not None: + state.replay_projections(queue, parsed_last_event_id) try: # Send initial connected event @@ -261,64 +251,40 @@ async def _event_generator( # noqa: PLR0915 logger.info("SSE: Sending connected event", data=data) yield {"data": data, "id": "0"} - if event_bus_stream is None: - while True: - await asyncio.sleep(10.0) + while True: + try: + with anyio.fail_after(10.0): + event_id, event = await queue.get() + except TimeoutError: heartbeat = ServerHeartbeatEvent() data = _serialize_event(heartbeat, wrap_payload=wrap_payload) yield {"data": data} - else: - while True: - try: - with anyio.fail_after(10.0): - raw_event = await event_bus_stream.get() - except TimeoutError: - heartbeat = ServerHeartbeatEvent() - data = _serialize_event(heartbeat, wrap_payload=wrap_payload) - yield {"data": data} - continue - except asyncio.QueueShutDown: - break - - from wolfharness.orchestrator.core import EventEnvelope - - inner_event: Any - inner_event = raw_event.event if isinstance(raw_event, EventEnvelope) else raw_event - - if isinstance(inner_event, CustomEvent): - if inner_event.event_data is None: - continue - event = cast(Event, inner_event.event_data) - else: - event = inner_event - - if not hasattr(event, "type"): - continue - if factory is not None and not isinstance( - event, ServerHeartbeatEvent | ServerConnectedEvent - ): - data = factory.wrap(event) - elif wrap_payload: - data = _serialize_event(event, wrap_payload=True) - else: - data = _serialize_event(event) - logger.debug( - "SSE: Sending event", - event_type=getattr(event, "type", "unknown"), - session_id=_extract_session_id(event) or "-", - ) - event_id = raw_event.event_id if isinstance(raw_event, EventEnvelope) else 0 - yield {"data": data, "id": str(event_id)} + continue + except asyncio.QueueShutDown: + break + + if not hasattr(event, "type"): + continue + if factory is not None and not isinstance( + event, ServerHeartbeatEvent | ServerConnectedEvent + ): + data = factory.wrap(event) + elif wrap_payload: + data = _serialize_event(event, wrap_payload=True) + else: + data = _serialize_event(event) + logger.debug( + "SSE: Sending event", + event_type=getattr(event, "type", "unknown"), + session_id=_extract_session_id(event) or "-", + ) + yield {"data": data, "id": str(event_id)} finally: - if event_bus_stream is not None: - session_pool = getattr(state.pool, "session_pool", None) - if session_pool is not None: - with contextlib.suppress(Exception): - await session_pool.event_bus.unsubscribe("__global_sse__", event_bus_stream) - # Remove from subscriber count with contextlib.suppress(ValueError): - state.event_subscribers.remove(_sentinel_queue) + state.event_subscribers.remove(queue) + queue.shutdown() # Cancel any pending questions when the SSE client disconnects + session_controller = getattr(state, "session_controller", None) if session_controller is not None: cancelled = session_controller.cancel_all_pending_questions() else: diff --git a/src/wolfharness_server/opencode_server/state.py b/src/wolfharness_server/opencode_server/state.py index 5676e965a..17c52bd0c 100644 --- a/src/wolfharness_server/opencode_server/state.py +++ b/src/wolfharness_server/opencode_server/state.py @@ -3,7 +3,9 @@ from __future__ import annotations import asyncio +from collections import deque from collections.abc import Callable, Coroutine +import contextlib from dataclasses import dataclass, field from pathlib import Path import time @@ -79,7 +81,11 @@ class ServerState: agent_lock: asyncio.Lock = field(default_factory=asyncio.Lock) reverted_messages: dict[str, list[MessageWithParts]] = field(default_factory=dict) messages: dict[str, list[MessageWithParts]] = field(default_factory=dict) - event_subscribers: list[asyncio.Queue[Event]] = field(default_factory=list) + event_subscribers: list[asyncio.Queue[tuple[int, Event]]] = field(default_factory=list) + _projection_buffers: dict[str, deque[tuple[int, Event]]] = field( + default_factory=dict, repr=False + ) + _projection_counter: int = field(default=0, repr=False) _event_factory: GlobalEventFactory | None = field(default=None, repr=False) on_first_subscriber: OnFirstSubscriberCallback | None = None _first_subscriber_triggered: bool = field(default=False, repr=False) @@ -93,7 +99,6 @@ class ServerState: _mcp_tool_change_task: Any = field(default=None, repr=False) session_pool_integration: Any = field(default=None) session_controller: SessionController | None = field(default=None) - event_bridge: Any = field(default=None, repr=False) _shell_env: Any = field(default=None, repr=False) @staticmethod @@ -180,22 +185,33 @@ def __post_init__(self) -> None: # Fallback: reference the same env (preserves remote env support) self._shell_env = agent_env - # Instantiate the OpenCodeEventBridge when a SessionController is - # available. The bridge dual-publishes events to SSE subscribers - # (backward compat) and the SessionPool EventBus. - if self.session_controller is not None: - event_bus = None - if self._pool is not None: - session_pool = getattr(self._pool, "session_pool", None) - if session_pool is not None: - event_bus = getattr(session_pool, "event_bus", None) - - if event_bus is not None: - from wolfharness_server.opencode_server.event_bridge import ( - OpenCodeEventBridge, - ) + @staticmethod + def extract_session_id(event: Event) -> str | None: + """Extract the session ID from an OpenCode event's properties. + + Most session-scoped events inherit from ``SessionIdProperties`` and + expose ``properties.session_id`` directly. However, some event types + nest session_id deeper: + + - ``PartUpdatedEvent`` → ``properties.part.session_id`` + - ``SessionCreatedEvent`` / ``SessionUpdatedEvent`` → ``properties.info.id`` + - ``MessageUpdatedEvent`` → ``properties.info.session_id`` + + The typed ``match`` implementation in ``global_routes._extract_session_id`` + is the single source of truth; it is imported lazily to avoid a circular + import (same pattern as :meth:`get_event_factory`). + + Args: + event: The OpenCode event to inspect. + + Returns: + The session ID string, or ``None`` if the event is global. + """ + from wolfharness_server.opencode_server.routes.global_routes import ( + _extract_session_id, + ) - self.event_bridge = OpenCodeEventBridge(self, event_bus) + return _extract_session_id(event) def get_event_factory(self) -> GlobalEventFactory: """Get or lazily create the GlobalEventFactory for event wrapping. @@ -370,19 +386,105 @@ async def cleanup_tasks(self) -> None: self.background_tasks.clear() async def broadcast_event(self, event: Event) -> None: - """Broadcast an event via the EventBus bridge. + """Broadcast an OpenCode protocol event directly to SSE subscribers. + + Projections are delivered straight to the connected global SSE + subscriber queues (the client routes per-session, matching the + previous ``scope="all"`` bus stream semantics). Each event is + assigned a monotonic id and appended to a per-session projection + buffer so a reconnecting client can replay via ``Last-Event-ID``. + + Events are deliberately NOT republished into the SessionPool + EventBus: the bus carries only native agent events, and republishing + derived projections into the source bus creates the feedback loop + that caused duplicate user-message rendering (issue #380). + + Args: + event: The OpenCode protocol event to broadcast. + """ + self._projection_counter += 1 + event_id = self._projection_counter + session_id = self.extract_session_id(event) + if session_id is not None: + self._projection_buffers.setdefault(session_id, deque(maxlen=100)).append(( + event_id, + event, + )) + + # Fanout is atomic on the event loop (no await in the loop). A + # stalled/slow SSE client must not abort delivery to the other + # subscribers, so overflow is handled per-queue with a drop-oldest + # policy (mirrors EventBus._enqueue). Dead queues are collected and + # pruned after the loop. + dead_queues: list[asyncio.Queue[tuple[int, Event]]] = [] + overflow_count = 0 + for queue in self.event_subscribers: + try: + queue.put_nowait((event_id, event)) + except asyncio.QueueShutDown: + dead_queues.append(queue) + except asyncio.QueueFull: + # Evict the oldest buffered item and retry once; if still + # full the client is irrecoverably stalled, so drop the + # event for it and keep fanning out to everyone else. + with contextlib.suppress(asyncio.QueueEmpty): + queue.get_nowait() + try: + queue.put_nowait((event_id, event)) + except asyncio.QueueFull: + overflow_count += 1 + logger.warning( + "SSE subscriber queue overflow — dropping event for stalled client", + event_type=getattr(event, "type", "unknown"), + queue_size=queue.qsize(), + ) + for queue in dead_queues: + with contextlib.suppress(ValueError): + self.event_subscribers.remove(queue) + logger.debug( + "SSE: Broadcast event to subscribers", + event_type=getattr(event, "type", "unknown"), + event_id=event_id, + session_id=session_id or "-", + subscriber_count=len(self.event_subscribers), + dropped=overflow_count, + ) + + def replay_projections( + self, + queue: asyncio.Queue[tuple[int, Event]], + last_event_id: int, + ) -> None: + """Enqueue buffered projections with ``event_id > last_event_id``. + + The per-session buffers are merged and sorted by the single global + projection counter so a reconnecting client receives a monotonic + ``id:`` sequence. - When :attr:`event_bridge` is present, delegates to the bridge which - publishes the event to the SessionPool EventBus. Otherwise, the - event is silently dropped (no event delivery path available). + Args: + queue: The subscriber queue to enqueue replayed events into. + last_event_id: The client's last seen projection id. """ - if self.event_bridge is not None: - await self.event_bridge.publish(event) - else: - logger.debug( - "broadcast_event: no event_bridge, skipping event", - event_type=getattr(event, "type", "unknown"), - ) + # Merge all session buffers and replay in global event_id order. + buffered: list[tuple[int, Event]] = [] + for session_buf in self._projection_buffers.values(): + buffered.extend(session_buf) + for event_id, event in sorted(buffered, key=lambda item: item[0]): + if event_id <= last_event_id: + continue + try: + queue.put_nowait((event_id, event)) + except asyncio.QueueShutDown: + return + except asyncio.QueueFull: + # Reconnecting client stalled mid-replay — stop instead of + # silently dropping a suffix of the sequence. + logger.warning( + "SSE replay queue full — aborting replay for stalled client", + last_event_id=last_event_id, + event_id=event_id, + ) + return async def mark_session_idle(self, session_id: str) -> None: """Mark a session idle and broadcast the matching status events.""" diff --git a/tests/e2e/test_user_message_no_duplicate_e2e.py b/tests/e2e/test_user_message_no_duplicate_e2e.py index 952569a9b..ef081e6ba 100644 --- a/tests/e2e/test_user_message_no_duplicate_e2e.py +++ b/tests/e2e/test_user_message_no_duplicate_e2e.py @@ -265,3 +265,108 @@ async def send_prompt() -> None: f"Found {len(sse_part_updated_events)} part.updated events. " f"Text parts from SSE: {sse_user_text_parts}" ) + + +async def test_attach_existing_session_first_prompt_renders_once( + subprocess_server_simple: SubprocessServer, +) -> None: + """L4a: Attach to a pre-existing session, first prompt renders once. + + Reproduces issue #380 exactly: ``serve-opencode`` + ``opencode attach`` + against a session that already has history. The session consumer is + running and the replay buffer is populated; the first prompt on the + attached stream must produce exactly ONE ``message.updated`` per + ``message_id`` (no EventBus loopback, no replay re-conversion). + """ + base_url = subprocess_server_simple.base_url + marker_a = "PRE_EXISTING_TURN" + marker_b = "ATTACH_FIRST_PROMPT" + + async with ( + httpx.AsyncClient(timeout=60.0) as sse_client, + httpx.AsyncClient(timeout=60.0) as http_client, + ): + session_id = await _create_session(base_url, http_client) + + # Turn 1: pre-existing history on the session. + resp = await http_client.post( + f"{base_url}/session/{session_id}/prompt_async", + json={"parts": [{"type": "text", "text": marker_a}]}, + ) + assert resp.status_code == 204, f"first prompt_async failed: {resp.status_code}" + # Wait for the session to go idle (turn complete). + deadline = time.monotonic() + 20.0 + while time.monotonic() < deadline: + resp = await http_client.get(f"{base_url}/session/{session_id}") + if resp.status_code == 200: + data = resp.json() + if data.get("status") == "idle": + break + await asyncio.sleep(0.5) + + # Turn 2: attach a fresh SSE stream (the "attach" connection) and send + # the first prompt on it. + sse_events: list[dict[str, Any]] = [] + saw_busy = asyncio.Event() + + async def collect_sse_events() -> None: + deadline = time.monotonic() + 30.0 + async with sse_client.stream("GET", f"{base_url}/event") as sse_response: + assert sse_response.status_code == 200 + async for line in sse_response.aiter_lines(): + event = _parse_sse_line(line) + if event is not None: + sse_events.append(event) + if _is_session_busy(event): + saw_busy.set() + if _is_session_idle(event) and saw_busy.is_set(): + break + if time.monotonic() > deadline: + break + + async def send_prompt() -> None: + await asyncio.sleep(0.5) # Wait for SSE to connect + resp = await http_client.post( + f"{base_url}/session/{session_id}/prompt_async", + json={"parts": [{"type": "text", "text": marker_b}]}, + ) + assert resp.status_code == 204, ( + f"attach prompt_async failed: {resp.status_code}: {resp.text}" + ) + + sse_task = asyncio.create_task(collect_sse_events(), name="attach_sse_collector") + prompt_task = asyncio.create_task(send_prompt(), name="attach_prompt_sender") + + try: + await asyncio.wait_for(asyncio.gather(sse_task, prompt_task), timeout=35.0) + except TimeoutError: + pytest.fail( + f"Timed out waiting for attach SSE collection. " + f"Collected {len(sse_events)} events. " + f"Event types: {[e.get('type') for e in sse_events]}" + ) + + # --- Assertion: exactly 1 message.updated per user message_id on attach --- + message_updated_infos = [ + info for event in sse_events if (info := _get_message_updated_info(event)) is not None + ] + sse_user_msgs = [info for info in message_updated_infos if info.get("role") == "user"] + + from collections import Counter + + user_id_counts = Counter(info["id"] for info in sse_user_msgs) + assert user_id_counts, "No user message.updated events collected on attach stream" + + duplicate_ids = {mid for mid, count in user_id_counts.items() if count > 1} + assert not duplicate_ids, ( + f"Duplicate message.updated deliveries on attach for message_ids: {duplicate_ids}. " + f"All user message infos: {sse_user_msgs}" + ) + + # The attach stream connects after turn 1 (replay=False on first connect), + # so it should see exactly ONE user message (turn 2's first prompt) and + # that message must be delivered exactly once. + assert len(user_id_counts) == 1, ( + f"Attach stream should see exactly 1 user message, got {len(user_id_counts)}. " + f"All user message infos: {sse_user_msgs}" + ) diff --git a/tests/orchestrator/test_event_bus.py b/tests/orchestrator/test_event_bus.py index a455cb223..6b14cefa9 100644 --- a/tests/orchestrator/test_event_bus.py +++ b/tests/orchestrator/test_event_bus.py @@ -577,6 +577,79 @@ async def test_event_ordering_mixed_sessions() -> None: assert e.source_session_id == "sess-1" +# --------------------------------------------------------------------------- +# Source provenance (source_hint / exclude_source) +# --------------------------------------------------------------------------- + + +@pytest.mark.anyio +async def test_publish_source_hint_stored_on_envelope(event_bus: EventBus) -> None: + """Envelope carries the publish-time source_hint for provenance filtering.""" + stream = await event_bus.subscribe("sess-1") + await event_bus.publish( + "sess-1", + RunStartedEvent(session_id="sess-1", run_id="run-1"), + source_hint="opencode_event_bridge", + ) + received = await _receive_one(stream) + assert received is not None + assert received.source_hint == "opencode_event_bridge" + # Native events without a hint default to None + await event_bus.publish("sess-1", RunStartedEvent(session_id="sess-1", run_id="run-2")) + received2 = await _receive_one(stream) + assert received2 is not None + assert received2.source_hint is None + + +@pytest.mark.anyio +async def test_subscribe_exclude_source_filters_fanout(event_bus: EventBus) -> None: + """Subscriber with exclude_source never receives events from that source.""" + # Control subscriber receives everything. + control = await event_bus.subscribe("sess-1") + # Excluding subscriber must not see opencode_event_bridge events. + excluding = await event_bus.subscribe( + "sess-1", exclude_source=frozenset({"opencode_event_bridge"}) + ) + + await event_bus.publish( + "sess-1", + RunStartedEvent(session_id="sess-1", run_id="bridge-run"), + source_hint="opencode_event_bridge", + ) + # Live fanout: excluded subscriber gets nothing, control gets the event. + assert await _receive_one(excluding, timeout=0.1) is None + control_received = await _receive_one(control) + assert control_received is not None + assert control_received.event.run_id == "bridge-run" + + +@pytest.mark.anyio +async def test_exclude_source_does_not_block_native_events(event_bus: EventBus) -> None: + """Native events (no source_hint) still reach the excluding subscriber.""" + stream = await event_bus.subscribe( + "sess-1", exclude_source=frozenset({"opencode_event_bridge"}) + ) + await event_bus.publish("sess-1", RunStartedEvent(session_id="sess-1", run_id="native-run")) + received = await _receive_one(stream) + assert received is not None + assert received.event.run_id == "native-run" + + +@pytest.mark.anyio +async def test_exclude_source_filters_replayed_events(event_bus: EventBus) -> None: + """Replay buffer delivery honors exclude_source (loopback isolation).""" + await event_bus.publish( + "sess-1", + RunStartedEvent(session_id="sess-1", run_id="bridge-run"), + source_hint="opencode_event_bridge", + ) + stream = await event_bus.subscribe( + "sess-1", exclude_source=frozenset({"opencode_event_bridge"}) + ) + received = await _drain_stream(stream) + assert len(received) == 0 + + # --------------------------------------------------------------------------- # Descendants scope # --------------------------------------------------------------------------- diff --git a/tests/servers/opencode_server/conftest.py b/tests/servers/opencode_server/conftest.py index cf7391833..d3a05aa6a 100644 --- a/tests/servers/opencode_server/conftest.py +++ b/tests/servers/opencode_server/conftest.py @@ -67,6 +67,7 @@ async def _subscribe( *, replay: bool = True, last_event_id: int | None = None, + exclude_source: frozenset[str] | None = None, ) -> asyncio.Queue[Any]: queue: asyncio.Queue[Any] = asyncio.Queue(maxsize=_stream_buffer_size) _subscribers.setdefault(session_id, []).append((queue, scope)) @@ -82,7 +83,7 @@ async def _unsubscribe(session_id: str, queue: asyncio.Queue[Any]) -> None: with contextlib.suppress(asyncio.QueueShutDown): queue.shutdown() - async def _publish(session_id: str, event: Any) -> None: + async def _publish(session_id: str, event: Any, *, source_hint: str | None = None) -> None: for subscriber_sid, subscribers in _subscribers.items(): for queue, scope in subscribers: if scope == "all" or subscriber_sid == session_id: @@ -587,10 +588,10 @@ async def _list_sessions(**kwargs: object) -> list[SessionData]: @pytest.fixture -def server_state(tmp_project_dir: Path, mock_agent: Mock) -> ServerState: # noqa: PLR0915 +def server_state(tmp_project_dir: Path, mock_agent: Mock) -> ServerState: """Create a server state for testing.""" # Extract session_controller from mock pool so _event_generator can - # subscribe to the EventBus and receive events broadcast via event_bridge. + # reach state fields (event_subscribers) and broadcast via broadcast_event. session_controller = None session_pool = getattr(mock_agent.host_context, "session_pool", None) if session_pool is not None: @@ -719,17 +720,8 @@ async def _mock_route_message( return result state.session_pool_integration.route_message = AsyncMock(side_effect=_mock_route_message) - # event_bridge is automatically set up by __post_init__ when - # session_controller is present, but ensure it's initialized for cases - # where the mock pool's event_bus isn't available at construction time. - if state.event_bridge is None and state._pool is not None: - from wolfharness_server.opencode_server.event_bridge import OpenCodeEventBridge - - session_pool = getattr(state._pool, "session_pool", None) - if session_pool is not None: - event_bus = getattr(session_pool, "event_bus", None) - if event_bus is not None: - state.event_bridge = OpenCodeEventBridge(state, event_bus) + # Projection delivery is direct: broadcast_event fans out to + # state.event_subscribers queues. No EventBus bridge is wired. return state diff --git a/tests/servers/opencode_server/test_event_bridge.py b/tests/servers/opencode_server/test_event_bridge.py index de6ecab8d..318f78fe8 100644 --- a/tests/servers/opencode_server/test_event_bridge.py +++ b/tests/servers/opencode_server/test_event_bridge.py @@ -1,8 +1,9 @@ -"""Tests for OpenCodeEventBridge behavior parity. +"""Tests for ServerState direct-wire projection delivery. -Validates that the event bridge correctly dual-publishes events to both -legacy SSE subscribers and the SessionPool EventBus, while preserving -backward compatibility for the legacy path. +Validates that ``broadcast_event`` delivers OpenCode protocol projections +directly to SSE subscriber queues (no EventBus round-trip), buffers them +per-session for ``Last-Event-ID`` replay, and never republishes projections +into the SessionPool EventBus (loopback elimination, issue #380). """ from __future__ import annotations @@ -14,7 +15,6 @@ import pytest from wolfharness.agents.events.events import ( - CustomEvent, RunFailedEvent, RunStartedEvent, StreamCompleteEvent, @@ -34,7 +34,10 @@ SessionStatus, SessionStatusEvent, ) -from wolfharness_server.opencode_server.models.events import ServerConnectedEvent +from wolfharness_server.opencode_server.models.events import ( + ServerConnectedEvent, + ServerHeartbeatEvent, +) from wolfharness_server.opencode_server.opencode_event_bridge import ( OpenCodeEventBridgeMixin, ) @@ -48,8 +51,6 @@ from collections.abc import AsyncIterator from pathlib import Path - from wolfharness_server.opencode_server.models.events import Event - # ============================================================================= # Fixtures @@ -57,260 +58,162 @@ @pytest.fixture -def bridged_state(tmp_project_dir: Path, mock_agent: Mock) -> ServerState: - """Create a ServerState with an active OpenCodeEventBridge.""" - from wolfharness.orchestrator.core import EventBus - - # Wire a real EventBus into the mock pool so __post_init__ can discover it +def state_with_pool(tmp_project_dir: Path, mock_agent: Mock) -> ServerState: + """Create a ServerState wired to a real SessionPool EventBus.""" mock_agent.host_context.session_pool.event_bus = EventBus() return ServerState( working_dir=str(tmp_project_dir), agent=mock_agent, - session_controller=Mock(), # non-None triggers bridge instantiation + session_controller=Mock(), ) -@pytest.fixture -def event_bus(bridged_state: ServerState) -> EventBus: - """Return the EventBus attached to the bridged state.""" - assert bridged_state.event_bridge is not None - return bridged_state.event_bridge._event_bus - - # ============================================================================= -# Legacy path tests (no session_controller) +# Direct-wire delivery tests # ============================================================================= @pytest.mark.anyio -async def test_legacy_path_no_event_push( - tmp_project_dir: Path, - mock_agent: Mock, +async def test_broadcast_event_fans_out_to_subscriber_queues( + state_with_pool: ServerState, ) -> None: - """Without a session_controller, broadcast_event is a no-op (no queue push). - - The legacy path that pushed events directly to event_subscribers queues - has been removed. Event delivery now flows exclusively through the EventBus - via the event bridge. When no bridge is available, broadcast_event logs - a debug message and returns without pushing to any queues. - """ - state = ServerState( - working_dir=str(tmp_project_dir), - agent=mock_agent, - session_controller=None, - ) - queue: asyncio.Queue[Any] = asyncio.Queue() - state.event_subscribers.append(queue) + """Projections are delivered to every registered SSE subscriber queue.""" + queue_a: asyncio.Queue[Any] = asyncio.Queue() + queue_b: asyncio.Queue[Any] = asyncio.Queue() + state_with_pool.event_subscribers.extend([queue_a, queue_b]) - event = SessionStatusEvent.create("sess-legacy", SessionStatus(type="busy")) - await state.broadcast_event(event) + event = SessionStatusEvent.create("sess-1", SessionStatus(type="busy")) + await state_with_pool.broadcast_event(event) - # No events should be pushed to subscriber queues - assert queue.qsize() == 0 + for queue in (queue_a, queue_b): + event_id, received = queue.get_nowait() + assert received is event + assert event_id > 0 + assert queue_a.qsize() == 0 @pytest.mark.anyio -async def test_legacy_path_no_bridge_created( - tmp_project_dir: Path, - mock_agent: Mock, +async def test_broadcast_event_drop_oldest_on_queue_full( + state_with_pool: ServerState, ) -> None: - """ServerState without session_controller has no event_bridge.""" - state = ServerState( - working_dir=str(tmp_project_dir), - agent=mock_agent, - session_controller=None, - ) - assert state.event_bridge is None - + """A full subscriber queue drops its oldest item, never aborts fanout. -# ============================================================================= -# SessionPool path tests (bridge active) -# ============================================================================= - - -@pytest.mark.anyio -async def test_session_pool_path_does_not_push_to_sse_queues( - bridged_state: ServerState, -) -> None: - """With the bridge active, events flow through EventBus only (not SSE queues). - - The legacy event_subscribers queue push has been removed from - event_bridge.publish(). Events are delivered exclusively through - the EventBus subscription mechanism. + Regression test for the review finding that ``broadcast_event`` only + caught ``QueueShutDown``: a stalled SSE client filling its + ``maxsize`` queue would propagate ``QueueFull`` into every call site + and abort delivery to the *other* subscribers. Production now applies + the same drop-oldest policy as ``EventBus._enqueue``. """ - queue: asyncio.Queue[Any] = asyncio.Queue() - bridged_state.event_subscribers.append(queue) + stalled: asyncio.Queue[Any] = asyncio.Queue(maxsize=1) + healthy: asyncio.Queue[Any] = asyncio.Queue() + state_with_pool.event_subscribers.extend([stalled, healthy]) - event = SessionStatusEvent.create("sess-pool", SessionStatus(type="busy")) - await bridged_state.broadcast_event(event) + first = SessionStatusEvent.create("sess-1", SessionStatus(type="busy")) + await state_with_pool.broadcast_event(first) + # Stall the client: fill its single-slot queue (holds the first event). + assert stalled.qsize() == 1 - # SSE subscriber queues should NOT receive events (dead code removed) - assert queue.qsize() == 0 + second = SessionIdleEvent.create("sess-1") + # Must not raise QueueFull; the stalled queue evicts its oldest item + # (drop-oldest policy), the healthy queue receives the new event. + await state_with_pool.broadcast_event(second) + # Stalled queue: oldest (first) evicted, second delivered. + _event_id, received = stalled.get_nowait() + assert received is second + assert stalled.qsize() == 0 -@pytest.mark.anyio -async def test_bridge_republishes_to_event_bus( - bridged_state: ServerState, - event_bus: EventBus, -) -> None: - """Events are republished to the EventBus as CustomEvent wrappers.""" - subscriber = await event_bus.subscribe("sess-pool") - - event = SessionStatusEvent.create("sess-pool", SessionStatus(type="busy")) - await bridged_state.broadcast_event(event) - - # Allow the async publish to propagate - await asyncio.sleep(0.05) - - envelope = subscriber.get_nowait() - assert isinstance(envelope, EventEnvelope) - wrapped = envelope.event - assert isinstance(wrapped, CustomEvent) - assert wrapped.event_data is event - assert wrapped.event_type == "opencode:session.status" + # Healthy queue received both events in order. + _id1, got_first = healthy.get_nowait() + _id2, got_second = healthy.get_nowait() + assert got_first is first + assert got_second is second @pytest.mark.anyio -async def test_bridge_wraps_different_event_types( - bridged_state: ServerState, - event_bus: EventBus, +async def test_broadcast_event_never_republishes_to_event_bus( + state_with_pool: ServerState, ) -> None: - """Various OpenCode event types are correctly wrapped.""" - subscriber = await event_bus.subscribe("sess-mixed") - - events: list[Event] = [ - SessionStatusEvent.create("sess-mixed", SessionStatus(type="busy")), - SessionIdleEvent.create("sess-mixed"), - ] - - for evt in events: - await bridged_state.broadcast_event(evt) - - await asyncio.sleep(0.05) - - for _i, evt in enumerate(events): - envelope = subscriber.get_nowait() - assert isinstance(envelope, EventEnvelope) - wrapped = envelope.event - assert isinstance(wrapped, CustomEvent) - assert wrapped.event_data is evt - expected_type = f"opencode:{evt.type}" - assert wrapped.event_type == expected_type + """EventBus receives ZERO projection events (loopback eliminated).""" + event_bus = state_with_pool.pool.session_pool.event_bus + subscriber = await event_bus.subscribe("sess-1") - -@pytest.mark.anyio -async def test_global_event_not_republished_to_event_bus( - bridged_state: ServerState, - event_bus: EventBus, -) -> None: - """Global events without session_id are NOT republished to EventBus.""" - # Use a dummy session just to have a subscriber queue; the event itself - # has no session_id so it should not be published there. - subscriber = await event_bus.subscribe("global-session") - - event = ServerConnectedEvent() - await bridged_state.broadcast_event(event) + await state_with_pool.broadcast_event( + SessionStatusEvent.create("sess-1", SessionStatus(type="busy")) + ) + await state_with_pool.broadcast_event(SessionIdleEvent.create("sess-1")) await asyncio.sleep(0.05) - # EventBus should receive nothing because the event has no session_id with pytest.raises(asyncio.QueueEmpty): subscriber.get_nowait() - # SSE subscriber queues also do NOT receive events (dead code removed) - queue: asyncio.Queue[Any] = asyncio.Queue() - bridged_state.event_subscribers.append(queue) - await bridged_state.broadcast_event(event) - assert queue.qsize() == 0 - - -# ============================================================================= -# Bridge unit tests -# ============================================================================= - @pytest.mark.anyio -async def test_bridge_publish_does_not_push_to_sse_queues( - bridged_state: ServerState, +async def test_projection_buffer_per_session_for_replay( + state_with_pool: ServerState, ) -> None: - """Bridge.publish does NOT push to SSE subscriber queues (dead code removed). + """Replayed projections honor event_id > last_event_id per session.""" + await state_with_pool.broadcast_event( + SessionStatusEvent.create("sess-a", SessionStatus(type="busy")) + ) + await state_with_pool.broadcast_event( + SessionStatusEvent.create("sess-a", SessionStatus(type="idle")) + ) + await state_with_pool.broadcast_event( + SessionStatusEvent.create("sess-b", SessionStatus(type="busy")) + ) - Event delivery flows exclusively through the EventBus. The legacy - event_subscribers queue push has been removed. - """ + # Replay all: three projections across two sessions queue: asyncio.Queue[Any] = asyncio.Queue() - bridged_state.event_subscribers.append(queue) - - event = SessionStatusEvent.create("sess-unit", SessionStatus(type="idle")) - assert bridged_state.event_bridge is not None - await bridged_state.event_bridge.publish(event) - - # SSE subscriber queues should NOT receive events - assert queue.qsize() == 0 + state_with_pool.replay_projections(queue, last_event_id=0) + replayed: list[tuple[int, Any]] = [] + while not queue.empty(): + replayed.append(queue.get_nowait()) + assert len(replayed) == 3 + ids = [event_id for event_id, _ in replayed] + assert ids == sorted(ids), "replay must preserve broadcast order" + + # Conditional replay: only projections after the first are replayed + queue2: asyncio.Queue[Any] = asyncio.Queue() + state_with_pool.replay_projections(queue2, last_event_id=ids[0]) + replayed2: list[tuple[int, Any]] = [] + while not queue2.empty(): + replayed2.append(queue2.get_nowait()) + assert [event_id for event_id, _ in replayed2] == ids[1:] @pytest.mark.anyio -async def test_bridge_extract_session_id_variations( - bridged_state: ServerState, +async def test_global_event_not_buffered_for_replay( + state_with_pool: ServerState, ) -> None: - """_extract_session_id handles events with and without session_id.""" - bridge = bridged_state.event_bridge - assert bridge is not None - - # Event with session_id - status_event = SessionStatusEvent.create("sess-1", SessionStatus(type="busy")) - assert bridge._extract_session_id(status_event) == "sess-1" + """Events without a session_id are delivered but not buffered.""" + queue: asyncio.Queue[Any] = asyncio.Queue() + state_with_pool.event_subscribers.append(queue) - # Event without session_id - connected_event = ServerConnectedEvent() - assert bridge._extract_session_id(connected_event) is None + await state_with_pool.broadcast_event(ServerConnectedEvent()) - # Edge case: object with no properties attribute - class NoProperties: - pass + _event_id, received = queue.get_nowait() + assert isinstance(received, ServerConnectedEvent) - assert bridge._extract_session_id(NoProperties()) is None # type: ignore[arg-type] + # Nothing buffered under any session → no replay + queue2: asyncio.Queue[Any] = asyncio.Queue() + state_with_pool.replay_projections(queue2, last_event_id=0) + assert queue2.empty() @pytest.mark.anyio -async def test_bridge_wrap_event_format( - bridged_state: ServerState, -) -> None: - """_wrap_event produces a correctly formatted CustomEvent.""" - bridge = bridged_state.event_bridge - assert bridge is not None - - event = SessionIdleEvent.create("sess-wrap") - wrapped = bridge._wrap_event(event) - - assert isinstance(wrapped, CustomEvent) - assert wrapped.event_data is event - assert wrapped.event_type == "opencode:session.idle" - assert wrapped.source == "opencode_event_bridge" - - -@pytest.mark.anyio -async def test_bridge_isolation_between_sessions( - bridged_state: ServerState, - event_bus: EventBus, -) -> None: - """Events for session A do not leak into session B's EventBus subscription.""" - sub_a = await event_bus.subscribe("sess-a") - sub_b = await event_bus.subscribe("sess-b") - - await bridged_state.broadcast_event( - SessionStatusEvent.create("sess-a", SessionStatus(type="busy")) - ) - await asyncio.sleep(0.05) +async def test_extract_session_id_variations() -> None: + """extract_session_id handles events with and without session_id.""" + status_event = SessionStatusEvent.create("sess-1", SessionStatus(type="busy")) + assert ServerState.extract_session_id(status_event) == "sess-1" - envelope = sub_a.get_nowait() - assert isinstance(envelope, EventEnvelope) - wrapped = envelope.event - assert wrapped.event_data.properties.session_id == "sess-a" + assert ServerState.extract_session_id(ServerConnectedEvent()) is None - with pytest.raises(asyncio.QueueEmpty): - sub_b.get_nowait() + # Events whose properties carry no session association return None. + heartbeat = ServerHeartbeatEvent() + assert ServerState.extract_session_id(heartbeat) is None # ============================================================================= diff --git a/tests/servers/opencode_server/test_event_pipeline_e2e.py b/tests/servers/opencode_server/test_event_pipeline_e2e.py index 0c4a4a22b..2fe963063 100644 --- a/tests/servers/opencode_server/test_event_pipeline_e2e.py +++ b/tests/servers/opencode_server/test_event_pipeline_e2e.py @@ -1,13 +1,13 @@ """E2E tests for the OpenCode event pipeline. Simulates a full agent streaming session from agent event emission through -EventBus → session-scoped consumer → event_bridge → EventBus → SSE subscriber. -Verifies every OpenCode event type produced by the pipeline reaches the scope="all" -subscriber (representing the SSE frontend). +EventBus → session-scoped consumer → adapter → ``ServerState.broadcast_event`` +→ SSE subscriber queues. Verifies every OpenCode event type produced by the +pipeline reaches the subscriber queues (representing the SSE frontend). Covers the regression where PartUpdatedEvent (which has session_id at properties.part.session_id, not properties.session_id) was silently dropped -by event_bridge._extract_session_id(). +by the session-id extractor. """ from __future__ import annotations @@ -69,21 +69,18 @@ def _stream_empty(queue: asyncio.Queue[Any]) -> bool: def _extract_opencode_events(sse_queue: Any) -> list[Any]: """Extract OpenCode events from the SSE subscriber queue. - The SSE subscriber receives EventEnvelope objects where the ``.event`` - field can be a CustomEvent wrapper (from event_bridge) or a raw agent - event. This helper filters for CustomEvent wrappers and extracts the - underlying OpenCode event from ``.event_data``. + The direct-wire subscriber queue holds ``(event_id, event)`` tuples of + raw OpenCode protocol events (no CustomEvent wrapping — projections are + no longer republished into the EventBus). """ - from wolfharness.agents.events.events import CustomEvent - from wolfharness.orchestrator.core import EventEnvelope - result: list[Any] = [] while not _stream_empty(sse_queue): - envelope = sse_queue.get_nowait() - if isinstance(envelope, EventEnvelope): - inner = envelope.event - if isinstance(inner, CustomEvent) and inner.event_data is not None: - result.append(inner.event_data) + item = sse_queue.get_nowait() + if isinstance(item, tuple): + _event_id, event = item + else: + event = item + result.append(event) return result @@ -140,13 +137,9 @@ async def session_pool(server_state: ServerState): # type: ignore[no-untyped-de ) await sp.start() - # Wire the EventBus into server_state so event_bridge can discover it - server_state._pool = pool_mock - pool_mock.session_pool = sp - # Re-initialize event_bridge now that event_bus is available - from wolfharness_server.opencode_server.event_bridge import OpenCodeEventBridge - - server_state.event_bridge = OpenCodeEventBridge(server_state, sp.event_bus) + # The SessionPool EventBus is wired into server_state at __post_init__; + # no projection bridge exists — projections fan out directly to + # server_state.event_subscribers queues. yield sp await sp.shutdown() @@ -158,7 +151,7 @@ async def session_pool(server_state: ServerState): # type: ignore[no-untyped-de class TestEventPipelineE2E: - """Full pipeline: agent event → EventBus → consumer → event_bridge → EventBus → SSE.""" + """Full pipeline: agent event → EventBus → consumer → broadcast_event → SSE queues.""" @pytest.mark.asyncio async def test_text_streaming_full_pipeline( @@ -184,7 +177,8 @@ async def test_text_streaming_full_pipeline( await _async_wait(0.1) # Subscribe a scope="all" subscriber to mimic SSE frontend - sse_queue = await session_pool.event_bus.subscribe("__global_sse__", scope="all") + sse_queue: asyncio.Queue[tuple[int, Any]] = asyncio.Queue(maxsize=1000) + server_state.event_subscribers.append(sse_queue) # Simulate agent emitting PartStartEvent (text) text_start = PartStartEvent(index=0, part=pydantic_text_part("Hello")) @@ -211,8 +205,8 @@ async def test_text_streaming_full_pipeline( # Assert: Text PartUpdatedEvent reached SSE (regression test for nested session_id) assert len(part_updated_events) >= 1, ( - "PartUpdatedEvent should reach scope='all' subscriber; " - "event_bridge._extract_session_id must traverse properties.part.session_id" + "PartUpdatedEvent should reach the SSE subscriber; " + "ServerState.extract_session_id must traverse properties.part.session_id" ) # The first PartUpdatedEvent should contain a TextPart @@ -257,7 +251,8 @@ async def test_tool_call_full_pipeline( await integration.create_session(session_id=session_id, agent_name="test-agent") await _async_wait(0.1) - sse_queue = await session_pool.event_bus.subscribe("__global_sse__", scope="all") + sse_queue: asyncio.Queue[tuple[int, Any]] = asyncio.Queue(maxsize=1000) + server_state.event_subscribers.append(sse_queue) # Simulate agent emitting ToolCallStartEvent tool_start = ToolCallStartEvent( @@ -358,7 +353,8 @@ async def test_tool_call_empty_input_then_progress_full_pipeline( await integration.create_session(session_id=session_id, agent_name="test-agent") await _async_wait(0.1) - sse_queue = await session_pool.event_bus.subscribe("__global_sse__", scope="all") + sse_queue: asyncio.Queue[tuple[int, Any]] = asyncio.Queue(maxsize=1000) + server_state.event_subscribers.append(sse_queue) # Step 1: ToolCallStartEvent with EMPTY raw_input (simulates streamed args) tool_start = ToolCallStartEvent( @@ -444,7 +440,8 @@ async def test_run_started_produces_session_status( ) -> None: """RunStartedEvent → SessionStatusEvent (busy). - Verifies lifecycle events pass through event_bridge correctly. + Verifies lifecycle events pass through the consumer → adapter → + broadcast path correctly. """ integration = OpenCodeSessionPoolIntegration( session_pool=session_pool, @@ -455,7 +452,8 @@ async def test_run_started_produces_session_status( await integration.create_session(session_id=session_id, agent_name="test-agent") await _async_wait(0.1) - sse_queue = await session_pool.event_bus.subscribe("__global_sse__", scope="all") + sse_queue: asyncio.Queue[tuple[int, Any]] = asyncio.Queue(maxsize=1000) + server_state.event_subscribers.append(sse_queue) run_started = RunStartedEvent(session_id=session_id, run_id="run-e2e-001") await session_pool.event_bus.publish(session_id, run_started) @@ -478,7 +476,8 @@ async def test_run_error_produces_session_error( ) -> None: """RunErrorEvent → SessionErrorEvent. - Verifies error events pass through event_bridge correctly. + Verifies error events pass through the consumer → adapter → + broadcast path correctly. """ from wolfharness.agents.events import RunErrorEvent @@ -491,7 +490,8 @@ async def test_run_error_produces_session_error( await integration.create_session(session_id=session_id, agent_name="test-agent") await _async_wait(0.1) - sse_queue = await session_pool.event_bus.subscribe("__global_sse__", scope="all") + sse_queue: asyncio.Queue[tuple[int, Any]] = asyncio.Queue(maxsize=1000) + server_state.event_subscribers.append(sse_queue) run_error = RunErrorEvent( code="TestError", @@ -510,18 +510,17 @@ async def test_run_error_produces_session_error( await integration._stop_event_consumer(session_id) @pytest.mark.asyncio - async def test_event_bridge_extracts_nested_session_id( + async def test_extract_session_id_nested_part_session( self, session_pool: SessionPool, server_state: ServerState, ) -> None: - """Regression test: _extract_session_id handles PartUpdatedEvent's nested session_id. + """Regression test: extract_session_id handles PartUpdatedEvent's nested session_id. PartUpdatedEventProperties has session_id at properties.part.session_id, not at properties.session_id. The extractor must traverse this nested path. """ - # Verify event_bridge is set up - assert server_state.event_bridge is not None + from wolfharness_server.opencode_server.state import ServerState # Create a PartUpdatedEvent with a TextPart (nested session_id) text_part = TextPart( @@ -533,9 +532,9 @@ async def test_event_bridge_extracts_nested_session_id( part_updated = PartUpdatedEvent.create(text_part) # Extract session_id — must succeed with the fix - session_id = server_state.event_bridge._extract_session_id(part_updated) + session_id = ServerState.extract_session_id(part_updated) assert session_id == "sess-e2e-nested", ( - f"_extract_session_id should return 'sess-e2e-nested' for PartUpdatedEvent " + f"extract_session_id should return 'sess-e2e-nested' for PartUpdatedEvent " f"with TextPart, got {session_id!r}" ) @@ -548,9 +547,10 @@ async def test_session_scoped_consumer_does_not_loopback( """Verify the consumer doesn't process the OpenCode events it publishes back. When _handle_event converts an agent event to an OpenCode event and - broadcasts it via event_bridge, the event_bridge publishes back to the - SAME EventBus. The consumer should NOT try to convert these OpenCode - events again (they're not RichAgentStreamEvent instances). + broadcasts it via ``ServerState.broadcast_event``, the projection goes + directly to SSE subscriber queues — never back into the EventBus. The + consumer should NOT try to convert these OpenCode events again (they're + not RichAgentStreamEvent instances). """ integration = OpenCodeSessionPoolIntegration( session_pool=session_pool, @@ -561,7 +561,8 @@ async def test_session_scoped_consumer_does_not_loopback( await integration.create_session(session_id=session_id, agent_name="test-agent") await _async_wait(0.1) - sse_queue = await session_pool.event_bus.subscribe("__global_sse__", scope="all") + sse_queue: asyncio.Queue[tuple[int, Any]] = asyncio.Queue(maxsize=1000) + server_state.event_subscribers.append(sse_queue) # Publish a sequence of agent events text_start = PartStartEvent(index=0, part=pydantic_text_part("Hello")) diff --git a/tests/servers/opencode_server/test_global_event.py b/tests/servers/opencode_server/test_global_event.py index fe61910b7..80d31bd99 100644 --- a/tests/servers/opencode_server/test_global_event.py +++ b/tests/servers/opencode_server/test_global_event.py @@ -3,6 +3,7 @@ from __future__ import annotations import asyncio +from collections import deque import contextlib import json from pathlib import Path @@ -270,15 +271,13 @@ class _MockState: def __init__(self, working_dir: str = "/tmp/test_wd") -> None: self.working_dir = working_dir - self.event_subscribers: list[asyncio.Queue[Event]] = [] + self.event_subscribers: list[asyncio.Queue[tuple[int, Event]]] = [] + self._projection_buffers: dict[str, deque[tuple[int, Event]]] = {} + self._projection_counter = 0 self._event_factory: GlobalEventFactory | None = None self._first_subscriber_triggered = False self.on_first_subscriber: Any = None - # Provide EventBus-based infrastructure so _event_generator uses - # the EventBus path instead of the heartbeat-only fallback. - self._mock_event_bus = _MockEventBus() self.session_controller = _MockSessionController() - self.pool = _MockPool(_MockSessionPool(self._mock_event_bus)) def get_event_factory(self) -> GlobalEventFactory: if self._event_factory is None: @@ -297,6 +296,55 @@ def cancel_all_pending_questions(self) -> list[str]: """No-op mock for SSE disconnect handler.""" return [] + @staticmethod + def extract_session_id(event: Event) -> str | None: + """Mirror ServerState.extract_session_id for mock-state tests.""" + from wolfharness_server.opencode_server.routes.global_routes import ( + _extract_session_id, + ) + + return _extract_session_id(event) + + async def broadcast_event(self, event: Event) -> None: + """Direct-wire fanout to subscriber queues (mirrors ServerState). + + Overflow uses the same drop-oldest policy as production so the mock + does not hide queue-full behavior from the suite. + """ + self._projection_counter += 1 + event_id = self._projection_counter + session_id = self.extract_session_id(event) + if session_id is not None: + self._projection_buffers.setdefault(session_id, deque(maxlen=100)).append(( + event_id, + event, + )) + for queue in self.event_subscribers: + try: + queue.put_nowait((event_id, event)) + except asyncio.QueueShutDown: + continue + except asyncio.QueueFull: + with contextlib.suppress(asyncio.QueueEmpty): + queue.get_nowait() + try: + queue.put_nowait((event_id, event)) + except asyncio.QueueFull: + continue + + def replay_projections(self, queue: asyncio.Queue[Any], last_event_id: int) -> None: + """Enqueue buffered projections with event_id > last_event_id.""" + buffered: list[tuple[int, Event]] = [] + for session_buf in self._projection_buffers.values(): + buffered.extend(session_buf) + for event_id, event in sorted(buffered, key=lambda item: item[0]): + if event_id <= last_event_id: + continue + try: + queue.put_nowait((event_id, event)) + except (asyncio.QueueShutDown, asyncio.QueueFull): + return + # get_next_event_id() removed — event_id now comes from EventBus # (EventEnvelope.event_id assigned at publish time, not ServerState). @@ -312,9 +360,9 @@ async def _collect_events( # Get the initial connected event item = await gen.__anext__() results.append(json.loads(item["data"])) - # Send additional events through the mock EventBus + # Send additional events through the direct-wire broadcast system for event in events_to_send: - await state._mock_event_bus.publish("__global_sse__", event) + await state.broadcast_event(event) item = await gen.__anext__() results.append(json.loads(item["data"])) return results @@ -474,8 +522,8 @@ async def test_event_endpoint_unicode_preserved() -> None: gen = _event_generator(state, wrap_payload=False) # Consume connected event await gen.__anext__() - # Send unicode session event via mock EventBus - await state._mock_event_bus.publish("__global_sse__", session_evt) + # Send unicode session event via direct-wire broadcast + await state.broadcast_event(session_evt) item = await gen.__anext__() raw_data = item["data"] assert "会话测试" in raw_data @@ -724,7 +772,7 @@ async def test_disconnect_events_not_delivered() -> None: # Put an event via mock EventBus — only gen2 should receive it event = SessionStatusEvent.create(session_id="disc1", status_type="busy") - await state._mock_event_bus.publish("__global_sse__", event) + await state.broadcast_event(event) item = await gen2.__anext__() data = json.loads(item["data"]) assert data["type"] == "session.status" @@ -1166,7 +1214,7 @@ async def test_concurrent_two_subscribers_both_receive_events() -> None: # Broadcast event to both subscribers via mock EventBus event = SessionStatusEvent.create(session_id="s_concurrent", status_type="busy") - await state._mock_event_bus.publish("__global_sse__", event) + await state.broadcast_event(event) item1 = await gen1.__anext__() item2 = await gen2.__anext__() @@ -1192,7 +1240,7 @@ async def test_concurrent_subscribers_receive_same_content() -> None: await gen2.__anext__() event = SessionStatusEvent.create(session_id="s_same", status_type="idle") - await state._mock_event_bus.publish("__global_sse__", event) + await state.broadcast_event(event) item1 = await gen1.__anext__() item2 = await gen2.__anext__() @@ -1224,7 +1272,7 @@ async def test_concurrent_event_ordering_preserved() -> None: ] for ev in events: - await state._mock_event_bus.publish("__global_sse__", ev) + await state.broadcast_event(ev) # Collect all 3 events from each subscriber received1 = [json.loads((await gen1.__anext__())["data"]) for _ in range(3)] @@ -1257,7 +1305,7 @@ async def test_concurrent_subscriber_receives_after_another_disconnects() -> Non # Send event via mock EventBus — remaining subscriber B should receive it event = SessionStatusEvent.create(session_id="s_survive", status_type="busy") - await state._mock_event_bus.publish("__global_sse__", event) + await state.broadcast_event(event) item_b = await gen_b.__anext__() data_b = json.loads(item_b["data"]) @@ -1294,56 +1342,51 @@ async def test_concurrent_all_get_server_connected() -> None: def _make_broadcast_state() -> ServerState: """Create a ServerState with a minimal mock agent for broadcast_event tests.""" - from unittest.mock import AsyncMock, Mock + from unittest.mock import Mock mock_env = Mock() mock_env.get_fs = Mock(return_value=Mock()) mock_agent = Mock() mock_agent.env = mock_env - state = ServerState(working_dir="/test", agent=mock_agent) - # Set up event_bridge mock so broadcast_event has a destination - state.event_bridge = Mock() - state.event_bridge.publish = AsyncMock() - return state + return ServerState(working_dir="/test", agent=mock_agent) @pytest.mark.anyio -async def test_broadcast_event_delegates_to_bridge() -> None: - """Broadcast delegates event to event_bridge.publish.""" +async def test_broadcast_event_fans_out_to_subscribers() -> None: + """Broadcast delivers projections to every registered subscriber queue.""" state = _make_broadcast_state() + queue: asyncio.Queue[Any] = asyncio.Queue() + state.event_subscribers.append(queue) event = SessionStatusEvent.create(session_id="abc", status_type="busy") await state.broadcast_event(event) - state.event_bridge.publish.assert_called_once_with(event) + event_id, received = queue.get_nowait() + assert received is event + assert event_id > 0 @pytest.mark.anyio -async def test_broadcast_event_no_bridge_no_error() -> None: - """Broadcast with no event_bridge does not raise.""" - from unittest.mock import Mock - - mock_env = Mock() - mock_env.get_fs = Mock(return_value=Mock()) - mock_agent = Mock() - mock_agent.env = mock_env - state = ServerState(working_dir="/test", agent=mock_agent) - assert state.event_bridge is None +async def test_broadcast_event_no_subscribers_no_error() -> None: + """Broadcast with no subscribers does not raise.""" + state = _make_broadcast_state() event = SessionStatusEvent.create(session_id="abc", status_type="busy") await state.broadcast_event(event) # Should not raise @pytest.mark.anyio -async def test_broadcast_event_bridge_error_propagates() -> None: - """If event_bridge.publish raises, broadcast_event propagates the error.""" +async def test_broadcast_event_ignores_dead_subscriber_queue() -> None: + """A shut-down subscriber queue is dropped, not raised on.""" state = _make_broadcast_state() - state.event_bridge.publish.side_effect = RuntimeError("bridge broken") + queue: asyncio.Queue[Any] = asyncio.Queue() + queue.shutdown() + state.event_subscribers.append(queue) event = SessionStatusEvent.create(session_id="abc", status_type="busy") - # The error propagates since broadcast_event does not catch it - with pytest.raises(RuntimeError, match="bridge broken"): - await state.broadcast_event(event) + await state.broadcast_event(event) # Should not raise + + assert queue not in state.event_subscribers # ============================================================================= diff --git a/tests/servers/opencode_server/test_message_id_alignment.py b/tests/servers/opencode_server/test_message_id_alignment.py index 35903f3fa..ad3939d03 100644 --- a/tests/servers/opencode_server/test_message_id_alignment.py +++ b/tests/servers/opencode_server/test_message_id_alignment.py @@ -470,23 +470,25 @@ def test_same_ms_ids_have_consistent_timestamps(self) -> None: # ============================================================================= -# C4: CustomEvent bypasses assistant registration +# Loopback elimination: no bridge CustomEvents reach the consumer # ============================================================================= -class TestCustomEventBypassesRegistration: - """Tests that CustomEvent does not trigger assistant message registration (C4). +class TestLoopbackEliminated: + """Tests reflecting post-loopback semantics (issue #380). - Issue C4: CustomEvent wraps SSE broadcast events (e.g. - SessionCreatedEvent) republished from the OpenCodeEventBridge. These - are not real agent events and must NOT trigger assistant message - registration. If they do, the assistant message is broadcast before - the agent runs, causing notification ID > assistant ID → QUEUED. + The OpenCodeEventBridge republish was removed: projections no longer + flow back into the EventBus, so no ``source="opencode_event_bridge"`` + CustomEvents reach the session consumer. Subscription-level isolation + (``exclude_source``) is covered in ``tests/servers/test_subagent_event_mixin.py``. + Tool-emitted CustomEvents (``source=None`` or tool name) fall through to + normal event processing and trigger assistant registration (C3) like any + other real agent event. """ @pytest.mark.asyncio - async def test_custom_event_does_not_register_assistant(self): - """CustomEvent should NOT trigger assistant message registration.""" + async def test_tool_custom_event_registers_assistant(self): + """A tool-emitted CustomEvent triggers assistant registration (C3).""" from wolfharness.agents.events.events import CustomEvent from wolfharness.orchestrator.event_bus import EventEnvelope from wolfharness_server.opencode_server.models import ( @@ -528,36 +530,36 @@ async def test_custom_event_does_not_register_assistant(self): integration._contexts["test-session"] = ctx integration._message_registered["test-session"] = False - # Send a CustomEvent (e.g., wrapping a SessionCreatedEvent) + # A tool-emitted CustomEvent (source=tool name) is a real agent event. custom_event = CustomEvent( - event_data={"type": "session.created"}, - event_type="opencode:session.created", + event_data={"type": "tool.visibility"}, + event_type="tool:visibility", + source="my_tool", ) envelope = EventEnvelope( event=custom_event, source_session_id="test-session", ) - await integration._handle_event("test-session", envelope) + import unittest.mock as _mock - # Assistant message should NOT have been registered - assert not integration._message_registered.get("test-session", False), ( - "CustomEvent should NOT trigger assistant message registration (C4)" + with _mock.patch( + "wolfharness_server.opencode_server.opencode_event_bridge.append_message_to_session", + new_callable=_mock.AsyncMock, + ): + await integration._handle_event("test-session", envelope) + + # Assistant message IS registered (the agent is producing events) + assert integration._message_registered.get("test-session", False), ( + "Tool-emitted CustomEvent should trigger assistant registration (C3)" ) - # broadcast_event should NOT have been called for MessageUpdatedEvent - broadcast_calls = server_state.broadcast_event.call_args_list - for call in broadcast_calls: - event_arg = call.args[0] if call.args else call.kwargs.get("event") - if hasattr(event_arg, "type") and event_arg.type == "message.updated": - pytest.fail("CustomEvent should NOT trigger MessageUpdatedEvent broadcast (C4)") @pytest.mark.asyncio async def test_real_agent_event_does_register_assistant(self): """RunStartedEvent (a real agent event) SHOULD trigger registration. - This is the positive control for C4: after skipping CustomEvent, - the next real agent event (RunStartedEvent) must trigger assistant - registration. + This is the positive control: a real agent event must trigger + assistant registration regardless of loopback state. """ from wolfharness.agents.events import RunStartedEvent from wolfharness.orchestrator.event_bus import EventEnvelope @@ -649,13 +651,7 @@ async def _empty_convert(event): @pytest.mark.asyncio async def test_custom_event_then_runstarted_registers_on_runstarted(self): - """CustomEvent followed by RunStartedEvent: registration on RunStarted only. - - This simulates the real timeline: - 1. SessionCreatedEvent → CustomEvent → skip (C4) - 2. System notifications - 3. RunStartedEvent → register assistant message - """ + """CustomEvent followed by RunStartedEvent: assistant stays registered.""" from wolfharness.agents.events import RunStartedEvent from wolfharness.agents.events.events import CustomEvent from wolfharness.orchestrator.event_bus import EventEnvelope @@ -708,23 +704,32 @@ async def _empty_convert(event): mock_adapter.convert_event = _empty_convert integration._adapters["test-session"] = mock_adapter - # Step 1: Send CustomEvent (SessionCreatedEvent) + # Step 1: Send a tool CustomEvent — registers the assistant (C3) custom_event = CustomEvent( - event_data={"type": "session.created"}, - event_type="opencode:session.created", + event_data={"type": "tool.visibility"}, + event_type="tool:visibility", + source="my_tool", ) envelope1 = EventEnvelope( event=custom_event, source_session_id="test-session", ) - await integration._handle_event("test-session", envelope1) - # After CustomEvent: NOT registered - assert not integration._message_registered.get("test-session", False), ( - "CustomEvent should NOT trigger registration (C4)" + import unittest.mock as _mock + + with _mock.patch( + "wolfharness_server.opencode_server.opencode_event_bridge.append_message_to_session", + new_callable=_mock.AsyncMock, + ): + await integration._handle_event("test-session", envelope1) + + # After CustomEvent: registered (the agent is producing events) + assert integration._message_registered.get("test-session", False), ( + "Tool CustomEvent should trigger registration (C3)" ) - # Step 2: Send RunStartedEvent + # Step 2: Send RunStartedEvent for a new turn — D1 reset fires, + # creating a fresh assistant message (still registered after). run_event = RunStartedEvent( session_id="test-session", run_id="run_001", @@ -735,17 +740,15 @@ async def _empty_convert(event): source_session_id="test-session", ) - import unittest.mock as _mock - with _mock.patch( "wolfharness_server.opencode_server.opencode_event_bridge.append_message_to_session", new_callable=_mock.AsyncMock, ): await integration._handle_event("test-session", envelope2) - # After RunStartedEvent: registered - assert integration._message_registered.get("test-session", False), ( - "RunStartedEvent after CustomEvent should trigger registration" + # After RunStartedEvent: fresh turn started (D1 reset → new message) + assert integration._contexts["test-session"].assistant_msg_id != "msg_timeline", ( + "RunStartedEvent after registration should start a fresh turn (D1)" ) diff --git a/tests/servers/opencode_server/test_sse_compliance.py b/tests/servers/opencode_server/test_sse_compliance.py index dab209e0c..b9d097081 100644 --- a/tests/servers/opencode_server/test_sse_compliance.py +++ b/tests/servers/opencode_server/test_sse_compliance.py @@ -15,6 +15,7 @@ from __future__ import annotations import asyncio +from collections import deque import json from typing import TYPE_CHECKING, Any @@ -147,15 +148,13 @@ class _MockState: def __init__(self, working_dir: str = "/tmp/test_wd") -> None: self.working_dir = working_dir - self.event_subscribers: list[asyncio.Queue[Event]] = [] + self.event_subscribers: list[asyncio.Queue[tuple[int, Event]]] = [] + self._projection_buffers: dict[str, deque[tuple[int, Event]]] = {} + self._projection_counter = 0 self._event_factory: GlobalEventFactory | None = None self._first_subscriber_triggered = False self.on_first_subscriber: Any = None - # Provide EventBus-based infrastructure so _event_generator uses - # the EventBus path instead of the heartbeat-only fallback. - self._mock_event_bus = _MockEventBus() self.session_controller = _MockSessionController() - self.pool = _MockPool(_MockSessionPool(self._mock_event_bus)) def get_event_factory(self) -> GlobalEventFactory: if self._event_factory is None: @@ -171,12 +170,58 @@ def get_event_factory(self) -> GlobalEventFactory: def create_background_task(self, coro: Any, name: str = "") -> asyncio.Task[Any]: return asyncio.ensure_future(coro) - # get_next_event_id() removed — event_id now comes from EventBus - # (EventEnvelope.event_id assigned at publish time, not ServerState). - def cancel_all_pending_questions(self) -> list[str]: return [] + @staticmethod + def extract_session_id(event: Event) -> str | None: + """Mirror ServerState.extract_session_id for mock-state tests.""" + from wolfharness_server.opencode_server.routes.global_routes import ( + _extract_session_id, + ) + + return _extract_session_id(event) + + async def broadcast_event(self, event: Event) -> None: + """Direct-wire fanout to subscriber queues (mirrors ServerState). + + Overflow uses the same drop-oldest policy as production so the mock + does not hide queue-full behavior from the suite. + """ + self._projection_counter += 1 + event_id = self._projection_counter + session_id = self.extract_session_id(event) + if session_id is not None: + self._projection_buffers.setdefault(session_id, deque(maxlen=100)).append(( + event_id, + event, + )) + for queue in self.event_subscribers: + try: + queue.put_nowait((event_id, event)) + except asyncio.QueueShutDown: + continue + except asyncio.QueueFull: + with contextlib.suppress(asyncio.QueueEmpty): + queue.get_nowait() + try: + queue.put_nowait((event_id, event)) + except asyncio.QueueFull: + continue + + def replay_projections(self, queue: asyncio.Queue[Any], last_event_id: int) -> None: + """Enqueue buffered projections with event_id > last_event_id.""" + buffered: list[tuple[int, Event]] = [] + for session_buf in self._projection_buffers.values(): + buffered.extend(session_buf) + for event_id, event in sorted(buffered, key=lambda item: item[0]): + if event_id <= last_event_id: + continue + try: + queue.put_nowait((event_id, event)) + except (asyncio.QueueShutDown, asyncio.QueueFull): + return + async def _collect_events( state: _MockState, @@ -189,9 +234,9 @@ async def _collect_events( # Get the initial connected event item = await gen.__anext__() results.append(json.loads(item["data"])) - # Send additional events through the mock EventBus + # Send additional events through the direct-wire broadcast system for event in events_to_send: - await state._mock_event_bus.publish("__global_sse__", event) + await state.broadcast_event(event) item = await gen.__anext__() results.append(json.loads(item["data"])) return results diff --git a/tests/servers/test_subagent_event_mixin.py b/tests/servers/test_subagent_event_mixin.py index f62c5fa3c..42a9cfeb0 100644 --- a/tests/servers/test_subagent_event_mixin.py +++ b/tests/servers/test_subagent_event_mixin.py @@ -96,7 +96,9 @@ async def test_start_consumer_subscribes_and_runs_loop( assert "sess-1" in consumer._session_groups assert "sess-1" in consumer._consumer_streams - mock_event_bus.subscribe.assert_awaited_once_with("sess-1", scope="descendants") + mock_event_bus.subscribe.assert_awaited_once_with( + "sess-1", scope="descendants", replay=True, exclude_source=None + ) await consumer.stop_event_consumer("sess-1") @@ -174,7 +176,7 @@ async def test_handle_event_dispatches_to_subclass(mock_event_bus: AsyncMock) -> await consumer.start_event_consumer("sess-1") for _ in range(100): - if mock_handle.await_count > 0: + if mock_handle.called: break await asyncio.sleep(0.01) @@ -183,6 +185,54 @@ async def test_handle_event_dispatches_to_subclass(mock_event_bus: AsyncMock) -> await consumer.stop_event_consumer("sess-1") +class _NoReplayConsumer(_TestConsumer): + """Consumer that opts out of EventBus replay (OpenCode-style policy).""" + + def _get_subscription_replay(self) -> bool: + return False + + def _get_subscription_exclude_source(self) -> frozenset[str] | None: + return frozenset({"opencode_event_bridge"}) + + +@pytest.mark.anyio +async def test_session_consumer_subscribes_replay_false(mock_event_bus: AsyncMock) -> None: + """OpenCode-style consumer subscribes with replay=False + exclude_source.""" + _make_queue_and_mock_subscribe(mock_event_bus) + consumer = _NoReplayConsumer(mock_event_bus) + await consumer.start_event_consumer("sess-1") + + mock_event_bus.subscribe.assert_awaited_once_with( + "sess-1", + scope="descendants", + replay=False, + exclude_source=frozenset({"opencode_event_bridge"}), + ) + + await consumer.stop_event_consumer("sess-1") + + +class _RecoveryReplayConsumer(_TestConsumer): + """Consumer that needs historical events (crash-recovery opt-in).""" + + def _get_subscription_replay(self) -> bool: + return True + + +@pytest.mark.anyio +async def test_recovery_path_opt_in_replay_true(mock_event_bus: AsyncMock) -> None: + """Recovery consumers explicitly opt back in to replay=True.""" + _make_queue_and_mock_subscribe(mock_event_bus) + consumer = _RecoveryReplayConsumer(mock_event_bus) + await consumer.start_event_consumer("sess-1") + + mock_event_bus.subscribe.assert_awaited_once_with( + "sess-1", scope="descendants", replay=True, exclude_source=None + ) + + await consumer.stop_event_consumer("sess-1") + + @pytest.mark.anyio async def test_consumer_shutdown_gracefully_stops_loop( mock_event_bus: AsyncMock,