refactor: replace React Query message cache with windowed message store

Introduces a new message-window-store module to manage message state with automatic windowing of visible and pending messages. This replaces manual React Query cache operations with a centralized, observable state system. close #39 Key changes: - New MessageWindowState tracks visible/pending messages with size limits - Automatic trimming of message windows (400 visible, 200 pending messages) - Pending message buffering when user scrolls away from bottom - Centralized status updates for optimistic messages - Thread component simplified with forwarded scroll and pending callbacks - Removes message count tracking from components This improves performance for chats with many messages and provides a cleaner separation of concerns between UI and state management.
2026-08-05 06:24:37 +00:00 · 2026-01-07 16:59:12 +08:00
parent 70242041d5
commit 0f29ee182c
10 changed files with 556 additions and 174 deletions
@@ -11,6 +11,7 @@ import { useSyncingState } from '@/hooks/useSyncingState'
 import { usePushNotifications } from '@/hooks/usePushNotifications'
 import { queryKeys } from '@/lib/query-keys'
 import { AppContextProvider } from '@/lib/app-context'
+import { fetchLatestMessages } from '@/lib/message-window-store'
 import { useAppGoBack } from '@/hooks/useAppGoBack'
 import { LoginPrompt } from '@/components/LoginPrompt'
 import { InstallPrompt } from '@/components/InstallPrompt'
@@ -152,11 +153,13 @@ export function App() {
        const invalidations = [
            queryClient.invalidateQueries({ queryKey: queryKeys.sessions }),
            ...(selectedSessionId ? [
-                queryClient.invalidateQueries({ queryKey: queryKeys.session(selectedSessionId) }),
-                queryClient.invalidateQueries({ queryKey: queryKeys.messages(selectedSessionId) })
+                queryClient.invalidateQueries({ queryKey: queryKeys.session(selectedSessionId) })
            ] : [])
        ]
-        Promise.all(invalidations)
+        const refreshMessages = (selectedSessionId && api)
+            ? fetchLatestMessages(api, selectedSessionId)
+            : Promise.resolve()
+        Promise.all([...invalidations, refreshMessages])
            .catch((error) => {
                console.error('Failed to invalidate queries on SSE connect:', error)
            })
@@ -166,7 +169,7 @@ export function App() {
                    endSync()
                }
            })
-    }, [queryClient, selectedSessionId, startSync, endSync])
+    }, [api, queryClient, selectedSessionId, startSync, endSync])

    const handleSseEvent = useCallback(() => {}, [])

