diff --git a/backend/apps/agents/core/ws_manager.py b/backend/apps/agents/core/ws_manager.py index 48fe69a0..0db3e8e1 100644 --- a/backend/apps/agents/core/ws_manager.py +++ b/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) diff --git a/backend/tests/test_stream_frames_skip_dashboard_socket.py b/backend/tests/test_stream_frames_skip_dashboard_socket.py new file mode 100644 index 00000000..2162100f --- /dev/null +++ b/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" diff --git a/frontend/src/shared/ws/BackgroundDeltaBuffer.test.ts b/frontend/src/shared/ws/BackgroundDeltaBuffer.test.ts new file mode 100644 index 00000000..3f3a5fed --- /dev/null +++ b/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'); +}); diff --git a/frontend/src/shared/ws/BackgroundDeltaBuffer.ts b/frontend/src/shared/ws/BackgroundDeltaBuffer.ts new file mode 100644 index 00000000..11fd3fa4 --- /dev/null +++ b/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; + } +} diff --git a/frontend/src/shared/ws/WebSocketManager.ts b/frontend/src/shared/ws/WebSocketManager.ts index b9da563d..625f88a1 100644 --- a/frontend/src/shared/ws/WebSocketManager.ts +++ b/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 | 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();