[eric] streaming: stream frames stop riding the dashboard socket and the backgrounded chat batches to 1Hz (ENG-329)

backend/apps/agents/core/ws_manager.py

@@ -129,11 +129,16 @@ class ConnectionManager:
                     await ws.send_text(payload_str)
                 except Exception:
                     logger.debug("send_to_session: send failed (will retry on reconnect)", exc_info=True)
-            for ws in list(self.global_connections):
-                try:
-                    await ws.send_text(payload_str)
-                except Exception:
-                    logger.debug("send_to_session: global send failed", exc_info=True)
+            # Stream frames never ride the dashboard socket: its only client drops them unread
+            # (skipStreamEvents), so fanning them out just taxed the renderer with a parse per
+            # token, twice for an expanded chat. Session sockets carry the stream; /ws/dashboard
+            # has no replay protocol, so nothing downstream misses them.
+            if event not in ("agent:stream_start", "agent:stream_delta", "agent:stream_end"):
+                for ws in list(self.global_connections):
+                    try:
+                        await ws.send_text(payload_str)
+                    except Exception:
+                        logger.debug("send_to_session: global send failed", exc_info=True)
             # Persist under the lock so a concurrent running status can't race past and overwrite with stale state.
             if event == "agent:status" and data.get("status") in TERMINAL_STATUSES:
                 seq_log.persist_terminal(session_id, payload_str)

backend/tests/test_stream_frames_skip_dashboard_socket.py

@@ -0,0 +1,49 @@
+"""Stream frames must never ride the dashboard socket: its only client (dashboardWs,
+skipStreamEvents=true) drops them unread, so the fan-out taxed the renderer with a JSON parse per
+streamed token and doubled the work for an expanded chat (its own socket carries the real copy).
+Both directions pinned: deltas stay session-only, and every other event still reaches the
+dashboard socket, or narrator pills and status chips would go blind."""
+
+import pytest
+from unittest.mock import AsyncMock
+
+from backend.apps.agents.core.ws_manager import ws_manager
+
+
+class P_FakeWs:
+    def __init__(self):
+        self.sent = []
+
+    async def send_text(self, payload: str):
+        self.sent.append(payload)
+
+
+@pytest.fixture()
+def wired():
+    session_ws = P_FakeWs()
+    dashboard_ws = P_FakeWs()
+    ws_manager.connections.setdefault("sess-1", []).append(session_ws)
+    ws_manager.global_connections.append(dashboard_ws)
+    yield session_ws, dashboard_ws
+    ws_manager.connections.get("sess-1", []).remove(session_ws)
+    if not ws_manager.connections.get("sess-1"):
+        ws_manager.connections.pop("sess-1", None)
+    ws_manager.global_connections.remove(dashboard_ws)
+
+
+@pytest.mark.asyncio
+async def test_stream_frames_reach_the_session_socket_only(wired):
+    session_ws, dashboard_ws = wired
+    for event in ("agent:stream_start", "agent:stream_delta", "agent:stream_end"):
+        await ws_manager.send_to_session("sess-1", event, {"message_id": "m1", "delta": "hi"})
+    assert len(session_ws.sent) == 3, "the chat's own socket must carry the full stream"
+    assert dashboard_ws.sent == [], "the dashboard socket must never see a stream frame"
+
+
+@pytest.mark.asyncio
+async def test_every_other_event_still_fans_out_globally(wired):
+    session_ws, dashboard_ws = wired
+    await ws_manager.send_to_session("sess-1", "agent:status", {"status": "running"})
+    await ws_manager.send_to_session("sess-1", "agent:message", {"message": {}})
+    assert len(session_ws.sent) == 2
+    assert len(dashboard_ws.sent) == 2, "non-stream events power collapsed cards; they must keep flowing"

frontend/src/shared/ws/BackgroundDeltaBuffer.test.ts

