From 1ca4ea044af0e4d7a542aa562e124c1ddcbdbda4 Mon Sep 17 00:00:00 2001 From: oxoxDev <164490987+oxoxDev@users.noreply.github.com> Date: Thu, 9 Apr 2026 02:57:25 +0530 Subject: [PATCH] fix(chat): deduplicate assistant messages by subscribing to canonical events only (#432) (#439) * fix(chat): deduplicate assistant messages by subscribing to canonical events only (#432) The Rust core emits socket events with both snake_case and colon:case aliases via emit_with_aliases(). The frontend was subscribing to both, causing every chat event to fire twice and producing duplicate assistant messages. - Subscribe only to canonical snake_case events (tool_call, chat_segment, chat_done, chat_error) instead of both naming conventions - Add safety-net dedup layer in Conversations.tsx using a seen-events map with TTL to guard against any remaining edge cases - Add unit tests verifying only canonical events are processed Closes #432 Co-Authored-By: Claude Opus 4.6 * style: apply prettier formatting to chatService test Co-Authored-By: Claude Opus 4.6 --------- Co-authored-by: Claude Opus 4.6 --- app/src/pages/Conversations.tsx | 43 +++++++++ .../services/__tests__/chatService.test.ts | 89 +++++++++++++++++++ app/src/services/chatService.ts | 39 ++++---- 3 files changed, 156 insertions(+), 15 deletions(-) create mode 100644 app/src/services/__tests__/chatService.test.ts diff --git a/app/src/pages/Conversations.tsx b/app/src/pages/Conversations.tsx index 0248508ef..2431fc280 100644 --- a/app/src/pages/Conversations.tsx +++ b/app/src/pages/Conversations.tsx @@ -229,6 +229,34 @@ const Conversations = () => { const autocompleteDebounceRef = useRef(null); const autocompleteRequestSeqRef = useRef(0); const sendingTimeoutRef = useRef | null>(null); + const seenChatEventsRef = useRef>(new Map()); + + const markChatEventSeen = (key: string): boolean => { + const now = Date.now(); + const cache = seenChatEventsRef.current; + const ttlMs = 10 * 60_000; + const maxEntries = 500; + + if (cache.has(key)) return false; + + cache.set(key, now); + + // Prune old entries first. + for (const [existingKey, timestamp] of cache) { + if (now - timestamp > ttlMs) { + cache.delete(existingKey); + } + } + + // Keep bounded memory in long sessions. + while (cache.size > maxEntries) { + const oldest = cache.keys().next().value; + if (!oldest) break; + cache.delete(oldest); + } + + return true; + }; const getAudioExtension = (mimeType: string): string => { const lower = mimeType.toLowerCase(); @@ -400,6 +428,9 @@ const Conversations = () => { const cleanup = subscribeChatEvents({ onToolCall: (event: ChatToolCallEvent) => { + const eventKey = `tool_call:${event.thread_id}:${event.request_id ?? 'none'}:${event.round}:${event.tool_name}`; + if (!markChatEventSeen(eventKey)) return; + setToolTimelineByThread(prev => { const existing = prev[event.thread_id] ?? []; return { @@ -417,6 +448,9 @@ const Conversations = () => { }); }, onToolResult: (event: ChatToolResultEvent) => { + const eventKey = `tool_result:${event.thread_id}:${event.request_id ?? 'none'}:${event.round}:${event.tool_name}:${event.success}`; + if (!markChatEventSeen(eventKey)) return; + setToolTimelineByThread(prev => { const existing = prev[event.thread_id] ?? []; if (existing.length === 0) return prev; @@ -441,6 +475,9 @@ const Conversations = () => { }); }, onSegment: (event: ChatSegmentEvent) => { + const eventKey = `segment:${event.thread_id}:${event.request_id}:${event.segment_index}`; + if (!markChatEventSeen(eventKey)) return; + // Rust delivers segments with delays already applied — just dispatch. if (event.reaction_emoji) { const pending = pendingReactionRef.current.get(event.thread_id); @@ -458,6 +495,9 @@ const Conversations = () => { dispatch(addInferenceResponse({ content: segmentText(event), threadId: event.thread_id })); }, onDone: event => { + const eventKey = `done:${event.thread_id}:${event.request_id ?? 'none'}`; + if (!markChatEventSeen(eventKey)) return; + // Update tool timeline setToolTimelineByThread(prev => { const existing = prev[event.thread_id] ?? []; @@ -508,6 +548,9 @@ const Conversations = () => { dispatch(setActiveThread(null)); }, onError: event => { + const eventKey = `error:${event.thread_id}:${event.request_id ?? 'none'}:${event.error_type}:${event.message}`; + if (!markChatEventSeen(eventKey)) return; + if (event.thread_id !== selectedThreadIdRef.current) return; if (sendingTimeoutRef.current) { clearTimeout(sendingTimeoutRef.current); diff --git a/app/src/services/__tests__/chatService.test.ts b/app/src/services/__tests__/chatService.test.ts new file mode 100644 index 000000000..c3ff5086f --- /dev/null +++ b/app/src/services/__tests__/chatService.test.ts @@ -0,0 +1,89 @@ +import { beforeEach, describe, expect, it, vi } from 'vitest'; + +import { subscribeChatEvents } from '../chatService'; +import { socketService } from '../socketService'; + +vi.mock('../socketService', () => ({ socketService: { getSocket: vi.fn() } })); + +type Handler = (...args: unknown[]) => void; + +function createMockSocket() { + const handlers = new Map(); + const on = vi.fn((event: string, cb: Handler) => { + const existing = handlers.get(event) ?? []; + existing.push(cb); + handlers.set(event, existing); + }); + const off = vi.fn((event: string, cb: Handler) => { + const existing = handlers.get(event) ?? []; + handlers.set( + event, + existing.filter(handler => handler !== cb) + ); + }); + const emit = (event: string, payload: unknown) => { + for (const handler of handlers.get(event) ?? []) { + handler(payload); + } + }; + + return { id: 'socket-1', on, off, emit }; +} + +describe('chatService.subscribeChatEvents', () => { + beforeEach(() => { + vi.clearAllMocks(); + }); + + it('subscribes to canonical snake_case chat events only', () => { + const socket = createMockSocket(); + vi.mocked(socketService.getSocket).mockReturnValue(socket as never); + + subscribeChatEvents({ + onToolCall: () => {}, + onToolResult: () => {}, + onSegment: () => {}, + onDone: () => {}, + onError: () => {}, + }); + + const subscribedEvents = socket.on.mock.calls.map(call => call[0]); + expect(subscribedEvents).toEqual([ + 'tool_call', + 'tool_result', + 'chat_segment', + 'chat_done', + 'chat_error', + ]); + expect(subscribedEvents).not.toContain('chat:tool_call'); + expect(subscribedEvents).not.toContain('chat:tool_result'); + expect(subscribedEvents).not.toContain('chat:segment'); + expect(subscribedEvents).not.toContain('chat:done'); + expect(subscribedEvents).not.toContain('chat:error'); + }); + + it('does not process alias events when only canonical subscriptions are active', () => { + const socket = createMockSocket(); + vi.mocked(socketService.getSocket).mockReturnValue(socket as never); + const onDone = vi.fn(); + + subscribeChatEvents({ onDone }); + + socket.emit('chat:done', { thread_id: 't1' }); + expect(onDone).not.toHaveBeenCalled(); + + socket.emit('chat_done', { thread_id: 't1' }); + expect(onDone).toHaveBeenCalledTimes(1); + }); + + it('removes all handlers on cleanup', () => { + const socket = createMockSocket(); + vi.mocked(socketService.getSocket).mockReturnValue(socket as never); + + const cleanup = subscribeChatEvents({ onToolCall: () => {}, onDone: () => {} }); + cleanup(); + + const unsubscribedEvents = socket.off.mock.calls.map(call => call[0]); + expect(unsubscribedEvents).toEqual(['tool_call', 'chat_done']); + }); +}); diff --git a/app/src/services/chatService.ts b/app/src/services/chatService.ts index 2e623cc4d..4c6d9941f 100644 --- a/app/src/services/chatService.ts +++ b/app/src/services/chatService.ts @@ -11,6 +11,7 @@ import { socketService } from './socketService'; export interface ChatToolCallEvent { thread_id: string; + request_id?: string; tool_name: string; skill_id: string; args: Record; @@ -19,6 +20,7 @@ export interface ChatToolCallEvent { export interface ChatToolResultEvent { thread_id: string; + request_id?: string; tool_name: string; skill_id: string; output: string; @@ -28,6 +30,7 @@ export interface ChatToolResultEvent { export interface ChatDoneEvent { thread_id: string; + request_id?: string; full_response: string; rounds_used: number; total_input_tokens: number; @@ -60,6 +63,7 @@ export function segmentText(event: ChatSegmentEvent): string { export interface ChatErrorEvent { thread_id: string; + request_id?: string; message: string; error_type: 'network' | 'timeout' | 'tool_error' | 'inference' | 'cancelled'; round: number | null; @@ -78,40 +82,45 @@ export function subscribeChatEvents(listeners: ChatEventListeners): () => void { if (!socket) return () => {}; const handlers: Array<[string, (...args: unknown[]) => void]> = []; + // Canonical convention for web-channel events is snake_case. + // The core emits aliases for compatibility, but subscribing once avoids + // processing the same logical event twice. + const EVENTS = { + toolCall: 'tool_call', + toolResult: 'tool_result', + segment: 'chat_segment', + done: 'chat_done', + error: 'chat_error', + } as const; if (listeners.onToolCall) { const cb = (payload: unknown) => listeners.onToolCall?.(payload as ChatToolCallEvent); - socket.on('chat:tool_call', cb); - socket.on('tool_call', cb); - handlers.push(['chat:tool_call', cb], ['tool_call', cb]); + socket.on(EVENTS.toolCall, cb); + handlers.push([EVENTS.toolCall, cb]); } if (listeners.onToolResult) { const cb = (payload: unknown) => listeners.onToolResult?.(payload as ChatToolResultEvent); - socket.on('chat:tool_result', cb); - socket.on('tool_result', cb); - handlers.push(['chat:tool_result', cb], ['tool_result', cb]); + socket.on(EVENTS.toolResult, cb); + handlers.push([EVENTS.toolResult, cb]); } if (listeners.onSegment) { const cb = (payload: unknown) => listeners.onSegment?.(payload as ChatSegmentEvent); - socket.on('chat:segment', cb); - socket.on('chat_segment', cb); - handlers.push(['chat:segment', cb], ['chat_segment', cb]); + socket.on(EVENTS.segment, cb); + handlers.push([EVENTS.segment, cb]); } if (listeners.onDone) { const cb = (payload: unknown) => listeners.onDone?.(payload as ChatDoneEvent); - socket.on('chat:done', cb); - socket.on('chat_done', cb); - handlers.push(['chat:done', cb], ['chat_done', cb]); + socket.on(EVENTS.done, cb); + handlers.push([EVENTS.done, cb]); } if (listeners.onError) { const cb = (payload: unknown) => listeners.onError?.(payload as ChatErrorEvent); - socket.on('chat:error', cb); - socket.on('chat_error', cb); - handlers.push(['chat:error', cb], ['chat_error', cb]); + socket.on(EVENTS.error, cb); + handlers.push([EVENTS.error, cb]); } return () => {