fix(opencode): eliminate EventBus loopback causing duplicate SSE projections

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.

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
 

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

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,14 +438,20 @@ 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.
         """
         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):

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. |

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