@@ -59,14 +59,18 @@ export function HappyThread(props: {
    disabled: boolean
    onRefresh: () => void
    onRetryMessage?: (localId: string) => void
+    onFlushPending: () => void
+    onAtBottomChange: (atBottom: boolean) => void
    isLoadingMessages: boolean
    messagesWarning: string | null
    hasMoreMessages: boolean
    isLoadingMoreMessages: boolean
    onLoadMore: () => Promise<unknown>
+    pendingCount: number
    rawMessagesCount: number
    normalizedMessagesCount: number
-    renderedMessagesCount: number
+    messagesVersion: number
+    forceScrollToken: number
 }) {
    const viewportRef = useRef<HTMLDivElement | null>(null)
    const topSentinelRef = useRef<HTMLDivElement | null>(null)
@@ -75,22 +79,25 @@ export function HappyThread(props: {
    const prevLoadingMoreRef = useRef(false)
    const loadStartedRef = useRef(false)
    const isLoadingMoreRef = useRef(props.isLoadingMoreMessages)
+    const atBottomRef = useRef(true)
+    const onAtBottomChangeRef = useRef(props.onAtBottomChange)
+    const onFlushPendingRef = useRef(props.onFlushPending)
+    const forceScrollTokenRef = useRef(props.forceScrollToken)

    // Smart scroll state: autoScroll enabled when user is near bottom
    const [autoScrollEnabled, setAutoScrollEnabled] = useState(true)
-    const [newMessageCount, setNewMessageCount] = useState(0)
-    const prevRenderedCountRef = useRef(props.renderedMessagesCount)
    const autoScrollEnabledRef = useRef(autoScrollEnabled)
-    const newMessageCountRef = useRef(newMessageCount)
-    const hasBootstrappedRef = useRef(false)

    // Keep refs in sync with state
    useEffect(() => {
        autoScrollEnabledRef.current = autoScrollEnabled
    }, [autoScrollEnabled])
    useEffect(() => {
-        newMessageCountRef.current = newMessageCount
-    }, [newMessageCount])
+        onAtBottomChangeRef.current = props.onAtBottomChange
+    }, [props.onAtBottomChange])
+    useEffect(() => {
+        onFlushPendingRef.current = props.onFlushPending
+    }, [props.onFlushPending])

    // Track scroll position to toggle autoScroll (stable listener using refs)
    useEffect(() => {
@@ -105,9 +112,16 @@ export function HappyThread(props: {

            if (isNearBottom) {
                if (!autoScrollEnabledRef.current) setAutoScrollEnabled(true)
-                if (newMessageCountRef.current > 0) setNewMessageCount(0)
-            } else {
-                if (autoScrollEnabledRef.current) setAutoScrollEnabled(false)
+            } else if (autoScrollEnabledRef.current) {
+                setAutoScrollEnabled(false)
+            }
+
+            if (isNearBottom !== atBottomRef.current) {
+                atBottomRef.current = isNearBottom
+                onAtBottomChangeRef.current(isNearBottom)
+                if (isNearBottom) {
+                    onFlushPendingRef.current()
+                }
            }
        }

@@ -115,43 +129,6 @@ export function HappyThread(props: {
        return () => viewport.removeEventListener('scroll', handleScroll)
    }, []) // Stable: no dependencies, reads from refs

-    // Track new messages when autoScroll is disabled
-    const wasLoadingMoreRef = useRef(props.isLoadingMoreMessages)
-    useEffect(() => {
-        const prevCount = prevRenderedCountRef.current
-        const currentCount = props.renderedMessagesCount
-        const wasLoadingMore = wasLoadingMoreRef.current
-        wasLoadingMoreRef.current = props.isLoadingMoreMessages
-
-        if (props.isLoadingMessages) {
-            prevRenderedCountRef.current = currentCount
-            return
-        }
-
-        if (!hasBootstrappedRef.current) {
-            hasBootstrappedRef.current = true
-            prevRenderedCountRef.current = currentCount
-            return
-        }
-
-        prevRenderedCountRef.current = currentCount
-
-        // Skip during loading states
-        if (props.isLoadingMoreMessages) {
-            return
-        }
-
-        // Skip if load-more just finished (older messages, not new ones)
-        if (wasLoadingMore) {
-            return
-        }
-
-        const newCount = currentCount - prevCount
-        if (newCount > 0 && !autoScrollEnabled) {
-            setNewMessageCount((prev) => prev + newCount)
-        }
-    }, [props.renderedMessagesCount, props.isLoadingMoreMessages, props.isLoadingMessages, autoScrollEnabled])
-
    // Scroll to bottom handler for the indicator button
    const scrollToBottom = useCallback(() => {
        const viewport = viewportRef.current
@@ -159,17 +136,29 @@ export function HappyThread(props: {
            viewport.scrollTo({ top: viewport.scrollHeight, behavior: 'smooth' })
        }
        setAutoScrollEnabled(true)
-        setNewMessageCount(0)
+        if (!atBottomRef.current) {
+            atBottomRef.current = true
+            onAtBottomChangeRef.current(true)
+        }
+        onFlushPendingRef.current()
    }, [])

    // Reset state when session changes
    useEffect(() => {
        setAutoScrollEnabled(true)
-        setNewMessageCount(0)
-        prevRenderedCountRef.current = 0
-        hasBootstrappedRef.current = false
+        atBottomRef.current = true
+        onAtBottomChangeRef.current(true)
+        forceScrollTokenRef.current = props.forceScrollToken
    }, [props.sessionId])

+    useEffect(() => {
+        if (forceScrollTokenRef.current === props.forceScrollToken) {
+            return
+        }
+        forceScrollTokenRef.current = props.forceScrollToken
+        scrollToBottom()
+    }, [props.forceScrollToken, scrollToBottom])
+
    const handleLoadMore = useCallback(() => {
        if (props.isLoadingMessages || !props.hasMoreMessages || props.isLoadingMoreMessages || loadLockRef.current) {
            return
@@ -242,7 +231,7 @@ export function HappyThread(props: {
        viewport.scrollTop = pending.scrollTop + delta
        pendingScrollRef.current = null
        loadLockRef.current = false
-    }, [props.rawMessagesCount])
+    }, [props.messagesVersion])

    useEffect(() => {
        isLoadingMoreRef.current = props.isLoadingMoreMessages
@@ -320,7 +309,7 @@ export function HappyThread(props: {
                        </div>
                    </div>
                </ThreadPrimitive.Viewport>
-                <NewMessagesIndicator count={newMessageCount} onClick={scrollToBottom} />
+                <NewMessagesIndicator count={props.pendingCount} onClick={scrollToBottom} />
            </ThreadPrimitive.Root>
        </HappyChatProvider>
    )
@@ -1,4 +1,4 @@
-import { useCallback, useEffect, useMemo, useRef } from 'react'
+import { useCallback, useEffect, useMemo, useRef, useState } from 'react'
 import { useNavigate } from '@tanstack/react-router'
 import { AssistantRuntimeProvider } from '@assistant-ui/react'
 import type { ApiClient } from '@/api/client'
@@ -24,10 +24,14 @@ export function SessionChat(props: {
    isLoadingMessages: boolean
    isLoadingMoreMessages: boolean
    isSending: boolean
+    pendingCount: number
+    messagesVersion: number
    onBack: () => void
    onRefresh: () => void
    onLoadMore: () => Promise<unknown>
    onSend: (text: string) => void
+    onFlushPending: () => void
+    onAtBottomChange: (atBottom: boolean) => void
    onRetryMessage?: (localId: string) => void
    autocompleteSuggestions?: (query: string) => Promise<Suggestion[]>
 }) {
@@ -36,6 +40,7 @@ export function SessionChat(props: {
    const controlsDisabled = !props.session.active
    const normalizedCacheRef = useRef<Map<string, { source: DecryptedMessage; normalized: NormalizedMessage | null }>>(new Map())
    const blocksByIdRef = useRef<Map<string, ChatBlock>>(new Map())
+    const [forceScrollToken, setForceScrollToken] = useState(0)
    const agentFlavor = props.session.metadata?.flavor ?? null
    const { abortSession, switchSession, setPermissionMode, setModelMode } = useSessionActions(
        props.api,
@@ -134,11 +139,16 @@ export function SessionChat(props: {
        })
    }, [navigate, props.session.id])

+    const handleSend = useCallback((text: string) => {
+        props.onSend(text)
+        setForceScrollToken((token) => token + 1)
+    }, [props.onSend])
+
    const runtime = useHappyRuntime({
        session: props.session,
        blocks: reconciled.blocks,
        isSending: props.isSending,
-        onSendMessage: props.onSend,
+        onSendMessage: handleSend,
        onAbort: handleAbort
    })

@@ -170,14 +180,18 @@ export function SessionChat(props: {
                        disabled={controlsDisabled}
                        onRefresh={props.onRefresh}
                        onRetryMessage={props.onRetryMessage}
+                        onFlushPending={props.onFlushPending}
+                        onAtBottomChange={props.onAtBottomChange}
                        isLoadingMessages={props.isLoadingMessages}
                        messagesWarning={props.messagesWarning}
                        hasMoreMessages={props.hasMoreMessages}
                        isLoadingMoreMessages={props.isLoadingMoreMessages}
                        onLoadMore={props.onLoadMore}
+                        pendingCount={props.pendingCount}
                        rawMessagesCount={props.messages.length}
                        normalizedMessagesCount={normalizedMessages.length}
-                        renderedMessagesCount={reconciled.blocks.length}
+                        messagesVersion={props.messagesVersion}
+                        forceScrollToken={forceScrollToken}
                    />

                    <HappyComposer
@@ -1,8 +1,12 @@
-import { useMutation, useQueryClient, type InfiniteData } from '@tanstack/react-query'
+import { useMutation } from '@tanstack/react-query'
 import type { ApiClient } from '@/api/client'
-import type { DecryptedMessage, MessagesResponse } from '@/types/api'
-import { makeClientSideId, upsertMessagesInCache } from '@/lib/messages'
-import { queryKeys } from '@/lib/query-keys'
+import type { DecryptedMessage } from '@/types/api'
+import { makeClientSideId } from '@/lib/messages'
+import {
+    appendOptimisticMessage,
+    getMessageWindowState,
+    updateMessageStatus,
+} from '@/lib/message-window-store'
 import { usePlatform } from '@/hooks/usePlatform'

 type SendMessageInput = {
@@ -12,36 +16,16 @@ type SendMessageInput = {
    createdAt: number
 }

-function updateMessageStatus(
-    data: InfiniteData<MessagesResponse> | undefined,
-    localId: string,
-    status: DecryptedMessage['status'],
-): InfiniteData<MessagesResponse> | undefined {
-    if (!data) return data
-
-    const pages = data.pages.map((page) => ({
-        ...page,
-        messages: page.messages.map((message) =>
-            message.localId === localId
-                ? { ...message, status }
-                : message
-        ),
-    }))
-
-    return {
-        ...data,
-        pages,
-    }
-}
-
 function findMessageByLocalId(
-    data: InfiniteData<MessagesResponse> | undefined,
+    sessionId: string,
    localId: string,
 ): DecryptedMessage | null {
-    if (!data) return null
-    for (const page of data.pages) {
-        const match = page.messages.find((message) => message.localId === localId)
-        if (match) return match
+    const state = getMessageWindowState(sessionId)
+    for (const message of state.messages) {
+        if (message.localId === localId) return message
+    }
+    for (const message of state.pending) {
+        if (message.localId === localId) return message
    }
    return null
 }
@@ -51,7 +35,6 @@ export function useSendMessage(api: ApiClient | null, sessionId: string | null):
    retryMessage: (localId: string) => void
    isSending: boolean
 } {
-    const queryClient = useQueryClient()
    const { haptic } = usePlatform()

    const mutation = useMutation({
@@ -72,23 +55,14 @@ export function useSendMessage(api: ApiClient | null, sessionId: string | null):
                originalText: input.text,
            }

-            queryClient.setQueryData<InfiniteData<MessagesResponse>>(
-                queryKeys.messages(input.sessionId),
-                (data) => upsertMessagesInCache(data, [optimisticMessage]),
-            )
+            appendOptimisticMessage(input.sessionId, optimisticMessage)
        },
        onSuccess: (_, input) => {
-            queryClient.setQueryData<InfiniteData<MessagesResponse>>(
-                queryKeys.messages(input.sessionId),
-                (data) => updateMessageStatus(data, input.localId, 'sent'),
-            )
+            updateMessageStatus(input.sessionId, input.localId, 'sent')
            haptic.notification('success')
        },
        onError: (_, input) => {
-            queryClient.setQueryData<InfiniteData<MessagesResponse>>(
-                queryKeys.messages(input.sessionId),
-                (data) => updateMessageStatus(data, input.localId, 'failed'),
-            )
+            updateMessageStatus(input.sessionId, input.localId, 'failed')
            haptic.notification('error')
        },
    })
@@ -109,14 +83,10 @@ export function useSendMessage(api: ApiClient | null, sessionId: string | null):
        if (!api || !sessionId) return
        if (mutation.isPending) return

-        const data = queryClient.getQueryData<InfiniteData<MessagesResponse>>(queryKeys.messages(sessionId))
-        const message = findMessageByLocalId(data, localId)
+        const message = findMessageByLocalId(sessionId, localId)
        if (!message?.originalText) return

-        queryClient.setQueryData<InfiniteData<MessagesResponse>>(
-            queryKeys.messages(sessionId),
-            (current) => updateMessageStatus(current, localId, 'sending'),
-        )
+        updateMessageStatus(sessionId, localId, 'sending')

        mutation.mutate({
            sessionId,
@@ -3,6 +3,7 @@ import { isPermissionModeAllowedForFlavor } from '@hapi/protocol'
 import type { ApiClient } from '@/api/client'
 import type { ModelMode, PermissionMode } from '@/types/api'
 import { queryKeys } from '@/lib/query-keys'
+import { clearMessageWindow } from '@/lib/message-window-store'

 export function useSessionActions(
    api: ApiClient | null,
@@ -100,7 +101,7 @@ export function useSessionActions(
        onSuccess: async () => {
            if (!sessionId) return
            queryClient.removeQueries({ queryKey: queryKeys.session(sessionId) })
-            queryClient.removeQueries({ queryKey: queryKeys.messages(sessionId) })
+            clearMessageWindow(sessionId)
            await queryClient.invalidateQueries({ queryKey: queryKeys.sessions })
        },
    })
@@ -1,9 +1,31 @@
-import { useInfiniteQuery } from '@tanstack/react-query'
-import { useMemo } from 'react'
+import { useCallback, useEffect, useSyncExternalStore } from 'react'
 import type { ApiClient } from '@/api/client'
-import type { DecryptedMessage, MessagesResponse } from '@/types/api'
-import { mergeMessages } from '@/lib/messages'
-import { queryKeys } from '@/lib/query-keys'
+import type { DecryptedMessage } from '@/types/api'
+import {
+    clearMessageWindow,
+    fetchLatestMessages,
+    fetchOlderMessages,
+    flushPendingMessages,
+    getMessageWindowState,
+    setAtBottom as setMessageWindowAtBottom,
+    subscribeMessageWindow,
+    type MessageWindowState,
+} from '@/lib/message-window-store'
+
+const EMPTY_STATE: MessageWindowState = {
+    sessionId: 'unknown',
+    messages: [],
+    pending: [],
+    pendingCount: 0,
+    hasMore: false,
+    oldestSeq: null,
+    newestSeq: null,
+    isLoading: false,
+    isLoadingMore: false,
+    warning: null,
+    atBottom: true,
+    messagesVersion: 0,
+}

 export function useMessages(api: ApiClient | null, sessionId: string | null): {
    messages: DecryptedMessage[]
@@ -11,51 +33,80 @@ export function useMessages(api: ApiClient | null, sessionId: string | null): {
    isLoading: boolean
    isLoadingMore: boolean
    hasMore: boolean
+    pendingCount: number
+    messagesVersion: number
    loadMore: () => Promise<unknown>
    refetch: () => Promise<unknown>
+    flushPending: () => Promise<void>
+    setAtBottom: (atBottom: boolean) => void
 } {
-    const resolvedSessionId = sessionId ?? 'unknown'
-    const query = useInfiniteQuery<MessagesResponse>({
-        queryKey: queryKeys.messages(resolvedSessionId),
-        queryFn: async ({ pageParam }) => {
-            if (!api || !sessionId) {
-                throw new Error('Messages unavailable')
+    const state = useSyncExternalStore(
+        useCallback((listener) => {
+            if (!sessionId) {
+                return () => {}
            }
-            const beforeSeq = typeof pageParam === 'number' ? pageParam : null
-            return await api.getMessages(sessionId, { limit: 50, beforeSeq })
-        },
-        initialPageParam: null,
-        getNextPageParam: (lastPage) =>
-            lastPage.page.hasMore ? lastPage.page.nextBeforeSeq : undefined,
-        enabled: Boolean(api && sessionId),
-    })
+            return subscribeMessageWindow(sessionId, listener)
+        }, [sessionId]),
+        useCallback(() => {
+            if (!sessionId) {
+                return EMPTY_STATE
+            }
+            return getMessageWindowState(sessionId)
+        }, [sessionId]),
+        () => EMPTY_STATE
+    )

-    const messages = useMemo(() => {
-        const pages = query.data?.pages ?? []
-        let merged: DecryptedMessage[] = []
-        for (const page of pages) {
-            merged = mergeMessages(merged, page.messages)
+    useEffect(() => {
+        if (!api || !sessionId) {
+            return
        }
-        return merged
-    }, [query.data?.pages])
+        void fetchLatestMessages(api, sessionId)
+    }, [api, sessionId])

-    const warning = useMemo(() => {
-        if (!query.error) return null
-        return query.error instanceof Error ? query.error.message : 'Failed to load messages'
-    }, [query.error])
+    useEffect(() => {
+        if (!sessionId) {
+            return
+        }
+        return () => {
+            clearMessageWindow(sessionId)
+        }
+    }, [sessionId])

-    const loadMore = async () => {
-        if (!query.hasNextPage || query.isFetchingNextPage) return
-        await query.fetchNextPage()
-    }
+    const loadMore = useCallback(async () => {
+        if (!api || !sessionId) return
+        if (!state.hasMore || state.isLoadingMore) return
+        await fetchOlderMessages(api, sessionId)
+    }, [api, sessionId, state.hasMore, state.isLoadingMore])
+
+    const refetch = useCallback(async () => {
+        if (!api || !sessionId) return
+        await fetchLatestMessages(api, sessionId)
+    }, [api, sessionId])
+
+    const flushPending = useCallback(async () => {
+        if (!sessionId) return
+        const needsRefresh = flushPendingMessages(sessionId)
+        if (needsRefresh && api) {
+            await fetchLatestMessages(api, sessionId)
+        }
+    }, [api, sessionId])
+
+    const setAtBottom = useCallback((atBottom: boolean) => {
+        if (!sessionId) return
+        setMessageWindowAtBottom(sessionId, atBottom)
+    }, [sessionId])

    return {
-        messages,
-        warning,
-        isLoading: query.isLoading,
-        isLoadingMore: query.isFetchingNextPage,
-        hasMore: Boolean(query.hasNextPage),
+        messages: state.messages,
+        warning: state.warning,
+        isLoading: state.isLoading,
+        isLoadingMore: state.isLoadingMore,
+        hasMore: state.hasMore,
+        pendingCount: state.pendingCount,
+        messagesVersion: state.messagesVersion,
        loadMore,
-        refetch: query.refetch,
+        refetch,
+        flushPending,
+        setAtBottom,
    }
 }
@@ -1,8 +1,8 @@
 import { useEffect, useMemo, useRef } from 'react'
-import { useQueryClient, type InfiniteData } from '@tanstack/react-query'
-import type { MessagesResponse, SyncEvent } from '@/types/api'
+import { useQueryClient } from '@tanstack/react-query'
+import type { SyncEvent } from '@/types/api'
 import { queryKeys } from '@/lib/query-keys'
-import { upsertMessagesInCache } from '@/lib/messages'
+import { clearMessageWindow, ingestIncomingMessages } from '@/lib/message-window-store'

 function isObject(value: unknown): value is Record<string, unknown> {
    return Boolean(value) && typeof value === 'object'
@@ -86,15 +86,7 @@ export function useSSE(options: {

        const handleSyncEvent = (event: SyncEvent) => {
            if (event.type === 'message-received') {
-                queryClient.setQueryData<InfiniteData<MessagesResponse>>(
-                    queryKeys.messages(event.sessionId),
-                    (data) => upsertMessagesInCache(data, [event.message])
-                )
-                // Mark stale so the initial query still fetches history when it mounts.
-                void queryClient.invalidateQueries({
-                    queryKey: queryKeys.messages(event.sessionId),
-                    refetchType: 'none'
-                })
+                ingestIncomingMessages(event.sessionId, [event.message])
            }

            if (event.type === 'session-added' || event.type === 'session-updated' || event.type === 'session-removed') {
@@ -102,7 +94,7 @@ export function useSSE(options: {
                if ('sessionId' in event) {
                    if (event.type === 'session-removed') {
                        void queryClient.removeQueries({ queryKey: queryKeys.session(event.sessionId) })
-                        void queryClient.removeQueries({ queryKey: queryKeys.messages(event.sessionId) })
+                        clearMessageWindow(event.sessionId)
                    } else {
                        void queryClient.invalidateQueries({ queryKey: queryKeys.session(event.sessionId) })
                    }
@@ -0,0 +1,351 @@
+import type { ApiClient } from '@/api/client'
+import type { DecryptedMessage, MessageStatus } from '@/types/api'
+import { mergeMessages } from '@/lib/messages'
+
+export type MessageWindowState = {
+    sessionId: string
+    messages: DecryptedMessage[]
+    pending: DecryptedMessage[]
+    pendingCount: number
+    hasMore: boolean
+    oldestSeq: number | null
+    newestSeq: number | null
+    isLoading: boolean
+    isLoadingMore: boolean
+    warning: string | null
+    atBottom: boolean
+    messagesVersion: number
+}
+
+export const VISIBLE_WINDOW_SIZE = 400
+export const PENDING_WINDOW_SIZE = 200
+const PAGE_SIZE = 50
+const PENDING_OVERFLOW_WARNING = 'New messages arrived while you were away. Scroll to bottom to refresh.'
+
+type InternalState = MessageWindowState & {
+    pendingOverflowCount: number
+}
+
+const states = new Map<string, InternalState>()
+const listeners = new Map<string, Set<() => void>>()
+
+function createState(sessionId: string): InternalState {
+    return {
+        sessionId,
+        messages: [],
+        pending: [],
+        pendingCount: 0,
+        hasMore: false,
+        oldestSeq: null,
+        newestSeq: null,
+        isLoading: false,
+        isLoadingMore: false,
+        warning: null,
+        atBottom: true,
+        messagesVersion: 0,
+        pendingOverflowCount: 0,
+    }
+}
+
+function getState(sessionId: string): InternalState {
+    const existing = states.get(sessionId)
+    if (existing) {
+        return existing
+    }
+    const created = createState(sessionId)
+    states.set(sessionId, created)
+    return created
+}
+
+function notify(sessionId: string): void {
+    const subs = listeners.get(sessionId)
+    if (!subs) return
+    for (const listener of subs) {
+        listener()
+    }
+}
+
+function setState(sessionId: string, next: InternalState): void {
+    states.set(sessionId, next)
+    notify(sessionId)
+}
+
+function updateState(sessionId: string, updater: (prev: InternalState) => InternalState): void {
+    const prev = getState(sessionId)
+    const next = updater(prev)
+    if (next !== prev) {
+        setState(sessionId, next)
+    }
+}
+
+function deriveSeqBounds(messages: DecryptedMessage[]): { oldestSeq: number | null; newestSeq: number | null } {
+    let oldest: number | null = null
+    let newest: number | null = null
+    for (const message of messages) {
+        if (typeof message.seq !== 'number') {
+            continue
+        }
+        if (oldest === null || message.seq < oldest) {
+            oldest = message.seq
+        }
+        if (newest === null || message.seq > newest) {
+            newest = message.seq
+        }
+    }
+    return { oldestSeq: oldest, newestSeq: newest }
+}
+
+function buildState(
+    prev: InternalState,
+    updates: {
+        messages?: DecryptedMessage[]
+        pending?: DecryptedMessage[]
+        pendingOverflowCount?: number
+        hasMore?: boolean
+        isLoading?: boolean
+        isLoadingMore?: boolean
+        warning?: string | null
+        atBottom?: boolean
+    }
+): InternalState {
+    const messages = updates.messages ?? prev.messages
+    const pending = updates.pending ?? prev.pending
+    const pendingOverflowCount = updates.pendingOverflowCount ?? prev.pendingOverflowCount
+    const pendingCount = pending.length + pendingOverflowCount
+    const { oldestSeq, newestSeq } = deriveSeqBounds(messages)
+    const messagesVersion = messages === prev.messages ? prev.messagesVersion : prev.messagesVersion + 1
+
+    return {
+        ...prev,
+        messages,
+        pending,
+        pendingOverflowCount,
+        pendingCount,
+        oldestSeq,
+        newestSeq,
+        hasMore: updates.hasMore !== undefined ? updates.hasMore : prev.hasMore,
+        isLoading: updates.isLoading !== undefined ? updates.isLoading : prev.isLoading,
+        isLoadingMore: updates.isLoadingMore !== undefined ? updates.isLoadingMore : prev.isLoadingMore,
+        warning: updates.warning !== undefined ? updates.warning : prev.warning,
+        atBottom: updates.atBottom !== undefined ? updates.atBottom : prev.atBottom,
+        messagesVersion,
+    }
+}
+
+function trimVisible(messages: DecryptedMessage[], mode: 'append' | 'prepend'): DecryptedMessage[] {
+    if (messages.length <= VISIBLE_WINDOW_SIZE) {
+        return messages
+    }
+    if (mode === 'prepend') {
+        return messages.slice(0, VISIBLE_WINDOW_SIZE)
+    }
+    return messages.slice(messages.length - VISIBLE_WINDOW_SIZE)
+}
+
+function trimPending(messages: DecryptedMessage[]): { pending: DecryptedMessage[]; dropped: number } {
+    if (messages.length <= PENDING_WINDOW_SIZE) {
+        return { pending: messages, dropped: 0 }
+    }
+    const dropped = messages.length - PENDING_WINDOW_SIZE
+    return { pending: messages.slice(messages.length - PENDING_WINDOW_SIZE), dropped }
+}
+
+function filterPendingAgainstVisible(pending: DecryptedMessage[], visible: DecryptedMessage[]): DecryptedMessage[] {
+    if (pending.length === 0 || visible.length === 0) {
+        return pending
+    }
+    const visibleIds = new Set(visible.map((message) => message.id))
+    return pending.filter((message) => !visibleIds.has(message.id))
+}
+
+function isOptimisticMessage(message: DecryptedMessage): boolean {
+    return Boolean(message.localId && message.id === message.localId)
+}
+
+function mergeIntoPending(
+    prev: InternalState,
+    incoming: DecryptedMessage[]
+): { pending: DecryptedMessage[]; pendingOverflowCount: number; warning: string | null } {
+    if (incoming.length === 0) {
+        return { pending: prev.pending, pendingOverflowCount: prev.pendingOverflowCount, warning: prev.warning }
+    }
+    const mergedPending = mergeMessages(prev.pending, incoming)
+    const filtered = filterPendingAgainstVisible(mergedPending, prev.messages)
+    const { pending, dropped } = trimPending(filtered)
+    const pendingOverflowCount = prev.pendingOverflowCount + dropped
+    const warning = dropped > 0 && !prev.warning ? PENDING_OVERFLOW_WARNING : prev.warning
+    return { pending, pendingOverflowCount, warning }
+}
+
+export function getMessageWindowState(sessionId: string): MessageWindowState {
+    return getState(sessionId)
+}
+
+export function subscribeMessageWindow(sessionId: string, listener: () => void): () => void {
+    const subs = listeners.get(sessionId) ?? new Set()
+    subs.add(listener)
+    listeners.set(sessionId, subs)
+    return () => {
+        const current = listeners.get(sessionId)
+        if (!current) return
+        current.delete(listener)
+        if (current.size === 0) {
+            listeners.delete(sessionId)
+            states.delete(sessionId)
+        }
+    }
+}
+
+export function clearMessageWindow(sessionId: string): void {
+    if (!states.has(sessionId)) {
+        return
+    }
+    setState(sessionId, createState(sessionId))
+}
+
+export async function fetchLatestMessages(api: ApiClient, sessionId: string): Promise<void> {
+    const initial = getState(sessionId)
+    if (initial.isLoading) {
+        return
+    }
+    updateState(sessionId, (prev) => buildState(prev, { isLoading: true, warning: null }))
+
+    try {
+        const response = await api.getMessages(sessionId, { limit: PAGE_SIZE, beforeSeq: null })
+        updateState(sessionId, (prev) => {
+            if (prev.atBottom) {
+                const merged = mergeMessages(prev.messages, [...prev.pending, ...response.messages])
+                const trimmed = trimVisible(merged, 'append')
+                return buildState(prev, {
+                    messages: trimmed,
+                    pending: [],
+                    pendingOverflowCount: 0,
+                    hasMore: response.page.hasMore,
+                    isLoading: false,
+                    warning: null,
+                })
+            }
+            const pendingResult = mergeIntoPending(prev, response.messages)
+            return buildState(prev, {
+                pending: pendingResult.pending,
+                pendingOverflowCount: pendingResult.pendingOverflowCount,
+                isLoading: false,
+                warning: pendingResult.warning,
+            })
+        })
+    } catch (error) {
+        const message = error instanceof Error ? error.message : 'Failed to load messages'
+        updateState(sessionId, (prev) => buildState(prev, { isLoading: false, warning: message }))
+    }
+}
+
+export async function fetchOlderMessages(api: ApiClient, sessionId: string): Promise<void> {
+    const initial = getState(sessionId)
+    if (initial.isLoadingMore || !initial.hasMore) {
+        return
+    }
+    if (initial.oldestSeq === null) {
+        return
+    }
+    updateState(sessionId, (prev) => buildState(prev, { isLoadingMore: true }))
+
+    try {
+        const response = await api.getMessages(sessionId, { limit: PAGE_SIZE, beforeSeq: initial.oldestSeq })
+        updateState(sessionId, (prev) => {
+            const merged = mergeMessages(response.messages, prev.messages)
+            const trimmed = trimVisible(merged, 'prepend')
+            return buildState(prev, {
+                messages: trimmed,
+                hasMore: response.page.hasMore,
+                isLoadingMore: false,
+            })
+        })
+    } catch (error) {
+        const message = error instanceof Error ? error.message : 'Failed to load messages'
+        updateState(sessionId, (prev) => buildState(prev, { isLoadingMore: false, warning: message }))
+    }
+}
+
+export function ingestIncomingMessages(sessionId: string, incoming: DecryptedMessage[]): void {
+    if (incoming.length === 0) {
+        return
+    }
+    updateState(sessionId, (prev) => {
+        if (prev.atBottom) {
+            const merged = mergeMessages(prev.messages, incoming)
+            const trimmed = trimVisible(merged, 'append')
+            const pending = filterPendingAgainstVisible(prev.pending, trimmed)
+            return buildState(prev, { messages: trimmed, pending })
+        }
+        const pendingResult = mergeIntoPending(prev, incoming)
+        return buildState(prev, {
+            pending: pendingResult.pending,
+            pendingOverflowCount: pendingResult.pendingOverflowCount,
+            warning: pendingResult.warning,
+        })
+    })
+}
+
+export function flushPendingMessages(sessionId: string): boolean {
+    const current = getState(sessionId)
+    if (current.pending.length === 0 && current.pendingOverflowCount === 0) {
+        return false
+    }
+    const needsRefresh = current.pendingOverflowCount > 0
+    updateState(sessionId, (prev) => {
+        const merged = mergeMessages(prev.messages, prev.pending)
+        const trimmed = trimVisible(merged, 'append')
+        return buildState(prev, {
+            messages: trimmed,
+            pending: [],
+            pendingOverflowCount: 0,
+            warning: needsRefresh ? (prev.warning ?? PENDING_OVERFLOW_WARNING) : prev.warning,
+        })
+    })
+    return needsRefresh
+}
+
+export function setAtBottom(sessionId: string, atBottom: boolean): void {
+    updateState(sessionId, (prev) => {
+        if (prev.atBottom === atBottom) {
+            return prev
+        }
+        return buildState(prev, { atBottom })
+    })
+}
+
+export function appendOptimisticMessage(sessionId: string, message: DecryptedMessage): void {
+    updateState(sessionId, (prev) => {
+        const merged = mergeMessages(prev.messages, [message])
+        const trimmed = trimVisible(merged, 'append')
+        const pending = filterPendingAgainstVisible(prev.pending, trimmed)
+        return buildState(prev, { messages: trimmed, pending, atBottom: true })
+    })
+}
+
+export function updateMessageStatus(sessionId: string, localId: string, status: MessageStatus): void {
+    if (!localId) {
+        return
+    }
+    updateState(sessionId, (prev) => {
+        let changed = false
+        const updateList = (list: DecryptedMessage[]) => {
+            return list.map((message) => {
+                if (message.localId !== localId || !isOptimisticMessage(message)) {
+                    return message
+                }
+                if (message.status === status) {
+                    return message
+                }
+                changed = true
+                return { ...message, status }
+            })
+        }
+        const messages = updateList(prev.messages)
+        const pending = updateList(prev.pending)
+        if (!changed) {
+            return prev
+        }
+        return buildState(prev, { messages, pending })
+    })
+}
@@ -16,6 +16,10 @@ function isUserMessage(msg: DecryptedMessage): boolean {
    return false
 }

+function isOptimisticMessage(msg: DecryptedMessage): boolean {
+    return Boolean(msg.localId && msg.id === msg.localId)
+}
+
 function compareMessages(a: DecryptedMessage, b: DecryptedMessage): number {
    const aSeq = typeof a.seq === 'number' ? a.seq : null
    const bSeq = typeof b.seq === 'number' ? b.seq : null
@@ -46,33 +50,32 @@ export function mergeMessages(existing: DecryptedMessage[], incoming: DecryptedM

    let merged = Array.from(byId.values())

-    const incomingLocalIds = new Set<string>()
+    const incomingStoredLocalIds = new Set<string>()
    for (const msg of incoming) {
-        if (msg.localId) {
-            incomingLocalIds.add(msg.localId)
+        if (msg.localId && !isOptimisticMessage(msg)) {
+            incomingStoredLocalIds.add(msg.localId)
        }
    }

-    // If we received a stored message with a localId, drop any optimistic bubble with the same localId.
-    if (incomingLocalIds.size > 0) {
+    // If we received stored messages with a localId, drop any optimistic bubbles with the same localId.
+    if (incomingStoredLocalIds.size > 0) {
        merged = merged.filter((msg) => {
-            if (!msg.localId || !incomingLocalIds.has(msg.localId)) {
+            if (!msg.localId || !incomingStoredLocalIds.has(msg.localId)) {
                return true
            }
-            return !msg.status
+            return !isOptimisticMessage(msg)
        })
    }

    // Fallback: if an optimistic message was marked as sent but we didn't get a localId echo,
    // drop it when a server user message appears close in time.
-    const optimisticMessages = merged.filter((m) => m.localId && m.status)
-    const nonOptimisticMessages = merged.filter((m) => !m.localId || !m.status)
+    const optimisticMessages = merged.filter((m) => isOptimisticMessage(m))
+    const nonOptimisticMessages = merged.filter((m) => !isOptimisticMessage(m))
    const result: DecryptedMessage[] = [...nonOptimisticMessages]

    for (const optimistic of optimisticMessages) {
        if (optimistic.status === 'sent') {
            const hasServerUserMessage = nonOptimisticMessages.some((m) =>
-                !m.status &&
                isUserMessage(m) &&
                Math.abs(m.createdAt - optimistic.createdAt) < 10_000
            )
@@ -133,6 +133,10 @@ function SessionPage() {
        hasMore: messagesHasMore,
        loadMore: loadMoreMessages,
        refetch: refetchMessages,
+        pendingCount,
+        messagesVersion,
+        flushPending,
+        setAtBottom,
    } = useMessages(api, sessionId)
    const {
        sendMessage,
@@ -169,10 +173,14 @@ function SessionPage() {
            isLoadingMessages={messagesLoading}
            isLoadingMoreMessages={messagesLoadingMore}
            isSending={isSending}
+            pendingCount={pendingCount}
+            messagesVersion={messagesVersion}
            onBack={goBack}
            onRefresh={refreshSelectedSession}
            onLoadMore={loadMoreMessages}
            onSend={sendMessage}
+            onFlushPending={flushPending}
+            onAtBottomChange={setAtBottom}
            onRetryMessage={retryMessage}
            autocompleteSuggestions={getSlashSuggestions}
        />