From c9064a5193de92bce3cdb0764f4e0473fd42aaa6 Mon Sep 17 00:00:00 2001 From: qer Date: Mon, 6 Jul 2026 13:04:51 +0800 Subject: [PATCH 01/11] fix(web): reconcile session from snapshot on reopen --- .changeset/fix-web-reopen-reconcile.md | 5 ++ .../composables/client/useWorkspaceState.ts | 7 ++- .../src/composables/useKimiWebClient.ts | 51 +++++++++++++------ 3 files changed, 43 insertions(+), 20 deletions(-) create mode 100644 .changeset/fix-web-reopen-reconcile.md diff --git a/.changeset/fix-web-reopen-reconcile.md b/.changeset/fix-web-reopen-reconcile.md new file mode 100644 index 0000000000..fd5eb32821 --- /dev/null +++ b/.changeset/fix-web-reopen-reconcile.md @@ -0,0 +1,5 @@ +--- +"@moonshot-ai/kimi-code": patch +--- + +web: Fix the end of a reply staying missing after reopening a session. diff --git a/apps/kimi-web/src/composables/client/useWorkspaceState.ts b/apps/kimi-web/src/composables/client/useWorkspaceState.ts index 00add89630..50e07ea3ea 100644 --- a/apps/kimi-web/src/composables/client/useWorkspaceState.ts +++ b/apps/kimi-web/src/composables/client/useWorkspaceState.ts @@ -972,10 +972,9 @@ export function useWorkspaceState(rawState: ExtendedState, deps: UseWorkspaceSta const result = await syncSessionFromSnapshot(sessionId); if (result === 'not-found') return; } else { - // Re-open: if the session was evicted from the subscription cap since - // last open, reopenSession rebuilds it from a snapshot (the kept cursor - // may have skipped per-session events); otherwise it re-subscribes from - // the tracked cursor and the daemon replays any missed durable events. + // Re-open: rebuild from a fresh snapshot rather than resuming from the + // tracked cursor — the daemon only replays durable events, so volatile + // streamed deltas lost to a WS hiccup would otherwise stay missing. const result = await reopenSession(sessionId); if (result === 'not-found') return; } diff --git a/apps/kimi-web/src/composables/useKimiWebClient.ts b/apps/kimi-web/src/composables/useKimiWebClient.ts index 9e149433e6..8ec05616e2 100644 --- a/apps/kimi-web/src/composables/useKimiWebClient.ts +++ b/apps/kimi-web/src/composables/useKimiWebClient.ts @@ -541,6 +541,7 @@ function forgetSession(sessionId: string): void { delete rawState.messagesHasMoreBySession[sessionId]; delete rawState.messagesLoadMoreErrorBySession[sessionId]; delete epochBySession[sessionId]; + lastLoadedUpdatedAt.delete(sessionId); sessionsKnownEmpty.delete(sessionId); // In-flight / queued prompt state: drop these too so a queued follow-up // can't be submitted to a session that was just archived when its turn later @@ -904,6 +905,12 @@ function connectEventsIfNeeded(): void { // reactive — only consulted when building the subscribe cursor. const epochBySession: Record = {}; +// Session `updatedAt` (ISO string) captured the last time we loaded this +// session's messages from a snapshot. Compared on re-open to skip the snapshot +// fetch + transcript remount when the session has not changed since. Not +// reactive — only read inside reopenSession. +const lastLoadedUpdatedAt = new Map(); + // Sessions created locally in this client instance are known to be empty until // they receive their first message. This is more reliable than the daemon's // messageCount field, which can be stale for old sessions and would otherwise @@ -1169,6 +1176,7 @@ async function syncSessionFromSnapshot(sessionId: string): Promise { if (sessionsWithStaleCursor.has(sessionId)) { return syncSessionFromSnapshot(sessionId); } - subscribeToSessionEvents(sessionId); - return 'ok'; + const current = rawState.sessions.find((s) => s.id === sessionId)?.updatedAt; + if (current !== undefined && current === lastLoadedUpdatedAt.get(sessionId)) { + resubscribeSessionEvents(sessionId); + return 'ok'; + } + return syncSessionFromSnapshot(sessionId); } // --------------------------------------------------------------------------- From 003d0de531784e0382c0b0335e03696b289a07b4 Mon Sep 17 00:00:00 2001 From: qer Date: Mon, 6 Jul 2026 13:37:02 +0800 Subject: [PATCH 02/11] fix(web): discard stale snapshot when a newer prompt races reopen --- apps/kimi-web/src/composables/useKimiWebClient.ts | 13 +++++++++++++ 1 file changed, 13 insertions(+) diff --git a/apps/kimi-web/src/composables/useKimiWebClient.ts b/apps/kimi-web/src/composables/useKimiWebClient.ts index 8ec05616e2..7fbfbb2760 100644 --- a/apps/kimi-web/src/composables/useKimiWebClient.ts +++ b/apps/kimi-web/src/composables/useKimiWebClient.ts @@ -1126,10 +1126,23 @@ async function pullSessionWarnings(sessionId: string): Promise { } async function syncSessionFromSnapshot(sessionId: string): Promise { + // Preconditions captured before the await, used to detect a newer local + // prompt/seq that appears while the snapshot is in flight (see guard below). + const seqBefore = rawState.lastSeqBySession[sessionId] ?? 0; + const promptBefore = rawState.promptIdBySession[sessionId]; try { const api = getKimiWebApi(); const snap = await api.getSessionSnapshot(sessionId); + // Discard a snapshot that went stale because a newer local prompt/seq + // arrived while it was in flight. On the re-open path `sessionLoading` is + // false and the composer stays usable, so a send can race this GET: the + // snapshot's `asOfSeq` predates the new prompt, and replacing messages here + // would wipe a live turn whose volatile deltas are not replayable. The live + // stream will populate messages; a later re-open reconciles again if needed. + if ((rawState.lastSeqBySession[sessionId] ?? 0) !== seqBefore) return 'ok'; + if (rawState.promptIdBySession[sessionId] !== promptBefore) return 'ok'; + // Drain any queued streaming deltas before the snapshot replaces // messagesBySession[sessionId]. The snapshot is authoritative (it already // contains everything up to asOfSeq); applying stale queued deltas on top From d051f9b144602be1bc4bccefda0e1dbef0373635 Mon Sep 17 00:00:00 2001 From: qer Date: Mon, 6 Jul 2026 13:58:11 +0800 Subject: [PATCH 03/11] fix(web): harden reopen snapshot against first-open and optimistic-send races --- .../src/composables/useKimiWebClient.ts | 25 ++++++++++++------- 1 file changed, 16 insertions(+), 9 deletions(-) diff --git a/apps/kimi-web/src/composables/useKimiWebClient.ts b/apps/kimi-web/src/composables/useKimiWebClient.ts index 7fbfbb2760..1e4b437676 100644 --- a/apps/kimi-web/src/composables/useKimiWebClient.ts +++ b/apps/kimi-web/src/composables/useKimiWebClient.ts @@ -1127,21 +1127,28 @@ async function pullSessionWarnings(sessionId: string): Promise { async function syncSessionFromSnapshot(sessionId: string): Promise { // Preconditions captured before the await, used to detect a newer local - // prompt/seq that appears while the snapshot is in flight (see guard below). + // prompt/seq or optimistic send that appears while the snapshot is in flight + // (see guard below). const seqBefore = rawState.lastSeqBySession[sessionId] ?? 0; const promptBefore = rawState.promptIdBySession[sessionId]; + const inFlightBefore = inFlightPromptSessions.has(sessionId); + const wasLoaded = hasLoadedMessages(sessionId); try { const api = getKimiWebApi(); const snap = await api.getSessionSnapshot(sessionId); - // Discard a snapshot that went stale because a newer local prompt/seq - // arrived while it was in flight. On the re-open path `sessionLoading` is - // false and the composer stays usable, so a send can race this GET: the - // snapshot's `asOfSeq` predates the new prompt, and replacing messages here - // would wipe a live turn whose volatile deltas are not replayable. The live - // stream will populate messages; a later re-open reconciles again if needed. - if ((rawState.lastSeqBySession[sessionId] ?? 0) !== seqBefore) return 'ok'; - if (rawState.promptIdBySession[sessionId] !== promptBefore) return 'ok'; + // Staleness guard, scoped to the re-open path (session already loaded): + // there the composer stays usable, so a send or a broadcast global event can + // race this GET and produce a snapshot whose `asOfSeq` predates newer local + // state — replacing messages would wipe a live turn (volatile deltas are not + // replayable) or the just-sent optimistic user message. On first open we must + // always install messages + subscribe, even if a global event advanced lastSeq + // during the await. When discarded, the live stream / next re-open reconciles. + if (wasLoaded) { + if ((rawState.lastSeqBySession[sessionId] ?? 0) !== seqBefore) return 'ok'; + if (rawState.promptIdBySession[sessionId] !== promptBefore) return 'ok'; + if (inFlightPromptSessions.has(sessionId) !== inFlightBefore) return 'ok'; + } // Drain any queued streaming deltas before the snapshot replaces // messagesBySession[sessionId]. The snapshot is authoritative (it already From d6c18a63c3d55f086afab21cb00341628dccc3cf Mon Sep 17 00:00:00 2001 From: qer Date: Mon, 6 Jul 2026 14:09:01 +0800 Subject: [PATCH 04/11] fix(web): keep evicted reopens subscribed when a snapshot races --- .../src/composables/useKimiWebClient.ts | 18 ++++++++++-------- 1 file changed, 10 insertions(+), 8 deletions(-) diff --git a/apps/kimi-web/src/composables/useKimiWebClient.ts b/apps/kimi-web/src/composables/useKimiWebClient.ts index 1e4b437676..ec8635c15e 100644 --- a/apps/kimi-web/src/composables/useKimiWebClient.ts +++ b/apps/kimi-web/src/composables/useKimiWebClient.ts @@ -1137,14 +1137,16 @@ async function syncSessionFromSnapshot(sessionId: string): Promise Date: Mon, 6 Jul 2026 14:21:43 +0800 Subject: [PATCH 05/11] fix(web): let resync snapshots bypass the reopen staleness guard --- .../src/composables/useKimiWebClient.ts | 21 +++++++++++++------ 1 file changed, 15 insertions(+), 6 deletions(-) diff --git a/apps/kimi-web/src/composables/useKimiWebClient.ts b/apps/kimi-web/src/composables/useKimiWebClient.ts index ec8635c15e..3799b6305b 100644 --- a/apps/kimi-web/src/composables/useKimiWebClient.ts +++ b/apps/kimi-web/src/composables/useKimiWebClient.ts @@ -1125,7 +1125,10 @@ async function pullSessionWarnings(sessionId: string): Promise { } } -async function syncSessionFromSnapshot(sessionId: string): Promise { +async function syncSessionFromSnapshot( + sessionId: string, + opts?: { force?: boolean }, +): Promise { // Preconditions captured before the await, used to detect a newer local // prompt/seq or optimistic send that appears while the snapshot is in flight // (see guard below). @@ -1143,10 +1146,12 @@ async function syncSessionFromSnapshot(sessionId: string): Promise + syncSessionFromSnapshot(sid, { force: true }), +); function hasLoadedMessages(sessionId: string): boolean { return Object.prototype.hasOwnProperty.call(rawState.messagesBySession, sessionId); From c12c42378c818319b70bcf15c2ef892da632afd9 Mon Sep 17 00:00:00 2001 From: qer Date: Mon, 6 Jul 2026 14:32:38 +0800 Subject: [PATCH 06/11] fix(web): force-apply the snapshot after an undo --- apps/kimi-web/src/composables/client/useWorkspaceState.ts | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/apps/kimi-web/src/composables/client/useWorkspaceState.ts b/apps/kimi-web/src/composables/client/useWorkspaceState.ts index 50e07ea3ea..6140c013a6 100644 --- a/apps/kimi-web/src/composables/client/useWorkspaceState.ts +++ b/apps/kimi-web/src/composables/client/useWorkspaceState.ts @@ -122,7 +122,7 @@ export interface UseWorkspaceStateDeps { ) => void; nextOptimisticMsgId: () => string; getEventConn: () => KimiEventConnection | null; - syncSessionFromSnapshot: (sessionId: string) => Promise; + syncSessionFromSnapshot: (sessionId: string, opts?: { force?: boolean }) => Promise; reopenSession: (sessionId: string) => Promise; hasLoadedMessages: (sessionId: string) => boolean; refreshSessionStatus: (sessionId: string) => Promise; @@ -1762,7 +1762,7 @@ export function useWorkspaceState(rawState: ExtendedState, deps: UseWorkspaceSta })(); try { await getKimiWebApi().undoSession(sid, count); - await syncSessionFromSnapshot(sid); + await syncSessionFromSnapshot(sid, { force: true }); return lastUserText; } catch (err) { pushOperationFailure('undo', err, { sessionId: sid }); From 57b06efe877a10103f0c112a73048d3678155e83 Mon Sep 17 00:00:00 2001 From: qer Date: Mon, 6 Jul 2026 14:45:05 +0800 Subject: [PATCH 07/11] fix(web): gate session reopen on durable seq instead of updatedAt --- .../src/composables/useKimiWebClient.ts | 30 +++++++++++-------- 1 file changed, 17 insertions(+), 13 deletions(-) diff --git a/apps/kimi-web/src/composables/useKimiWebClient.ts b/apps/kimi-web/src/composables/useKimiWebClient.ts index 3799b6305b..383d3a89a6 100644 --- a/apps/kimi-web/src/composables/useKimiWebClient.ts +++ b/apps/kimi-web/src/composables/useKimiWebClient.ts @@ -541,7 +541,7 @@ function forgetSession(sessionId: string): void { delete rawState.messagesHasMoreBySession[sessionId]; delete rawState.messagesLoadMoreErrorBySession[sessionId]; delete epochBySession[sessionId]; - lastLoadedUpdatedAt.delete(sessionId); + lastLoadedSeq.delete(sessionId); sessionsKnownEmpty.delete(sessionId); // In-flight / queued prompt state: drop these too so a queued follow-up // can't be submitted to a session that was just archived when its turn later @@ -905,11 +905,13 @@ function connectEventsIfNeeded(): void { // reactive — only consulted when building the subscribe cursor. const epochBySession: Record = {}; -// Session `updatedAt` (ISO string) captured the last time we loaded this +// Durable `seq` (event journal offset) captured the last time we loaded this // session's messages from a snapshot. Compared on re-open to skip the snapshot -// fetch + transcript remount when the session has not changed since. Not -// reactive — only read inside reopenSession. -const lastLoadedUpdatedAt = new Map(); +// fetch + transcript remount when nothing durable happened since. Not reactive +// — only read inside reopenSession. Preferred over `updatedAt`: every durable +// event (including `turn.ended`, which does not bump `updatedAt`) advances seq, +// so it can never miss a change that matters here. +const lastLoadedSeq = new Map(); // Sessions created locally in this client instance are known to be empty until // they receive their first message. This is more reliable than the daemon's @@ -1203,7 +1205,7 @@ async function syncSessionFromSnapshot( [sessionId]: snap.asOfSeq, }; epochBySession[sessionId] = snap.epoch; - lastLoadedUpdatedAt.set(sessionId, snap.session.updatedAt); + lastLoadedSeq.set(sessionId, snap.asOfSeq); connectEventsIfNeeded(); if (eventConn) { @@ -1326,18 +1328,20 @@ function resubscribeSessionEvents(sessionId: string): void { * * But rebuilding on EVERY re-open remounts the transcript (scroll reset + * entrance-animation replay) even when nothing changed — a daily annoyance. - * The session `updatedAt` is a zero-fetch, always-fresh signal: it advances on - * the durable `turn.ended` at the latest, so it can never miss a content - * change that matters here. Gate on it: + * Gate on the durable event `seq` captured at the last snapshot load: every + * durable event (including `turn.ended`, which does NOT bump `updatedAt`) + * advances it, so unlike `updatedAt` it can never miss a content change — + * including a mid-turn snapshot followed by a lost tail whose reconnect replay + * is only `turn.ended`. * - evicted subscription → cursor untrustworthy, always rebuild; - * - updatedAt unchanged → cheap cursor resubscribe, no remount; - * - updatedAt advanced → rebuild from snapshot, restoring any lost tail. */ + * - seq unchanged → cheap cursor resubscribe, no remount; + * - seq advanced → rebuild from snapshot, restoring any lost tail. */ async function reopenSession(sessionId: string): Promise { if (sessionsWithStaleCursor.has(sessionId)) { return syncSessionFromSnapshot(sessionId); } - const current = rawState.sessions.find((s) => s.id === sessionId)?.updatedAt; - if (current !== undefined && current === lastLoadedUpdatedAt.get(sessionId)) { + const currentSeq = rawState.lastSeqBySession[sessionId] ?? 0; + if (currentSeq === (lastLoadedSeq.get(sessionId) ?? 0)) { resubscribeSessionEvents(sessionId); return 'ok'; } From 49609017a88c34c124ed4444329c789d654c7b1f Mon Sep 17 00:00:00 2001 From: qer Date: Mon, 6 Jul 2026 15:24:12 +0800 Subject: [PATCH 08/11] refactor(web): always rebuild reopened sessions from a snapshot --- .../src/composables/useKimiWebClient.ts | 55 ++----------------- 1 file changed, 6 insertions(+), 49 deletions(-) diff --git a/apps/kimi-web/src/composables/useKimiWebClient.ts b/apps/kimi-web/src/composables/useKimiWebClient.ts index 383d3a89a6..6553427fdd 100644 --- a/apps/kimi-web/src/composables/useKimiWebClient.ts +++ b/apps/kimi-web/src/composables/useKimiWebClient.ts @@ -541,7 +541,6 @@ function forgetSession(sessionId: string): void { delete rawState.messagesHasMoreBySession[sessionId]; delete rawState.messagesLoadMoreErrorBySession[sessionId]; delete epochBySession[sessionId]; - lastLoadedSeq.delete(sessionId); sessionsKnownEmpty.delete(sessionId); // In-flight / queued prompt state: drop these too so a queued follow-up // can't be submitted to a session that was just archived when its turn later @@ -905,14 +904,6 @@ function connectEventsIfNeeded(): void { // reactive — only consulted when building the subscribe cursor. const epochBySession: Record = {}; -// Durable `seq` (event journal offset) captured the last time we loaded this -// session's messages from a snapshot. Compared on re-open to skip the snapshot -// fetch + transcript remount when nothing durable happened since. Not reactive -// — only read inside reopenSession. Preferred over `updatedAt`: every durable -// event (including `turn.ended`, which does not bump `updatedAt`) advances seq, -// so it can never miss a change that matters here. -const lastLoadedSeq = new Map(); - // Sessions created locally in this client instance are known to be empty until // they receive their first message. This is more reliable than the daemon's // messageCount field, which can be stale for old sessions and would otherwise @@ -1205,7 +1196,6 @@ async function syncSessionFromSnapshot( [sessionId]: snap.asOfSeq, }; epochBySession[sessionId] = snap.epoch; - lastLoadedSeq.set(sessionId, snap.asOfSeq); connectEventsIfNeeded(); if (eventConn) { @@ -1301,50 +1291,17 @@ function dropWsSubscription(sessionId: string): void { sessionsWithStaleCursor.delete(sessionId); } -/** Re-subscribe to live events from the tracked cursor (cheap, no remount). - * Used on re-open when the session provably has not changed since we last - * loaded it, so there is nothing new to rebuild from a snapshot. */ -function resubscribeSessionEvents(sessionId: string): void { - connectEventsIfNeeded(); - if (eventConn) { - // Apply any queued streaming deltas before re-subscribing so the transcript - // is current. (Volatile — never replayed by the server, but flushing here is - // cheap and future-proofs the cursor if the batching set ever changes.) - enqueueEvent.flush(); - const seq = rawState.lastSeqBySession[sessionId] ?? 0; - const epoch = epochBySession[sessionId]; - eventConn.subscribe(sessionId, { seq, epoch }); - retainWsSubscription(sessionId); - } -} - -/** Re-open an already-loaded session. +/** Re-open an already-loaded session: always rebuild from a fresh snapshot. * * Volatile `assistant.delta` frames are never journaled or replayed: if a * transport hiccup covered the tail of a turn while the user was away, the * local transcript silently lost the model's final text, and a cursor - * resubscribe has nothing to recover it with. So re-open must be able to - * rebuild from the authoritative snapshot. - * - * But rebuilding on EVERY re-open remounts the transcript (scroll reset + - * entrance-animation replay) even when nothing changed — a daily annoyance. - * Gate on the durable event `seq` captured at the last snapshot load: every - * durable event (including `turn.ended`, which does NOT bump `updatedAt`) - * advances it, so unlike `updatedAt` it can never miss a content change — - * including a mid-turn snapshot followed by a lost tail whose reconnect replay - * is only `turn.ended`. - * - evicted subscription → cursor untrustworthy, always rebuild; - * - seq unchanged → cheap cursor resubscribe, no remount; - * - seq advanced → rebuild from snapshot, restoring any lost tail. */ + * resubscribe has nothing to recover it with. Fetching the authoritative + * snapshot on every re-open keeps the logic trivially correct (no freshness + * heuristics); the snapshot is cheap server-side (LRU on the wire file), and + * the staleness guard inside syncSessionFromSnapshot discards the fetched + * snapshot if newer local activity (a send / live events) raced the GET. */ async function reopenSession(sessionId: string): Promise { - if (sessionsWithStaleCursor.has(sessionId)) { - return syncSessionFromSnapshot(sessionId); - } - const currentSeq = rawState.lastSeqBySession[sessionId] ?? 0; - if (currentSeq === (lastLoadedSeq.get(sessionId) ?? 0)) { - resubscribeSessionEvents(sessionId); - return 'ok'; - } return syncSessionFromSnapshot(sessionId); } From a5e312c793bca8ad8022257a1fa533015873c716 Mon Sep 17 00:00:00 2001 From: qer Date: Mon, 6 Jul 2026 15:47:24 +0800 Subject: [PATCH 09/11] fix(web): sharpen reopen snapshot discard and skip rebuilds mid-stream --- .../src/composables/useKimiWebClient.ts | 50 +++++++++++++------ 1 file changed, 34 insertions(+), 16 deletions(-) diff --git a/apps/kimi-web/src/composables/useKimiWebClient.ts b/apps/kimi-web/src/composables/useKimiWebClient.ts index 6553427fdd..8eb70359ab 100644 --- a/apps/kimi-web/src/composables/useKimiWebClient.ts +++ b/apps/kimi-web/src/composables/useKimiWebClient.ts @@ -1122,10 +1122,9 @@ async function syncSessionFromSnapshot( sessionId: string, opts?: { force?: boolean }, ): Promise { - // Preconditions captured before the await, used to detect a newer local - // prompt/seq or optimistic send that appears while the snapshot is in flight - // (see guard below). - const seqBefore = rawState.lastSeqBySession[sessionId] ?? 0; + // Preconditions captured before the await, used to detect an optimistic send + // or prompt change that appears while the snapshot is in flight (see guard + // below). const promptBefore = rawState.promptIdBySession[sessionId]; const inFlightBefore = inFlightPromptSessions.has(sessionId); const wasLoaded = hasLoadedMessages(sessionId); @@ -1135,17 +1134,22 @@ async function syncSessionFromSnapshot( // Staleness guard, scoped to the re-open path (session already loaded AND // not evicted from the subscription cap): there the composer stays usable, so - // a send or a broadcast global event can race this GET and produce a snapshot - // whose `asOfSeq` predates newer local state — replacing messages would wipe a - // live turn (volatile deltas are not replayable) or the just-sent optimistic - // user message. On first open we must always install + subscribe, even if a - // global event advanced lastSeq during the await; an evicted session is - // already unsubscribed, so it must always rebuild + re-subscribe; and a forced - // resync (delta-gap recovery) must always apply the authoritative snapshot, - // even though the triggering cursor advance is exactly what the guard watches. - // When discarded, the live stream / next re-open reconciles. + // a send can race this GET — replacing messages with a snapshot whose + // `asOfSeq` predates newer local state would wipe a live turn (volatile + // deltas are not replayable) or the just-sent optimistic user message. + // - seq: discard only when the LOCAL cursor is ahead of `snap.asOfSeq` — + // a durable event that raced the GET but is already included in the + // snapshot must not cancel the install (that would skip the repair this + // path exists for). + // - prompt / optimistic send: a submission that started mid-fetch is not + // in the snapshot yet, so discard. + // First opens always install + subscribe, even if events advanced lastSeq + // during the await; an evicted session is already unsubscribed, so it must + // always rebuild + re-subscribe; and a forced resync (delta-gap recovery) or + // post-undo sync must always apply the authoritative snapshot. When + // discarded, the live stream / next re-open reconciles. if (!opts?.force && wasLoaded && !sessionsWithStaleCursor.has(sessionId)) { - if ((rawState.lastSeqBySession[sessionId] ?? 0) !== seqBefore) return 'ok'; + if ((rawState.lastSeqBySession[sessionId] ?? 0) > snap.asOfSeq) return 'ok'; if (rawState.promptIdBySession[sessionId] !== promptBefore) return 'ok'; if (inFlightPromptSessions.has(sessionId) !== inFlightBefore) return 'ok'; } @@ -1291,7 +1295,7 @@ function dropWsSubscription(sessionId: string): void { sessionsWithStaleCursor.delete(sessionId); } -/** Re-open an already-loaded session: always rebuild from a fresh snapshot. +/** Re-open an already-loaded session: rebuild from a fresh snapshot. * * Volatile `assistant.delta` frames are never journaled or replayed: if a * transport hiccup covered the tail of a turn while the user was away, the @@ -1300,8 +1304,22 @@ function dropWsSubscription(sessionId: string): void { * snapshot on every re-open keeps the logic trivially correct (no freshness * heuristics); the snapshot is cheap server-side (LRU on the wire file), and * the staleness guard inside syncSessionFromSnapshot discards the fetched - * snapshot if newer local activity (a send / live events) raced the GET. */ + * snapshot if newer local activity (a send / live events) raced the GET. + * + * Exception: a RUNNING session that is still subscribed keeps its live stream. + * Rebuilding mid-stream can drop volatile deltas that arrived after the + * snapshot was assembled (queued deltas are flushed onto the old transcript and + * then overwritten, with no replay path). The live stream is already feeding + * this session; any lost tail is repaired on the next idle re-open. Evicted + * sessions are NOT exempt — they have no live stream, so the snapshot (with its + * seeded in-flight turn) is strictly better than staying unsubscribed. */ async function reopenSession(sessionId: string): Promise { + const running = + rawState.sessions.find((s) => s.id === sessionId)?.status === 'running' || + inFlightPromptSessions.has(sessionId); + if (running && !sessionsWithStaleCursor.has(sessionId)) { + return 'ok'; + } return syncSessionFromSnapshot(sessionId); } From dc8b992fab627d6ab2359e02f07cb4285ff9a436 Mon Sep 17 00:00:00 2001 From: qer Date: Mon, 6 Jul 2026 20:56:20 +0800 Subject: [PATCH 10/11] refactor(web): unconditionally apply session snapshots, drop the staleness guard --- .../composables/client/useWorkspaceState.ts | 4 +- .../src/composables/useKimiWebClient.ts | 66 +++---------------- 2 files changed, 11 insertions(+), 59 deletions(-) diff --git a/apps/kimi-web/src/composables/client/useWorkspaceState.ts b/apps/kimi-web/src/composables/client/useWorkspaceState.ts index 6140c013a6..50e07ea3ea 100644 --- a/apps/kimi-web/src/composables/client/useWorkspaceState.ts +++ b/apps/kimi-web/src/composables/client/useWorkspaceState.ts @@ -122,7 +122,7 @@ export interface UseWorkspaceStateDeps { ) => void; nextOptimisticMsgId: () => string; getEventConn: () => KimiEventConnection | null; - syncSessionFromSnapshot: (sessionId: string, opts?: { force?: boolean }) => Promise; + syncSessionFromSnapshot: (sessionId: string) => Promise; reopenSession: (sessionId: string) => Promise; hasLoadedMessages: (sessionId: string) => boolean; refreshSessionStatus: (sessionId: string) => Promise; @@ -1762,7 +1762,7 @@ export function useWorkspaceState(rawState: ExtendedState, deps: UseWorkspaceSta })(); try { await getKimiWebApi().undoSession(sid, count); - await syncSessionFromSnapshot(sid, { force: true }); + await syncSessionFromSnapshot(sid); return lastUserText; } catch (err) { pushOperationFailure('undo', err, { sessionId: sid }); diff --git a/apps/kimi-web/src/composables/useKimiWebClient.ts b/apps/kimi-web/src/composables/useKimiWebClient.ts index 8eb70359ab..001d6e04f6 100644 --- a/apps/kimi-web/src/composables/useKimiWebClient.ts +++ b/apps/kimi-web/src/composables/useKimiWebClient.ts @@ -1118,42 +1118,11 @@ async function pullSessionWarnings(sessionId: string): Promise { } } -async function syncSessionFromSnapshot( - sessionId: string, - opts?: { force?: boolean }, -): Promise { - // Preconditions captured before the await, used to detect an optimistic send - // or prompt change that appears while the snapshot is in flight (see guard - // below). - const promptBefore = rawState.promptIdBySession[sessionId]; - const inFlightBefore = inFlightPromptSessions.has(sessionId); - const wasLoaded = hasLoadedMessages(sessionId); +async function syncSessionFromSnapshot(sessionId: string): Promise { try { const api = getKimiWebApi(); const snap = await api.getSessionSnapshot(sessionId); - // Staleness guard, scoped to the re-open path (session already loaded AND - // not evicted from the subscription cap): there the composer stays usable, so - // a send can race this GET — replacing messages with a snapshot whose - // `asOfSeq` predates newer local state would wipe a live turn (volatile - // deltas are not replayable) or the just-sent optimistic user message. - // - seq: discard only when the LOCAL cursor is ahead of `snap.asOfSeq` — - // a durable event that raced the GET but is already included in the - // snapshot must not cancel the install (that would skip the repair this - // path exists for). - // - prompt / optimistic send: a submission that started mid-fetch is not - // in the snapshot yet, so discard. - // First opens always install + subscribe, even if events advanced lastSeq - // during the await; an evicted session is already unsubscribed, so it must - // always rebuild + re-subscribe; and a forced resync (delta-gap recovery) or - // post-undo sync must always apply the authoritative snapshot. When - // discarded, the live stream / next re-open reconciles. - if (!opts?.force && wasLoaded && !sessionsWithStaleCursor.has(sessionId)) { - if ((rawState.lastSeqBySession[sessionId] ?? 0) > snap.asOfSeq) return 'ok'; - if (rawState.promptIdBySession[sessionId] !== promptBefore) return 'ok'; - if (inFlightPromptSessions.has(sessionId) !== inFlightBefore) return 'ok'; - } - // Drain any queued streaming deltas before the snapshot replaces // messagesBySession[sessionId]. The snapshot is authoritative (it already // contains everything up to asOfSeq); applying stale queued deltas on top @@ -1226,11 +1195,7 @@ async function syncSessionFromSnapshot( } } -// Resync (delta-gap recovery) must always apply the authoritative snapshot, so -// it bypasses the staleness guard that only the re-open path needs. -const snapshotSyncRunner = createCoalescedAsyncRunner((sid) => - syncSessionFromSnapshot(sid, { force: true }), -); +const snapshotSyncRunner = createCoalescedAsyncRunner(syncSessionFromSnapshot); function hasLoadedMessages(sessionId: string): boolean { return Object.prototype.hasOwnProperty.call(rawState.messagesBySession, sessionId); @@ -1295,31 +1260,18 @@ function dropWsSubscription(sessionId: string): void { sessionsWithStaleCursor.delete(sessionId); } -/** Re-open an already-loaded session: rebuild from a fresh snapshot. +/** Re-open an already-loaded session: always rebuild from a fresh snapshot. * * Volatile `assistant.delta` frames are never journaled or replayed: if a * transport hiccup covered the tail of a turn while the user was away, the * local transcript silently lost the model's final text, and a cursor - * resubscribe has nothing to recover it with. Fetching the authoritative - * snapshot on every re-open keeps the logic trivially correct (no freshness - * heuristics); the snapshot is cheap server-side (LRU on the wire file), and - * the staleness guard inside syncSessionFromSnapshot discards the fetched - * snapshot if newer local activity (a send / live events) raced the GET. - * - * Exception: a RUNNING session that is still subscribed keeps its live stream. - * Rebuilding mid-stream can drop volatile deltas that arrived after the - * snapshot was assembled (queued deltas are flushed onto the old transcript and - * then overwritten, with no replay path). The live stream is already feeding - * this session; any lost tail is repaired on the next idle re-open. Evicted - * sessions are NOT exempt — they have no live stream, so the snapshot (with its - * seeded in-flight turn) is strictly better than staying unsubscribed. */ + * resubscribe has nothing to recover it with. Always fetching the authoritative + * snapshot keeps the logic trivially correct (no freshness heuristics, no + * races to reason about); the snapshot is cheap server-side (LRU on the wire + * file). Trade-off: a snapshot GET in flight during a steep local send can + * momentarily overwrite that optimistic message — the user notices immediately + * and the next re-open (or a refresh) reconciles. */ async function reopenSession(sessionId: string): Promise { - const running = - rawState.sessions.find((s) => s.id === sessionId)?.status === 'running' || - inFlightPromptSessions.has(sessionId); - if (running && !sessionsWithStaleCursor.has(sessionId)) { - return 'ok'; - } return syncSessionFromSnapshot(sessionId); } From 26d60df9b7fd415c5aefbb3fc44cf003b205e67e Mon Sep 17 00:00:00 2001 From: qer Date: Mon, 6 Jul 2026 21:19:27 +0800 Subject: [PATCH 11/11] fix(web): preserve loaded older messages when reopen snapshots apply --- .../src/composables/useKimiWebClient.ts | 8 +++- apps/kimi-web/src/lib/snapshotMessages.ts | 27 ++++++++++++ apps/kimi-web/test/lib-logic.test.ts | 42 ++++++++++++++++++- 3 files changed, 75 insertions(+), 2 deletions(-) create mode 100644 apps/kimi-web/src/lib/snapshotMessages.ts diff --git a/apps/kimi-web/src/composables/useKimiWebClient.ts b/apps/kimi-web/src/composables/useKimiWebClient.ts index 001d6e04f6..123bda2599 100644 --- a/apps/kimi-web/src/composables/useKimiWebClient.ts +++ b/apps/kimi-web/src/composables/useKimiWebClient.ts @@ -13,6 +13,7 @@ import { type WorkspaceSortMode, } from '../lib/workspaceOrder'; import { mergeWorkspaces } from '../lib/mergeWorkspaces'; +import { mergeSnapshotMessages } from '../lib/snapshotMessages'; import { createCoalescedAsyncRunner } from '../lib/snapshotSync'; import { loadUnread, @@ -1137,7 +1138,12 @@ async function syncSessionFromSnapshot(sessionId: string): Promise { + const createdAtMs = Date.parse(message.createdAt); + return !Number.isNaN(createdAtMs) && createdAtMs < earliestSnapshotMs; + }); + + return older.length > 0 ? [...older, ...snapshot] : snapshot; +} diff --git a/apps/kimi-web/test/lib-logic.test.ts b/apps/kimi-web/test/lib-logic.test.ts index d04269b014..00ca0f0132 100644 --- a/apps/kimi-web/test/lib-logic.test.ts +++ b/apps/kimi-web/test/lib-logic.test.ts @@ -8,6 +8,7 @@ import { parseDiff } from '../src/lib/parseDiff'; import { buildDiffLines } from '../src/lib/diffLines'; import { buildEditDiffLines } from '../src/lib/toolDiff'; import { createCoalescedAsyncRunner } from '../src/lib/snapshotSync'; +import { mergeSnapshotMessages } from '../src/lib/snapshotMessages'; import { normalizeToolName, toolSummary } from '../src/lib/toolMeta'; import { coerceThinkingForModel, @@ -17,7 +18,7 @@ import { modelThinkingAvailability, segmentsFor, } from '../src/lib/modelThinking'; -import type { AppModel } from '../src/api/types'; +import type { AppMessage, AppModel } from '../src/api/types'; import { resolveToolRenderer } from '../src/components/chat/tool-calls/toolRegistry'; import AgentTool from '../src/components/chat/tool-calls/AgentTool.vue'; import EditTool from '../src/components/chat/tool-calls/EditTool.vue'; @@ -369,3 +370,42 @@ describe('modelThinking', () => { }); }); }); + +describe('mergeSnapshotMessages', () => { + function msg(id: string, createdAt: string): AppMessage { + return { id, sessionId: 's1', role: 'assistant', content: [], createdAt }; + } + + it('keeps loaded messages older than the snapshot window', () => { + const loaded = [ + msg('old-1', '2026-01-01T00:00:00.000Z'), + msg('old-2', '2026-01-02T00:00:00.000Z'), + msg('recent-live', '2026-01-03T00:00:00.000Z'), + ]; + const snapshot = [ + msg('m0', '2026-01-03T00:00:00.000Z'), + msg('m1', '2026-01-04T00:00:00.000Z'), + ]; + expect(mergeSnapshotMessages(loaded, snapshot).map((m) => m.id)).toEqual([ + 'old-1', + 'old-2', + 'm0', + 'm1', + ]); + }); + + it('returns the snapshot when there is no older loaded prefix', () => { + const loaded = [msg('recent-live', '2026-01-03T00:00:00.000Z')]; + const snapshot = [ + msg('m0', '2026-01-03T00:00:00.000Z'), + msg('m1', '2026-01-04T00:00:00.000Z'), + ]; + expect(mergeSnapshotMessages(loaded, snapshot)).toBe(snapshot); + }); + + it('returns the snapshot when either side is empty', () => { + const snapshot = [msg('m0', '2026-01-03T00:00:00.000Z')]; + expect(mergeSnapshotMessages([], snapshot)).toBe(snapshot); + expect(mergeSnapshotMessages(snapshot, [])).toEqual([]); + }); +});