From 17ee052d9a9063b332508a47e3bd56d784c09ba7 Mon Sep 17 00:00:00 2001 From: Ananovo Date: Wed, 9 Sep 2026 09:23:31 +0800 Subject: [PATCH] fix(web): preserve the visible chat window during rewind (#1766) * fix(web): preserve chat window during rewind * fix(web): scope rewind invalidation preservation * fix(web): clear unknown rewind boundaries * fix(web): deduplicate rewind invalidations * fix(web): retain rewind dedupe history --- docs/api/client-contract/sse.md | 2 +- hub/src/sync/syncEngine.ts | 8 ++- shared/src/resume.test.ts | 12 ++++ shared/src/schemas.ts | 4 +- web/src/App.tsx | 8 ++- web/src/components/SessionChat.tsx | 12 +++- web/src/lib/message-window-store.test.ts | 92 ++++++++++++++++++++++++ web/src/lib/message-window-store.ts | 74 ++++++++++++++++++- web/src/router.tsx | 8 ++- 9 files changed, 208 insertions(+), 12 deletions(-) diff --git a/docs/api/client-contract/sse.md b/docs/api/client-contract/sse.md index 99836b73..4723e4f5 100644 --- a/docs/api/client-contract/sse.md +++ b/docs/api/client-contract/sse.md @@ -140,7 +140,7 @@ Schema: `SyncEventSchema` in `shared/src/schemas.ts` (discriminated on `type`). | `session-updated` | `sessionId`, `data?: Session \| SessionPatch` | See [Versioned patch algorithm](#versioned-patch-algorithm). | | `session-removed` | `sessionId` | Drop the session from the list, drop its detail cache, clear its message window. | | `message-received` | `sessionId`, `message: DecryptedMessage` | Ingest into the message window; advance the tail cursor (see [pagination](./pagination.md)). Also fired for the caller's own send (the localId echo). | -| `messages-invalidated` | `sessionId` | Message history changed **structurally** (rewind, fork, import, clear). Session scope: discard the whole window and run a fresh tail sync. Global scope: refetch the session list. | +| `messages-invalidated` | `sessionId`; rewind may also include `reason: 'rewind'` and `truncateFromLocalId` | Message history changed **structurally** (rewind, fork, import, clear). For a rewind, retain only the known prefix through the client boundary before tail-syncing; for every other invalidation, discard the whole window and run a fresh tail sync. Global scope: refetch the session list. | | `scheduled-matured` | `sessionId` | A scheduled message became due and was handed to the agent. Refetch list/queue indicators. | | `session-ended` | `sessionId`, `reason?: 'completed'\|'terminated'\|'error'\|'handoff'\|'cleared'` | Session lifecycle signal (the `session-updated` flow still carries the state change). | | `machine-updated` | `machineId`, `data?: Machine \| MachinePatch \| null` | Full `Machine`: upsert (remove when `active:false`). `null`: machine removed. Patch `{active?, activeAt?, updatedAt?}`: `active:false` ⇒ remove, otherwise refetch machines. `data` absent ⇒ refetch. | diff --git a/hub/src/sync/syncEngine.ts b/hub/src/sync/syncEngine.ts index 8d50c86c..f5f2ba17 100644 --- a/hub/src/sync/syncEngine.ts +++ b/hub/src/sync/syncEngine.ts @@ -1652,7 +1652,13 @@ export class SyncEngine { ) this.scrubHistoryLocators(sessionId, namespace) this.sessionCache.rebuildTodosFromTranscript(sessionId) - this.eventPublisher.emit({ type: 'messages-invalidated', sessionId, namespace }) + this.eventPublisher.emit({ + type: 'messages-invalidated', + sessionId, + namespace, + reason: 'rewind', + truncateFromLocalId: rpcResult.truncateFromLocalId ?? messageLocalId + }) this.sessionCache.refreshSession(sessionId) return { type: 'success' } } catch (error) { diff --git a/shared/src/resume.test.ts b/shared/src/resume.test.ts index d653fd26..009fc46e 100644 --- a/shared/src/resume.test.ts +++ b/shared/src/resume.test.ts @@ -81,6 +81,18 @@ describe('resume schemas', () => { }).success).toBe(true) }) + it('accepts rewind details on messages-invalidated events', () => { + expect(SyncEventSchema.parse({ + type: 'messages-invalidated', + sessionId: 'session-1', + reason: 'rewind', + truncateFromLocalId: 'user-message-1' + })).toMatchObject({ + reason: 'rewind', + truncateFromLocalId: 'user-message-1' + }) + }) + it('validates structured session and machine update patches', () => { expect(SyncEventSchema.safeParse({ type: 'session-updated', diff --git a/shared/src/schemas.ts b/shared/src/schemas.ts index 167f95d6..a3cc160d 100644 --- a/shared/src/schemas.ts +++ b/shared/src/schemas.ts @@ -558,7 +558,9 @@ export const SyncEventSchema = z.discriminatedUnion('type', [ message: DecryptedMessageSchema }), SessionChangedSchema.extend({ - type: z.literal('messages-invalidated') + type: z.literal('messages-invalidated'), + reason: z.literal('rewind').optional(), + truncateFromLocalId: z.string().min(1).optional() }), SessionChangedSchema.extend({ type: z.literal('scheduled-matured') diff --git a/web/src/App.tsx b/web/src/App.tsx index ef770dca..bffc3869 100644 --- a/web/src/App.tsx +++ b/web/src/App.tsx @@ -16,7 +16,7 @@ import { useViewportHeight } from '@/hooks/useViewportHeight' import { useVisibilityReporter } from '@/hooks/useVisibilityReporter' import { queryKeys } from '@/lib/query-keys' import { AppContextProvider } from '@/lib/app-context' -import { clearMessageWindow, syncTailMessages } from '@/lib/message-window-store' +import { clearMessageWindow, rewindMessageWindow, syncTailMessages } from '@/lib/message-window-store' import { useAppGoBack } from '@/hooks/useAppGoBack' import { useTranslation } from '@/lib/use-translation' import { VoiceProvider } from '@/lib/voice-context' @@ -291,7 +291,11 @@ function AppInner() { if (!api || event.sessionId !== selectedSessionId) { return } - clearMessageWindow(event.sessionId) + if (event.reason === 'rewind' && event.truncateFromLocalId) { + rewindMessageWindow(event.sessionId, event.truncateFromLocalId) + } else { + clearMessageWindow(event.sessionId) + } void syncTailMessages(api, event.sessionId) }, [api, selectedSessionId]) diff --git a/web/src/components/SessionChat.tsx b/web/src/components/SessionChat.tsx index a9818cd3..30679410 100644 --- a/web/src/components/SessionChat.tsx +++ b/web/src/components/SessionChat.tsx @@ -54,8 +54,8 @@ import { } from '@/lib/messageDelivery' import type { MessageDeliveryMode } from '@hapi/protocol' import { isSteeringSupportedForSession } from '@hapi/protocol' -import type { OlderLoadOutcome } from '@/lib/message-window-store' import { createAttachmentAdapter } from '@/lib/attachmentAdapter' +import { rewindMessageWindow, type OlderLoadOutcome } from '@/lib/message-window-store' import { ShareSeedConsumer } from '@/components/ShareSeedConsumer' import { createScratchlistAttachmentAdapter, @@ -515,7 +515,7 @@ type SessionChatProps = { historyVersion: number tailRevision: number onBack: () => void - onRefresh: () => void + onRefresh: () => void | Promise onLoadMore: (onBeforeApply?: (historyVersion: number) => boolean) => Promise onCancelLoadMore: () => void // Returns the accepted mutation's attempt id, or false when @@ -594,7 +594,13 @@ function SessionChatInner(props: SessionChatProps) { setHistoryActionPending(true) try { await props.api.rewindConversation(props.session.id, messageLocalId) - props.onRefresh() + // Apply the deterministic local part immediately so the removed + // suffix cannot flash back while the authoritative refresh runs. + rewindMessageWindow(props.session.id, messageLocalId) + await props.onRefresh() + // Force the same tail behavior as a successful send after the + // refreshed message window has been committed. + setForceScrollToken((token) => token + 1) } catch (error) { if (isRewindForkFallbackError(error)) { setRewindForkFallback(messageLocalId) diff --git a/web/src/lib/message-window-store.test.ts b/web/src/lib/message-window-store.test.ts index 121ee296..713546f8 100644 --- a/web/src/lib/message-window-store.test.ts +++ b/web/src/lib/message-window-store.test.ts @@ -11,9 +11,11 @@ import { getMessageWindowState, getQueuedReconcileCandidateLocalIds, ingestIncomingMessages, + invalidateMessageWindow, markMessagesConsumed, reconcileQueuedLocalIds, removeOptimisticMessage, + rewindMessageWindow, setMessageViewMode, syncTailMessages, updateMessageStatus, @@ -247,6 +249,96 @@ afterEach(() => { }) describe('message tail synchronization', () => { + it('removes the rewound suffix immediately and applies duplicate invalidations once', async () => { + const id = sessionId('rewind-suffix') + const prefix = makeAgentMessage({ id: 'prefix', seq: 1, at: 1_000 }) + const target = makeUserMessage({ + id: 'target', + seq: 2, + localId: 'target-local-id', + createdAt: 2_000, + invokedAt: 2_000 + }) + const suffix = makeAgentMessage({ id: 'suffix', seq: 3, at: 3_000 }) + const getMessages = vi.fn(async () => latestResponse([prefix, target, suffix], { epoch: 1 })) + + await syncTailMessages(createApi(getMessages), id) + rewindMessageWindow(id, 'target-local-id') + rewindMessageWindow(id, 'target-local-id') + + expect(getMessageWindowState(id).messages.map((message) => message.id)).toEqual(['prefix']) + expect(getMessageWindowState(id).isSyncingTail).toBe(true) + }) + + it('clears the window when the rewind boundary is outside the loaded page', async () => { + const id = sessionId('rewind-boundary-not-loaded') + const current = makeAgentMessage({ id: 'current', seq: 40, at: 40_000 }) + const getMessages = vi.fn(async () => latestResponse([current], { epoch: 1 })) + + await syncTailMessages(createApi(getMessages), id) + rewindMessageWindow(id, 'boundary-not-loaded') + + expect(getMessageWindowState(id).messages).toEqual([]) + }) + + it('deduplicates delayed rewind events across consecutive boundaries', async () => { + const id = sessionId('rewind-delayed-event') + const prefix = makeAgentMessage({ id: 'prefix', seq: 1, at: 1_000 }) + const firstBoundary = makeUserMessage({ + id: 'first-boundary', + seq: 2, + localId: 'first-boundary-local-id', + createdAt: 2_000, + invokedAt: 2_000 + }) + const secondBoundary = makeUserMessage({ + id: 'second-boundary', + seq: 3, + localId: 'second-boundary-local-id', + createdAt: 3_000, + invokedAt: 3_000 + }) + const suffix = makeAgentMessage({ id: 'suffix', seq: 4, at: 4_000 }) + const getMessages = vi.fn(async () => latestResponse( + [prefix, firstBoundary, secondBoundary, suffix], + { epoch: 1 } + )) + + await syncTailMessages(createApi(getMessages), id) + rewindMessageWindow(id, 'second-boundary-local-id') + rewindMessageWindow(id, 'first-boundary-local-id') + rewindMessageWindow(id, 'second-boundary-local-id') + + expect(getMessageWindowState(id).messages.map((message) => message.id)).toEqual(['prefix']) + }) + + it('retains the current window while a latest reset is in flight', async () => { + const id = sessionId('invalidation-preserves-window') + const current = makeAgentMessage({ id: 'current', seq: 10, at: 10_000 }) + const latest = makeAgentMessage({ id: 'latest', seq: 20, at: 20_000 }) + const response = deferred() + const getMessages = vi.fn() + .mockResolvedValueOnce(latestResponse([current], { epoch: 1 })) + .mockImplementationOnce(async () => await response.promise) + const api = createApi(getMessages) + + await syncTailMessages(api, id) + invalidateMessageWindow(id) + + expect(getMessageWindowState(id).messages.map((message) => message.id)).toEqual(['current']) + expect(getMessageWindowState(id).isSyncingTail).toBe(true) + + const syncing = syncTailMessages(api, id) + await vi.waitFor(() => expect(getMessages).toHaveBeenCalledTimes(2)) + expect(getMessages.mock.calls[1]?.[1]).toEqual({ limit: 200 }) + + response.resolve(latestResponse([latest], { epoch: 2 })) + await syncing + + expect(getMessageWindowState(id).messages.map((message) => message.id)).toEqual(['latest']) + expect(getMessageWindowState(id).isSyncingTail).toBe(false) + }) + it('renders a persisted window immediately, then requests the latest tail on re-entry', async () => { const id = sessionId('reentry') const cached = makeAgentMessage({ id: 'cached', seq: 40, at: 40_000 }) diff --git a/web/src/lib/message-window-store.ts b/web/src/lib/message-window-store.ts index 14fc630e..e7a6f8d3 100644 --- a/web/src/lib/message-window-store.ts +++ b/web/src/lib/message-window-store.ts @@ -80,6 +80,7 @@ type TailSyncController = { const states = new Map() const listeners = new Map void>>() const tailSyncControllers = new Map() +const appliedRewindLocalIds = new Map>() const NOTIFY_THROTTLE_MS = 150 const PERSIST_THROTTLE_MS = 200 @@ -312,7 +313,14 @@ function notifyImmediate(sessionId: string): void { function setState(sessionId: string, next: InternalState, immediate = false): void { states.set(sessionId, next) - schedulePersist(sessionId) + // A latest-reset state still contains the previous server snapshot. Do not + // persist that stale window while the authoritative replacement is in + // flight; a reload during the reset must not resurrect removed messages. + if (next.requiresLatestReset) { + pendingPersistSessionIds.delete(sessionId) + } else { + schedulePersist(sessionId) + } if (immediate) { notifyImmediate(sessionId) } else { @@ -1091,6 +1099,70 @@ export function clearMessageWindow(sessionId: string): void { }, true) } +function markMessageWindowForLatestReset(sessionId: string, messages: DecryptedMessage[]): void { + const previous = states.get(sessionId) + if (!previous) return + + tailSyncControllers.delete(sessionId) + clearPersistedState(sessionId) + setState(sessionId, buildState(previous, { + messages, + epoch: null, + oldestPositionAt: null, + oldestPositionSeq: null, + newestPositionAt: null, + newestPositionSeq: null, + requiresLatestReset: true, + preferLatestOnActivation: false, + isSyncingTail: true, + isLoadingMore: false, + warning: null, + syncGeneration: previous.syncGeneration + 1, + olderGeneration: previous.olderGeneration + 1 + }), true) +} + +/** + * Mark the current window stale without exposing an empty transcript while a + * latest snapshot is fetched. The next tail sync sees `requiresLatestReset` + * and replaces server rows atomically with the authoritative response. + */ +export function invalidateMessageWindow(sessionId: string): void { + const previous = states.get(sessionId) + if (!previous) return + + markMessageWindowForLatestReset(sessionId, previous.messages) +} + +/** + * Apply the known local effect of a successful Rewind before the server + * snapshot arrives. Rewind removes the boundary message and every later row; + * retaining the earlier prefix keeps the chat usable and lets the current + * bottom position clamp directly to the new tail. + */ +export function rewindMessageWindow(sessionId: string, messageLocalId: string): void { + const previous = states.get(sessionId) + if (!previous) return + + const applied = appliedRewindLocalIds.get(sessionId) ?? new Set() + if (applied.has(messageLocalId)) return + applied.add(messageLocalId) + appliedRewindLocalIds.set(sessionId, applied) + + const boundaryIndex = previous.messages.findIndex((message) => message.localId === messageLocalId) + if (boundaryIndex < 0) { + // The boundary may be outside the current latest window. Without a + // local boundary, retaining rows could show messages removed by the + // rewind until the authoritative tail sync completes. + clearMessageWindow(sessionId) + return + } + + const messages = previous.messages.slice(0, boundaryIndex) + + markMessageWindowForLatestReset(sessionId, messages) +} + export function seedMessageWindowFromSession(fromSessionId: string, toSessionId: string): void { if (!fromSessionId || !toSessionId || fromSessionId === toSessionId) return const source = getState(fromSessionId) diff --git a/web/src/router.tsx b/web/src/router.tsx index 7cf2361c..a182f5aa 100644 --- a/web/src/router.tsx +++ b/web/src/router.tsx @@ -739,9 +739,11 @@ function SessionPage() { getSlashSuggestions, ]) - const refreshSelectedSession = useCallback(() => { - void refetchSession() - void refetchMessages() + const refreshSelectedSession = useCallback(async () => { + await Promise.all([ + refetchSession(), + refetchMessages(), + ]) }, [refetchMessages, refetchSession]) const handleInitialOutlineConsumed = useCallback(() => {