From bc28fa9ac871ffaf0232586ac82cb0f10821d552 Mon Sep 17 00:00:00 2001 From: Bohdan Triapitsyn Date: Wed, 28 Jan 2026 23:54:23 +0200 Subject: [PATCH] refactor(chat): replace message sync with soft resync scheduling Remove useMessageSync hook from App initialization Introduce scheduleSoftResync to respect cooldowns before resyncs Preserve hasMoreAbove handling and totalAvailableMessages for loading older messages --- packages/ui/src/App.tsx | 3 +- packages/ui/src/hooks/useEventStream.ts | 146 +++++++++----- packages/ui/src/hooks/useMessageSync.ts | 257 ------------------------ packages/ui/src/stores/messageStore.ts | 36 +++- 4 files changed, 125 insertions(+), 317 deletions(-) delete mode 100644 packages/ui/src/hooks/useMessageSync.ts diff --git a/packages/ui/src/App.tsx b/packages/ui/src/App.tsx index 9ba50994..9fd8cbce 100644 --- a/packages/ui/src/App.tsx +++ b/packages/ui/src/App.tsx @@ -9,7 +9,6 @@ import { ErrorBoundary } from '@/components/ui/ErrorBoundary'; import { useEventStream } from '@/hooks/useEventStream'; import { useKeyboardShortcuts } from '@/hooks/useKeyboardShortcuts'; import { useMenuActions } from '@/hooks/useMenuActions'; -import { useMessageSync } from '@/hooks/useMessageSync'; import { useSessionStatusBootstrap } from '@/hooks/useSessionStatusBootstrap'; import { useSessionAutoCleanup } from '@/hooks/useSessionAutoCleanup'; import { useRouter } from '@/hooks/useRouter'; @@ -176,7 +175,7 @@ function App({ apis }: AppProps) { useMenuActions(handleToggleMemoryDebug); - useMessageSync(); + useSessionStatusBootstrap(); useSessionAutoCleanup(); diff --git a/packages/ui/src/hooks/useEventStream.ts b/packages/ui/src/hooks/useEventStream.ts index 728ad87b..344afc57 100644 --- a/packages/ui/src/hooks/useEventStream.ts +++ b/packages/ui/src/hooks/useEventStream.ts @@ -314,9 +314,10 @@ export const useEventStream = () => { console.info('[useEventStream] Bootstrapping state:', reason); } try { + const activeLimit = getActiveSessionWindow(); await Promise.all([ loadSessions(), - currentSessionId ? resyncMessages(currentSessionId, reason, Infinity) : Promise.resolve(), + currentSessionId ? resyncMessages(currentSessionId, reason, activeLimit) : Promise.resolve(), ]); } catch (error) { console.warn('[useEventStream] Bootstrap failed:', reason, error); @@ -325,6 +326,27 @@ export const useEventStream = () => { [currentSessionId, loadSessions, resyncMessages] ); + const scheduleSoftResync = React.useCallback( + (sessionId: string, reason: string, limit = getActiveSessionWindow()): Promise => { + if (!sessionId) return Promise.resolve(); + + const memory = useSessionStore.getState().sessionMemoryState.get(sessionId); + const cooldownUntil = memory?.streamingCooldownUntil; + const now = Date.now(); + if (typeof cooldownUntil === 'number' && cooldownUntil > now) { + const delay = Math.min(3000, Math.max(0, cooldownUntil - now)); + return new Promise((resolve) => { + setTimeout(() => { + resyncMessages(sessionId, reason, limit).finally(resolve); + }, delay); + }); + } + + return resyncMessages(sessionId, reason, limit); + }, + [resyncMessages] + ); + const trackMessage = React.useCallback((messageId: string, event?: string, extraData?: Record) => { if (streamDebugEnabled()) { console.debug(`[MessageTracker] ${messageId}: ${event}`, extraData); @@ -1635,21 +1657,21 @@ export const useEventStream = () => { // already-running sessions (e.g., started via CLI before UI opened) void refreshSessionActivityStatus(); - if (shouldRefresh) { - void bootstrapState('sse_reconnected'); - } else { - const sessionId = currentSessionIdRef.current; - if (sessionId) { - setTimeout(() => { - resyncMessages(sessionId, 'sse_reconnected', Infinity) + if (shouldRefresh) { + void bootstrapState('sse_reconnected'); + } else { + const sessionId = currentSessionIdRef.current; + if (sessionId) { + setTimeout(() => { + scheduleSoftResync(sessionId, 'sse_reconnected', getActiveSessionWindow()) .then(() => requestSessionMetadataRefresh(sessionId)) - .catch((error) => { + .catch((error: unknown) => { console.warn('[useEventStream] Failed to resync messages after reconnect:', error); }); - }, 0); + }, 0); + } } - } - }; + }; if (streamDebugEnabled()) { console.info('[useEventStream] Connecting to event source (SDK SSE only):', { @@ -1712,7 +1734,7 @@ export const useEventStream = () => { stopStream, publishStatus, checkConnection, - resyncMessages, + scheduleSoftResync, requestSessionMetadataRefresh, handleEvent, effectiveDirectory, @@ -1798,8 +1820,8 @@ export const useEventStream = () => { }, 5000); }; - const handleVisibilityChange = () => { - visibilityStateRef.current = resolveVisibilityState(); + const handleVisibilityChange = () => { + visibilityStateRef.current = resolveVisibilityState(); if (visibilityStateRef.current === 'visible') { clearPauseTimeout(); @@ -1808,7 +1830,7 @@ export const useEventStream = () => { console.info('[useEventStream] Visibility restored, triggering soft refresh...'); const sessionId = currentSessionIdRef.current; if (sessionId) { - resyncMessages(sessionId, 'visibility_restore', getActiveSessionWindow()).catch(() => {}); + scheduleSoftResync(sessionId, 'visibility_restore', getActiveSessionWindow()); requestSessionMetadataRefresh(sessionId); } @@ -1822,8 +1844,8 @@ export const useEventStream = () => { } }; - const handleWindowFocus = () => { - visibilityStateRef.current = resolveVisibilityState(); + const handleWindowFocus = () => { + visibilityStateRef.current = resolveVisibilityState(); if (visibilityStateRef.current === 'visible') { clearPauseTimeout(); @@ -1832,13 +1854,11 @@ export const useEventStream = () => { if (pendingResumeRef.current || !unsubscribeRef.current) { console.info('[useEventStream] Window focused after pause, triggering soft refresh...'); const sessionId = currentSessionIdRef.current; - if (sessionId) { - requestSessionMetadataRefresh(sessionId); - resyncMessages(sessionId, 'window_focus', getActiveSessionWindow()) - .then(() => console.info('[useEventStream] Messages refreshed on focus')) - .catch((err) => console.warn('[useEventStream] Failed to refresh messages:', err)); - } - void refreshSessionActivityStatus(); + if (sessionId) { + requestSessionMetadataRefresh(sessionId); + scheduleSoftResync(sessionId, 'window_focus', getActiveSessionWindow()); + } + void refreshSessionActivityStatus(); publishStatus('connecting', 'Resuming stream'); startStream({ resetAttempts: true }); @@ -1846,31 +1866,54 @@ export const useEventStream = () => { } }; - const handleOnline = () => { - onlineStatusRef.current = true; - maybeBootstrapIfStale('network_restored'); - if (pendingResumeRef.current || !unsubscribeRef.current) { - publishStatus('connecting', 'Network restored'); - startStream({ resetAttempts: true }); + const handleOnline = () => { + onlineStatusRef.current = true; + maybeBootstrapIfStale('network_restored'); + if (pendingResumeRef.current || !unsubscribeRef.current) { + publishStatus('connecting', 'Network restored'); + startStream({ resetAttempts: true }); + } + }; + + const handleOffline = () => { + onlineStatusRef.current = false; + pendingResumeRef.current = true; + publishStatus('offline', 'Waiting for network'); + stopStream(); + }; + + const handlePageHide = () => { + pendingResumeRef.current = true; + stopStream(); + publishStatus('paused', 'Paused while hidden'); + }; + + const handlePageShow = (event: PageTransitionEvent) => { + // If page was restored from bfcache, SSE is definitely gone. + pendingResumeRef.current = pendingResumeRef.current || Boolean(event.persisted); + visibilityStateRef.current = resolveVisibilityState(); + if (visibilityStateRef.current === 'visible') { + const sessionId = currentSessionIdRef.current; + if (sessionId) { + void scheduleSoftResync(sessionId, 'page_show', getActiveSessionWindow()); + requestSessionMetadataRefresh(sessionId); + } + void refreshSessionActivityStatus(); + startStream({ resetAttempts: true }); + } + }; + + if (typeof document !== 'undefined') { + document.addEventListener('visibilitychange', handleVisibilityChange); + } + + if (typeof window !== 'undefined') { + window.addEventListener('online', handleOnline); + window.addEventListener('offline', handleOffline); + window.addEventListener('focus', handleWindowFocus); + window.addEventListener('pagehide', handlePageHide); + window.addEventListener('pageshow', handlePageShow as EventListener); } - }; - - const handleOffline = () => { - onlineStatusRef.current = false; - pendingResumeRef.current = true; - publishStatus('offline', 'Waiting for network'); - stopStream(); - }; - - if (typeof document !== 'undefined') { - document.addEventListener('visibilitychange', handleVisibilityChange); - } - - if (typeof window !== 'undefined') { - window.addEventListener('online', handleOnline); - window.addEventListener('offline', handleOffline); - window.addEventListener('focus', handleWindowFocus); - } const startTimer = setTimeout(() => { startStream({ resetAttempts: true }); @@ -1922,6 +1965,8 @@ export const useEventStream = () => { window.removeEventListener('online', handleOnline); window.removeEventListener('offline', handleOffline); window.removeEventListener('focus', handleWindowFocus); + window.removeEventListener('pagehide', handlePageHide); + window.removeEventListener('pageshow', handlePageShow as EventListener); } clearPauseTimeout(); @@ -1967,6 +2012,7 @@ export const useEventStream = () => { shouldHoldConnection, loadSessions, maybeBootstrapIfStale, - resyncMessages + resyncMessages, + scheduleSoftResync ]); }; diff --git a/packages/ui/src/hooks/useMessageSync.ts b/packages/ui/src/hooks/useMessageSync.ts deleted file mode 100644 index 75bfe6aa..00000000 --- a/packages/ui/src/hooks/useMessageSync.ts +++ /dev/null @@ -1,257 +0,0 @@ -import React from 'react'; -import type { AssistantMessage, Message, Part } from '@opencode-ai/sdk/v2'; -import { useSessionStore } from '@/stores/useSessionStore'; -import { getMemoryLimits } from '@/stores/types/sessionTypes'; -import { opencodeClient } from '@/lib/opencode/client'; -import { readSessionCursor } from '@/lib/messageCursorPersistence'; -import { extractTextFromPart } from '@/stores/utils/messageUtils'; - -type SessionMessageRecord = { info: Message; parts: Part[] }; - -const isAssistantMessage = (message: Message): message is AssistantMessage => message.role === 'assistant'; - -const getCompletionTimestamp = (record: { info: Message }): number | undefined => { - const message = record.info; - if (!isAssistantMessage(message)) { - return undefined; - } - const completed = message.time?.completed; - return typeof completed === 'number' ? completed : undefined; -}; - -const isUserMessageInfo = (info?: Message): boolean => { - if (!info) return false; - if (info.role === 'user') return true; - const infoExt = info as { clientRole?: string; userMessageMarker?: boolean }; - if (infoExt.clientRole === 'user') return true; - return Boolean(infoExt.userMessageMarker); -}; - -const normalizeMessageText = (message?: SessionMessageRecord): string => { - const parts = Array.isArray(message?.parts) ? message.parts : []; - const raw = parts - .map((part) => extractTextFromPart(part)) - .join(' ') - .replace(/\s+/g, ' ') - .trim(); - return raw; -}; - -const findServerIndexForLocalUserMessage = ( - localMessage: SessionMessageRecord | undefined, - serverMessages: SessionMessageRecord[] -): number => { - if (!localMessage || !isUserMessageInfo(localMessage.info)) { - return -1; - } - - const localText = normalizeMessageText(localMessage); - const localCreated = - typeof localMessage.info?.time?.created === 'number' ? localMessage.info.time.created : undefined; - - let bestIndex = -1; - let bestScore = Number.POSITIVE_INFINITY; - - serverMessages.forEach((candidate, index) => { - if (!isUserMessageInfo(candidate.info)) { - return; - } - - const candidateText = normalizeMessageText(candidate); - const textMatches = Boolean(localText && candidateText && candidateText === localText); - const candidateCreated = - typeof candidate.info?.time?.created === 'number' ? candidate.info.time.created : undefined; - const timeDelta = - typeof localCreated === 'number' && typeof candidateCreated === 'number' - ? Math.abs(candidateCreated - localCreated) - : null; - - if (textMatches) { - const score = typeof timeDelta === 'number' ? timeDelta : 0; - if (score < bestScore) { - bestIndex = index; - bestScore = score; - } - return; - } - - if (!localText && !candidateText && typeof timeDelta === 'number' && timeDelta < 1500 && timeDelta < bestScore) { - bestIndex = index; - bestScore = timeDelta; - } - }); - - if (bestIndex !== -1) { - return bestIndex; - } - - if (typeof localCreated === 'number') { - return serverMessages.findIndex((candidate) => { - if (!isUserMessageInfo(candidate.info)) return false; - const candidateCreated = - typeof candidate.info?.time?.created === 'number' ? candidate.info.time.created : undefined; - if (typeof candidateCreated !== 'number') return false; - return Math.abs(candidateCreated - localCreated) < 1000; - }); - } - - return -1; -}; - -export const useMessageSync = () => { - const { - currentSessionId, - messages, - streamingMessageIds - } = useSessionStore(); - - const streamingMessageId = React.useMemo(() => { - if (!currentSessionId) return null; - return streamingMessageIds.get(currentSessionId) ?? null; - }, [currentSessionId, streamingMessageIds]); - - const syncTimeoutRef = React.useRef(undefined); - const lastSyncRef = React.useRef(0); - - const syncMessages = React.useCallback(async () => { - if (!currentSessionId) return; - if (streamingMessageId) return; - - const now = Date.now(); - if (now - lastSyncRef.current < 2000) return; - lastSyncRef.current = now; - - try { - - const currentMessages = (messages.get(currentSessionId) || []) as SessionMessageRecord[]; - - const memoryState = useSessionStore.getState().sessionMemoryState.get(currentSessionId); - const memLimits = getMemoryLimits(); - const targetLimit = memoryState?.isStreaming ? memLimits.VIEWPORT_MESSAGES : memLimits.HISTORICAL_MESSAGES; - const fetchLimit = targetLimit + memLimits.FETCH_BUFFER; - const latestMessages = (await opencodeClient.getSessionMessages(currentSessionId, fetchLimit)) as SessionMessageRecord[]; - const cursorRecord = await readSessionCursor(currentSessionId); - - if (!latestMessages) return; - - const lastLocalMessage = currentMessages[currentMessages.length - 1]; - - if (lastLocalMessage) { - const directIndex = latestMessages.findIndex((m) => m.info.id === lastLocalMessage.info.id); - const fuzzyIndex = - directIndex === -1 - ? findServerIndexForLocalUserMessage(lastLocalMessage, latestMessages) - : directIndex; - const lastLocalIndex = fuzzyIndex; - - if (lastLocalIndex !== -1) { - - if (lastLocalIndex < latestMessages.length - 1) { - const newMessages = latestMessages.slice(lastLocalIndex + 1); - console.log(`[SYNC] Found ${newMessages.length} new messages to append`); - - const updatedMessages = [...currentMessages, ...newMessages]; - const { syncMessages } = useSessionStore.getState(); - syncMessages(currentSessionId, updatedMessages); - } else { - - const serverLastMessage = latestMessages[lastLocalIndex]; - const localLastMessage = currentMessages[currentMessages.length - 1]; - - const serverCompleted = getCompletionTimestamp(serverLastMessage); - const localCompleted = getCompletionTimestamp(localLastMessage); - - if (serverCompleted && !localCompleted) { - console.log('[SYNC] Last message completed on server'); - - const updatedMessages = [...currentMessages.slice(0, -1), serverLastMessage]; - const { syncMessages } = useSessionStore.getState(); - syncMessages(currentSessionId, updatedMessages); - } - } - } else { - - if (isUserMessageInfo(lastLocalMessage.info)) { - const messagesToLoad = latestMessages.slice(-targetLimit); - console.log('[SYNC] Local user message missing by ID; merging latest messages for deduplication'); - const { syncMessages } = useSessionStore.getState(); - syncMessages(currentSessionId, messagesToLoad); - } else { - console.log('[SYNC] Local messages not found on server - skipping sync'); - } - } - } else if (cursorRecord) { - const cursorIndex = latestMessages.findIndex(m => m.info.id === cursorRecord.messageId); - - if (cursorIndex !== -1) { - if (cursorIndex < latestMessages.length - 1) { - const newMessages = latestMessages.slice(cursorIndex + 1); - const limited = newMessages.slice(-targetLimit); - if (limited.length > 0) { - console.log(`[SYNC] Restoring ${limited.length} messages after cursor`); - const { syncMessages } = useSessionStore.getState(); - syncMessages(currentSessionId, limited); - } - } - } else if (latestMessages.length > 0) { - console.log('[SYNC] Cursor not found on server response, loading recent messages'); - const messagesToLoad = latestMessages.slice(-targetLimit); - const { syncMessages } = useSessionStore.getState(); - syncMessages(currentSessionId, messagesToLoad); - } - } else if (latestMessages.length > 0) { - - const messagesToLoad = latestMessages.slice(-targetLimit); - console.log(`[SYNC] Loading last ${messagesToLoad.length} messages`); - const { syncMessages } = useSessionStore.getState(); - syncMessages(currentSessionId, messagesToLoad); - } - } catch (error) { - - console.debug('Background sync failed:', error); - } - }, [currentSessionId, messages, streamingMessageId]); - - React.useEffect(() => { - const handleFocus = () => { - console.log('[FOCUS] Window focused - checking for updates'); - syncMessages(); - }; - - window.addEventListener('focus', handleFocus); - return () => window.removeEventListener('focus', handleFocus); - }, [syncMessages]); - - React.useEffect(() => { - if (!currentSessionId || streamingMessageId) return; - - const scheduleSync = () => { - - if (document.visibilityState === 'visible') { - syncMessages(); - } - - syncTimeoutRef.current = setTimeout(scheduleSync, 30000); - }; - - syncTimeoutRef.current = setTimeout(scheduleSync, 30000); - - return () => { - if (syncTimeoutRef.current) { - clearTimeout(syncTimeoutRef.current); - } - }; - }, [currentSessionId, streamingMessageId, syncMessages]); - - React.useEffect(() => { - const handleVisibilityChange = () => { - if (document.visibilityState === 'visible') { - console.log('[FOCUS] Tab became visible - checking for updates'); - syncMessages(); - } - }; - - document.addEventListener('visibilitychange', handleVisibilityChange); - return () => document.removeEventListener('visibilitychange', handleVisibilityChange); - }, [syncMessages]); -}; diff --git a/packages/ui/src/stores/messageStore.ts b/packages/ui/src/stores/messageStore.ts index fe049032..38badd2a 100644 --- a/packages/ui/src/stores/messageStore.ts +++ b/packages/ui/src/stores/messageStore.ts @@ -398,6 +398,10 @@ export const useMessageStore = create()( const revertMessageId = getSessionRevertMessageId(sessionId); const messagesWithoutReverted = filterRevertedMessages(allMessages, revertMessageId); + // If we fetched more than we keep (usually via buffer), there are older messages above. + // This intentionally ignores watermark filtering so "Load older" can remain available. + const hasMoreAbove = messagesWithoutReverted.length > targetLimit; + const watermark = get().sessionMemoryState.get(sessionId)?.trimmedHeadMaxId; const afterWatermark = watermark @@ -485,8 +489,8 @@ export const useMessageStore = create()( isStreaming: false, lastAccessedAt: Date.now(), backgroundMessageCount: 0, - totalAvailableMessages: allMessages.length, - hasMoreAbove: allMessages.length > messagesToKeep.length, + totalAvailableMessages: previousMemoryState?.totalAvailableMessages, + hasMoreAbove, trimmedHeadMaxId: previousMemoryState?.trimmedHeadMaxId, streamingCooldownUntil: undefined, }); @@ -2257,6 +2261,9 @@ export const useMessageStore = create()( const updatedMemoryState = { ...memoryState, viewportAnchor: anchor - start, + // If we trimmed older messages out of the in-memory window, + // keep "Load older" available even if the last fetch didn't exceed its limit. + hasMoreAbove: Boolean(memoryState.hasMoreAbove) || start > 0, trimmedHeadMaxId: computeMaxTrimmedHeadId(removedOlder, memoryState.trimmedHeadMaxId), }; newMemoryState.set(sessionId, updatedMemoryState); @@ -2402,11 +2409,18 @@ export const useMessageStore = create()( return; } - if (memoryState.totalAvailableMessages && currentMessages.length >= memoryState.totalAvailableMessages) { - return; - } + const memLimits = getMemoryLimits(); + // OpenCode may default to "last N" when limit is omitted. + // For "Load older" we progressively increase the tail window. + const desiredLimit = Math.max( + currentMessages.length + memLimits.VIEWPORT_MESSAGES + memLimits.FETCH_BUFFER, + memLimits.HISTORICAL_MESSAGES + memLimits.FETCH_BUFFER, + ); - const allMessages = await executeWithSessionDirectory(sessionId, () => opencodeClient.getSessionMessages(sessionId)); + const allMessages = await executeWithSessionDirectory( + sessionId, + () => opencodeClient.getSessionMessages(sessionId, desiredLimit) + ); if (direction === "up" && currentMessages.length > 0) { const dedupedMessages = dedupeMessagesById(allMessages); @@ -2430,7 +2444,10 @@ export const useMessageStore = create()( ...memoryState, viewportAnchor: memoryState.viewportAnchor + addedCount, hasMoreAbove: indexInAll - loadCount > 0, - totalAvailableMessages: dedupedMessages.length, + totalAvailableMessages: Math.max( + memoryState.totalAvailableMessages ?? 0, + dedupedMessages.length + ), }); return { @@ -2444,7 +2461,10 @@ export const useMessageStore = create()( newMemoryState.set(sessionId, { ...memoryState, hasMoreAbove: false, - totalAvailableMessages: dedupedMessages.length, + totalAvailableMessages: Math.max( + memoryState.totalAvailableMessages ?? 0, + dedupedMessages.length + ), }); return { sessionMemoryState: newMemoryState }; });