refactor: move bridge content buffer to server-side single source of truth

Eliminate dual accumulation by piggybacking assistant response text on
the server's ws_state:idle SSE event.  The bridge no longer maintains
its own _ws_content_buffer — it reads content directly from the idle
event and passes it through to TurnCompleteEvent unchanged.

Server-side: WebUI accumulates tokens in on_content_token(), joins and
includes in the idle broadcast, then resets (with 256 KB cap).

Downstream consumers (Discord bidi DM forwarding, catch-up) are
unaffected — TurnCompleteEvent.content is still populated.

docs/architecture.md

@@ -1202,13 +1202,13 @@ The bridge dispatches it to `POST /v1/api/cancel` on the server owning the works
 which sets the cooperative cancel flag and unblocks any pending approval/plan waits.
 
 **Completion detection:** The bridge tracks which `correlation_id` maps to which
-`ws_id` for active sends. Content tokens from the per-ws SSE stream are accumulated
-in a per-workstream buffer (`_ws_content_buffer`, capped at 256 KB). When the global
-SSE reports `ws_state → idle` for a tracked workstream, the bridge emits a synthetic
-`TurnCompleteEvent` carrying the correlation ID and the accumulated response text in
-`content`. This catch-up mechanism lets downstream consumers (e.g. the Discord bot)
-recover the full response even when individual `ContentEvent`s were missed due to the
-race between the two independent SSE connections.
+`ws_id` for active sends. The server accumulates content tokens in the WebUI and
+piggybacks the full response text onto the `ws_state → idle` global SSE event.
+When the bridge receives this event, it emits a synthetic `TurnCompleteEvent`
+carrying the correlation ID and the server-provided `content`. This lets downstream
+consumers (e.g. the Discord bot) recover the full response when individual
+`ContentEvent`s were missed, and serves as the primary delivery path for
+bidirectional notification DM forwarding.
 
 **Multi-node routing:** Each bridge retrieves its `node_id` from the server's
 `/health` endpoint on startup (with exponential backoff retry). The server

docs/diagrams/07-message-routing.puml

@@ -39,8 +39,8 @@ BridgeA -> Redis : PUBLISH turnstone:events:abc12345\nAckEvent(status:"ok")
 ... SSE events flow: content, tool_output_chunk, tool_result, status, state_change ...
 
 BridgeA -> Redis : PUBLISH turnstone:events:abc12345\nContentEvent, ToolResultEvent, ...
-BridgeA -> Redis : PUBLISH turnstone:events:global\nStateChangeEvent(state:"idle")
-BridgeA -> Redis : PUBLISH turnstone:events:abc12345\nTurnCompleteEvent
+BridgeA -> Redis : PUBLISH turnstone:events:global\nStateChangeEvent(state:"idle", content:"...")
+BridgeA -> Redis : PUBLISH turnstone:events:abc12345\nTurnCompleteEvent(content:"...")
 
 == Scenario B: Directed Message to Specific Node ==
 

sdk/typescript/src/events.ts

