Broadcast user messages to parallel clients

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:

nerve/gateway/server.py

@@ -617,6 +617,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(

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 = []

web/src/api/websocket.ts

@@ -24,6 +24,7 @@ export type WSMessage =
   | { type: 'notification_answered'; notification_id: string; session_id: string; answer: string; answered_by: string; approval_status?: 'answered' | 'snoozed'; dispatch_ok?: boolean; snooze_until?: string }
   | { type: 'notification_expired'; notification_id: string; session_id: string; notification_type: string; title: string }
   | { 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' }[] }

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, handleModelChanged } 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, handleNotificationExpired, handleBackgroundTasksUpdate } from './handlers/auxiliaryHandlers';
 
@@ -733,6 +733,7 @@ export const useChatStore = create<ChatState>((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);

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<WSMessage, { type: 'user_message' }>,
+  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] }));
+}