diff options
Diffstat (limited to 'src')
| -rw-r--r-- | src/adapters/idb/index.test.ts | 15 | ||||
| -rw-r--r-- | src/adapters/idb/index.ts | 23 | ||||
| -rw-r--r-- | src/adapters/ws/logic.test.ts | 15 | ||||
| -rw-r--r-- | src/adapters/ws/logic.ts | 3 | ||||
| -rw-r--r-- | src/app/App.test.ts | 8 | ||||
| -rw-r--r-- | src/app/store.svelte.ts | 48 | ||||
| -rw-r--r-- | src/app/store.test.ts | 36 | ||||
| -rw-r--r-- | src/core/chunks/trim.test.ts | 21 | ||||
| -rw-r--r-- | src/core/chunks/trim.ts | 22 | ||||
| -rw-r--r-- | src/core/wire/conformance.test.ts | 4 | ||||
| -rw-r--r-- | src/features/chat/store.svelte.ts | 122 | ||||
| -rw-r--r-- | src/features/chat/store.test.ts | 103 | ||||
| -rw-r--r-- | src/features/chat/test-helpers.ts | 21 | ||||
| -rw-r--r-- | src/features/conversation-cache/cache.test.ts | 19 | ||||
| -rw-r--r-- | src/features/conversation-cache/cache.ts | 56 | ||||
| -rw-r--r-- | src/features/conversation-cache/types.ts | 4 | ||||
| -rw-r--r-- | src/features/layout/adapter/http.test.ts | 2 |
17 files changed, 440 insertions, 82 deletions
diff --git a/src/adapters/idb/index.test.ts b/src/adapters/idb/index.test.ts index 04e2177..aa56f15 100644 --- a/src/adapters/idb/index.test.ts +++ b/src/adapters/idb/index.test.ts @@ -128,3 +128,18 @@ describe("createIdbChunkStore", () => { expect(loaded2[0]?.chunk).toEqual(textChunk("b1")); }); }); + +it("persists epoch across adapter recreation and append, including empty histories", async () => { + const indexedDB = new IDBFactory(); + const store = createIdbChunkStore({ indexedDB }); + await store.append("c", [makeChunk(1, "legacy")]); + expect(await store.historyEpoch("c")).toBeUndefined(); + await store.reset("c", 2); + const reopened = createIdbChunkStore({ indexedDB }); + expect(await reopened.load("c")).toEqual([]); + expect(await reopened.historyEpoch("c")).toBe(2); + await reopened.append("c", [makeChunk(101, "new")]); + expect(await reopened.historyEpoch("c")).toBe(2); + await reopened.delete("c"); + expect(await reopened.historyEpoch("c")).toBeUndefined(); +}); diff --git a/src/adapters/idb/index.ts b/src/adapters/idb/index.ts index 3dffef9..8b1848a 100644 --- a/src/adapters/idb/index.ts +++ b/src/adapters/idb/index.ts @@ -20,6 +20,7 @@ interface ChunkRecord { interface MetaRecord { conversationId: string; lastAccess: number; + historyEpoch?: number; } export interface CreateIdbChunkStoreOptions { @@ -92,6 +93,26 @@ export function createIdbChunkStore(opts?: CreateIdbChunkStoreOptions): Conversa } return { + async historyEpoch(conversationId) { + const db = await getDb(); + const tx = db.transaction(META_STORE, "readonly"); + const meta = await requestToPromise<MetaRecord | undefined>( + tx.objectStore(META_STORE).get(conversationId), + ); + await txComplete(tx); + return meta?.historyEpoch; + }, + async reset(conversationId, historyEpoch) { + const db = await getDb(); + const tx = db.transaction([CHUNKS_STORE, META_STORE], "readwrite"); + tx.objectStore(CHUNKS_STORE).delete(keyRangeFor(conversationId)); + tx.objectStore(META_STORE).put({ + conversationId, + historyEpoch, + lastAccess: Date.now(), + } satisfies MetaRecord); + await txComplete(tx); + }, async load(conversationId: string): Promise<readonly StoredChunk[]> { const db = await getDb(); const tx = db.transaction(CHUNKS_STORE, "readonly"); @@ -122,7 +143,9 @@ export function createIdbChunkStore(opts?: CreateIdbChunkStoreOptions): Conversa } satisfies ChunkRecord); } + const meta = await requestToPromise<MetaRecord | undefined>(metaStore.get(conversationId)); metaStore.put({ + ...meta, conversationId, lastAccess: Date.now(), } satisfies MetaRecord); diff --git a/src/adapters/ws/logic.test.ts b/src/adapters/ws/logic.test.ts index 2179ad0..4be2cd6 100644 --- a/src/adapters/ws/logic.test.ts +++ b/src/adapters/ws/logic.test.ts @@ -562,3 +562,18 @@ describe("parseServerMessage — subagent tab groups", () => { ).toBeNull(); }); }); + +it("preserves a compaction epoch and rejects absent or invalid epochs", () => { + const msg = { + type: "conversation.compacted", + conversationId: "c", + newConversationId: "archive", + messagesSummarized: 2, + messagesKept: 1, + historyEpoch: 7, + }; + expect(parseServerMessage(JSON.stringify(msg))).toEqual(msg); + for (const historyEpoch of [undefined, -1, 1.5, "7"]) { + expect(parseServerMessage(JSON.stringify({ ...msg, historyEpoch }))).toBeNull(); + } +}); diff --git a/src/adapters/ws/logic.ts b/src/adapters/ws/logic.ts index cecc66c..858deaa 100644 --- a/src/adapters/ws/logic.ts +++ b/src/adapters/ws/logic.ts @@ -180,7 +180,10 @@ export function parseServerMessage(data: string): WsServerMessage | null { if (typeof parsed.newConversationId !== "string") return null; if (typeof parsed.messagesSummarized !== "number") return null; if (typeof parsed.messagesKept !== "number") return null; + if (!Number.isSafeInteger(parsed.historyEpoch) || (parsed.historyEpoch as number) < 0) + return null; const msg: ConversationCompactedMessage = { + historyEpoch: parsed.historyEpoch as number, type: "conversation.compacted", conversationId: parsed.conversationId, newConversationId: parsed.newConversationId, diff --git a/src/app/App.test.ts b/src/app/App.test.ts index 26f5cdb..183b8b4 100644 --- a/src/app/App.test.ts +++ b/src/app/App.test.ts @@ -104,7 +104,9 @@ function fakeFetchImpl(): typeof fetch { { status: 200 }, ); } - return new Response(JSON.stringify({ chunks: [], latestSeq: 0 }), { status: 200 }); + return new Response(JSON.stringify({ chunks: [], historyEpoch: 0, latestSeq: 0 }), { + status: 200, + }); }; } @@ -564,7 +566,9 @@ describe("App component interaction tests", () => { status: 200, }); } - return new Response(JSON.stringify({ chunks: [], latestSeq: 0 }), { status: 200 }); + return new Response(JSON.stringify({ chunks: [], historyEpoch: 0, latestSeq: 0 }), { + status: 200, + }); }; const ws = fakeSocket(); diff --git a/src/app/store.svelte.ts b/src/app/store.svelte.ts index c354e11..2eb13ef 100644 --- a/src/app/store.svelte.ts +++ b/src/app/store.svelte.ts @@ -704,7 +704,16 @@ function createHistorySync(httpBase: string, fetchImpl: typeof fetch): HistorySy if (!res.ok) { throw new Error(`History sync failed: ${res.status}`); } - return (await res.json()) as ConversationHistoryResponse; + const data = (await res.json()) as ConversationHistoryResponse; + if ( + !data || + !Array.isArray(data.chunks) || + !Number.isSafeInteger(data.historyEpoch) || + data.historyEpoch < 0 + ) { + throw new Error("Invalid history response"); + } + return data; }; } @@ -1483,6 +1492,13 @@ export function createAppStore(opts?: CreateAppStoreOptions): AppStore { if (!res.ok) return; const data = (await res.json()) as ConversationListResponse; + if (!data || !Array.isArray(data.conversations)) { + reportError( + "Failed to load open conversations", + new Error("Invalid conversations response"), + ); + return; + } // Update the status map from the authoritative backend list. const newStatuses = new Map<string, ConversationStatus>(); for (const conv of data.conversations) { @@ -1628,27 +1644,17 @@ export function createAppStore(opts?: CreateAppStoreOptions): AppStore { } }, onConversationCompacted(msg: ConversationCompactedMessage): void { - // Compaction keeps the conversation ID — the old full history is forked - // to an archive (newConversationId). Just reload the same conversation's - // history (dispose stale store + cache + re-fetch). const cid = msg.conversationId; - const wasActive = tabsStore.activeConversationId === cid; - void cache.delete(cid); - if (!chatStores.has(cid) || !loadedConversationIds.delete(cid)) return; - const store = chatStores.get(cid); - if (store !== undefined) { - store.dispose(); - } - const tab = tabsStore.tabs.find((t) => t.conversationId === cid); - const fresh = createChatFor( - cid, - tab?.model ?? activeModel, - tab?.workspaceId ?? activeWorkspaceId, - ); - chatStores.set(cid, fresh); - ensureLoaded(cid); - if (wasActive) { - refreshActiveChat(); + const store = chatStores.get(cid) ?? watchStores.get(cid); + if (store && loadedConversationIds.has(cid)) { + store.resync(msg.historyEpoch); + } else { + store?.invalidateHistory(msg.historyEpoch); + // Unseen conversations stay lazy. Invalidation is queued before their + // next load; epoch-guarded commits reject any older in-flight response. + void cache + .reset(cid, msg.historyEpoch) + .catch((err) => reportError("Failed to reset conversation cache", err)); } }, onReopen() { diff --git a/src/app/store.test.ts b/src/app/store.test.ts index f5d4723..f257f3a 100644 --- a/src/app/store.test.ts +++ b/src/app/store.test.ts @@ -195,7 +195,8 @@ function fakeFetchImpl(opts?: FakeFetchOptions): typeof fetch { } if (/^\/conversations\/[^/]+$/.test(pathname) && searchParams.has("sinceSeq")) { return Response.json( - history[url] ?? ({ chunks: [], latestSeq: 0 } satisfies ConversationHistoryResponse), + history[url] ?? + ({ chunks: [], historyEpoch: 0, latestSeq: 0 } satisfies ConversationHistoryResponse), ); } } @@ -682,6 +683,7 @@ describe("createAppStore", () => { { seq: 1, role: "user", chunk: { type: "text", text: "hi" } }, { seq: 2, role: "assistant", chunk: { type: "text", text: "hello!" } }, ], + historyEpoch: 0, latestSeq: 2, }; const base = fakeFetchImpl(); @@ -2743,6 +2745,7 @@ describe("conversation load-on-demand", () => { { seq: 1, role: "user", turnId: "t1", chunk: { type: "text", text: "hi" } }, { seq: 2, role: "assistant", turnId: "t1", chunk: { type: "text", text: "hello" } }, ], + historyEpoch: 0, latestSeq: 2, }); } @@ -2907,12 +2910,21 @@ describe("conversation load-on-demand", () => { const { store, requests } = await setup(ws); try { ws.resolveOpen(); + ws.feedServerMessage({ + type: "chat.delta", + event: { type: "turn-start", conversationId: "lazy-1", turnId: "old" }, + }); + ws.feedServerMessage({ + type: "chat.delta", + event: { type: "text-delta", conversationId: "lazy-1", turnId: "old", delta: "stale" }, + }); for (const id of ["lazy-1", "unknown", "lazy-0"]) { ws.feedServerMessage({ type: "conversation.compacted", conversationId: id, newConversationId: `${id}-archive`, messagesSummarized: 10, + historyEpoch: 0, messagesKept: 2, }); } @@ -2957,3 +2969,25 @@ describe("conversation load-on-demand", () => { } }); }); + +it.each([ + null, + {}, + { conversations: {} }, +])("reports malformed conversation lists: %j", async (body) => { + const ws = fakeSocket(); + const base = fakeFetchImpl(); + const store = createAppStore({ + socketFactory: () => ws, + indexedDB: new IDBFactory(), + localStorage: createFakeStorage(), + fetchImpl: async (input, init) => + String(input).includes("/conversations?status=") ? Response.json(body) : base(input, init), + }); + try { + ws.resolveOpen(); + await vi.waitFor(() => expect(store.fatalError).toContain("Invalid conversations response")); + } finally { + store.dispose(); + } +}); diff --git a/src/core/chunks/trim.test.ts b/src/core/chunks/trim.test.ts index 7c4bbef..24c75fd 100644 --- a/src/core/chunks/trim.test.ts +++ b/src/core/chunks/trim.test.ts @@ -146,7 +146,7 @@ describe("windowTranscript", () => { it("is the identity within the window", () => { const state = stateWith(chunks(1, 50)); expect(windowTranscript(state, 75)).toBe(state); - expect(selectHasEarlier(state)).toBe(false); + expect(selectHasEarlier(state, 1)).toBe(false); }); }); @@ -178,21 +178,21 @@ describe("restoreEarlier", () => { expect(selectHasEarlier(restored)).toBe(true); }); - it("restoring down to seq 1 reaches the contractual origin (hasEarlier clears)", () => { + it("restoring down to a known boundary clears hasEarlier", () => { const windowed = windowTranscript(stateWith(chunks(1, 100)), 75); // hidden: 1..25 const restored = restoreEarlier(windowed, chunks(1, 100), 64); expect(restored.committed).toHaveLength(100); expect(restored.committed[0]?.seq).toBe(1); expect(restored.hiddenBeforeSeq).toBe(1); // floor at the origin — inert expect(restored.hiddenThinkingCount).toBe(0); - expect(selectHasEarlier(restored)).toBe(false); + expect(selectHasEarlier(restored, 1)).toBe(false); }); it("is the identity when nothing older is known locally (server may still hold more)", () => { const windowed = windowTranscript(stateWith(chunks(50, 200)), 75); const restored = restoreEarlier(windowed, [], 64); expect(restored).toBe(windowed); - // seqs are 1-based gap-free: window starts at 126 ⇒ older chunks DO exist. + // The server boundary is unknown, so another page may exist. expect(selectHasEarlier(restored)).toBe(true); }); @@ -203,7 +203,7 @@ describe("restoreEarlier", () => { it("works on a server-windowed transcript (no local watermark)", () => { // A cold-cache fresh load with `?limit=` commits a suffix (seq 809..1000) - // with hiddenBeforeSeq still 0 — hasEarlier derives from seq > 1, and a + // with hiddenBeforeSeq still 0 — the server boundary is unknown, and a // backfilled run merges below it. const state = stateWith(chunks(809, 1000)); expect(state.hiddenBeforeSeq).toBe(0); @@ -230,6 +230,15 @@ describe("restoreEarlier", () => { const restored = restoreEarlier(trimmed, original, 1000); expect(restored.committed).toEqual(original); expect(restored.hiddenBeforeSeq).toBe(1); - expect(selectHasEarlier(restored)).toBe(false); + expect(selectHasEarlier(restored, 1)).toBe(false); }); }); + +it("uses a non-1 history boundary rather than the seq origin", () => { + const state = stateWith(chunks(101, 110)); + expect(selectHasEarlier(state, 101)).toBe(false); + expect(selectHasEarlier(state, null)).toBe(true); + const windowed = windowTranscript(state, 5); + expect(selectHasEarlier(windowed, 101)).toBe(true); + expect(selectHasEarlier(restoreEarlier(windowed, state.committed, 5), 101)).toBe(false); +}); diff --git a/src/core/chunks/trim.ts b/src/core/chunks/trim.ts index 9846357..259c101 100644 --- a/src/core/chunks/trim.ts +++ b/src/core/chunks/trim.ts @@ -150,8 +150,7 @@ function oldestLoadedSeq(state: TranscriptState): number { * `?beforeSeq=` backfill; chunks at/inside the window are ignored). The newest * `count` of them are merged back in front of `committed`, and the watermark * follows the new window start so history merges still can't resurrect what - * remains unloaded. Identity when the window already starts at seq 1 (the - * contractual origin) or nothing older is known locally. + * remains unloaded. Identity when nothing older is known locally. */ export function restoreEarlier( state: TranscriptState, @@ -159,7 +158,7 @@ export function restoreEarlier( count: number, ): TranscriptState { const oldest = oldestLoadedSeq(state); - if (oldest <= 1) return state; + if (oldest === 0) return state; const below = earlier.filter((c) => c.seq < oldest).sort((a, b) => a.seq - b.seq); if (below.length === 0) return state; const keep = below.slice(-Math.max(1, count)); @@ -172,14 +171,11 @@ export function restoreEarlier( }; } -/** - * Whether earlier history exists below the loaded window — drives the - * "Show earlier messages" affordance. Derived from the [email protected] CONTRACT - * that per-conversation seqs are 1-based and gap-free: a loaded window that - * starts above seq 1 means older chunks exist (locally cached or server-side), - * whether the window came from a local trim or a server-windowed (`?limit=`) - * fresh load. - */ -export function selectHasEarlier(state: TranscriptState): boolean { - return oldestLoadedSeq(state) > 1; +/** Earlier availability uses the known server boundary, never a seq origin. */ +export function selectHasEarlier( + state: TranscriptState, + earliestSeq: number | null = null, +): boolean { + const oldest = oldestLoadedSeq(state); + return oldest > 0 && (earliestSeq === null || oldest > earliestSeq); } diff --git a/src/core/wire/conformance.test.ts b/src/core/wire/conformance.test.ts index 2b98ca6..e74a663 100644 --- a/src/core/wire/conformance.test.ts +++ b/src/core/wire/conformance.test.ts @@ -184,6 +184,7 @@ describe("classifies every WsServerMessage type", () => { conversationId: "c1", newConversationId: "c2", messagesSummarized: 10, + historyEpoch: 0, messagesKept: 5, }, ]; @@ -268,6 +269,7 @@ describe("ConversationHistoryResponse shape is constructible", () => { it("constructs a response with chunks", () => { const resp: ConversationHistoryResponse = { chunks: [{ seq: 1, role: "user", chunk: { type: "text", text: "hi" } }], + historyEpoch: 0, latestSeq: 1, }; expect(resp.chunks).toHaveLength(1); @@ -275,7 +277,7 @@ describe("ConversationHistoryResponse shape is constructible", () => { }); it("constructs an empty (caught-up) response", () => { - const resp: ConversationHistoryResponse = { chunks: [], latestSeq: 5 }; + const resp: ConversationHistoryResponse = { chunks: [], historyEpoch: 0, latestSeq: 5 }; expect(resp.chunks).toHaveLength(0); expect(resp.latestSeq).toBe(5); }); diff --git a/src/features/chat/store.svelte.ts b/src/features/chat/store.svelte.ts index 01ad9f9..dab1ba2 100644 --- a/src/features/chat/store.svelte.ts +++ b/src/features/chat/store.svelte.ts @@ -209,7 +209,9 @@ export interface ChatStore { * running, the server's post-subscribe replay re-asserts `generating`. The * app store pairs this with a `chat.subscribe` for the conversation. */ - resync(): void; + resync(historyEpoch?: number): void; + /** Discard pre-replacement transcript state without loading an unseen conversation. */ + invalidateHistory(historyEpoch: number): void; dispose(): void; } @@ -223,6 +225,23 @@ export function createChatStore(deps: ChatStoreDependencies): ChatStore { let cacheTimer = $state<CacheTimerState>(initialCacheTimerState()); const now = deps.now ?? Date.now; let disposed = false; + let rerunRequested = false; + let historyEpoch: number | undefined; + let revision = 0; + let reloadWindow = false; + let earliestSeq = $state<number | null>(null); + + function resetTranscript(epoch: number): void { + historyEpoch = epoch; + revision++; + reloadWindow = true; + transcript = initialState(); + earliestSeq = null; + pendingEvents = []; + } + function hasEarlier(): boolean { + return selectHasEarlier(transcript, earliestSeq); + } /** The most recent turn's id (resolves a seeded turn gap when it is stamped). */ function lastTurnIdOf(m: MetricsState): string | undefined { @@ -313,21 +332,58 @@ export function createChatStore(deps: ChatStoreDependencies): ChatStore { * between the cache and the fetched window. */ async function syncTail(coldLimit?: number): Promise<void> { - if (disposed || _pendingSync) return; - _pendingSync = true; - try { - const since = await deps.cache.sinceSeq(deps.conversationId); - const window = since === 0 && coldLimit !== undefined ? { limit: coldLimit } : undefined; - const res = await deps.historySync(deps.conversationId, since, window); - const merged = await deps.cache.commit(deps.conversationId, res.chunks); - transcript = applyHistory(transcript, merged); - maybeTrim(); - _error = null; - } catch (err) { - _error = err instanceof Error ? err.message : String(err); - } finally { - _pendingSync = false; + if (disposed) return; + if (_pendingSync) { + rerunRequested = true; + return; } + _pendingSync = true; + do { + rerunRequested = false; + try { + const currentRevision = revision; + const cachedEpoch = await deps.cache.historyEpoch(deps.conversationId); + if (historyEpoch !== undefined && cachedEpoch !== historyEpoch) + await deps.cache.reset(deps.conversationId, historyEpoch); + const since = + cachedEpoch === undefined ? 0 : await deps.cache.sinceSeq(deps.conversationId); + const limit = + since === 0 ? (reloadWindow ? initialWindowSize(chatLimit) : coldLimit) : undefined; + const res = await deps.historySync( + deps.conversationId, + since, + limit === undefined ? undefined : { limit }, + ); + if (disposed) break; + if (revision !== currentRevision) { + rerunRequested = true; + continue; + } + if ( + (cachedEpoch !== undefined && cachedEpoch !== res.historyEpoch) || + (historyEpoch !== undefined && historyEpoch !== res.historyEpoch) + ) { + resetTranscript(res.historyEpoch); + await deps.cache.reset(deps.conversationId, res.historyEpoch); + if (since !== 0) { + rerunRequested = true; + continue; + } + } + reloadWindow = false; + historyEpoch = res.historyEpoch; + const merged = await deps.cache.commit(deps.conversationId, res.chunks, res.historyEpoch); + if (disposed || historyEpoch !== res.historyEpoch) continue; + if (since === 0 && (limit === undefined || res.chunks.length < limit)) + earliestSeq = merged[0]?.seq ?? 0; + transcript = applyHistory(transcript, merged); + maybeTrim(); + _error = null; + } catch (err) { + _error = err instanceof Error ? err.message : String(err); + } + } while (rerunRequested && !disposed); + _pendingSync = false; } async function syncMetrics(): Promise<void> { @@ -351,14 +407,25 @@ export function createChatStore(deps: ChatStoreDependencies): ChatStore { * newest `count` of them). Shared by `showEarlier` and the raise-refill. */ async function backfillOlder(oldest: number, want: number): Promise<readonly StoredChunk[]> { + const currentRevision = revision; let earlier = (await deps.cache.load(deps.conversationId)).filter((c) => c.seq < oldest); + if (disposed || revision !== currentRevision) return []; const oldestKnown = earlier[0]?.seq ?? oldest; - if (earlier.length < want && oldestKnown > 1) { + if (earlier.length < want && (earliestSeq === null || oldestKnown > earliestSeq)) { const res = await deps.historySync(deps.conversationId, 0, { beforeSeq: oldestKnown, limit: want - earlier.length, }); - const merged = await deps.cache.commit(deps.conversationId, res.chunks); + if (disposed || revision !== currentRevision) return []; + if (res.historyEpoch !== historyEpoch) { + resetTranscript(res.historyEpoch); + await deps.cache.reset(deps.conversationId, res.historyEpoch); + await syncTail(initialWindowSize(chatLimit)); + return []; + } + const merged = await deps.cache.commit(deps.conversationId, res.chunks, res.historyEpoch); + if (disposed || revision !== currentRevision) return []; + if (res.chunks.length < want - earlier.length) earliestSeq = merged[0]?.seq ?? 0; earlier = merged.filter((c) => c.seq < oldest); } return earlier; @@ -368,15 +435,15 @@ export function createChatStore(deps: ChatStoreDependencies): ChatStore { * Refill toward the fresh-load window after a limit RAISE: pull older * history (cache first, then server) so the loaded set grows to match what a * fresh `load()` would show at the new limit. No-op when already at the - * origin (seq 1) or already within the window. `restoreEarlier` re-derives + * known history boundary or already within the window. `restoreEarlier` re-derives * the window start at apply time, so a delta landing during the await can't * corrupt the merge. NOT gated (refilling prepends above the viewport; the * caller preserves scroll position). */ async function refill(): Promise<void> { if (disposed) return; + if (!hasEarlier()) return; const oldest = transcript.committed[0]?.seq ?? transcript.hiddenBeforeSeq; - if (oldest <= 1) return; const want = initialWindowSize(chatLimit) - transcript.committed.length; if (want <= 0) return; try { @@ -418,7 +485,7 @@ export function createChatStore(deps: ChatStoreDependencies): ChatStore { return _model; }, get hasEarlier(): boolean { - return selectHasEarlier(transcript); + return hasEarlier(); }, get thinkingKeyBase(): number { return transcript.hiddenThinkingCount; @@ -524,8 +591,11 @@ export function createChatStore(deps: ChatStoreDependencies): ChatStore { // server instead (CR-5 `?limit=`), so a huge conversation never ships // whole. The post-sync window re-asserts the cap either way. const windowSize = initialWindowSize(chatLimit); + const currentRevision = revision; + const cachedEpoch = await deps.cache.historyEpoch(deps.conversationId); const cached = await deps.cache.load(deps.conversationId); - if (cached.length > 0) { + if (!disposed && revision === currentRevision && cachedEpoch !== undefined) { + historyEpoch = cachedEpoch; transcript = windowTranscript(applyHistory(transcript, cached), windowSize); } await syncTail(windowSize); @@ -536,7 +606,7 @@ export function createChatStore(deps: ChatStoreDependencies): ChatStore { async showEarlier(): Promise<void> { if (disposed) return; const oldest = transcript.committed[0]?.seq ?? transcript.hiddenBeforeSeq; - if (oldest <= 1) return; + if (!hasEarlier()) return; const want = unloadCount(chatLimit); try { const earlier = await backfillOlder(oldest, want); @@ -547,8 +617,14 @@ export function createChatStore(deps: ChatStoreDependencies): ChatStore { } }, - resync(): void { + invalidateHistory(epoch): void { + if (!disposed && (historyEpoch === undefined || epoch > historyEpoch)) resetTranscript(epoch); + }, + + resync(epoch?: number): void { if (disposed) return; + if (epoch !== undefined && (historyEpoch === undefined || epoch > historyEpoch)) + resetTranscript(epoch); // A turn may have sealed while we were disconnected (missed `turn-sealed`): // clear the now-stale spinner BEFORE re-subscribing, so a finished turn // doesn't spin forever. A still-running turn's replay re-asserts it. diff --git a/src/features/chat/store.test.ts b/src/features/chat/store.test.ts index 8604107..bb1e304 100644 --- a/src/features/chat/store.test.ts +++ b/src/features/chat/store.test.ts @@ -1,5 +1,8 @@ import type { AgentEvent, ImageInput, StepId, StoredChunk } from "@dispatch/wire"; +import { IDBFactory, IDBKeyRange } from "fake-indexeddb"; import { describe, expect, it, vi } from "vitest"; +import { createIdbChunkStore } from "../../adapters/idb"; +import { createConversationCache } from "../conversation-cache"; import { createChatStore } from "./store.svelte"; import { createFakeCache, @@ -528,7 +531,7 @@ describe("createChatStore", () => { store.dispose(); }); - it("overlapping syncs are guarded", async () => { + it("overlapping syncs coalesce into a sequential rerun", async () => { const transport = createFakeTransport(); const _historySync = createFakeHistorySync(); const metricsSync = createFakeMetricsSync(); @@ -546,7 +549,7 @@ describe("createChatStore", () => { if (callCount === 1) { await firstSyncPromise; } - return { chunks: [], latestSeq: sinceSeq }; + return { chunks: [], latestSeq: sinceSeq, historyEpoch: 0 }; }; const store = createChatStore({ @@ -573,7 +576,7 @@ describe("createChatStore", () => { // Release the first sync resolveFirstSync?.(); - await new Promise((r) => setTimeout(r, 10)); + await vi.waitFor(() => expect(callCount).toBe(2)); store.dispose(); }); @@ -1645,6 +1648,7 @@ describe("createChatStore", () => { }); await store.load(); // only 50 chunks → all loaded, window starts at seq 1 expect(store.chunks).toHaveLength(50); + await store.showEarlier(); // Probe server exhaustion; seq 1 alone is not a boundary. expect(store.hasEarlier).toBe(false); const callsAfterLoad = historySync.calls.length; @@ -1892,3 +1896,96 @@ describe("createChatStore — cacheTimer", () => { busy.dispose(); }); }); + +describe("history epochs", () => { + it.each([ + "tail", + "backfill", + ])("recovers missed compaction on %s and exhausts non-1 history", async (path) => { + const history = createFakeHistorySync(); + const cache = createFakeCache(); + history.returnChunks = Array.from({ length: 20 }, (_, i) => makeStoredChunk(i + 1)); + const store = createChatStore({ + conversationId: CONV_ID, + transport: createFakeTransport().impl, + historySync: history.impl, + metricsSync: createFakeMetricsSync().impl, + cache: cache.impl, + chatLimit: 10, + }); + await store.load(); + history.historyEpoch = 1; + history.returnChunks = [makeStoredChunk(101), makeStoredChunk(102)]; + if (path === "tail") { + store.resync(); + await vi.waitFor(() => expect(store.chunks[0]?.seq).toBe(101)); + } else await store.showEarlier(); + expect(store.chunks.map((c) => c.seq)).toEqual([101, 102]); + expect((await cache.impl.load(CONV_ID)).map((c) => c.seq)).toEqual([101, 102]); + expect(history.calls.at(-1)).toMatchObject({ sinceSeq: 0, window: { limit: 7 } }); + expect(store.hasEarlier).toBe(false); + const calls = history.calls.length; + await store.showEarlier(); + expect(history.calls).toHaveLength(calls); + store.dispose(); + }); +}); + +it("a compaction broadcast during an in-flight fetch discards the old response", async () => { + const history = createFakeHistorySync(); + history.returnChunks = [makeStoredChunk(101)]; + let resolve!: (res: import("@dispatch/transport-contract").ConversationHistoryResponse) => void; + const first = new Promise<import("@dispatch/transport-contract").ConversationHistoryResponse>( + (r) => { + resolve = r; + }, + ); + const sync = vi + .fn<import("./ports").HistorySync>() + .mockImplementationOnce(() => first) + .mockImplementation(history.impl); + const cache = createFakeCache(); + const store = createChatStore({ + conversationId: CONV_ID, + transport: createFakeTransport().impl, + historySync: sync, + metricsSync: createFakeMetricsSync().impl, + cache: cache.impl, + }); + const loading = store.load(); + await vi.waitFor(() => expect(sync).toHaveBeenCalledTimes(1)); + history.historyEpoch = 1; + store.resync(1); + resolve({ chunks: [makeStoredChunk(1)], latestSeq: 1, historyEpoch: 0 }); + await loading; + expect(sync).toHaveBeenCalledTimes(2); + expect(store.chunks.map((c) => c.seq)).toEqual([101]); + expect((await cache.impl.load(CONV_ID)).map((c) => c.seq)).toEqual([101]); + store.dispose(); +}); + +it("reloads unknown legacy cache epochs rather than merging pre-upgrade rows", async () => { + vi.stubGlobal("IDBKeyRange", IDBKeyRange); + const storage = createIdbChunkStore({ indexedDB: new IDBFactory() }); + const cache = createConversationCache(storage); + await storage.append(CONV_ID, [makeStoredChunk(1)]); + const history = createFakeHistorySync(); + history.returnChunks = [makeStoredChunk(101)]; + const store = createChatStore({ + conversationId: CONV_ID, + transport: createFakeTransport().impl, + historySync: history.impl, + metricsSync: createFakeMetricsSync().impl, + cache, + }); + try { + await store.load(); + expect(history.calls[0]).toMatchObject({ sinceSeq: 0 }); + expect(store.chunks.map((c) => c.seq)).toEqual([101]); + expect(await cache.historyEpoch(CONV_ID)).toBe(0); + expect((await cache.load(CONV_ID)).map((c) => c.seq)).toEqual([101]); + } finally { + store.dispose(); + vi.unstubAllGlobals(); + } +}); diff --git a/src/features/chat/test-helpers.ts b/src/features/chat/test-helpers.ts index c99d1f4..c1b020b 100644 --- a/src/features/chat/test-helpers.ts +++ b/src/features/chat/test-helpers.ts @@ -43,14 +43,22 @@ export interface FakeHistorySync { readonly calls: Array<{ conversationId: string; sinceSeq: number; window?: HistoryWindow }>; /** Set the chunks to return on the next call. */ returnChunks: readonly StoredChunk[]; + historyEpoch: number; readonly impl: HistorySync; } export function createFakeHistorySync(): FakeHistorySync { const calls: Array<{ conversationId: string; sinceSeq: number; window?: HistoryWindow }> = []; let returnChunks: readonly StoredChunk[] = []; + let historyEpoch = 0; return { calls, + get historyEpoch() { + return historyEpoch; + }, + set historyEpoch(v: number) { + historyEpoch = v; + }, get returnChunks() { return returnChunks; }, @@ -72,7 +80,7 @@ export function createFakeHistorySync(): FakeHistorySync { chunks = chunks.slice(-window.limit); } const latestSeq = chunks.length > 0 ? Math.max(...chunks.map((c) => c.seq)) : sinceSeq; - return { chunks, latestSeq }; + return { chunks, latestSeq, historyEpoch }; }, }; } @@ -122,13 +130,22 @@ export interface FakeCache { export function createFakeCache(): FakeCache { const store = new Map<string, StoredChunk[]>(); + const epochs = new Map<string, number>(); return { store, impl: { + async historyEpoch(id) { + return epochs.get(id) ?? 0; + }, + async reset(id, epoch) { + store.delete(id); + epochs.set(id, epoch); + }, async load(conversationId) { return store.get(conversationId) ?? []; }, - async commit(conversationId, incoming) { + async commit(conversationId, incoming, epoch) { + if (epoch !== undefined) epochs.set(conversationId, epoch); const existing = store.get(conversationId) ?? []; const seen = new Set(existing.map((c) => c.seq)); const toAppend = incoming.filter((c) => !seen.has(c.seq)); diff --git a/src/features/conversation-cache/cache.test.ts b/src/features/conversation-cache/cache.test.ts index ce1c60c..4306d51 100644 --- a/src/features/conversation-cache/cache.test.ts +++ b/src/features/conversation-cache/cache.test.ts @@ -15,8 +15,16 @@ const chunk = (seq: number, role: "user" | "assistant" = "user"): StoredChunk => */ function createFakeStore(): ConversationChunkStore { const store = new Map<string, StoredChunk[]>(); + const epochs = new Map<string, number>(); return { + async historyEpoch(id) { + return epochs.get(id); + }, + async reset(id, epoch) { + store.delete(id); + epochs.set(id, epoch); + }, async load(conversationId) { return store.get(conversationId) ?? []; }, @@ -33,6 +41,7 @@ function createFakeStore(): ConversationChunkStore { async delete(conversationId) { store.delete(conversationId); + epochs.delete(conversationId); }, async index() { @@ -195,3 +204,13 @@ describe("cache.delete", () => { expect(result).toEqual([]); }); }); + +it("serializes invalidation and rejects stale epoch commits", async () => { + const store = createFakeStore(); + const cache = createConversationCache(store); + await store.append("c", [chunk(1)]); // Legacy rows have no trusted epoch. + expect(await cache.commit("c", [chunk(101)], 1)).toEqual([chunk(101)]); + await Promise.all([cache.commit("c", [chunk(102)], 1), cache.reset("c", 2)]); + expect(await cache.commit("c", [chunk(103)], 1)).toEqual([]); + expect(await store.historyEpoch("c")).toBe(2); +}); diff --git a/src/features/conversation-cache/cache.ts b/src/features/conversation-cache/cache.ts index 4953944..431a2a7 100644 --- a/src/features/conversation-cache/cache.ts +++ b/src/features/conversation-cache/cache.ts @@ -3,6 +3,9 @@ import { nextSinceSeq, reconcileCache, selectEvictions } from "./logic"; import type { ConversationChunkStore } from "./types"; export interface ConversationCache { + historyEpoch(conversationId: string): Promise<number | undefined>; + reset(conversationId: string, historyEpoch: number): Promise<void>; + /** Load all cached chunks for a conversation. */ load(conversationId: string): Promise<readonly StoredChunk[]>; @@ -10,7 +13,11 @@ export interface ConversationCache { * Load + reconcile + append new chunks. * Returns the merged cache (the new authoritative cache for this conversation). */ - commit(conversationId: string, incoming: readonly StoredChunk[]): Promise<readonly StoredChunk[]>; + commit( + conversationId: string, + incoming: readonly StoredChunk[], + historyEpoch?: number, + ): Promise<readonly StoredChunk[]>; /** Return the `?sinceSeq=` cursor for the next incremental sync. */ sinceSeq(conversationId: string): Promise<number>; @@ -43,21 +50,52 @@ export function createConversationCache( ): ConversationCache { const maxChunks = opts?.maxChunks ?? DEFAULT_MAX_CHUNKS; + // Serialize mutation per conversation, including invalidation, so an old fetch + // cannot resurrect rows after a compaction broadcast has invalidated them. + const pending = new Map<string, Promise<unknown>>(); + function exclusive<T>(id: string, run: () => Promise<T>): Promise<T> { + const next = (pending.get(id) ?? Promise.resolve()).catch(() => {}).then(run); + pending.set(id, next); + void next + .finally(() => { + if (pending.get(id) === next) pending.delete(id); + }) + .catch(() => {}); + return next; + } return { + historyEpoch: async (id) => { + await pending.get(id); + return store.historyEpoch(id); + }, + reset: (id, epoch) => + exclusive(id, async () => { + const current = await store.historyEpoch(id); + if (current === undefined || current < epoch) await store.reset(id, epoch); + }), async load(conversationId) { + await pending.get(conversationId); return store.load(conversationId); }, - async commit(conversationId, incoming) { - const cached = await store.load(conversationId); - const { merged, toAppend } = reconcileCache(cached, incoming); - if (toAppend.length > 0) { - await store.append(conversationId, toAppend); - } - return merged; + commit(conversationId, incoming, epoch) { + return exclusive(conversationId, async () => { + if (epoch !== undefined) { + const current = await store.historyEpoch(conversationId); + if (current !== undefined && current > epoch) return store.load(conversationId); + if (current !== epoch) await store.reset(conversationId, epoch); + } + const cached = await store.load(conversationId); + const { merged, toAppend } = reconcileCache(cached, incoming); + if (toAppend.length > 0) { + await store.append(conversationId, toAppend); + } + return merged; + }); }, async sinceSeq(conversationId) { + await pending.get(conversationId); const cached = await store.load(conversationId); return nextSinceSeq(cached); }, @@ -72,7 +110,7 @@ export function createConversationCache( }, async delete(conversationId) { - await store.delete(conversationId); + await exclusive(conversationId, () => store.delete(conversationId)); }, }; } diff --git a/src/features/conversation-cache/types.ts b/src/features/conversation-cache/types.ts index 345ab71..a273f34 100644 --- a/src/features/conversation-cache/types.ts +++ b/src/features/conversation-cache/types.ts @@ -17,6 +17,10 @@ export interface ConversationCacheIndexEntry { * All methods MUST be idempotent on `seq`: re-appending an existing seq is a no-op. */ export interface ConversationChunkStore { + historyEpoch(conversationId: string): Promise<number | undefined>; + /** Atomically replace cached rows with empty history at this epoch. */ + reset(conversationId: string, historyEpoch: number): Promise<void>; + /** Load all cached chunks for a conversation, seq-ordered. Returns [] if absent. */ load(conversationId: string): Promise<readonly StoredChunk[]>; diff --git a/src/features/layout/adapter/http.test.ts b/src/features/layout/adapter/http.test.ts index cc7d7e9..30e194a 100644 --- a/src/features/layout/adapter/http.test.ts +++ b/src/features/layout/adapter/http.test.ts @@ -77,7 +77,7 @@ describe("createLayoutHttp — getTree", () => { it("a malformed 200 (SPA-fallback JSON without a tree) degrades to notFound", async () => { const http = createLayoutHttp( "http://backend.test", - fetchReturning(200, { chunks: [], latestSeq: 0 }), + fetchReturning(200, { chunks: [], historyEpoch: 0, latestSeq: 0 }), ); const result = await http.getTree("ws-a"); expect(result).toMatchObject({ ok: false, notFound: true }); |