@@ -115,6 +115,8 @@ export interface WsStateEvent {
   context_ratio: number;
   activity: string;
   activity_state: string;
+  /** Full assistant response text — populated on idle transitions only. */
+  content?: string;
 }
 
 export interface WsActivityEvent {

tests/test_bridge_events.py

@@ -69,32 +69,28 @@ class TestIdleTurnComplete:
         assert len(turn_completes) == 0
 
 
-class TestContentBuffer:
-    """Bridge should accumulate content tokens and attach to TurnCompleteEvent."""
+class TestContentPassthrough:
+    """Bridge should pass through content from the server's idle SSE event."""
 
-    def test_content_buffer_accumulated_in_turn_complete(self):
-        """Content events should be accumulated and included in TurnCompleteEvent."""
+    def test_content_passed_through_in_turn_complete(self):
+        """Content from idle event should be included in TurnCompleteEvent."""
         bridge = _make_bridge()
 
-        # Simulate content events from per-ws SSE
-        bridge._handle_ws_event("ws-1", {"type": "content", "text": "Hello "})
-        bridge._handle_ws_event("ws-1", {"type": "content", "text": "world"})
-
         published = []
         with patch.object(
             bridge, "_publish_ws", side_effect=lambda ws, ev: published.append((ws, ev))
         ):
-            bridge._handle_global_event({"type": "ws_state", "ws_id": "ws-1", "state": "idle"})
+            bridge._handle_global_event(
+                {"type": "ws_state", "ws_id": "ws-1", "state": "idle", "content": "Hello world"}
+            )
 
         turn_completes = [(ws, ev) for ws, ev in published if isinstance(ev, TurnCompleteEvent)]
         assert len(turn_completes) == 1
         _, ev = turn_completes[0]
         assert ev.content == "Hello world"
-        # Buffer should be cleared
-        assert "ws-1" not in bridge._ws_content_buffer
 
-    def test_content_buffer_empty_for_no_content_turn(self):
-        """TurnCompleteEvent.content should be empty when no content events fired."""
+    def test_content_empty_when_not_in_event(self):
+        """TurnCompleteEvent.content should be empty when idle event has no content."""
         bridge = _make_bridge()
 
         published = []
@@ -108,17 +104,7 @@ class TestContentBuffer:
         _, ev = turn_completes[0]
         assert ev.content == ""
 
-    def test_content_buffer_cleared_on_ws_closed(self):
-        """ws_closed should clean up the content buffer."""
-        bridge = _make_bridge()
-
-        bridge._handle_ws_event("ws-1", {"type": "content", "text": "orphan"})
-        assert "ws-1" in bridge._ws_content_buffer
-
-        bridge._handle_global_event({"type": "ws_closed", "ws_id": "ws-1"})
-        assert "ws-1" not in bridge._ws_content_buffer
-
-    def test_content_buffer_publishes_content_event(self):
+    def test_content_event_still_published(self):
         """Content events should still be published to per-ws channel."""
         bridge = _make_bridge()
 
@@ -132,25 +118,3 @@ class TestContentBuffer:
         assert len(content_events) == 1
         _, ev = content_events[0]
         assert ev.text == "hello"
-
-    def test_multi_round_content_accumulates(self):
-        """Content from multiple tool-use rounds accumulates in a single turn."""
-        bridge = _make_bridge()
-
-        # Round 1
-        bridge._handle_ws_event("ws-1", {"type": "content", "text": "I'll run "})
-        bridge._handle_ws_event("ws-1", {"type": "stream_end"})
-        # Round 2 (after tool execution)
-        bridge._handle_ws_event("ws-1", {"type": "content", "text": "the command."})
-        bridge._handle_ws_event("ws-1", {"type": "stream_end"})
-
-        published = []
-        with patch.object(
-            bridge, "_publish_ws", side_effect=lambda ws, ev: published.append((ws, ev))
-        ):
-            bridge._handle_global_event({"type": "ws_state", "ws_id": "ws-1", "state": "idle"})
-
-        turn_completes = [(ws, ev) for ws, ev in published if isinstance(ev, TurnCompleteEvent)]
-        assert len(turn_completes) == 1
-        _, ev = turn_completes[0]
-        assert ev.content == "I'll run the command."

tests/test_webui_content.py

@@ -0,0 +1,155 @@
+"""Tests for WebUI content accumulation — server-side single source of truth."""
+
+import queue
+
+import pytest
+
+from turnstone.server import WebUI
+
+
+@pytest.fixture(autouse=True)
+def _reset_global_queue():
+    """Ensure WebUI._global_queue is set for tests and cleaned up after."""
+    WebUI._global_queue = queue.Queue()
+    yield
+    WebUI._global_queue = None
+
+
+def _make_ui() -> WebUI:
+    """Create a WebUI with a global queue for capturing broadcast events."""
+    return WebUI(ws_id="ws-test")
+
+
+def _drain_global() -> list[dict]:
+    """Drain all events from the global queue."""
+    events = []
+    assert WebUI._global_queue is not None
+    while not WebUI._global_queue.empty():
+        events.append(WebUI._global_queue.get_nowait())
+    return events
+
+
+class TestContentAccumulation:
+    """WebUI should accumulate content tokens and include in idle broadcast."""
+
+    def test_content_token_accumulates(self):
+        """on_content_token should append to _ws_turn_content."""
+        ui = _make_ui()
+        ui.on_content_token("Hello ")
+        ui.on_content_token("world")
+        assert ui._ws_turn_content == ["Hello ", "world"]
+
+    def test_idle_broadcast_includes_content(self):
+        """_broadcast_state('idle') should include joined content and reset."""
+        ui = _make_ui()
+        ui.on_content_token("Hello ")
+        ui.on_content_token("world")
+        ui._broadcast_state("idle")
+
+        events = _drain_global()
+        idle_events = [e for e in events if e.get("state") == "idle"]
+        assert len(idle_events) == 1
+        assert idle_events[0]["content"] == "Hello world"
+        # Accumulator should be reset
+        assert ui._ws_turn_content == []
+        assert ui._ws_turn_content_size == 0
+
+    def test_error_broadcast_resets_without_content(self):
+        """_broadcast_state('error') should reset accumulator without content in event."""
+        ui = _make_ui()
+        ui.on_content_token("partial")
+        ui._broadcast_state("error")
+
+        events = _drain_global()
+        error_events = [e for e in events if e.get("state") == "error"]
+        assert len(error_events) == 1
+        assert "content" not in error_events[0]
+        assert ui._ws_turn_content == []
+        assert ui._ws_turn_content_size == 0
+
+    def test_thinking_broadcast_does_not_touch_accumulator(self):
+        """_broadcast_state('thinking') should not affect the accumulator."""
+        ui = _make_ui()
+        ui.on_content_token("in progress")
+        ui._broadcast_state("thinking")
+
+        assert ui._ws_turn_content == ["in progress"]
+        events = _drain_global()
+        thinking_events = [e for e in events if e.get("state") == "thinking"]
+        assert len(thinking_events) == 1
+        assert "content" not in thinking_events[0]
+
+    def test_multi_round_accumulation(self):
+        """Content from multiple streaming rounds accumulates before idle."""
+        ui = _make_ui()
+        # Round 1
+        ui.on_content_token("I'll check ")
+        ui.on_content_token("that. ")
+        # Round 2 (after tool execution)
+        ui.on_content_token("Here's ")
+        ui.on_content_token("the result.")
+        ui._broadcast_state("idle")
+
+        events = _drain_global()
+        idle_events = [e for e in events if e.get("state") == "idle"]
+        assert len(idle_events) == 1
+        assert idle_events[0]["content"] == "I'll check that. Here's the result."
+
+    def test_empty_content_on_idle_without_tokens(self):
+        """idle with no content tokens should include empty content string."""
+        ui = _make_ui()
+        ui._broadcast_state("idle")
+
+        events = _drain_global()
+        idle_events = [e for e in events if e.get("state") == "idle"]
+        assert len(idle_events) == 1
+        assert idle_events[0]["content"] == ""
+
+    def test_cancellation_preserves_partial_content(self):
+        """Partial content accumulated before cancel should appear in idle event."""
+        ui = _make_ui()
+        ui.on_content_token("I'll ")
+        ui.on_content_token("start by...")
+        # Cancellation triggers idle broadcast with partial content
+        ui._broadcast_state("idle")
+
+        events = _drain_global()
+        idle_events = [e for e in events if e.get("state") == "idle"]
+        assert len(idle_events) == 1
+        assert idle_events[0]["content"] == "I'll start by..."
+
+    def test_consecutive_turns_isolated(self):
+        """Content from turn 1 should not leak into turn 2."""
+        ui = _make_ui()
+        # Turn 1
+        ui.on_content_token("first response")
+        ui._broadcast_state("idle")
+        _drain_global()
+
+        # Turn 2
+        ui.on_content_token("second response")
+        ui._broadcast_state("idle")
+
+        events = _drain_global()
+        idle_events = [e for e in events if e.get("state") == "idle"]
+        assert len(idle_events) == 1
+        assert idle_events[0]["content"] == "second response"
+
+    def test_content_cap_prevents_unbounded_growth(self):
+        """Content exceeding the cap should stop accumulating."""
+        from turnstone.server import _MAX_TURN_CONTENT_CHARS
+
+        ui = _make_ui()
+        # Fill to capacity
+        chunk = "x" * 1024
+        for _ in range(_MAX_TURN_CONTENT_CHARS // 1024 + 10):
+            ui.on_content_token(chunk)
+
+        assert ui._ws_turn_content_size <= _MAX_TURN_CONTENT_CHARS + 1024
+        ui._broadcast_state("idle")
+
+        events = _drain_global()
+        idle_events = [e for e in events if e.get("state") == "idle"]
+        assert len(idle_events) == 1
+        # Content should be capped, not contain everything
+        assert len(idle_events[0]["content"]) <= _MAX_TURN_CONTENT_CHARS + 1024

turnstone/mq/bridge.py

@@ -16,7 +16,6 @@ import os
 import threading
 import time
 import uuid
-from collections import deque
 from typing import TYPE_CHECKING, Any
 
 import httpx
@@ -58,11 +57,6 @@ log = logging.getLogger("turnstone.mq.bridge")
 # Server's default safe tools (auto-approved without user confirmation)
 DEFAULT_SAFE_TOOLS = frozenset(["read_file", "search", "man", "memory", "recall"])
 
-# Maximum total character count of the per-ws content buffer.  Prevents
-# unbounded memory growth if a workstream produces very long responses or
-# the idle event never fires (e.g. bug / disconnect).
-_MAX_CONTENT_BUFFER_CHARS = 256 * 1024
-
 
 class Bridge:
     """Connects a message broker to turnstone-server's HTTP API.
@@ -112,12 +106,6 @@ class Bridge:
         self._active_sends: dict[str, str] = {}  # ws_id → correlation_id
         self._pending_approvals: dict[str, str] = {}  # ws_id → request_id
         self._pending_plan_reviews: dict[str, str] = {}  # ws_id → request_id
-        # Content buffer: accumulates assistant text per workstream from
-        # per-ws SSE.  Attached to TurnCompleteEvent when idle is detected
-        # via the global SSE so downstream consumers can catch up if the
-        # streaming path missed events (race between the two SSE connections).
-        self._ws_content_buffer: dict[str, deque[str]] = {}
-        self._ws_content_buffer_size: dict[str, int] = {}  # running char total
         self._running = True
 
     @property
@@ -599,23 +587,7 @@ class Bridge:
         etype = data.get("type", "")
 
         if etype == "content":
-            text = data.get("text", "")
-            if text:
-                with self._lock:
-                    if ws_id not in self._ws_content_buffer:
-                        self._ws_content_buffer[ws_id] = deque()
-                        self._ws_content_buffer_size[ws_id] = 0
-                    buf = self._ws_content_buffer[ws_id]
-                    buf.append(text)
-                    self._ws_content_buffer_size[ws_id] += len(text)
-                    # Cap buffer per workstream to prevent DoS from
-                    # extremely long responses or missing idle events.
-                    while (
-                        self._ws_content_buffer_size[ws_id] > _MAX_CONTENT_BUFFER_CHARS
-                        and len(buf) > 1
-                    ):
-                        self._ws_content_buffer_size[ws_id] -= len(buf.popleft())
-            self._publish_ws(ws_id, ContentEvent(ws_id=ws_id, text=text))
+            self._publish_ws(ws_id, ContentEvent(ws_id=ws_id, text=data.get("text", "")))
         elif etype == "reasoning":
             self._publish_ws(ws_id, ReasoningEvent(ws_id=ws_id, text=data.get("text", "")))
         elif etype == "tool_info":
@@ -853,14 +825,12 @@ class Bridge:
             if state == "idle":
                 with self._lock:
                     cid = self._active_sends.pop(ws_id, None)
-                    content_parts = self._ws_content_buffer.pop(ws_id, deque())
-                    self._ws_content_buffer_size.pop(ws_id, None)
                 self._publish_ws(
                     ws_id,
                     TurnCompleteEvent(
                         ws_id=ws_id,
                         correlation_id=cid or "",
-                        content="".join(content_parts),
+                        content=data.get("content", ""),
                     ),
                 )
 
@@ -877,8 +847,6 @@ class Bridge:
                 self._ws_auto_approve.pop(ws_id, None)
                 self._ws_approve_tools.pop(ws_id, None)
                 self._active_sends.pop(ws_id, None)
-                self._ws_content_buffer.pop(ws_id, None)
-                self._ws_content_buffer_size.pop(ws_id, None)
 
     # -- heartbeat -----------------------------------------------------------
 

turnstone/mq/protocol.py

@@ -269,10 +269,11 @@ class TurnCompleteEvent(OutboundEvent):
     the ws_state transition to 'idle'.  ``correlation_id`` is set for
     MQ-initiated turns and empty for turns initiated from the server UI.
 
-    ``content`` carries the full assistant response text accumulated from
-    per-ws SSE content tokens.  Downstream consumers (e.g. Discord bot) can
-    use it as a catch-up when the streaming path missed events due to the
-    race between the global SSE and per-ws SSE connections.
+    ``content`` carries the full assistant response text piggybacked on
+    the server's idle SSE event (accumulated server-side in WebUI).
+    Downstream consumers (e.g. Discord bot) use it for catch-up when the
+    streaming path missed events, and as the primary delivery path for
+    bidirectional notification DM forwarding.
     """
 
     type: str = "turn_complete"

turnstone/sdk/events.py

@@ -174,6 +174,7 @@ class WsStateEvent(ServerEvent):
     context_ratio: float = 0.0
     activity: str = ""
     activity_state: str = ""
+    content: str = ""  # populated on idle transitions only
 
 
 @dataclass

turnstone/server.py

@@ -67,6 +67,8 @@ _HTML = (_STATIC_DIR / "index.html").read_text(encoding="utf-8")
 # WebUI — implements SessionUI for browser-based interaction
 # ---------------------------------------------------------------------------
 
+_MAX_TURN_CONTENT_CHARS = 256 * 1024  # cap piggybacked content on idle events
+
 
 class WebUI:
     """Browser-based UI using SSE for streaming and HTTP POST for actions.
@@ -107,6 +109,10 @@ class WebUI:
         self._pending_verdicts: list[dict[str, Any]] = []
         # Last user decision for late-arriving verdicts (set in resolve_approval)
         self._last_verdict_decision: str = ""
+        # Content accumulator — tokens appended in on_content_token(), joined
+        # and piggybacked onto the ws_state:idle global SSE event, then reset.
+        self._ws_turn_content: list[str] = []
+        self._ws_turn_content_size: int = 0
 
     def _enqueue(self, data: dict[str, Any]) -> None:
         with self._listeners_lock:
@@ -135,17 +141,23 @@ class WebUI:
                 ctx = self._ws_context_ratio
                 activity = self._ws_current_activity
                 activity_state = self._ws_activity_state
-            WebUI._global_queue.put(
-                {
-                    "type": "ws_state",
-                    "ws_id": self.ws_id,
-                    "state": state,
-                    "tokens": tokens,
-                    "context_ratio": ctx,
-                    "activity": activity,
-                    "activity_state": activity_state,
-                }
-            )
+            event: dict[str, Any] = {
+                "type": "ws_state",
+                "ws_id": self.ws_id,
+                "state": state,
+                "tokens": tokens,
+                "context_ratio": ctx,
+                "activity": activity,
+                "activity_state": activity_state,
+            }
+            if state == "idle":
+                event["content"] = "".join(self._ws_turn_content)
+                self._ws_turn_content = []
+                self._ws_turn_content_size = 0
+            elif state == "error":
+                self._ws_turn_content = []
+                self._ws_turn_content_size = 0
+            WebUI._global_queue.put(event)
 
     def _broadcast_activity(self) -> None:
         """Send an activity-change event to the global SSE channel."""
@@ -178,6 +190,9 @@ class WebUI:
         self._enqueue({"type": "reasoning", "text": text})
 
     def on_content_token(self, text: str) -> None:
+        if self._ws_turn_content_size < _MAX_TURN_CONTENT_CHARS:
+            self._ws_turn_content.append(text)
+            self._ws_turn_content_size += len(text)
         self._enqueue({"type": "content", "text": text})
 
     def on_stream_end(self) -> None: