From f144eff46ee4c328a0cf1ba66446de7076415b67 Mon Sep 17 00:00:00 2001 From: Bohdan Triapitsyn Date: Sat, 20 Dec 2025 02:34:01 +0200 Subject: [PATCH] feat: enhance message handling with deduplication and resync functionality --- .../ui/src/components/chat/MessageList.tsx | 12 +++- packages/ui/src/hooks/useEventStream.ts | 48 +++++++++++--- packages/ui/src/stores/messageStore.ts | 62 +++++++++++++++++-- 3 files changed, 108 insertions(+), 14 deletions(-) diff --git a/packages/ui/src/components/chat/MessageList.tsx b/packages/ui/src/components/chat/MessageList.tsx index 895b7ac3..567f6a35 100644 --- a/packages/ui/src/components/chat/MessageList.tsx +++ b/packages/ui/src/components/chat/MessageList.tsx @@ -39,7 +39,17 @@ const MessageList: React.FC = ({ }, [permissions, onMessageContentChange]); const displayMessages = React.useMemo(() => { - return messages.filter((message) => !isFullySyntheticMessage(message.parts)); + const seenIds = new Set(); + return messages.filter((message) => { + const messageId = message.info?.id; + if (typeof messageId === 'string') { + if (seenIds.has(messageId)) { + return false; + } + seenIds.add(messageId); + } + return !isFullySyntheticMessage(message.parts); + }); }, [messages]); const { getContextForMessage } = useTurnGrouping(displayMessages); diff --git a/packages/ui/src/hooks/useEventStream.ts b/packages/ui/src/hooks/useEventStream.ts index 16bc4376..09077cd7 100644 --- a/packages/ui/src/hooks/useEventStream.ts +++ b/packages/ui/src/hooks/useEventStream.ts @@ -24,6 +24,7 @@ declare global { const ENABLE_EMPTY_RESPONSE_DETECTION = false; const TEXT_SHRINK_TOLERANCE = 50; +const RESYNC_DEBOUNCE_MS = 750; const textLengthCache = new WeakMap(); const computeTextLength = (parts: Part[] | undefined | null): number => { @@ -160,6 +161,32 @@ export const useEventStream = () => { [setEventStreamStatus] ); + const resyncMessages = React.useCallback( + (sessionId: string, reason: string) => { + if (!sessionId) { + return Promise.resolve(); + } + const now = Date.now(); + if (resyncInFlightRef.current) { + return resyncInFlightRef.current; + } + if (now - lastResyncAtRef.current < RESYNC_DEBOUNCE_MS) { + return Promise.resolve(); + } + const task = loadMessages(sessionId) + .catch((error) => { + console.warn(`[useEventStream] Failed to resync messages (${reason}):`, error); + }) + .finally(() => { + resyncInFlightRef.current = null; + lastResyncAtRef.current = Date.now(); + }); + resyncInFlightRef.current = task; + return task; + }, + [loadMessages] + ); + const bootstrapState = React.useCallback( async (reason: string) => { if (streamDebugEnabled()) { @@ -168,13 +195,13 @@ export const useEventStream = () => { try { await Promise.all([ loadSessions(), - currentSessionId ? loadMessages(currentSessionId) : Promise.resolve(), + currentSessionId ? resyncMessages(currentSessionId, reason) : Promise.resolve(), ]); } catch (error) { console.warn('[useEventStream] Bootstrap failed:', reason, error); } }, - [currentSessionId, loadMessages, loadSessions] + [currentSessionId, loadSessions, resyncMessages] ); const trackMessage = React.useCallback((messageId: string, event?: string, extraData?: Record) => { @@ -196,6 +223,8 @@ export const useEventStream = () => { const metadataRefreshTimestampsRef = React.useRef>(new Map()); const sessionRefreshTimeoutRef = React.useRef(null); const isCleaningUpRef = React.useRef(false); + const resyncInFlightRef = React.useRef | null>(null); + const lastResyncAtRef = React.useRef(0); const resolveVisibilityState = React.useCallback((): 'visible' | 'hidden' => { if (typeof document === 'undefined') return 'visible'; @@ -980,7 +1009,7 @@ export const useEventStream = () => { const sessionId = currentSessionIdRef.current; if (sessionId) { setTimeout(() => { - loadMessages(sessionId) + resyncMessages(sessionId, 'sse_reconnected') .then(() => requestSessionMetadataRefresh(sessionId)) .catch((error) => { console.warn('[useEventStream] Failed to resync messages after reconnect:', error); @@ -1029,7 +1058,7 @@ export const useEventStream = () => { stopStream, publishStatus, checkConnection, - loadMessages, + resyncMessages, requestSessionMetadataRefresh, requestSessionListRefresh, completeStreamingMessage, @@ -1126,10 +1155,11 @@ export const useEventStream = () => { if (pendingResumeRef.current || !unsubscribeRef.current) { console.info('[useEventStream] Visibility restored, triggering soft refresh...'); const sessionId = currentSessionIdRef.current; - if (sessionId) { - loadMessages(sessionId).catch(() => {}); - requestSessionMetadataRefresh(sessionId); - } + if (sessionId) { + resyncMessages(sessionId, 'visibility_restore').catch(() => {}); + requestSessionMetadataRefresh(sessionId); + } + void loadSessions(); void refreshSessionActivityStatus(); publishStatus('connecting', 'Resuming stream'); @@ -1153,7 +1183,7 @@ export const useEventStream = () => { const sessionId = currentSessionIdRef.current; if (sessionId) { requestSessionMetadataRefresh(sessionId); - loadMessages(sessionId) + resyncMessages(sessionId, 'window_focus') .then(() => console.info('[useEventStream] Messages refreshed on focus')) .catch((err) => console.warn('[useEventStream] Failed to refresh messages:', err)); } diff --git a/packages/ui/src/stores/messageStore.ts b/packages/ui/src/stores/messageStore.ts index 0c290977..dcce78aa 100644 --- a/packages/ui/src/stores/messageStore.ts +++ b/packages/ui/src/stores/messageStore.ts @@ -140,6 +140,58 @@ const mergePreferExistingParts = (existing: Part[] = [], incoming: Part[] = []): return merged; }; +const mergeDuplicateMessage = ( + existing: { info: any; parts: Part[] }, + incoming: { info: any; parts: Part[] } +): { info: any; parts: Part[] } => { + const existingParts = Array.isArray(existing.parts) ? existing.parts : []; + const incomingParts = Array.isArray(incoming.parts) ? incoming.parts : []; + const existingLen = computePartsTextLength(existingParts); + const incomingLen = computePartsTextLength(incomingParts); + const existingStop = hasStopReasonStop(existingParts); + const incomingStop = hasStopReasonStop(incomingParts); + + let parts = incomingParts; + if (existingStop && existingLen >= incomingLen) { + parts = mergePreferExistingParts(existingParts, incomingParts); + } else if (incomingStop && incomingLen >= existingLen) { + parts = mergePreferExistingParts(incomingParts, existingParts); + } else if (existingLen >= incomingLen) { + parts = existingParts; + } + + return { + ...incoming, + info: { + ...existing.info, + ...incoming.info, + }, + parts, + }; +}; + +const dedupeMessagesById = (messages: { info: any; parts: Part[] }[]) => { + const deduped: { info: any; parts: Part[] }[] = []; + const indexById = new Map(); + + for (const message of messages) { + const messageId = typeof message?.info?.id === "string" ? message.info.id : null; + if (!messageId) { + deduped.push(message); + continue; + } + const existingIndex = indexById.get(messageId); + if (existingIndex === undefined) { + indexById.set(messageId, deduped.length); + deduped.push(message); + continue; + } + deduped[existingIndex] = mergeDuplicateMessage(deduped[existingIndex], message); + } + + return deduped; +}; + const computeMaxTrimmedHeadId = (removed: Array<{ info: any }>, previous?: string): string | undefined => { let maxId = previous; let maxSortable = previous ? extractSortableId(previous) : null; @@ -372,7 +424,7 @@ export const useMessageStore = create()( }; }); - const mergedMessages = normalizedMessages; + const mergedMessages = dedupeMessagesById(normalizedMessages); const previousIds = new Set(previousMessages.map((msg) => msg.info.id)); const nextIds = new Set(mergedMessages.map((msg) => msg.info.id)); @@ -1849,7 +1901,7 @@ export const useMessageStore = create()( }; }); - const mergedMessages = normalizedMessages; + const mergedMessages = dedupeMessagesById(normalizedMessages); const previousIds = new Set(previousMessages.map((msg) => msg.info.id)); const nextIds = new Set(mergedMessages.map((msg) => msg.info.id)); @@ -2187,13 +2239,15 @@ export const useMessageStore = create()( set((state) => { const updatedMessages = [...newMessages, ...currentMessages]; + const dedupedMessages = dedupeMessagesById(updatedMessages); const newMessagesMap = new Map(state.messages); - newMessagesMap.set(sessionId, updatedMessages); + newMessagesMap.set(sessionId, dedupedMessages); + const addedCount = Math.max(0, dedupedMessages.length - currentMessages.length); const newMemoryState = new Map(state.sessionMemoryState); newMemoryState.set(sessionId, { ...memoryState, - viewportAnchor: memoryState.viewportAnchor + newMessages.length, + viewportAnchor: memoryState.viewportAnchor + addedCount, hasMoreAbove: indexInAll - loadCount > 0, totalAvailableMessages: allMessages.length, });