@@ -0,0 +1,43 @@
+// The 1Hz background-stream buffer (ENG-329): text must NEVER be lost or reordered, only batched.
+// Eviction on message switch is the ordering invariant; a buffer that held two messages could
+// interleave them wrong on flush.
+import { test } from 'node:test';
+import assert from 'node:assert';
+import { BackgroundDeltaBuffer } from './BackgroundDeltaBuffer';
+
+test('same-message deltas coalesce into one payload, byte-exact', () => {
+  const b = new BackgroundDeltaBuffer();
+  assert.strictEqual(b.add('m1', 'Hello '), null);
+  assert.strictEqual(b.add('m1', 'wor'), null);
+  assert.strictEqual(b.add('m1', 'ld'), null);
+  assert.deepStrictEqual(b.take(), { messageId: 'm1', text: 'Hello world' });
+  assert.strictEqual(b.hasPending, false);
+});
+
+test('a delta for a different message evicts the pending one first', () => {
+  const b = new BackgroundDeltaBuffer();
+  b.add('m1', 'first');
+  const evicted = b.add('m2', 'second');
+  assert.deepStrictEqual(evicted, { messageId: 'm1', text: 'first' });
+  assert.deepStrictEqual(b.take(), { messageId: 'm2', text: 'second' });
+});
+
+test('take on empty is null and pendingMessageId tracks the buffer', () => {
+  const b = new BackgroundDeltaBuffer();
+  assert.strictEqual(b.take(), null);
+  assert.strictEqual(b.pendingMessageId, null);
+  b.add('m9', 'x');
+  assert.strictEqual(b.pendingMessageId, 'm9');
+});
+
+test('nothing is lost across an evict-then-take sequence (byte accounting)', () => {
+  const b = new BackgroundDeltaBuffer();
+  const seen: string[] = [];
+  for (const [mid, d] of [['a', '1'], ['a', '2'], ['b', '3'], ['b', '4'], ['a', '5']] as const) {
+    const ev = b.add(mid, d);
+    if (ev) seen.push(ev.text);
+  }
+  const last = b.take();
+  if (last) seen.push(last.text);
+  assert.strictEqual(seen.join(''), '12345');
+});

frontend/src/shared/ws/BackgroundDeltaBuffer.ts

@@ -0,0 +1,42 @@
+export interface PendingDelta {
+  messageId: string;
+  text: string;
+}
+
+// Coalesces stream deltas for the ONE backgrounded keep-alive session socket: nobody is reading
+// that transcript, so frame-rate Redux dispatches are pure heat (ENG-329). Deltas for the same
+// message concatenate; a delta for a DIFFERENT message evicts the pending one so per-session
+// ordering can never invert. The socket flushes on a 1s timer, on reopen, and on any non-delta
+// event, so no text is ever lost, it just lands in 1Hz batches while backgrounded.
+export class BackgroundDeltaBuffer {
+  private pending: PendingDelta | null = null;
+
+  // Buffer this delta; returns a delta that must dispatch FIRST to preserve order, or null.
+  add(messageId: string, delta: string): PendingDelta | null {
+    if (this.pending && this.pending.messageId !== messageId) {
+      const evicted = this.pending;
+      this.pending = { messageId, text: delta };
+      return evicted;
+    }
+    if (this.pending) {
+      this.pending.text += delta;
+      return null;
+    }
+    this.pending = { messageId, text: delta };
+    return null;
+  }
+
+  take(): PendingDelta | null {
+    const p = this.pending;
+    this.pending = null;
+    return p;
+  }
+
+  get hasPending(): boolean {
+    return this.pending !== null;
+  }
+
+  get pendingMessageId(): string | null {
+    return this.pending?.messageId ?? null;
+  }
+}

frontend/src/shared/ws/WebSocketManager.ts

@@ -30,6 +30,7 @@ import {
   clearTurnLabel,
 } from '../state/agentsSlice';
 import { streamStart, streamDelta, streamEnd, clearStreamingForSession } from '../state/streamingSlice';
+import { BackgroundDeltaBuffer } from './BackgroundDeltaBuffer';
 import { addBrowserCardFromBackend, setBrowserDocked, markBrowserCardEnding, keepBrowserCardOpen, placeBesideCard, placeBelowCard, placeBrowserBesideChat, setBrowserCardPosition, setGlowingBrowserCards, fadeGlowingBrowserCards, clearGlowingBrowserCards, removeBrowserCard, GRID_GAP, WORKFLOW_CARD_GAP, openWorkflowsApp, openWorkflowMonitor } from '../state/dashboardLayoutSlice';
 import { upsertOutput } from '../state/outputsSlice';
 import { setCardPosition } from '../state/dashboardLayoutSlice';
