BuzzbeeSCD/openchamber
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: enhance message handling with deduplication and resync functionality

Browse Source
This commit is contained in:
Bohdan Triapitsyn
2025-12-20 02:34:01 +02:00
parent 53c103a222
commit f144eff46e
3 changed files with 108 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/ui/src/components/chat/MessageList.tsx
+11 -1
View File
@@ -39,7 +39,17 @@ const MessageList: React.FC<MessageListProps> = ({
}, [permissions, onMessageContentChange]);
const displayMessages = React.useMemo(() => {
return messages.filter((message) => !isFullySyntheticMessage(message.parts));
const seenIds = new Set<string>();
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);
packages/ui/src/hooks/useEventStream.ts
+39 -9
View File
@@ -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<Part[], number>();
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<string, unknown>) => {
@@ -196,6 +223,8 @@ export const useEventStream = () => {
const metadataRefreshTimestampsRef = React.useRef<Map<string, number>>(new Map());
const sessionRefreshTimeoutRef = React.useRef<NodeJS.Timeout | null>(null);
const isCleaningUpRef = React.useRef(false);
const resyncInFlightRef = React.useRef<Promise<void> | 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));
}
packages/ui/src/stores/messageStore.ts
+58 -4
View File
@@ -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<string, number>();
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<MessageStore>()(
};
});
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<MessageStore>()(
};
});
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<MessageStore>()(
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,
});