From 580dda347231b76ed282a49620305c513754013e Mon Sep 17 00:00:00 2001 From: Pervakov Grigorii Date: Tue, 30 Jun 2026 16:33:07 +0200 Subject: [PATCH] Broadcast user messages to parallel clients A message sent from one web client was only rendered optimistically on the sender; other connected clients (and the same client in a second tab) saw the assistant reply with no preceding user bubble until they reloaded. Echo the user's message over the session channel from the inbound WS handler, excluding the sender (which already showed it). Parallel clients append the bubble live via a new user_message event; engine.run still persists it, so reloads continue to source it from history. Adds an exclude arg to StreamBroadcaster.broadcast plus a test. Co-Authored-By: Claude Opus 4.7 --- nerve/agent/streaming.py | 11 +++++++++-- nerve/gateway/server.py | 11 +++++++++++ tests/test_streaming.py | 17 +++++++++++++++++ web/src/api/websocket.ts | 1 + web/src/stores/chatStore.ts | 3 ++- web/src/stores/handlers/sessionHandlers.ts | 14 ++++++++++++++ 6 files changed, 54 insertions(+), 3 deletions(-) diff --git a/nerve/agent/streaming.py b/nerve/agent/streaming.py index 6fd8456..a8f615e 100644 --- a/nerve/agent/streaming.py +++ b/nerve/agent/streaming.py @@ -82,8 +82,13 @@ async def unregister(self, session_id: str, callback_id: str) -> None: if not self._listeners[session_id]: del self._listeners[session_id] - async def broadcast(self, session_id: str, message: dict[str, Any]) -> None: - """Send a message to all listeners of a session. Also buffers if active.""" + async def broadcast(self, session_id: str, message: dict[str, Any], exclude: str | None = None) -> None: + """Send a message to all listeners of a session. Also buffers if active. + + ``exclude`` skips the listener with that callback id — used to echo an + event to every client except the one that originated it (and already + rendered it optimistically). + """ # Terminal events close the open-turn flag so the engine's # backstop in run() knows no synthetic "done" is needed. if message.get("type") in ("done", "stopped", "error"): @@ -101,6 +106,8 @@ async def broadcast(self, session_id: str, message: dict[str, Any]) -> None: listeners = list(self._listeners.get(session_id, [])) for callback_id, callback in listeners: + if callback_id == exclude: + continue try: await callback(session_id, message) except Exception as e: diff --git a/nerve/gateway/server.py b/nerve/gateway/server.py index b9e27f1..e63a8e5 100644 --- a/nerve/gateway/server.py +++ b/nerve/gateway/server.py @@ -606,6 +606,17 @@ async def ws_broadcast(session_id: str, message: dict): _engine.db, file_ids, ) + # Echo the message to every *other* client of this session so + # parallel tabs render the user bubble live (the sender already + # showed it optimistically). engine.run persists it, so reloads + # get it from history regardless. + await broadcaster.broadcast(session_id, { + "type": "user_message", + "session_id": session_id, + "content": user_text, + "blocks": image_refs or None, + }, exclude=client_id) + # Run agent in background, store task for stop support task = asyncio.create_task( _engine.run( diff --git a/tests/test_streaming.py b/tests/test_streaming.py index 5499224..40e22e6 100644 --- a/tests/test_streaming.py +++ b/tests/test_streaming.py @@ -54,6 +54,23 @@ async def h2(sid, msg): assert len(r1) == 1 assert len(r2) == 1 + async def test_broadcast_exclude_skips_origin_listener(self): + bc = StreamBroadcaster() + sender_msgs, other_msgs = [], [] + + async def sender(sid, msg): + sender_msgs.append(msg) + + async def other(sid, msg): + other_msgs.append(msg) + + await bc.register("s1", "sender", sender) + await bc.register("s1", "other", other) + await bc.broadcast("s1", {"type": "user_message"}, exclude="sender") + + assert sender_msgs == [] # origin listener is skipped + assert len(other_msgs) == 1 # every other listener still receives + async def test_failed_listener_doesnt_block(self): bc = StreamBroadcaster() received = [] diff --git a/web/src/api/websocket.ts b/web/src/api/websocket.ts index f0877b6..d33aa17 100644 --- a/web/src/api/websocket.ts +++ b/web/src/api/websocket.ts @@ -23,6 +23,7 @@ export type WSMessage = | { type: 'notification'; notification_id: string; notification_type: 'notify' | 'question' | 'approval'; session_id: string; title: string; body: string; priority: string; options: string[] | null; option_labels?: Record; target_kind?: string; target_id?: string; silenced?: boolean; silence_reason?: string; silence_pattern?: string; silenced_by?: string } | { type: 'notification_answered'; notification_id: string; session_id: string; answer: string; answered_by: string; approval_status?: 'answered' | 'snoozed'; dispatch_ok?: boolean } | { type: 'answer_injected'; session_id: string; notification_id: string; title: string; answer: string; answered_by: string; content: string } + | { type: 'user_message'; session_id: string; content: string; blocks?: { type: string; url?: string; filename?: string; media_type?: string; size?: number }[] | null } | { type: 'session_running'; session_id: string; is_running: boolean } | { type: 'session_awaiting_input'; session_id: string; awaiting: boolean } | { type: 'background_tasks_update'; session_id: string; tasks: { task_id: string; label: string; tool: string; status: 'running' | 'done' | 'failed' | 'timeout' }[] } diff --git a/web/src/stores/chatStore.ts b/web/src/stores/chatStore.ts index 17df747..8240125 100644 --- a/web/src/stores/chatStore.ts +++ b/web/src/stores/chatStore.ts @@ -10,7 +10,7 @@ import { cancelAutoClose, clearAllAutoCloseTimers, MAX_COMPLETED_TABS } from './ import { extractTodosFromMessages, extractCCTasksFromMessages } from './helpers/bufferReplay'; // Handlers import { handleThinking, handleToken, handleToolUse, handleToolResult, handleDone, handleStopped, handleError, handleWakeup, handleAutoTurn } from './handlers/streamingHandlers'; -import { handleSessionUpdated, handleSessionStatus, handleSessionSwitched, handleSessionForked, handleSessionResumed, handleSessionArchived, handleSessionRunning, handleSessionAwaitingInput, handleAnswerInjected } from './handlers/sessionHandlers'; +import { handleSessionUpdated, handleSessionStatus, handleSessionSwitched, handleSessionForked, handleSessionResumed, handleSessionArchived, handleSessionRunning, handleSessionAwaitingInput, handleAnswerInjected, handleUserMessage } from './handlers/sessionHandlers'; import { handlePlanUpdate, handleSubagentStart, handleSubagentComplete, handleHoaProgress, handleWorkflowProgress } from './handlers/panelHandlers'; import { handleInteraction, handleFileChanged, handleNotification, handleNotificationAnswered, handleBackgroundTasksUpdate } from './handlers/auxiliaryHandlers'; @@ -707,6 +707,7 @@ export const useChatStore = create((set, get) => ({ case 'session_running': return handleSessionRunning(msg, get, set); case 'session_awaiting_input': return handleSessionAwaitingInput(msg, get, set); case 'answer_injected': return handleAnswerInjected(msg, get, set); + case 'user_message': return handleUserMessage(msg, get, set); // Panels case 'plan_update': return handlePlanUpdate(msg, get, set); case 'subagent_start': return handleSubagentStart(msg, get, set); diff --git a/web/src/stores/handlers/sessionHandlers.ts b/web/src/stores/handlers/sessionHandlers.ts index a27cc8f..3c03c30 100644 --- a/web/src/stores/handlers/sessionHandlers.ts +++ b/web/src/stores/handlers/sessionHandlers.ts @@ -1,6 +1,7 @@ import type { WSMessage } from '../../api/websocket'; import type { PanelTab } from '../../types/chat'; import { applyStreamEvent, rebuildPanelTabsFromBuffer, deriveStatus, extractTodosFromBuffer, extractCCTasksFromBuffer } from '../helpers/bufferReplay'; +import { hydrateMessage } from '../../utils/hydrateMessage'; import type { Get, Set } from './types'; // ------------------------------------------------------------------ // @@ -262,3 +263,16 @@ export function handleAnswerInjected( })); } } + +export function handleUserMessage( + msg: Extract, + get: Get, + set: Set, +): void { + // A message sent from another client of this session — render the user + // bubble live. The sender is excluded server-side, so this never duplicates + // its own optimistic message. hydrateMessage rebuilds any image/file blocks. + if (msg.session_id !== get().activeSession) return; + const hydrated = hydrateMessage({ role: 'user', content: msg.content, blocks: msg.blocks ?? undefined }); + set(s => ({ messages: [...s.messages, hydrated] })); +}