diff --git a/nerve/agent/streaming.py b/nerve/agent/streaming.py index 66646ac..efb1a18 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 f191050..1b07a70 100644 --- a/nerve/gateway/server.py +++ b/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( 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 3413028..b991524 100644 --- a/web/src/api/websocket.ts +++ b/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' }[] } diff --git a/web/src/stores/chatStore.ts b/web/src/stores/chatStore.ts index 15bfbf1..e43bd11 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, 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((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] })); +}