@@ -88,6 +89,9 @@ class WebSocketManager {
   private ws: WebSocket | null = null;
   private url: string;
   private skipStreamEvents: boolean;
+  private backgrounded = false;
+  private bgBuffer = new BackgroundDeltaBuffer();
+  private bgFlushTimer: ReturnType<typeof setTimeout> | null = null;
   private sessionId: string | null;
 
   // Resume state. lastSeq is the highest server-assigned seq this client has applied; it's sent on every (re)connect so the server can replay missed events. Persists for the lifetime of this WebSocketManager instance, when the user navigates away and a new createSessionWs() is constructed, lastSeq starts at 0 and we get a full replay.
@@ -179,9 +183,41 @@ class WebSocketManager {
       firstAgentResponseMarked = true;
       try { (window as any).openswarm?.markFirstAgentResponse?.(); } catch { /* not in Electron */ }
     }
+    if (this.backgrounded) {
+      const evicted = this.bgBuffer.add(messageId, delta);
+      if (evicted) store.dispatch(streamDelta({ sessionId, messageId: evicted.messageId, delta: evicted.text }));
+      this.armBgFlush();
+      return;
+    }
     store.dispatch(streamDelta({ sessionId, messageId, delta }));
   }
 
+  // Backgrounded = kept alive across a chat hop with nobody watching; deltas batch to 1Hz there
+  // (ENG-329). Reopen flushes synchronously BEFORE the transcript remounts, so no text is lost.
+  setBackgrounded(value: boolean) {
+    this.backgrounded = value;
+    if (!value) this.flushBgDelta();
+  }
+
+  private armBgFlush() {
+    if (this.bgFlushTimer !== null) return;
+    this.bgFlushTimer = setTimeout(() => {
+      this.bgFlushTimer = null;
+      this.flushBgDelta();
+    }, 1000);
+  }
+
+  private flushBgDelta() {
+    if (this.bgFlushTimer !== null) {
+      clearTimeout(this.bgFlushTimer);
+      this.bgFlushTimer = null;
+    }
+    const p = this.bgBuffer.take();
+    if (p && this.sessionId) {
+      store.dispatch(streamDelta({ sessionId: this.sessionId, messageId: p.messageId, delta: p.text }));
+    }
+  }
+
   connect() {
     if (this.ws?.readyState === WebSocket.OPEN) return;
     this.explicitlyClosed = false;
@@ -253,6 +289,7 @@ class WebSocketManager {
   }
 
   disconnect() {
+    this.flushBgDelta();
     this.explicitlyClosed = true;
     if (this.reconnectTimer) {
       clearTimeout(this.reconnectTimer);
@@ -405,6 +442,13 @@ class WebSocketManager {
       }
     }
 
+    // A buffered background delta must land BEFORE any other event for this socket, or a
+    // stream_end / finalized message could overtake its own text and truncate the transcript.
+    if (this.bgBuffer.hasPending
+        && !(event === 'agent:stream_delta' && data?.message_id === this.bgBuffer.pendingMessageId)) {
+      this.flushBgDelta();
+    }
+
     switch (event) {
       case 'agent:test_state':
         // broadcast_global puts everything under data (no top-level session_id).
@@ -1032,6 +1076,7 @@ export function acquireSessionWs(sessionId: string): WebSocketManager {
   if (_backgroundedSessionWs?.sessionId === sessionId) {
     const ws = _backgroundedSessionWs.ws;
     _backgroundedSessionWs = null;
+    ws.setBackgrounded(false);
     return ws;
   }
   warnIfNotCanonicalSessionId(sessionId, 'acquireSessionWs');
@@ -1045,6 +1090,7 @@ export function releaseSessionWs(sessionId: string, ws: WebSocketManager, keepAl
     _backgroundedSessionWs = null;
   }
   if (keepAlive) {
+    ws.setBackgrounded(true);
     _backgroundedSessionWs = { sessionId, ws };
   } else {
     ws.disconnect();