From 2686cf6bcd247d17528922e36aa380a521c964e0 Mon Sep 17 00:00:00 2001 From: Adam Malczewski Date: Wed, 21 Oct 2026 02:31:38 +0900 Subject: fix(review-3.6): invalidate history caches by epoch and coalesce syncs --- src/adapters/idb/index.test.ts | 15 +++++++++++++++ src/adapters/idb/index.ts | 23 +++++++++++++++++++++++ src/adapters/ws/logic.test.ts | 15 +++++++++++++++ src/adapters/ws/logic.ts | 3 +++ 4 files changed, 56 insertions(+) (limited to 'src/adapters') 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( + 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 { 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(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, -- cgit v1.2.3 From 0bdb04b0aa0c0500e68e2e985b1a2880730ad7e8 Mon Sep 17 00:00:00 2001 From: Adam Malczewski Date: Wed, 21 Oct 2026 14:46:06 +0900 Subject: fix(review-followup-fec): reconcile epoch rollback and retain reconnect backoff --- src/adapters/ws/index.test.ts | 77 ++++++++++- src/adapters/ws/index.ts | 17 ++- src/features/chat/store.svelte.ts | 68 ++++++++-- src/features/chat/store.test.ts | 176 ++++++++++++++++++++++++++ src/features/chat/test-helpers.ts | 6 + src/features/conversation-cache/cache.test.ts | 35 +++++ src/features/conversation-cache/cache.ts | 24 +++- 7 files changed, 383 insertions(+), 20 deletions(-) (limited to 'src/adapters') diff --git a/src/adapters/ws/index.test.ts b/src/adapters/ws/index.test.ts index 9d821e2..6f5e4ef 100644 --- a/src/adapters/ws/index.test.ts +++ b/src/adapters/ws/index.test.ts @@ -6,7 +6,7 @@ interface FakeSocket extends WebSocketLike { sent: string[]; resolveOpen(): void; invokeMessage(data: string): void; - invokeClose(): void; + invokeClose(code?: number): void; } function fakeSocket(): FakeSocket { @@ -44,8 +44,8 @@ function fakeSocket(): FakeSocket { invokeMessage(data: string) { onmessage?.({ data }); }, - invokeClose() { - onclose?.({ code: 1000, reason: "" }); + invokeClose(code = 1000) { + onclose?.({ code, reason: "" }); }, sent, }; @@ -405,3 +405,74 @@ describe("createSurfaceSocket", () => { expect(JSON.parse(ws.sent[0] ?? "")).toEqual({ type: "chat.send", message: "queued" }); }); }); + +describe("reconnect backoff after open", () => { + it.each([1013, 1006, 1011])("grows and caps delays across repeated early %s closes", (code) => { + vi.useFakeTimers(); + const random = vi.spyOn(Math, "random").mockReturnValue(0.5); + const sockets: FakeSocket[] = []; + const handle = createSurfaceSocket({ + url: "ws://test", + onMessage: vi.fn(), + socketFactory: () => { + const ws = fakeSocket(); + sockets.push(ws); + return ws; + }, + }); + try { + for (const delay of [500, 1000, 2000, 4000, 8000, 16000, 30000, 30000]) { + const count = sockets.length; + sockets.at(-1)?.resolveOpen(); + vi.advanceTimersByTime(100); + sockets.at(-1)?.invokeClose(code); + vi.advanceTimersByTime(delay - 1); + expect(sockets).toHaveLength(count); + vi.advanceTimersByTime(1); + expect(sockets).toHaveLength(count + 1); + } + } finally { + handle.close(); + random.mockRestore(); + vi.useRealTimers(); + } + }); + + it.each([ + "healthy", + "normal", + "going away", + ])("resets accumulated backoff after a %s connection", (mode) => { + vi.useFakeTimers(); + const random = vi.spyOn(Math, "random").mockReturnValue(0.5); + const sockets: FakeSocket[] = []; + const handle = createSurfaceSocket({ + url: "ws://test", + onMessage: vi.fn(), + socketFactory: () => { + const ws = fakeSocket(); + sockets.push(ws); + return ws; + }, + }); + try { + sockets[0]?.resolveOpen(); + sockets[0]?.invokeClose(1013); + vi.advanceTimersByTime(500); + sockets[1]?.resolveOpen(); + if (mode === "healthy") vi.advanceTimersByTime(30000); + sockets[1]?.invokeClose(mode === "normal" ? 1000 : mode === "going away" ? 1001 : 1013); + vi.advanceTimersByTime(499); + expect(sockets).toHaveLength(2); + vi.advanceTimersByTime(1); + expect(sockets).toHaveLength(3); + handle.close(); + vi.advanceTimersByTime(60000); + expect(sockets).toHaveLength(3); + } finally { + handle.close(); + random.mockRestore(); + vi.useRealTimers(); + } + }); +}); diff --git a/src/adapters/ws/index.ts b/src/adapters/ws/index.ts index 76d9e75..eeb7901 100644 --- a/src/adapters/ws/index.ts +++ b/src/adapters/ws/index.ts @@ -67,7 +67,12 @@ export function createSurfaceSocket(opts: SurfaceSocketOptions): SurfaceSocketHa let disposed = false; let reconnectAttempt = 0; let reconnectTimer: ReturnType | null = null; + let healthyTimer: ReturnType | null = null; let isOpen = false; + function clearHealthyTimer(): void { + if (healthyTimer !== null) clearTimeout(healthyTimer); + healthyTimer = null; + } const queue: string[] = []; function connect(isReconnect: boolean): void { @@ -77,7 +82,12 @@ export function createSurfaceSocket(opts: SurfaceSocketOptions): SurfaceSocketHa socket.onopen = () => { if (disposed) return; isOpen = true; - reconnectAttempt = 0; + // Opening is not proof of recovery: replay may immediately overflow the + // server's slow-consumer bound. Only a sustained open resets the streak. + healthyTimer = setTimeout(() => { + healthyTimer = null; + reconnectAttempt = 0; + }, 30_000); for (const raw of queue.splice(0)) { socket?.send(raw); } @@ -112,8 +122,10 @@ export function createSurfaceSocket(opts: SurfaceSocketOptions): SurfaceSocketHa } }; - socket.onclose = () => { + socket.onclose = (ev) => { + clearHealthyTimer(); if (disposed) return; + if (isOpen && (ev.code === 1000 || ev.code === 1001)) reconnectAttempt = 0; isOpen = false; scheduleReconnect(); }; @@ -143,6 +155,7 @@ export function createSurfaceSocket(opts: SurfaceSocketOptions): SurfaceSocketHa }, close(): void { disposed = true; + clearHealthyTimer(); if (reconnectTimer !== null) { clearTimeout(reconnectTimer); reconnectTimer = null; diff --git a/src/features/chat/store.svelte.ts b/src/features/chat/store.svelte.ts index 6cfa484..f48d441 100644 --- a/src/features/chat/store.svelte.ts +++ b/src/features/chat/store.svelte.ts @@ -339,15 +339,28 @@ export function createChatStore(deps: ChatStoreDependencies): ChatStore { return; } _pendingSync = true; + // One request plus at most two sequential reruns. Continued churn waits for + // the next reconnect/seal/explicit resync, never a microtask refetch loop. + let attempts = 0; do { + attempts++; rerunRequested = false; try { const currentRevision = revision; - const cachedEpoch = await deps.cache.historyEpoch(deps.conversationId); - if (historyEpoch !== undefined && cachedEpoch !== historyEpoch) + let cachedEpoch = await deps.cache.historyEpoch(deps.conversationId); + if (historyEpoch !== undefined && cachedEpoch !== historyEpoch) { await deps.cache.reset(deps.conversationId, historyEpoch); + cachedEpoch = await deps.cache.historyEpoch(deps.conversationId); + } const since = - cachedEpoch === undefined ? 0 : await deps.cache.sinceSeq(deps.conversationId); + reloadWindow || cachedEpoch === undefined + ? 0 + : await deps.cache.sinceSeq(deps.conversationId); + if (disposed) break; + if (revision !== currentRevision) { + rerunRequested = true; + continue; + } const limit = since === 0 ? (reloadWindow ? initialWindowSize(chatLimit) : coldLimit) : undefined; const res = await deps.historySync( @@ -360,21 +373,43 @@ export function createChatStore(deps: ChatStoreDependencies): ChatStore { rerunRequested = true; continue; } - if ( + const mismatch = (cachedEpoch !== undefined && cachedEpoch !== res.historyEpoch) || - (historyEpoch !== undefined && historyEpoch !== res.historyEpoch) - ) { - resetTranscript(res.historyEpoch); - await deps.cache.reset(deps.conversationId, res.historyEpoch); - if (since !== 0) { + (historyEpoch !== undefined && historyEpoch !== res.historyEpoch); + if (mismatch && since !== 0) { + // A cursor response cannot replace a different history. Confirm with + // a new full-window request; don't mistake a lower epoch for staleness. + reloadWindow = true; + rerunRequested = true; + continue; + } + let merged: readonly StoredChunk[]; + if (mismatch || cachedEpoch === undefined) { + const replaced = await deps.cache.replace( + deps.conversationId, + res.chunks, + res.historyEpoch, + cachedEpoch, + ); + const replacedEpoch = await deps.cache.historyEpoch(deps.conversationId); + if (disposed) break; + if (revision !== currentRevision || !replaced || replacedEpoch !== res.historyEpoch) { + rerunRequested = true; + continue; + } + if (mismatch) resetTranscript(res.historyEpoch); + merged = res.chunks; + } else { + merged = await deps.cache.commit(deps.conversationId, res.chunks, res.historyEpoch); + const committedEpoch = await deps.cache.historyEpoch(deps.conversationId); + if (disposed) break; + if (revision !== currentRevision || committedEpoch !== res.historyEpoch) { 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); @@ -383,7 +418,10 @@ export function createChatStore(deps: ChatStoreDependencies): ChatStore { } catch (err) { _error = err instanceof Error ? err.message : String(err); } - } while (rerunRequested && !disposed); + } while (rerunRequested && !disposed && attempts < 3); + if (rerunRequested && !disposed) { + _error = "Conversation history changed repeatedly; sync paused until the next refresh."; + } _pendingSync = false; } @@ -419,8 +457,10 @@ export function createChatStore(deps: ChatStoreDependencies): ChatStore { }); if (disposed || revision !== currentRevision) return []; if (res.historyEpoch !== historyEpoch) { - resetTranscript(res.historyEpoch); - await deps.cache.reset(deps.conversationId, res.historyEpoch); + // Confirm a backfill mismatch with a fresh full window as well. Bump + // the local revision so an overlapping old tail cannot be applied. + revision++; + reloadWindow = true; await syncTail(initialWindowSize(chatLimit)); return []; } diff --git a/src/features/chat/store.test.ts b/src/features/chat/store.test.ts index 8c94316..de84b8a 100644 --- a/src/features/chat/store.test.ts +++ b/src/features/chat/store.test.ts @@ -2023,3 +2023,179 @@ it("reloads unknown legacy cache epochs rather than merging pre-upgrade rows", a vi.unstubAllGlobals(); } }); + +describe("server epoch rollback", () => { + it.each([ + "tail", + "backfill", + ])("converges on %s with a real cache and a fresh bounded fetch", async (path) => { + vi.stubGlobal("IDBKeyRange", IDBKeyRange); + const storage = createIdbChunkStore({ indexedDB: new IDBFactory() }); + const cache = createConversationCache(storage); + const old = Array.from({ length: 20 }, (_, i) => makeStoredChunk(i + 1)); + await cache.commit(CONV_ID, path === "backfill" ? old.slice(-7) : old, 5); + const history = createFakeHistorySync(); + history.historyEpoch = 5; + const store = createChatStore({ + conversationId: CONV_ID, + transport: createFakeTransport().impl, + historySync: history.impl, + metricsSync: createFakeMetricsSync().impl, + cache, + chatLimit: 10, + }); + try { + await store.load(); + history.historyEpoch = 0; + // Recreated ids can reuse seqs with completely different content. + const restored: StoredChunk = { + seq: 1, + role: "user", + chunk: { type: "text", text: "restored" }, + }; + history.returnChunks = [restored]; + const before = history.calls.length; + if (path === "tail") { + store.resync(); + await vi.waitFor(() => expect(store.pendingSync).toBe(false)); + } else await store.showEarlier(); + expect(history.calls.length - before).toBeLessThanOrEqual(2); + expect(history.calls.at(-1)).toMatchObject({ sinceSeq: 0, window: { limit: 7 } }); + expect(await cache.historyEpoch(CONV_ID)).toBe(0); + expect(await cache.load(CONV_ID)).toEqual([restored]); + expect(store.chunks.map((c) => c.chunk)).toEqual([restored.chunk]); + expect(store.hasEarlier).toBe(false); + expect(store.error).toBeNull(); + } finally { + store.dispose(); + vi.unstubAllGlobals(); + } + }); + + it("bounds reruns even when every response is invalidated by another compaction", async () => { + let calls = 0; + let churn = true; + let store!: ReturnType; + const sync: import("./ports").HistorySync = async () => { + calls++; + // Stop the old broken loop safely rather than hanging the test worker. + if (churn && calls <= 6) store.resync(calls); + return { chunks: [], latestSeq: 0, historyEpoch: calls - 1 }; + }; + store = createChatStore({ + conversationId: CONV_ID, + transport: createFakeTransport().impl, + historySync: sync, + metricsSync: createFakeMetricsSync().impl, + cache: createFakeCache().impl, + }); + try { + await store.load(); + expect(calls).toBe(3); + expect(store.pendingSync).toBe(false); + expect(store.error).toMatch(/history changed repeatedly/i); + churn = false; + store.resync(); + await vi.waitFor(() => expect(store.pendingSync).toBe(false)); + expect(calls).toBe(4); + expect(store.error).toBeNull(); + } finally { + store.dispose(); + } + }); +}); + +it.each([ + "tail", + "backfill", +])("rejects a delayed old %s after a lower-epoch history is observed", async (delayed) => { + vi.stubGlobal("IDBKeyRange", IDBKeyRange); + const cache = createConversationCache(createIdbChunkStore({ indexedDB: new IDBFactory() })); + await cache.commit( + CONV_ID, + Array.from({ length: 7 }, (_, i) => makeStoredChunk(14 + i)), + 5, + ); + const restored: StoredChunk = { + seq: 1, + role: "user", + chunk: { type: "text", text: "new history" }, + }; + type Response = import("@dispatch/transport-contract").ConversationHistoryResponse; + let resolve!: (res: Response) => void; + const slow = new Promise((r) => { + resolve = r; + }); + let delayNext = false; + let epoch = 5; + const sync = vi.fn().mockImplementation(async () => { + if (delayNext) { + delayNext = false; + return slow; + } + return { + chunks: epoch === 5 ? [] : [restored], + latestSeq: epoch === 5 ? 20 : 1, + historyEpoch: epoch, + }; + }); + const store = createChatStore({ + conversationId: CONV_ID, + transport: createFakeTransport().impl, + historySync: sync, + metricsSync: createFakeMetricsSync().impl, + cache, + chatLimit: 10, + }); + try { + await store.load(); + delayNext = true; + let earlier: Promise; + if (delayed === "tail") { + store.resync(); + await vi.waitFor(() => expect(sync).toHaveBeenCalledTimes(2)); + epoch = 0; + earlier = store.showEarlier(); + await earlier; // Invalidates the old tail and requests a sequential rerun. + } else { + earlier = store.showEarlier(); + await vi.waitFor(() => expect(sync).toHaveBeenCalledTimes(2)); + epoch = 0; + store.resync(); + await vi.waitFor(() => expect(store.chunks.map((c) => c.chunk)).toEqual([restored.chunk])); + } + resolve({ chunks: [makeStoredChunk(21)], latestSeq: 21, historyEpoch: 5 }); + await earlier; + await vi.waitFor(() => expect(store.pendingSync).toBe(false)); + expect(await cache.historyEpoch(CONV_ID)).toBe(0); + expect(await cache.load(CONV_ID)).toEqual([restored]); + expect(store.chunks.map((c) => c.chunk)).toEqual([restored.chunk]); + expect(sync).toHaveBeenCalledTimes(4); + } finally { + store.dispose(); + vi.unstubAllGlobals(); + } +}); + +it("replaces an empty high-epoch cache with the server's empty lower epoch in one request", async () => { + const cache = createConversationCache(createIdbChunkStore({ indexedDB: new IDBFactory() })); + await cache.reset(CONV_ID, 8); + const history = createFakeHistorySync(); + const store = createChatStore({ + conversationId: CONV_ID, + transport: createFakeTransport().impl, + historySync: history.impl, + metricsSync: createFakeMetricsSync().impl, + cache, + }); + try { + await store.load(); + expect(history.calls).toHaveLength(1); + expect(await cache.historyEpoch(CONV_ID)).toBe(0); + expect(await cache.load(CONV_ID)).toEqual([]); + expect(store.chunks).toEqual([]); + expect(store.hasEarlier).toBe(false); + } finally { + store.dispose(); + } +}); diff --git a/src/features/chat/test-helpers.ts b/src/features/chat/test-helpers.ts index c1b020b..41cfb3b 100644 --- a/src/features/chat/test-helpers.ts +++ b/src/features/chat/test-helpers.ts @@ -141,6 +141,12 @@ export function createFakeCache(): FakeCache { store.delete(id); epochs.set(id, epoch); }, + async replace(id, incoming, epoch, expectedEpoch) { + if ((epochs.get(id) ?? 0) !== expectedEpoch) return false; + store.set(id, [...incoming]); + epochs.set(id, epoch); + return true; + }, async load(conversationId) { return store.get(conversationId) ?? []; }, diff --git a/src/features/conversation-cache/cache.test.ts b/src/features/conversation-cache/cache.test.ts index 4306d51..72b4e20 100644 --- a/src/features/conversation-cache/cache.test.ts +++ b/src/features/conversation-cache/cache.test.ts @@ -214,3 +214,38 @@ it("serializes invalidation and rejects stale epoch commits", async () => { expect(await cache.commit("c", [chunk(103)], 1)).toEqual([]); expect(await store.historyEpoch("c")).toBe(2); }); + +it("replaces a confirmed full history at a lower epoch, without trusting stale commits", async () => { + const store = createFakeStore(); + const cache = createConversationCache(store); + await cache.commit("c", [chunk(9)], 5); + expect(await cache.replace("c", [chunk(1)], 1, 5)).toBe(true); + expect(await cache.historyEpoch("c")).toBe(1); + expect(await cache.load("c")).toEqual([chunk(1)]); + expect(await cache.commit("c", [chunk(9)], 5)).toEqual([chunk(1)]); + // A delayed full response started against epoch 5 must not restore it. + expect(await cache.replace("c", [chunk(9)], 5, 5)).toBe(false); + expect(await cache.load("c")).toEqual([chunk(1)]); +}); + +it("serializes a guarded replacement behind newer invalidation", async () => { + const store = createFakeStore(); + const cache = createConversationCache(store); + await cache.commit("c", [chunk(9)], 5); + const resetting = cache.reset("c", 6); + const replacing = cache.replace("c", [chunk(1)], 1, 5); + await resetting; + expect(await replacing).toBe(false); + expect(await cache.historyEpoch("c")).toBe(6); + expect(await cache.load("c")).toEqual([]); +}); + +it("allows an empty replacement and a legacy cache only if its observed epoch still matches", async () => { + const store = createFakeStore(); + const cache = createConversationCache(store); + await store.append("c", [chunk(9)]); + expect(await cache.replace("c", [], 0, undefined)).toBe(true); + expect(await cache.sinceSeq("c")).toBe(0); + expect(await cache.historyEpoch("c")).toBe(0); + expect(await cache.replace("c", [chunk(9)], 2, undefined)).toBe(false); +}); diff --git a/src/features/conversation-cache/cache.ts b/src/features/conversation-cache/cache.ts index 431a2a7..600a6e6 100644 --- a/src/features/conversation-cache/cache.ts +++ b/src/features/conversation-cache/cache.ts @@ -6,6 +6,19 @@ export interface ConversationCache { historyEpoch(conversationId: string): Promise; reset(conversationId: string, historyEpoch: number): Promise; + /** + * Replace from a fresh full-history response, including an epoch decrease. + * Only succeeds if the cache still has the epoch observed BEFORE that request. + * Unlike monotonic reset/commit, this is for confirmed server authority, not + * broadcasts or cursor/backfill responses. False means the response is stale. + */ + replace( + conversationId: string, + incoming: readonly StoredChunk[], + historyEpoch: number, + expectedEpoch: number | undefined, + ): Promise; + /** Load all cached chunks for a conversation. */ load(conversationId: string): Promise; @@ -73,6 +86,13 @@ export function createConversationCache( const current = await store.historyEpoch(id); if (current === undefined || current < epoch) await store.reset(id, epoch); }), + replace: (id, incoming, epoch, expectedEpoch) => + exclusive(id, async () => { + if ((await store.historyEpoch(id)) !== expectedEpoch) return false; + await store.reset(id, epoch); + if (incoming.length > 0) await store.append(id, incoming); + return true; + }), async load(conversationId) { await pending.get(conversationId); return store.load(conversationId); @@ -82,7 +102,9 @@ export function createConversationCache( return exclusive(conversationId, async () => { if (epoch !== undefined) { const current = await store.historyEpoch(conversationId); - if (current !== undefined && current > epoch) return store.load(conversationId); + // Once an epoch is known, only explicit invalidation or guarded + // full-history replacement may change it (in either direction). + if (current !== undefined && current !== epoch) return store.load(conversationId); if (current !== epoch) await store.reset(conversationId, epoch); } const cached = await store.load(conversationId); -- cgit v1.2.3 From e9786bf1dbf258b68f793870d02378c390939d3d Mon Sep 17 00:00:00 2001 From: Adam Malczewski Date: Wed, 21 Oct 2026 14:52:42 +0900 Subject: fix(review-followup-fed): reconcile compaction-queued send acknowledgements --- src/adapters/ws/index.test.ts | 23 +++++ src/adapters/ws/index.ts | 7 +- src/adapters/ws/logic.test.ts | 32 ++++++ src/adapters/ws/logic.ts | 16 +++ src/app/chat-registry.svelte.ts | 1 + src/app/store.test.ts | 85 +++++++++++++++ src/app/ws-handlers.ts | 5 +- src/core/chunks/index.ts | 1 + src/core/chunks/local-echo.test.ts | 45 ++++++++ src/core/chunks/reducer.ts | 31 +++++- src/core/chunks/types.ts | 2 + src/core/wire/conformance.test.ts | 7 ++ src/core/wire/conformance.ts | 2 + src/features/chat/queued.test.ts | 207 +++++++++++++++++++++++++++++++++++++ src/features/chat/store.svelte.ts | 20 +++- src/features/chat/store.test.ts | 63 +++++++++++ 16 files changed, 536 insertions(+), 11 deletions(-) create mode 100644 src/core/chunks/local-echo.test.ts create mode 100644 src/features/chat/queued.test.ts (limited to 'src/adapters') diff --git a/src/adapters/ws/index.test.ts b/src/adapters/ws/index.test.ts index 6f5e4ef..6461493 100644 --- a/src/adapters/ws/index.test.ts +++ b/src/adapters/ws/index.test.ts @@ -476,3 +476,26 @@ describe("reconnect backoff after open", () => { } }); }); + +it("routes chat.queued only through the typed chat callback, never surface/error handling", () => { + const ws = fakeSocket(); + const onChat = vi.fn(); + const onMessage = vi.fn(); + const handle = createSurfaceSocket({ + url: "ws://test", + socketFactory: () => ws, + onChat, + onMessage, + }); + const ack = { + type: "chat.queued", + conversationId: "c1", + reason: "compaction", + messageId: "q1", + requestId: "r1", + }; + ws.invokeMessage(JSON.stringify(ack)); + expect(onChat).toHaveBeenCalledExactlyOnceWith(ack); + expect(onMessage).not.toHaveBeenCalled(); + handle.close(); +}); diff --git a/src/adapters/ws/index.ts b/src/adapters/ws/index.ts index eeb7901..25e9911 100644 --- a/src/adapters/ws/index.ts +++ b/src/adapters/ws/index.ts @@ -1,6 +1,7 @@ import type { ChatDeltaMessage, ChatErrorMessage, + ChatQueuedMessage, ConversationCompactedMessage, ConversationOpenMessage, ConversationStatusChangedMessage, @@ -24,7 +25,7 @@ export interface WebSocketLike { export interface SurfaceSocketOptions { url: string; onMessage: (msg: SurfaceServerMessage) => void; - onChat?: (msg: ChatDeltaMessage | ChatErrorMessage) => void; + onChat?: (msg: ChatDeltaMessage | ChatErrorMessage | ChatQueuedMessage) => void; /** Broadcast when a conversation is "opened" (e.g. CLI `--open` flag). */ onConversationOpen?: (msg: ConversationOpenMessage) => void; /** Broadcast when a conversation's lifecycle status changes (active/idle/closed). */ @@ -100,8 +101,8 @@ export function createSurfaceSocket(opts: SurfaceSocketOptions): SurfaceSocketHa if (disposed) return; const msg = parseServerMessage(ev.data); if (msg !== null) { - if (msg.type === "chat.delta" || msg.type === "chat.error") { - opts.onChat?.(msg as ChatDeltaMessage | ChatErrorMessage); + if (msg.type === "chat.delta" || msg.type === "chat.error" || msg.type === "chat.queued") { + opts.onChat?.(msg as ChatDeltaMessage | ChatErrorMessage | ChatQueuedMessage); } else if (msg.type === "conversation.open") { opts.onConversationOpen?.(msg as ConversationOpenMessage); } else if (msg.type === "conversation.statusChanged") { diff --git a/src/adapters/ws/logic.test.ts b/src/adapters/ws/logic.test.ts index 4be2cd6..db6444a 100644 --- a/src/adapters/ws/logic.test.ts +++ b/src/adapters/ws/logic.test.ts @@ -577,3 +577,35 @@ it("preserves a compaction epoch and rejects absent or invalid epochs", () => { expect(parseServerMessage(JSON.stringify({ ...msg, historyEpoch }))).toBeNull(); } }); + +describe("chat.queued admission acknowledgement", () => { + const ack = { + type: "chat.queued", + conversationId: "c1", + reason: "compaction", + messageId: "q1", + }; + + it.each([undefined, "request-1"])("preserves optional requestId %s", (requestId) => { + const msg = { ...ack, ...(requestId !== undefined ? { requestId } : {}) }; + expect(parseServerMessage(JSON.stringify(msg))).toEqual(msg); + }); + + it.each([ + { conversationId: undefined }, + { conversationId: 12 }, + { messageId: undefined }, + { messageId: null }, + { reason: undefined }, + { reason: "other" }, + { requestId: null }, + { requestId: 12 }, + ])("rejects malformed acknowledgement %j", (bad) => { + expect(parseServerMessage(JSON.stringify({ ...ack, ...bad }))).toBeNull(); + }); + + it("serializes a correlated send", () => { + const msg = { type: "chat.send" as const, message: "hello", requestId: "r1" }; + expect(JSON.parse(serialize(msg))).toEqual(msg); + }); +}); diff --git a/src/adapters/ws/logic.ts b/src/adapters/ws/logic.ts index 858deaa..0efcd90 100644 --- a/src/adapters/ws/logic.ts +++ b/src/adapters/ws/logic.ts @@ -1,6 +1,7 @@ import type { ChatDeltaMessage, ChatErrorMessage, + ChatQueuedMessage, ConversationCompactedMessage, ConversationOpenMessage, ConversationStatusChangedMessage, @@ -25,6 +26,7 @@ const VALID_SERVER_TYPES = new Set([ "error", "chat.delta", "chat.error", + "chat.queued", "conversation.open", "conversation.statusChanged", "conversation.compacted", @@ -111,6 +113,20 @@ export function parseServerMessage(data: string): WsServerMessage | null { if (typeof event.type !== "string") return null; return { type: "chat.delta", event: event as unknown as ChatDeltaMessage["event"] }; } + case "chat.queued": { + if (typeof parsed.conversationId !== "string") return null; + if (typeof parsed.messageId !== "string") return null; + if (parsed.reason !== "compaction") return null; + if (parsed.requestId !== undefined && typeof parsed.requestId !== "string") return null; + const msg: ChatQueuedMessage = { + type: "chat.queued", + conversationId: parsed.conversationId, + messageId: parsed.messageId, + reason: "compaction", + ...(parsed.requestId !== undefined ? { requestId: parsed.requestId } : {}), + }; + return msg; + } case "chat.error": { if (typeof parsed.message !== "string") return null; const conversationId = parsed.conversationId; diff --git a/src/app/chat-registry.svelte.ts b/src/app/chat-registry.svelte.ts index 8e97af1..a9b28ff 100644 --- a/src/app/chat-registry.svelte.ts +++ b/src/app/chat-registry.svelte.ts @@ -78,6 +78,7 @@ export function createChatRegistry(deps: { function createChatFor(conversationId: string, model: string, workspaceId: string): ChatStore { return createChatStore({ conversationId, + createRequestId: randomId, model, workspaceId, transport: { diff --git a/src/app/store.test.ts b/src/app/store.test.ts index f257f3a..40a0fa7 100644 --- a/src/app/store.test.ts +++ b/src/app/store.test.ts @@ -2991,3 +2991,88 @@ it.each([ store.dispose(); } }); + +describe("compaction queued acknowledgement routing", () => { + it.each([ + "surface-first", + "ack-first", + "drained-before-ack", + ])("uses only authoritative queue snapshots: %s", (order) => { + const ws = fakeSocket(); + const store = createAppStore({ + socketFactory: () => ws, + fetchImpl: fakeFetchImpl(), + localStorage: createFakeStorage(), + chatFlushScheduler: null, + }); + ws.resolveOpen(); + store.send("queued text", [{ url: "data:image/png;base64,AAAA" }]); + const cid = activeConversationId(store); + const send = parseSent(ws).find((m) => (m as { type: string }).type === "chat.send") as { + requestId: string; + }; + expect(send.requestId).toEqual(expect.any(String)); + const sourceChat = store.activeChat; + const watch = store.watchConversation("other-conversation"); + watch.send("unrelated local echo"); + ws.feedSurfaceMessage({ + type: "catalog", + catalog: [{ id: "message-queue", region: "inline", title: "Queue", scope: "conversation" }], + }); + const spec = (messages: { id: string; text: string; timestamp: number }[]) => ({ + id: "message-queue", + region: "inline", + title: "Queue", + fields: [{ kind: "custom" as const, rendererId: "message-queue", payload: { messages } }], + }); + const queued = spec([{ id: "q1", text: "queued text", timestamp: 1 }]); + const drained = spec([]); + const ack = () => + ws.feedServerMessage({ + type: "chat.queued", + conversationId: cid, + reason: "compaction", + messageId: "q1", + requestId: send.requestId, + }); + if (order === "ack-first") { + ack(); + expect(store.surface("message-queue", cid)).toBeNull(); + } + ws.feedSurfaceMessage({ type: "surface", conversationId: cid, spec: queued }); + expect(store.surface("message-queue", cid)).toEqual(queued); + if (order === "drained-before-ack") { + ws.feedSurfaceMessage({ + type: "update", + update: { surfaceId: "message-queue", conversationId: cid, spec: drained }, + }); + ws.feedServerMessage({ + type: "chat.delta", + event: { type: "turn-start", conversationId: cid, turnId: "t1" }, + }); + ws.feedServerMessage({ + type: "chat.delta", + event: { type: "user-message", conversationId: cid, turnId: "t1", text: "queued text" }, + }); + } + ack(); + ack(); + expect(store.surface("message-queue", cid)).toEqual( + order === "drained-before-ack" ? drained : queued, + ); + expect(sourceChat.chunks.map((c) => c.chunk)).toEqual( + order === "drained-before-ack" ? [{ type: "text", text: "queued text" }] : [], + ); + expect(sourceChat.error).toBeNull(); + expect(watch.chunks.map((c) => c.chunk)).toEqual([ + { type: "text", text: "unrelated local echo" }, + ]); + expect(parseSent(ws).filter((m) => (m as { type: string }).type === "chat.send")).toHaveLength( + 2, + ); + expect(parseSent(ws).filter((m) => (m as { type: string }).type === "chat.queue")).toHaveLength( + 0, + ); + store.dispose(); + }); +}); diff --git a/src/app/ws-handlers.ts b/src/app/ws-handlers.ts index c76c9c1..aa1e03e 100644 --- a/src/app/ws-handlers.ts +++ b/src/app/ws-handlers.ts @@ -1,6 +1,7 @@ import type { ChatDeltaMessage, ChatErrorMessage, + ChatQueuedMessage, ConversationCompactedMessage, ConversationOpenMessage, ConversationStatusChangedMessage, @@ -52,7 +53,7 @@ export function createWsHandlers(deps: { layoutStore, tileLayoutStore, } = deps; - function handleChatMessage(msg: ChatDeltaMessage | ChatErrorMessage): void { + function handleChatMessage(msg: ChatDeltaMessage | ChatErrorMessage | ChatQueuedMessage): void { let targetId: string | undefined; if (msg.type === "chat.delta") { targetId = msg.event.conversationId; @@ -68,6 +69,8 @@ export function createWsHandlers(deps: { } } + if (msg.type === "chat.queued") return; + // fallback: try all stores (chat.error without conversationId) for (const store of registry.chatStores.values()) { store.handleDelta(msg); diff --git a/src/core/chunks/index.ts b/src/core/chunks/index.ts index 3822a79..49381d5 100644 --- a/src/core/chunks/index.ts +++ b/src/core/chunks/index.ts @@ -7,6 +7,7 @@ export { clearGenerating, foldEvent, initialState, + removeLocalEcho, } from "./reducer"; export type { ProviderRetryView } from "./retry-banner"; export { formatRetryDelay, viewProviderRetry } from "./retry-banner"; diff --git a/src/core/chunks/local-echo.test.ts b/src/core/chunks/local-echo.test.ts new file mode 100644 index 0000000..4b5eed4 --- /dev/null +++ b/src/core/chunks/local-echo.test.ts @@ -0,0 +1,45 @@ +import { describe, expect, it } from "vitest"; +import { appendUserMessage, applyHistory, foldEvent, initialState, removeLocalEcho } from "./index"; + +describe("request-local optimistic echo", () => { + it("retains identity through turn-start tags without conflating identical sends", () => { + let state = appendUserMessage(initialState(), "same", [{ url: "data:first" }], "r1"); + state = appendUserMessage(state, "same", [{ url: "data:second" }], "r2"); + state = foldEvent(state, { type: "turn-start", conversationId: "c1", turnId: "t1" }); + state = removeLocalEcho(state, "r1"); + expect(state.provisional.map((p) => p.localRequestId)).toEqual(["r2", "r2"]); + expect(state.provisional.map((p) => p.chunk)).toEqual([ + { type: "text", text: "same" }, + { type: "image", url: "data:second" }, + ]); + expect(state.generating).toBe(true); + expect(removeLocalEcho(state, "r1")).toBe(state); + }); + + it("a live user-message confirms only its deduplicated text, not every identical local send", () => { + let state = appendUserMessage(initialState(), "same", [{ url: "data:first" }], "r1"); + state = appendUserMessage(state, "same", [{ url: "data:second" }], "r2"); + state = foldEvent(state, { type: "turn-start", conversationId: "c1", turnId: "t1" }); + state = foldEvent(state, { + type: "user-message", + conversationId: "c1", + turnId: "t1", + text: "same", + }); + state = removeLocalEcho(removeLocalEcho(state, "r1"), "r2"); + expect(state.provisional).toEqual([ + { role: "user", turnId: "t1", chunk: { type: "text", text: "same" } }, + ]); + }); + + it("does not remove committed rows even if their content is identical", () => { + let state = appendUserMessage(initialState(), "same", undefined, "r1"); + state = applyHistory(state, [{ seq: 1, role: "user", chunk: { type: "text", text: "same" } }]); + const committed = state.committed; + state = removeLocalEcho(state, "r1"); + expect(state.committed).toBe(committed); + expect(state.provisional).toEqual([]); + expect(removeLocalEcho(state, "r1")).toBe(state); + expect(removeLocalEcho(initialState(), "r1")).toEqual(initialState()); + }); +}); diff --git a/src/core/chunks/reducer.ts b/src/core/chunks/reducer.ts index 0af2ddd..53af9c1 100644 --- a/src/core/chunks/reducer.ts +++ b/src/core/chunks/reducer.ts @@ -246,11 +246,17 @@ function reduceEvent(state: TranscriptState, event: AgentEvent): TranscriptState // text) emits no `user-message` and is not de-duped here. if (event.text.length === 0) return state; const run = trailingRun(state.provisional, "user"); - const alreadyEchoed = run.some((p) => p.chunk.type === "text" && p.chunk.text === event.text); - if (alreadyEchoed) { + const echoed = run.find((p) => p.chunk.type === "text" && p.chunk.text === event.text); + if (echoed !== undefined) { return { ...state, - provisional: tagTrailingUserEcho(state.provisional, event.turnId), + // Preserve the live text independently of the local send. A delayed + // queued ack may remove its images, but cannot erase this server event. + provisional: tagTrailingUserEcho(state.provisional, event.turnId).map((p) => { + if (p.chunk !== echoed.chunk || p.localRequestId === undefined) return p; + const { localRequestId: _local, ...confirmed } = p; + return confirmed; + }), generating: true, }; } @@ -463,6 +469,7 @@ export function appendUserMessage( state: TranscriptState, text: string, images?: readonly ImageInput[], + localRequestId?: string, ): TranscriptState { const provisional = flushAccumulating(state.provisional, state.accumulating); const userChunks: Chunk[] = []; @@ -484,7 +491,23 @@ export function appendUserMessage( } return { ...state, - provisional: [...provisional, ...userChunks.map((chunk) => ({ role: "user", chunk }) as const)], + provisional: [ + ...provisional, + ...userChunks.map( + (chunk) => + ({ + role: "user", + chunk, + ...(localRequestId !== undefined ? { localRequestId } : {}), + }) as const, + ), + ], accumulating: null, }; } + +/** Remove only still-local artifacts of one accepted queued send; never history/live text. */ +export function removeLocalEcho(state: TranscriptState, requestId: string): TranscriptState { + const provisional = state.provisional.filter((p) => p.localRequestId !== requestId); + return provisional.length === state.provisional.length ? state : { ...state, provisional }; +} diff --git a/src/core/chunks/types.ts b/src/core/chunks/types.ts index 949857a..f7bfe01 100644 --- a/src/core/chunks/types.ts +++ b/src/core/chunks/types.ts @@ -10,6 +10,8 @@ export interface AccumulatingChunk { /** A provisional chunk that has no authoritative seq yet. */ export interface ProvisionalChunk { + /** Local send identity only; turn-start tagging does NOT confirm this content. */ + readonly localRequestId?: string; readonly role: Role; readonly chunk: Chunk; /** diff --git a/src/core/wire/conformance.test.ts b/src/core/wire/conformance.test.ts index 1c5829b..1179fbc 100644 --- a/src/core/wire/conformance.test.ts +++ b/src/core/wire/conformance.test.ts @@ -174,6 +174,12 @@ describe("classifies every WsServerMessage type", () => { event: { type: "done" as const, conversationId: "c", turnId: "t", reason: "r" }, }, { type: "chat.error" as const, message: "e" }, + { + type: "chat.queued" as const, + conversationId: "conv-1", + reason: "compaction" as const, + messageId: "queued-1", + }, { type: "conversation.open" as const, conversationId: "c1", workspaceId: "w1" }, { type: "conversation.statusChanged" as const, @@ -198,6 +204,7 @@ describe("classifies every WsServerMessage type", () => { "error", "chat.delta", "chat.error", + "chat.queued", "conversation.open", "conversation.statusChanged", "conversation.compacted", diff --git a/src/core/wire/conformance.ts b/src/core/wire/conformance.ts index 9847081..78313f9 100644 --- a/src/core/wire/conformance.ts +++ b/src/core/wire/conformance.ts @@ -85,6 +85,8 @@ export function assertWsServerMessageExhaustive(msg: WsServerMessage): string { return "error"; case "chat.delta": return "chat.delta"; + case "chat.queued": + return "chat.queued"; case "chat.error": return "chat.error"; case "conversation.open": diff --git a/src/features/chat/queued.test.ts b/src/features/chat/queued.test.ts new file mode 100644 index 0000000..3b579c6 --- /dev/null +++ b/src/features/chat/queued.test.ts @@ -0,0 +1,207 @@ +import type { ChatQueuedMessage } from "@dispatch/transport-contract"; +import type { AgentEvent } from "@dispatch/wire"; +import { describe, expect, it, vi } from "vitest"; +import { createChatStore } from "./store.svelte"; +import { + createFakeCache, + createFakeHistorySync, + createFakeMetricsSync, + createFakeTransport, +} from "./test-helpers"; + +const images = [ + { url: "data:image/png;base64,AAAA", mimeType: "image/png" }, + { url: "data:image/png;base64,BBBB" }, +]; + +function fixture(batched = false) { + const transport = createFakeTransport(); + const history = createFakeHistorySync(); + let nextId = 0; + const scheduled: (() => void)[] = []; + const store = createChatStore({ + conversationId: "c1", + createRequestId: () => `r${++nextId}`, + transport: transport.impl, + historySync: history.impl, + metricsSync: createFakeMetricsSync().impl, + cache: createFakeCache().impl, + ...(batched ? { scheduleFlush: (flush: () => void) => scheduled.push(flush) } : {}), + }); + function ack(requestId: string | undefined = "r1", conversationId = "c1") { + const msg: ChatQueuedMessage = { + type: "chat.queued", + conversationId, + reason: "compaction", + messageId: "q1", + ...(requestId !== undefined ? { requestId } : {}), + }; + store.handleDelta(msg); + } + function event(event: AgentEvent) { + store.handleDelta({ type: "chat.delta", event }); + } + function start(text?: string, turnId = "t1") { + event({ type: "turn-start", conversationId: "c1", turnId }); + if (text !== undefined) event({ type: "user-message", conversationId: "c1", turnId, text }); + } + return { store, transport, history, scheduled, ack, event, start }; +} + +describe("compaction-deferred chat.send", () => { + it.each([ + "hello", + "", + ])("removes just the accepted local text/images for %j without error, retry or queue sends", (text) => { + const f = fixture(); + f.store.send(text, images); + expect(f.store.chunks).toHaveLength(text ? 3 : 2); + f.ack(); + f.ack(); + expect(f.store.chunks).toEqual([]); + expect(f.store.generating).toBe(false); + expect(f.store.error).toBeNull(); + expect(f.transport.sent).toHaveLength(1); + expect(f.transport.sent[0]).toMatchObject({ requestId: "r1", images, message: text }); + expect(f.transport.sentQueue).toEqual([]); + f.store.dispose(); + }); + + it("uses distinct request identities for identical sends and preserves unrelated content", () => { + const f = fixture(); + f.start("another client's prompt"); + f.event({ type: "text-delta", conversationId: "c1", turnId: "t1", delta: "working" }); + f.store.send("identical", images); + f.store.send("identical", images); + expect(f.transport.sent.map((m) => m.requestId)).toEqual(["r1", "r2"]); + f.ack("r1"); + f.ack("r1"); + expect(f.store.chunks.map((c) => c.chunk)).toEqual([ + { type: "text", text: "another client's prompt" }, + { type: "text", text: "working" }, + { type: "text", text: "identical" }, + ...images.map((i) => ({ type: "image", ...i })), + ]); + f.ack("r2"); + expect(f.store.chunks.map((c) => c.chunk)).toEqual([ + { type: "text", text: "another client's prompt" }, + { type: "text", text: "working" }, + ]); + expect(f.store.generating).toBe(true); + f.store.dispose(); + }); + + it.each([ + "missing", + "unknown", + "wrong-conversation", + ])("tolerates %s correlation without deleting any echo", (kind) => { + const f = fixture(); + f.store.send("hello", images); + f.store.handleDelta({ + type: "chat.queued", + conversationId: kind === "wrong-conversation" ? "c2" : "c1", + reason: "compaction", + messageId: "q1", + ...(kind === "missing" ? {} : { requestId: kind === "unknown" ? "r999" : "r1" }), + }); + expect(f.store.chunks).toHaveLength(3); + expect(f.store.error).toBeNull(); + f.store.dispose(); + }); + + it.each([ + false, + true, + ])("preserves a live confirmed prompt before delayed ack (batched=%s)", (batched) => { + const f = fixture(batched); + f.store.send("hello", images); + f.start("hello"); + f.event({ type: "text-delta", conversationId: "c1", turnId: "t1", delta: "answer" }); + f.ack(); + f.ack(); + for (const flush of f.scheduled) flush(); + expect(f.store.chunks.map((c) => c.chunk)).toEqual([ + { type: "text", text: "hello" }, + { type: "text", text: "answer" }, + ]); + expect(f.store.generating).toBe(true); + expect(f.store.error).toBeNull(); + f.store.dispose(); + }); + + it("turn-start tagging alone does not protect a still-local echo", () => { + const f = fixture(true); + f.store.send("hello", images); + f.start(); + f.ack(); + expect(f.store.chunks).toEqual([]); + expect(f.store.generating).toBe(true); + f.event({ type: "user-message", conversationId: "c1", turnId: "t1", text: "hello" }); + for (const flush of f.scheduled) flush(); + expect(f.store.chunks.map((c) => c.chunk)).toEqual([{ type: "text", text: "hello" }]); + f.store.dispose(); + }); + + it("keeps a combined queue opening and steering; removes only the original individual echoes", () => { + const f = fixture(true); + f.store.send("one", images); + f.store.send("two", images); + f.start("one\n\ntwo"); + f.event({ type: "steering", conversationId: "c1", turnId: "t1", text: "steer" }); + f.ack("r2"); + f.ack("r1"); + expect(f.store.chunks.map((c) => c.chunk)).toEqual([ + { type: "text", text: "one\n\ntwo" }, + { type: "text", text: "steer" }, + ]); + expect(f.store.generating).toBe(true); + f.store.dispose(); + }); + + it.each([ + "ack-first", + "reset-first", + "history-first", + ])("handles epoch replacement idempotently: %s", async (order) => { + const f = fixture(); + await f.store.load(); + f.store.send("hello", images); + if (order === "ack-first") f.ack(); + f.history.historyEpoch = 1; + f.history.returnChunks = [ + { seq: 1, role: "user", turnId: "t1", chunk: { type: "text", text: "hello" } }, + { seq: 2, role: "user", turnId: "t1", chunk: { type: "image", url: "/images/c1/saved.png" } }, + ]; + f.store.resync(1); + if (order === "reset-first") f.ack(); + await vi.waitFor(() => expect(f.store.pendingSync).toBe(false)); + f.start("hello"); + const before = f.store.chunks; + f.ack(); + f.ack(); + expect(f.store.chunks).toEqual(before); + expect(f.store.chunks.map((c) => c.chunk)).toEqual(f.history.returnChunks.map((c) => c.chunk)); + expect(f.store.generating).toBe(true); + expect(f.store.error).toBeNull(); + expect(f.transport.sent).toHaveLength(1); + f.store.dispose(); + }); + + it("keeps normal immediate-send text/image deduplication and authoritative history", async () => { + const f = fixture(); + f.store.send("hello", images); + f.start("hello"); + expect(f.store.chunks).toHaveLength(3); + f.history.returnChunks = [ + { seq: 1, role: "user", turnId: "t1", chunk: { type: "text", text: "hello" } }, + { seq: 2, role: "user", turnId: "t1", chunk: { type: "image", url: "/images/c1/saved.png" } }, + ]; + f.event({ type: "turn-sealed", conversationId: "c1", turnId: "t1" }); + await vi.waitFor(() => expect(f.store.pendingSync).toBe(false)); + expect(f.store.chunks.map((c) => c.chunk)).toEqual(f.history.returnChunks.map((c) => c.chunk)); + expect(f.store.chunks.every((c) => !c.provisional)).toBe(true); + expect(f.transport.sent).toHaveLength(1); + f.store.dispose(); + }); +}); diff --git a/src/features/chat/store.svelte.ts b/src/features/chat/store.svelte.ts index f48d441..66ee5ee 100644 --- a/src/features/chat/store.svelte.ts +++ b/src/features/chat/store.svelte.ts @@ -2,6 +2,7 @@ import type { ChatDeltaMessage, ChatErrorMessage, ChatQueueCancelMessage, + ChatQueuedMessage, ChatQueueMessage, ChatSendMessage, } from "@dispatch/transport-contract"; @@ -15,6 +16,7 @@ import { initialState, initialWindowSize, normalizeChatLimit, + removeLocalEcho, restoreEarlier, selectChunks, selectGenerating, @@ -43,6 +45,8 @@ import type { ChatTransport, HistorySync, MetricsSync } from "./ports"; export interface ChatStoreDependencies { readonly conversationId: string; + /** Injected unique client-local correlation token for each new send. */ + readonly createRequestId: () => string; readonly model?: string; readonly transport: ChatTransport; readonly historySync: HistorySync; @@ -142,7 +146,7 @@ export interface ChatStore { * Ignored while a turn is generating; a running turn gap keeps the later start. */ noteLastActivity(at: number): void; - handleDelta(msg: ChatDeltaMessage | ChatErrorMessage): void; + handleDelta(msg: ChatDeltaMessage | ChatErrorMessage | ChatQueuedMessage): void; /** * Send a user message (start a turn via `chat.send`). Optimistically echoes * the text + any `images` as provisional user chunks (`[text, image, …]` in @@ -540,7 +544,7 @@ export function createChatStore(deps: ChatStoreDependencies): ChatStore { cacheTimer = seedTurnGap(cacheTimer, at); }, - handleDelta(msg: ChatDeltaMessage | ChatErrorMessage): void { + handleDelta(msg: ChatDeltaMessage | ChatErrorMessage | ChatQueuedMessage): void { if (msg.type === "chat.error") { if (msg.conversationId !== undefined && msg.conversationId !== deps.conversationId) { return; @@ -548,6 +552,14 @@ export function createChatStore(deps: ChatStoreDependencies): ChatStore { _error = msg.message; return; } + if (msg.type === "chat.queued") { + if (disposed || msg.conversationId !== deps.conversationId) return; + // A queued live opening may already have confirmed the echoed text. + // Fold it first; turn-start alone must not protect still-local artifacts. + flushEvents(); + if (msg.requestId !== undefined) transcript = removeLocalEcho(transcript, msg.requestId); + return; + } if (msg.event.conversationId !== deps.conversationId) { return; } @@ -566,10 +578,12 @@ export function createChatStore(deps: ChatStoreDependencies): ChatStore { flushEvents(); // Sending ends the wait since the last turn: stamp it on that turn. cacheTimer = stopTurnGap(cacheTimer, now(), () => lastTurnIdOf(metrics)); - transcript = appendUserMessage(transcript, text, images); + const requestId = deps.createRequestId(); + transcript = appendUserMessage(transcript, text, images, requestId); maybeTrim(); const msg: ChatSendMessage = { type: "chat.send", + requestId, conversationId: deps.conversationId, message: text, ...(_model !== undefined ? { model: _model } : {}), diff --git a/src/features/chat/store.test.ts b/src/features/chat/store.test.ts index de84b8a..c24592d 100644 --- a/src/features/chat/store.test.ts +++ b/src/features/chat/store.test.ts @@ -32,6 +32,7 @@ describe("createChatStore", () => { const metricsSync = createFakeMetricsSync(); const cache = createFakeCache(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -64,6 +65,7 @@ describe("createChatStore", () => { const metricsSync = createFakeMetricsSync(); const cache = createFakeCache(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -107,6 +109,7 @@ describe("createChatStore", () => { const metricsSync = createFakeMetricsSync(); const cache = createFakeCache(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -131,6 +134,7 @@ describe("createChatStore", () => { const metricsSync = createFakeMetricsSync(); const cache = createFakeCache(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, model: "openai/gpt-4", transport: transport.impl, @@ -153,6 +157,7 @@ describe("createChatStore", () => { const metricsSync = createFakeMetricsSync(); const cache = createFakeCache(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -185,6 +190,7 @@ describe("createChatStore", () => { it("send omits images on the wire when none are staged (backward compatible)", () => { const transport = createFakeTransport(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: createFakeHistorySync().impl, @@ -201,6 +207,7 @@ describe("createChatStore", () => { it("send omits images on the wire for an empty array", () => { const transport = createFakeTransport(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: createFakeHistorySync().impl, @@ -217,6 +224,7 @@ describe("createChatStore", () => { it("send allows an images-only message (empty text)", () => { const transport = createFakeTransport(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: createFakeHistorySync().impl, @@ -241,6 +249,7 @@ describe("createChatStore", () => { const metricsSync = createFakeMetricsSync(); const cache = createFakeCache(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -263,6 +272,7 @@ describe("createChatStore", () => { it.each([" describe this ", "", " "])("queues images with text %j", (text) => { const transport = createFakeTransport(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, workspaceId: "images-workspace", transport: transport.impl, @@ -298,6 +308,7 @@ describe("createChatStore", () => { const metricsSync = createFakeMetricsSync(); const cache = createFakeCache(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -318,6 +329,7 @@ describe("createChatStore", () => { const metricsSync = createFakeMetricsSync(); const cache = createFakeCache(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -340,6 +352,7 @@ describe("createChatStore", () => { const metricsSync = createFakeMetricsSync(); const cache = createFakeCache(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -362,6 +375,7 @@ describe("createChatStore", () => { const metricsSync = createFakeMetricsSync(); const cache = createFakeCache(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -387,6 +401,7 @@ describe("createChatStore", () => { const metricsSync = createFakeMetricsSync(); const cache = createFakeCache(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -408,6 +423,7 @@ describe("createChatStore", () => { const metricsSync = createFakeMetricsSync(); const cache = createFakeCache(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -433,6 +449,7 @@ describe("createChatStore", () => { const metricsSync = createFakeMetricsSync(); const cache = createFakeCache(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -462,6 +479,7 @@ describe("createChatStore", () => { historySync.returnChunks = [makeStoredChunk(3, "assistant")]; const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -490,6 +508,7 @@ describe("createChatStore", () => { historySync.returnChunks = [makeStoredChunk(1, "assistant")]; const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -511,6 +530,7 @@ describe("createChatStore", () => { const metricsSync = createFakeMetricsSync(); const cache = createFakeCache(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -543,6 +563,7 @@ describe("createChatStore", () => { const metricsSync = createFakeMetricsSync(); const cache = createFakeCache(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -587,6 +608,7 @@ describe("createChatStore", () => { }; const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: slowHistorySync, @@ -621,6 +643,7 @@ describe("createChatStore", () => { const metricsSync = createFakeMetricsSync(); const cache = createFakeCache(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -666,6 +689,7 @@ describe("createChatStore", () => { const metricsSync = createFakeMetricsSync(); const cache = createFakeCache(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, model: "openai/gpt-4", transport: transport.impl, @@ -690,6 +714,7 @@ describe("createChatStore", () => { const metricsSync = createFakeMetricsSync(); const cache = createFakeCache(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -713,6 +738,7 @@ describe("createChatStore", () => { const metricsSync = createFakeMetricsSync(); const cache = createFakeCache(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -743,6 +769,7 @@ describe("createChatStore", () => { const metricsSync = createFakeMetricsSync(); const cache = createFakeCache(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -763,6 +790,7 @@ describe("createChatStore", () => { const metricsSync = createFakeMetricsSync(); const cache = createFakeCache(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -787,6 +815,7 @@ describe("createChatStore", () => { const metricsSync = createFakeMetricsSync(); const cache = createFakeCache(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -825,6 +854,7 @@ describe("createChatStore", () => { const metricsSync = createFakeMetricsSync(); const cache = createFakeCache(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -886,6 +916,7 @@ describe("createChatStore", () => { const metricsSync = createFakeMetricsSync(); const cache = createFakeCache(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -930,6 +961,7 @@ describe("createChatStore", () => { const metricsSync = createFakeMetricsSync(); const cache = createFakeCache(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -1002,6 +1034,7 @@ describe("createChatStore", () => { const metricsSync = createFakeMetricsSync(); const cache = createFakeCache(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -1068,6 +1101,7 @@ describe("createChatStore", () => { ]; const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -1092,6 +1126,7 @@ describe("createChatStore", () => { const metricsSync = createFakeMetricsSync(); const cache = createFakeCache(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -1123,6 +1158,7 @@ describe("createChatStore", () => { const metricsSync = createFakeMetricsSync(); const cache = createFakeCache(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -1147,6 +1183,7 @@ describe("createChatStore", () => { const metricsSync = createFakeMetricsSync(); const cache = createFakeCache(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -1181,6 +1218,7 @@ describe("createChatStore", () => { const metricsSync = createFakeMetricsSync(); const cache = createFakeCache(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -1227,6 +1265,7 @@ describe("createChatStore", () => { const cache = createFakeCache(); let atBottom = false; // reader scrolled up const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -1281,6 +1320,7 @@ describe("createChatStore", () => { const cache = createFakeCache(); let atBottom = false; const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -1319,6 +1359,7 @@ describe("createChatStore", () => { ); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -1348,6 +1389,7 @@ describe("createChatStore", () => { historySync.returnChunks = Array.from({ length: 500 }, (_, i) => makeStoredChunk(i + 1)); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -1383,6 +1425,7 @@ describe("createChatStore", () => { historySync.returnChunks = [makeStoredChunk(3)]; const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -1407,6 +1450,7 @@ describe("createChatStore", () => { historySync.returnChunks = Array.from({ length: 500 }, (_, i) => makeStoredChunk(i + 1)); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -1444,6 +1488,7 @@ describe("createChatStore", () => { ); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -1476,6 +1521,7 @@ describe("createChatStore", () => { ); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -1507,6 +1553,7 @@ describe("createChatStore", () => { ); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -1539,6 +1586,7 @@ describe("createChatStore", () => { const metricsSync = createFakeMetricsSync(); const cache = createFakeCache(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -1576,6 +1624,7 @@ describe("createChatStore", () => { Array.from({ length: 200 }, (_, i) => makeStoredChunk(i + 1)), ); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -1607,6 +1656,7 @@ describe("createChatStore", () => { // Server holds 200; cold-cache load at limit 100 → window 75 → seqs 126..200. historySync.returnChunks = Array.from({ length: 200 }, (_, i) => makeStoredChunk(i + 1)); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -1634,6 +1684,7 @@ describe("createChatStore", () => { const metricsSync = createFakeMetricsSync(); const cache = createFakeCache(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -1673,6 +1724,7 @@ describe("createChatStore", () => { Array.from({ length: 50 }, (_, i) => makeStoredChunk(i + 1)), ); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -1700,6 +1752,7 @@ describe("createChatStore", () => { const metricsSync = createFakeMetricsSync(); const cache = createFakeCache(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -1729,6 +1782,7 @@ describe("createChatStore", () => { const metricsSync = createFakeMetricsSync(); const cache = createFakeCache(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -1750,6 +1804,7 @@ describe("createChatStore — batched delta application (scheduleFlush)", () => const scheduled: (() => void)[] = []; const historySync = createFakeHistorySync(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: createFakeTransport().impl, historySync: historySync.impl, @@ -1831,6 +1886,7 @@ describe("createChatStore — batched delta application (scheduleFlush)", () => describe("createChatStore — cacheTimer", () => { function makeStore(now: () => number) { return createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: createFakeTransport().impl, historySync: createFakeHistorySync().impl, @@ -1940,6 +1996,7 @@ describe("history epochs", () => { const cache = createFakeCache(); history.returnChunks = Array.from({ length: 20 }, (_, i) => makeStoredChunk(i + 1)); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: createFakeTransport().impl, historySync: history.impl, @@ -1980,6 +2037,7 @@ it("a compaction broadcast during an in-flight fetch discards the old response", .mockImplementation(history.impl); const cache = createFakeCache(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: createFakeTransport().impl, historySync: sync, @@ -2006,6 +2064,7 @@ it("reloads unknown legacy cache epochs rather than merging pre-upgrade rows", a const history = createFakeHistorySync(); history.returnChunks = [makeStoredChunk(101)]; const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: createFakeTransport().impl, historySync: history.impl, @@ -2037,6 +2096,7 @@ describe("server epoch rollback", () => { const history = createFakeHistorySync(); history.historyEpoch = 5; const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: createFakeTransport().impl, historySync: history.impl, @@ -2083,6 +2143,7 @@ describe("server epoch rollback", () => { return { chunks: [], latestSeq: 0, historyEpoch: calls - 1 }; }; store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: createFakeTransport().impl, historySync: sync, @@ -2140,6 +2201,7 @@ it.each([ }; }); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: createFakeTransport().impl, historySync: sync, @@ -2182,6 +2244,7 @@ it("replaces an empty high-epoch cache with the server's empty lower epoch in on await cache.reset(CONV_ID, 8); const history = createFakeHistorySync(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: createFakeTransport().impl, historySync: history.impl, -- cgit v1.2.3 From ac43e92d174ed1f081c387944e9044643ee9efc2 Mon Sep 17 00:00:00 2001 From: Adam Malczewski Date: Wed, 21 Oct 2026 15:36:54 +0900 Subject: fix(review-followup-fee): recover omitted chat payloads through history --- src/adapters/ws/index.test.ts | 37 +++ src/adapters/ws/index.ts | 20 +- src/adapters/ws/logic.test.ts | 29 ++ src/adapters/ws/logic.ts | 14 + src/app/store.test.ts | 53 +++ src/app/views/ChatPane.svelte | 1 + src/app/ws-handlers.ts | 7 +- src/core/chunks/groups.test.ts | 22 ++ src/core/chunks/groups.ts | 14 +- src/core/chunks/index.ts | 1 + src/core/chunks/local-echo.test.ts | 2 +- src/core/chunks/reconcile.test.ts | 143 ++++++++ src/core/chunks/reconcile.ts | 54 +++ src/core/chunks/reducer.test.ts | 26 +- src/core/chunks/reducer.ts | 174 +++++++++- src/core/chunks/selectors.test.ts | 30 +- src/core/chunks/selectors.ts | 13 +- src/core/chunks/types.ts | 8 + src/core/wire/conformance.test.ts | 38 ++- src/core/wire/conformance.ts | 2 + src/features/chat/logic/tool-summary.ts | 8 +- src/features/chat/recovery.test.ts | 529 ++++++++++++++++++++++++++++++ src/features/chat/store.svelte.ts | 128 +++++++- src/features/chat/ui/ChatView.svelte | 10 +- src/features/chat/ui/ChatView.test.ts | 44 +++ src/features/heartbeat/ui/RunModal.svelte | 1 + 26 files changed, 1364 insertions(+), 44 deletions(-) create mode 100644 src/core/chunks/reconcile.test.ts create mode 100644 src/core/chunks/reconcile.ts create mode 100644 src/features/chat/recovery.test.ts (limited to 'src/adapters') diff --git a/src/adapters/ws/index.test.ts b/src/adapters/ws/index.test.ts index 6461493..e4e4ba4 100644 --- a/src/adapters/ws/index.test.ts +++ b/src/adapters/ws/index.test.ts @@ -499,3 +499,40 @@ it("routes chat.queued only through the typed chat callback, never surface/error expect(onMessage).not.toHaveBeenCalled(); handle.close(); }); + +it("routes resync notices to chat without reconnect/resend or surface errors", () => { + vi.useFakeTimers(); + const ws = fakeSocket(); + const onChat = vi.fn(); + const onMessage = vi.fn(); + const onReopen = vi.fn(); + const factory = vi.fn(() => ws); + const handle = createSurfaceSocket({ + url: "ws://test", + socketFactory: factory, + onChat, + onMessage, + onReopen, + }); + try { + ws.resolveOpen(); + const notice = { + type: "chat.resync-required", + conversationId: "c1", + turnId: "t1", + reason: "replay-limit", + }; + ws.invokeMessage(JSON.stringify(notice)); + ws.invokeMessage(JSON.stringify(notice)); + vi.advanceTimersByTime(1000); + expect(onChat).toHaveBeenCalledTimes(2); + expect(onChat).toHaveBeenCalledWith(notice); + expect(onMessage).not.toHaveBeenCalled(); + expect(onReopen).not.toHaveBeenCalled(); + expect(factory).toHaveBeenCalledTimes(1); + expect(ws.sent).toEqual([]); + } finally { + handle.close(); + vi.useRealTimers(); + } +}); diff --git a/src/adapters/ws/index.ts b/src/adapters/ws/index.ts index 25e9911..b1e17cf 100644 --- a/src/adapters/ws/index.ts +++ b/src/adapters/ws/index.ts @@ -2,6 +2,7 @@ import type { ChatDeltaMessage, ChatErrorMessage, ChatQueuedMessage, + ChatResyncRequiredMessage, ConversationCompactedMessage, ConversationOpenMessage, ConversationStatusChangedMessage, @@ -25,7 +26,9 @@ export interface WebSocketLike { export interface SurfaceSocketOptions { url: string; onMessage: (msg: SurfaceServerMessage) => void; - onChat?: (msg: ChatDeltaMessage | ChatErrorMessage | ChatQueuedMessage) => void; + onChat?: ( + msg: ChatDeltaMessage | ChatErrorMessage | ChatQueuedMessage | ChatResyncRequiredMessage, + ) => void; /** Broadcast when a conversation is "opened" (e.g. CLI `--open` flag). */ onConversationOpen?: (msg: ConversationOpenMessage) => void; /** Broadcast when a conversation's lifecycle status changes (active/idle/closed). */ @@ -101,8 +104,19 @@ export function createSurfaceSocket(opts: SurfaceSocketOptions): SurfaceSocketHa if (disposed) return; const msg = parseServerMessage(ev.data); if (msg !== null) { - if (msg.type === "chat.delta" || msg.type === "chat.error" || msg.type === "chat.queued") { - opts.onChat?.(msg as ChatDeltaMessage | ChatErrorMessage | ChatQueuedMessage); + if ( + msg.type === "chat.delta" || + msg.type === "chat.error" || + msg.type === "chat.queued" || + msg.type === "chat.resync-required" + ) { + opts.onChat?.( + msg as + | ChatDeltaMessage + | ChatErrorMessage + | ChatQueuedMessage + | ChatResyncRequiredMessage, + ); } else if (msg.type === "conversation.open") { opts.onConversationOpen?.(msg as ConversationOpenMessage); } else if (msg.type === "conversation.statusChanged") { diff --git a/src/adapters/ws/logic.test.ts b/src/adapters/ws/logic.test.ts index db6444a..4bc2aad 100644 --- a/src/adapters/ws/logic.test.ts +++ b/src/adapters/ws/logic.test.ts @@ -609,3 +609,32 @@ describe("chat.queued admission acknowledgement", () => { expect(JSON.parse(serialize(msg))).toEqual(msg); }); }); + +describe("chat.resync-required", () => { + it.each(["replay-limit", "payload-limit"])("parses %s with optional turn identity", (reason) => { + const msg = { type: "chat.resync-required", conversationId: "c1", reason }; + expect(parseServerMessage(JSON.stringify(msg))).toEqual(msg); + expect(parseServerMessage(JSON.stringify({ ...msg, turnId: "t1" }))).toEqual({ + ...msg, + turnId: "t1", + }); + }); + it.each([ + {}, + { conversationId: null }, + { conversationId: 1 }, + { reason: "unknown" }, + { reason: null }, + { turnId: null }, + { turnId: 42 }, + ])("rejects malformed notice %j", (bad) => { + const msg = { + type: "chat.resync-required", + conversationId: "c1", + reason: "replay-limit", + ...bad, + }; + if (Object.keys(bad).length === 0) delete (msg as { conversationId?: unknown }).conversationId; + expect(parseServerMessage(JSON.stringify(msg))).toBeNull(); + }); +}); diff --git a/src/adapters/ws/logic.ts b/src/adapters/ws/logic.ts index 0efcd90..abc0464 100644 --- a/src/adapters/ws/logic.ts +++ b/src/adapters/ws/logic.ts @@ -2,6 +2,7 @@ import type { ChatDeltaMessage, ChatErrorMessage, ChatQueuedMessage, + ChatResyncRequiredMessage, ConversationCompactedMessage, ConversationOpenMessage, ConversationStatusChangedMessage, @@ -27,6 +28,7 @@ const VALID_SERVER_TYPES = new Set([ "chat.delta", "chat.error", "chat.queued", + "chat.resync-required", "conversation.open", "conversation.statusChanged", "conversation.compacted", @@ -113,6 +115,18 @@ export function parseServerMessage(data: string): WsServerMessage | null { if (typeof event.type !== "string") return null; return { type: "chat.delta", event: event as unknown as ChatDeltaMessage["event"] }; } + case "chat.resync-required": { + if (typeof parsed.conversationId !== "string") return null; + if (parsed.turnId !== undefined && typeof parsed.turnId !== "string") return null; + if (parsed.reason !== "replay-limit" && parsed.reason !== "payload-limit") return null; + const msg: ChatResyncRequiredMessage = { + type: "chat.resync-required", + conversationId: parsed.conversationId, + reason: parsed.reason, + ...(parsed.turnId !== undefined ? { turnId: parsed.turnId } : {}), + }; + return msg; + } case "chat.queued": { if (typeof parsed.conversationId !== "string") return null; if (typeof parsed.messageId !== "string") return null; diff --git a/src/app/store.test.ts b/src/app/store.test.ts index 40a0fa7..33f93bb 100644 --- a/src/app/store.test.ts +++ b/src/app/store.test.ts @@ -3076,3 +3076,56 @@ describe("compaction queued acknowledgement routing", () => { store.dispose(); }); }); + +it("routes omitted replay only to the matching watch/tab without WS resend or an error banner", async () => { + const ws = fakeSocket(); + const store = createAppStore({ + socketFactory: () => ws, + fetchImpl: fakeFetchImpl(), + indexedDB: new IDBFactory(), + localStorage: createFakeStorage(), + chatFlushScheduler: null, + }); + try { + ws.resolveOpen(); + const tab = store.activeChat; + const watch = store.watchConversation("omitted-watch"); + const sentBefore = ws.sent.length; + ws.feedServerMessage({ + type: "chat.resync-required", + conversationId: "unknown", + turnId: "unknown-turn", + reason: "replay-limit", + }); + expect(watch.recovering).toBe(false); + expect(tab.recovering).toBe(false); + ws.feedServerMessage({ + type: "chat.resync-required", + conversationId: "omitted-watch", + turnId: "t1", + reason: "replay-limit", + }); + expect(watch.recovering).toBe(true); + expect(watch.generating).toBe(true); + expect(tab.recovering).toBe(false); + expect(watch.error).toBeNull(); + expect(tab.error).toBeNull(); + expect(store.fatalError).toBeNull(); + expect(ws.sent).toHaveLength(sentBefore); + ws.feedServerMessage({ + type: "chat.delta", + event: { + type: "turn-ended", + conversationId: "omitted-watch", + turnId: "t1", + reason: "aborted", + }, + }); + await vi.waitFor(() => expect(watch.pendingSync).toBe(false)); + expect(watch.recovering).toBe(false); + expect(watch.generating).toBe(false); + expect(ws.sent).toHaveLength(sentBefore); + } finally { + store.dispose(); + } +}); diff --git a/src/app/views/ChatPane.svelte b/src/app/views/ChatPane.svelte index d1630af..62b9344 100644 --- a/src/app/views/ChatPane.svelte +++ b/src/app/views/ChatPane.svelte @@ -80,6 +80,7 @@ {#key store.activeConversationId} { expect(groups.map((g) => g.kind)).toEqual(["single", "tool"]); }); }); + +it("keeps windowed committed orphan calls in Show earlier even while another payload recovers", () => { + const groups = groupRenderedChunks( + [ + { + seq: 50, + role: "tool", + provisional: false, + chunk: { + type: "tool-result", + toolCallId: "older", + toolName: "run_shell", + content: "output", + isError: false, + }, + }, + ], + true, + ); + expect(groups[0]).toMatchObject({ kind: "tool", entry: { call: null } }); + if (groups[0]?.kind === "tool") expect(groups[0].entry.recovering).toBeUndefined(); +}); diff --git a/src/core/chunks/groups.ts b/src/core/chunks/groups.ts index c8493cc..6e0f8ea 100644 --- a/src/core/chunks/groups.ts +++ b/src/core/chunks/groups.ts @@ -3,6 +3,7 @@ import type { RenderedChunk } from "./types"; /** One tool call and its response. Null call = unloaded history; null result = pending. */ export interface ToolEntry { + readonly recovering?: boolean; readonly toolCallId: string; readonly toolName: string; readonly call: ToolCallChunk | null; @@ -26,7 +27,10 @@ export type RenderGroup = * the stable t render keys are always unique, even on corrupt replay. * Pure: input → output; no DOM, effects, or Svelte. */ -export function groupRenderedChunks(rendered: readonly RenderedChunk[]): readonly RenderGroup[] { +export function groupRenderedChunks( + rendered: readonly RenderedChunk[], + recovering = false, +): readonly RenderGroup[] { const calls = new Map(); const results = new Map(); for (const rc of rendered) { @@ -55,7 +59,13 @@ export function groupRenderedChunks(rendered: readonly RenderedChunk[]): readonl const turnId = callRow?.turnId ?? resultRow?.turnId; groups.push({ kind: "tool", - entry: { toolCallId: c.toolCallId, toolName: call?.toolName ?? c.toolName, call, result }, + entry: { + toolCallId: c.toolCallId, + toolName: call?.toolName ?? c.toolName, + call, + result, + ...(recovering && call === null && resultRow?.provisional ? { recovering: true } : {}), + }, provisional: Boolean(callRow?.provisional || resultRow?.provisional), ...(stepId !== undefined ? { stepId } : {}), ...(turnId !== undefined ? { turnId } : {}), diff --git a/src/core/chunks/index.ts b/src/core/chunks/index.ts index 49381d5..03d687c 100644 --- a/src/core/chunks/index.ts +++ b/src/core/chunks/index.ts @@ -7,6 +7,7 @@ export { clearGenerating, foldEvent, initialState, + markHistoryRecovering, removeLocalEcho, } from "./reducer"; export type { ProviderRetryView } from "./retry-banner"; diff --git a/src/core/chunks/local-echo.test.ts b/src/core/chunks/local-echo.test.ts index 4b5eed4..565017c 100644 --- a/src/core/chunks/local-echo.test.ts +++ b/src/core/chunks/local-echo.test.ts @@ -28,7 +28,7 @@ describe("request-local optimistic echo", () => { }); state = removeLocalEcho(removeLocalEcho(state, "r1"), "r2"); expect(state.provisional).toEqual([ - { role: "user", turnId: "t1", chunk: { type: "text", text: "same" } }, + { role: "user", turnId: "t1", opening: true, chunk: { type: "text", text: "same" } }, ]); }); diff --git a/src/core/chunks/reconcile.test.ts b/src/core/chunks/reconcile.test.ts new file mode 100644 index 0000000..3d90a51 --- /dev/null +++ b/src/core/chunks/reconcile.test.ts @@ -0,0 +1,143 @@ +import type { AgentEvent, StepId, StoredChunk } from "@dispatch/wire"; +import { describe, expect, it } from "vitest"; +import { applyHistory, foldEvent, initialState, markHistoryRecovering } from "./reducer"; +import { selectChunks } from "./selectors"; +import type { TranscriptState } from "./types"; + +const stepId = "t1#1" as StepId; +function fold(s: TranscriptState, e: AgentEvent) { + return foldEvent(s, e); +} +const call = (toolCallId = "a", turnId = "t1", step = stepId): AgentEvent => ({ + type: "tool-call", + conversationId: "c1", + turnId, + stepId: step, + toolCallId, + toolName: "run_shell", + input: "partial", +}); +const result = (toolCallId = "a"): AgentEvent => ({ + type: "tool-result", + conversationId: "c1", + turnId: "t1", + stepId, + toolCallId, + toolName: "run_shell", + content: "partial", + isError: false, +}); +const stored: StoredChunk[] = [ + { seq: 50, role: "assistant", turnId: "t1", chunk: { type: "text", text: "full" } }, + { + seq: 51, + role: "assistant", + turnId: "t1", + chunk: { + type: "tool-call", + stepId, + toolCallId: "a", + toolName: "run_shell", + input: { command: "full" }, + }, + }, + { + seq: 52, + role: "tool", + turnId: "t1", + chunk: { + type: "tool-result", + stepId, + toolCallId: "a", + toolName: "run_shell", + content: "full result", + isError: false, + }, + }, +]; + +describe("history recovery identities", () => { + it.each([ + "before-live", + "after-live", + ])("reconciles committed tool/text identity %s, not payload equality", (order) => { + let s = markHistoryRecovering(initialState(), "t1"); + if (order === "before-live") s = applyHistory(s, stored); + s = fold(s, { type: "text-delta", conversationId: "c1", turnId: "t1", delta: "short" }); + s = fold(s, call()); + s = fold(s, result()); + if (order === "after-live") s = applyHistory(s, stored); + expect(selectChunks(s).map((c) => c.chunk)).toEqual(stored.map((c) => c.chunk)); + // Idempotent refresh; different step's identical text must remain live. + s = fold(s, { type: "text-delta", conversationId: "c1", turnId: "t1", delta: "full" }); + s = applyHistory(s, stored); + expect(selectChunks(s).at(-1)).toMatchObject({ + provisional: true, + streaming: true, + chunk: { type: "text", text: "full" }, + }); + }); + + it("does not conflate call/result, another turn, or another step's tool identity", () => { + let s = markHistoryRecovering(initialState(), "t1"); + s = fold(s, call("a", "other")); + s = fold(s, call("a", "t1", "other-step" as StepId)); + s = fold(s, result("a")); + s = applyHistory(s, stored); + expect(s.provisional).toHaveLength(2); + expect(s.provisional.map((p) => p.turnId)).toEqual(["other", "t1"]); + expect(s.provisional.every((p) => p.chunk.type === "tool-call")).toBe(true); + }); + + it("terminal sync removes only that turn and keeps optimistic/unrelated/carried content", () => { + let s = markHistoryRecovering(initialState(), "t1"); + s = fold(s, call()); + s = fold(s, { type: "turn-sealed", conversationId: "c1", turnId: "t1" }); + s = fold(s, { type: "turn-start", conversationId: "c1", turnId: "t2" }); + s = fold(s, { type: "text-delta", conversationId: "c1", turnId: "t2", delta: "queued tail" }); + s = applyHistory(s, stored, ["t1"]); + expect(s.generating).toBe(true); + expect(s.currentTurnId).toBe("t2"); + expect(s.accumulating?.text).toBe("queued tail"); + expect(s.historyRecoveryTurnIds).toEqual([]); + }); + + it("does not resurrect watermark-hidden history; older content stays available to Show earlier", () => { + let s = markHistoryRecovering({ ...initialState(), hiddenBeforeSeq: 51 }, "t1"); + s = fold(s, call()); + s = applyHistory(s, stored); + expect(s.committed.map((c) => c.seq)).toEqual([51, 52]); + expect(s.provisional).toEqual([]); + expect(s.hiddenBeforeSeq).toBe(51); + }); +}); + +it("keeps identical Stop→carry openings as distinct turn identities", () => { + let s = markHistoryRecovering(initialState(), "t1"); + s = fold(s, { type: "user-message", conversationId: "c1", turnId: "t1", text: "same" }); + s = fold(s, { type: "turn-ended", conversationId: "c1", turnId: "t1", reason: "aborted" }); + s = fold(s, { type: "user-message", conversationId: "c1", turnId: "t2", text: "same" }); + s = fold(s, { type: "turn-start", conversationId: "c1", turnId: "t2" }); + s = applyHistory( + s, + [{ seq: 1, role: "user", turnId: "t1", chunk: { type: "text", text: "same" } }], + ["t1"], + ); + expect(selectChunks(s).map((c) => [c.turnId, c.provisional])).toEqual([ + ["t1", false], + ["t2", true], + ]); + expect(s.generating).toBe(true); +}); + +it("reconciles steering by its completed tool batch, not repeated user text", () => { + let s = markHistoryRecovering(initialState(), "t1"); + s = fold(s, result()); + s = fold(s, { type: "steering", conversationId: "c1", turnId: "t1", text: "short" }); + s = applyHistory(s, [ + ...stored, + { seq: 53, role: "user", turnId: "t1", chunk: { type: "text", text: "full steering" } }, + ]); + expect(s.provisional).toEqual([]); + expect(selectChunks(s).at(-1)?.chunk).toEqual({ type: "text", text: "full steering" }); +}); diff --git a/src/core/chunks/reconcile.ts b/src/core/chunks/reconcile.ts new file mode 100644 index 0000000..277d25b --- /dev/null +++ b/src/core/chunks/reconcile.ts @@ -0,0 +1,54 @@ +import type { StoredChunk } from "@dispatch/wire"; +import type { ProvisionalChunk, TranscriptState } from "./types"; + +/** + * A persisted tool call proves its entire assistant step committed atomically. + * Text/thinking wire chunks have no stepId: their provisional provenance is + * learned from tool-call/step-complete, never from text equality. This also + * handles a window starting midway through the step and an omitted prefix. + * Also usable when HTTP finished before replay/live flush. + */ +export function historyBackedProvisional( + state: TranscriptState, + committed: readonly StoredChunk[] = state.committed, +): ReadonlySet { + const backed = new Set(); + if (!state.historyRecoveryTurnIds?.length) return backed; + const steps = new Set(); + const steeringAnchors = new Set(); + let lastResult: StoredChunk | undefined; + const toolRows = new Map>(); + for (const row of committed) { + const c = row.chunk; + if (row.role === "assistant" || row.turnId !== lastResult?.turnId) lastResult = undefined; + if (c.type === "tool-result") lastResult = row; + if (row.role === "user" && c.type === "text" && lastResult?.chunk.type === "tool-result") { + steeringAnchors.add(`${row.turnId}\0${lastResult.chunk.toolCallId}`); + } + if (c.type === "tool-call" || c.type === "tool-result") { + const key = `${row.turnId}\0${c.type}\0${c.toolCallId}`; + const ids = toolRows.get(key) ?? new Set(); + ids.add(c.stepId); + toolRows.set(key, ids); + if (c.type === "tool-call" && c.stepId !== undefined) steps.add(`${row.turnId}\0${c.stepId}`); + } + } + for (const p of state.provisional) { + if (p.turnId === undefined || !state.historyRecoveryTurnIds?.includes(p.turnId)) continue; + const c = p.chunk; + if (c.type === "tool-call" || c.type === "tool-result") { + const ids = toolRows.get(`${p.turnId}\0${c.type}\0${c.toolCallId}`); + if (ids !== undefined && (c.stepId === undefined || ids.has(undefined) || ids.has(c.stepId))) + backed.add(p); + } else if (p.afterToolCallId !== undefined && p.role === "user") { + if (steeringAnchors.has(`${p.turnId}\0${p.afterToolCallId}`)) backed.add(p); + } else if (p.opening && c.type === "text") { + // The opening is the first user row of this turn; steering is NOT opening. + const first = committed.find((row) => row.turnId === p.turnId); + if (first?.role === "user" && first.chunk.type === "text") backed.add(p); + } else if (p.stepId !== undefined && p.role === "assistant") { + if (steps.has(`${p.turnId}\0${p.stepId}`)) backed.add(p); + } + } + return backed; +} diff --git a/src/core/chunks/reducer.test.ts b/src/core/chunks/reducer.test.ts index ae0aa73..22b94d4 100644 --- a/src/core/chunks/reducer.test.ts +++ b/src/core/chunks/reducer.test.ts @@ -1053,7 +1053,7 @@ describe("turnId provenance on live chunks", () => { }); describe("foldEvent — additive wire compatibility", () => { - it.each(["turn-ended", "future-event"])("preserves transcript identity for %s", (type) => { + it.each(["future-event"])("preserves transcript identity for %s", (type) => { const state = foldEvent(foldEvent(initialState(), turnStart("t1")), textDelta("t1", "partial")); const before = structuredClone(state); const event = { type, conversationId: "c1", turnId: "t1", reason: "sealed" }; @@ -1061,3 +1061,27 @@ describe("foldEvent — additive wire compatibility", () => { expect(state).toEqual(before); }); }); + +describe("foldEvent — authoritative turn terminal", () => { + it.each([ + "sealed", + "aborted", + "error", + "rejected", + ] as const)("ends generation on %s", (reason) => { + const before = foldEvent( + foldEvent(initialState(), turnStart("t1")), + textDelta("t1", "partial"), + ); + const next = foldEvent(before, { + type: "turn-ended", + conversationId: "c1", + turnId: "t1", + reason, + }); + expect(next.generating).toBe(false); + expect(next.accumulating).toBeNull(); + expect(selectChunks(next).map((c) => c.chunk)).toEqual([{ type: "text", text: "partial" }]); + expect(before.accumulating?.text).toBe("partial"); + }); +}); diff --git a/src/core/chunks/reducer.ts b/src/core/chunks/reducer.ts index 53af9c1..ee5858f 100644 --- a/src/core/chunks/reducer.ts +++ b/src/core/chunks/reducer.ts @@ -1,5 +1,6 @@ import type { AgentEvent, Chunk, ImageInput, Role, StoredChunk } from "@dispatch/wire"; import { assertChunkExhaustive } from "../wire/conformance"; +import { historyBackedProvisional } from "./reconcile"; import type { AccumulatingChunk, ProvisionalChunk, TranscriptState } from "./types"; /** The initial empty transcript state. */ @@ -40,7 +41,13 @@ function flushAccumulating( if (acc === null) return provisional; const chunk: Chunk = acc.kind === "text" ? { type: "text", text: acc.text } : { type: "thinking", text: acc.text }; - return [...provisional, withTurn({ role: "assistant", chunk }, acc.turnId)]; + return [ + ...provisional, + withTurn( + { role: "assistant", chunk, ...(acc.stepId !== undefined ? { stepId: acc.stepId } : {}) }, + acc.turnId, + ), + ]; } /** Attach a `turnId` to a provisional chunk (omitted when unknown). */ @@ -129,8 +136,8 @@ function trailingRun( /** * Merge authoritative seq-keyed chunks into the committed history. * Dedupes by seq (new wins), keeps seq-monotonic order, idempotent. - * When sealedTurnId is set, drops all provisional chunks (now superseded) - * and clears sealedTurnId. + * A sealed/finally-synced turn supersedes only its own provisional chunks; + * unrelated turns and an already-running queued carry remain live. * * Chunks below the chat-limit unload watermark (`hiddenBeforeSeq`) are * REJECTED: a full-cache or tail merge must not resurrect what the trim @@ -139,6 +146,7 @@ function trailingRun( export function applyHistory( state: TranscriptState, chunks: readonly StoredChunk[], + finishedTurnIds: readonly string[] = [], ): TranscriptState { const seqMap = new Map(); for (const c of state.committed) seqMap.set(c.seq, c); @@ -149,14 +157,49 @@ export function applyHistory( seqMap.set(c.seq, c); } const committed = Array.from(seqMap.values()).sort((a, b) => a.seq - b.seq); + const identityRows = new Map(seqMap); + for (const c of chunks) identityRows.set(c.seq, c); + const backed = historyBackedProvisional( + state, + [...identityRows.values()].sort((a, b) => a.seq - b.seq), + ); + const finished = new Set(finishedTurnIds); + if (state.sealedTurnId !== null) finished.add(state.sealedTurnId); + state = { + ...state, + provisional: state.provisional.filter((p) => !backed.has(p) && !finished.has(p.turnId ?? "")), + accumulating: + state.accumulating?.turnId !== undefined && finished.has(state.accumulating.turnId) + ? null + : state.accumulating, + }; if (state.sealedTurnId !== null) { + // Legacy (non-recovery) histories may carry no turn provenance at all. + // Retain their seal cleanup, but never erase an already-live queued carry. + const legacySeal = + (state.historyRecoveryTurnIds?.length ?? 0) === 0 && finishedTurnIds.length === 0; return { ...state, committed, - provisional: [], - accumulating: null, + provisional: legacySeal + ? state.provisional.filter( + (p) => + state.generating && + p.turnId === state.currentTurnId && + p.turnId !== state.sealedTurnId, + ) + : state.provisional, + accumulating: + legacySeal && state.accumulating?.turnId === state.sealedTurnId ? null : state.accumulating, sealedTurnId: null, + ...(state.historyRecoveryTurnIds !== undefined + ? { + historyRecoveryTurnIds: state.historyRecoveryTurnIds.filter( + (id) => !finishedTurnIds.includes(id), + ), + } + : {}), }; } @@ -167,7 +210,14 @@ export function applyHistory( // match the trailing provisional user-run against the trailing committed // user-run by content equality and drop the whole echo when it is fully // backed. Leaves the accumulating (streaming) chunk untouched. - if (addedNew && state.generating && state.provisional.length > 0) { + if ( + addedNew && + state.generating && + state.provisional.length > 0 && + !state.provisional.some( + (p) => p.turnId !== undefined && state.historyRecoveryTurnIds?.includes(p.turnId), + ) + ) { const provRun = trailingRun(state.provisional, "user"); if (provRun.length > 0) { const commRun = trailingRun(committed, "user"); @@ -179,7 +229,11 @@ export function applyHistory( commRun.length >= provRun.length && provRun.every((p, i) => { const c = commRun[i]; - return c !== undefined && chunkContentEquals(p.chunk, c.chunk); + return ( + c !== undefined && + (c.turnId === undefined || p.turnId === undefined || c.turnId === p.turnId) && + chunkContentEquals(p.chunk, c.chunk) + ); }); if (fullyBacked) { const dropStart = state.provisional.length - provRun.length; @@ -189,7 +243,17 @@ export function applyHistory( } } - return { ...state, committed }; + return { + ...state, + committed, + ...(state.historyRecoveryTurnIds !== undefined + ? { + historyRecoveryTurnIds: state.historyRecoveryTurnIds.filter( + (id) => !finishedTurnIds.includes(id), + ), + } + : {}), + }; } /** @@ -227,7 +291,11 @@ function reduceEvent(state: TranscriptState, event: AgentEvent): TranscriptState case "turn-start": return { ...state, - provisional: tagTrailingUserEcho(state.provisional, event.turnId), + provisional: tagTrailingUserEcho( + flushAccumulating(state.provisional, state.accumulating), + event.turnId, + ), + accumulating: null, currentTurnId: event.turnId, generating: true, }; @@ -246,16 +314,21 @@ function reduceEvent(state: TranscriptState, event: AgentEvent): TranscriptState // text) emits no `user-message` and is not de-duped here. if (event.text.length === 0) return state; const run = trailingRun(state.provisional, "user"); - const echoed = run.find((p) => p.chunk.type === "text" && p.chunk.text === event.text); + const echoed = run.find( + (p) => + (p.turnId === undefined || p.turnId === event.turnId) && + p.chunk.type === "text" && + p.chunk.text === event.text, + ); if (echoed !== undefined) { return { ...state, // Preserve the live text independently of the local send. A delayed // queued ack may remove its images, but cannot erase this server event. provisional: tagTrailingUserEcho(state.provisional, event.turnId).map((p) => { - if (p.chunk !== echoed.chunk || p.localRequestId === undefined) return p; + if (p.chunk !== echoed.chunk) return p; const { localRequestId: _local, ...confirmed } = p; - return confirmed; + return { ...confirmed, opening: true }; }), generating: true, }; @@ -265,7 +338,12 @@ function reduceEvent(state: TranscriptState, event: AgentEvent): TranscriptState ...state, provisional: [ ...provisional, - { role: "user", chunk: { type: "text", text: event.text }, turnId: event.turnId }, + { + role: "user", + chunk: { type: "text", text: event.text }, + turnId: event.turnId, + opening: true, + }, ], accumulating: null, generating: true, @@ -274,7 +352,7 @@ function reduceEvent(state: TranscriptState, event: AgentEvent): TranscriptState case "text-delta": { const acc = state.accumulating; - if (acc !== null && acc.kind === "text") { + if (acc !== null && acc.kind === "text" && acc.turnId === event.turnId) { return { ...state, accumulating: { ...acc, text: acc.text + event.delta }, @@ -292,7 +370,7 @@ function reduceEvent(state: TranscriptState, event: AgentEvent): TranscriptState case "reasoning-delta": { const acc = state.accumulating; - if (acc !== null && acc.kind === "thinking") { + if (acc !== null && acc.kind === "thinking" && acc.turnId === event.turnId) { return { ...state, accumulating: { ...acc, text: acc.text + event.delta }, @@ -361,8 +439,24 @@ function reduceEvent(state: TranscriptState, event: AgentEvent): TranscriptState return { ...state, latestUsage: event.usage }; case "step-complete": - // Timing metadata — no content chunk; handled by the telemetry reducer. - return state; + // A persistence boundary may lag this notification. Stamp provenance and + // finish the generation chunk, but keep it until history really backs it. + return { + ...state, + provisional: flushAccumulating(state.provisional, state.accumulating), + accumulating: null, + }; + + case "turn-ended": + return { + ...state, + provisional: + state.accumulating?.turnId === event.turnId + ? flushAccumulating(state.provisional, state.accumulating) + : state.provisional, + accumulating: state.accumulating?.turnId === event.turnId ? null : state.accumulating, + generating: state.currentTurnId === event.turnId ? false : state.generating, + }; case "done": { const provisional = flushAccumulating(state.provisional, state.accumulating); @@ -394,11 +488,21 @@ function reduceEvent(state: TranscriptState, event: AgentEvent): TranscriptState // into the transcript by the sender). if (event.text.length === 0) return state; const provisional = flushAccumulating(state.provisional, state.accumulating); + const preceding = [...state.committed, ...provisional] + .reverse() + .find((p) => p.turnId === event.turnId && p.chunk.type === "tool-result"); + const afterToolCallId = + preceding?.chunk.type === "tool-result" ? preceding.chunk.toolCallId : undefined; return { ...state, provisional: [ ...provisional, - { role: "user", chunk: { type: "text", text: event.text }, turnId: event.turnId }, + { + role: "user", + chunk: { type: "text", text: event.text }, + turnId: event.turnId, + ...(afterToolCallId !== undefined ? { afterToolCallId } : {}), + }, ], accumulating: null, generating: true, @@ -440,9 +544,31 @@ const RETRY_CLEARING_EVENTS: ReadonlySet = new Set([ "error", "done", "turn-sealed", + "turn-ended", ]); export function foldEvent(state: TranscriptState, event: AgentEvent): TranscriptState { + if ( + event.type === "tool-call" || + event.type === "tool-result" || + event.type === "step-complete" + ) { + state = { + ...state, + provisional: state.provisional.map((p) => + p.turnId === event.turnId && + p.role === "assistant" && + p.stepId === undefined && + (p.chunk.type === "text" || p.chunk.type === "thinking") + ? { ...p, stepId: event.stepId } + : p, + ), + accumulating: + state.accumulating?.turnId === event.turnId + ? { ...state.accumulating, stepId: event.stepId } + : state.accumulating, + }; + } const next = reduceEvent(state, event); if (event.type === "provider-retry") return next; // set by reduceEvent; not a clearing event if (RETRY_CLEARING_EVENTS.has(event.type) && state.providerRetry !== null) { @@ -511,3 +637,15 @@ export function removeLocalEcho(state: TranscriptState, requestId: string): Tran const provisional = state.provisional.filter((p) => p.localRequestId !== requestId); return provisional.length === state.provisional.length ? state : { ...state, provisional }; } + +/** A resync notice is lifecycle evidence, never fabricated transcript content. */ +export function markHistoryRecovering(state: TranscriptState, turnId: string): TranscriptState { + return { + ...state, + currentTurnId: turnId, + generating: true, + historyRecoveryTurnIds: state.historyRecoveryTurnIds?.includes(turnId) + ? state.historyRecoveryTurnIds + : [...(state.historyRecoveryTurnIds ?? []), turnId], + }; +} diff --git a/src/core/chunks/selectors.test.ts b/src/core/chunks/selectors.test.ts index 57c1415..05dcc4e 100644 --- a/src/core/chunks/selectors.test.ts +++ b/src/core/chunks/selectors.test.ts @@ -1,6 +1,6 @@ import type { AgentEvent, StepId, StoredChunk } from "@dispatch/wire"; import { describe, expect, it } from "vitest"; -import { applyHistory, foldEvent, initialState } from "./reducer"; +import { applyHistory, foldEvent, initialState, markHistoryRecovering } from "./reducer"; import { selectChunks } from "./selectors"; import type { TranscriptState } from "./types"; @@ -131,6 +131,34 @@ describe("selectChunks — de-dup of a late-join replay against committed histor expect(rendered.filter((r) => r.provisional)).toHaveLength(3); }); + it("preserves an uncommitted accumulating tail even when its text repeats history", () => { + const committed: StoredChunk[] = [ + ...committedStep0, + { seq: 5, role: "assistant", chunk: { type: "text", text: "final" }, turnId: T }, + ]; + let s = applyHistory(initialState(), committed); + s = markHistoryRecovering(s, T); + s = fold(s, [userMsg, turnStart, think("plan"), call("a"), result("a"), text("final")]); + const rendered = selectChunks(s); + expect(rendered).toHaveLength(6); + expect(rendered.slice(0, 5).every((r) => !r.provisional)).toBe(true); + expect(rendered[5]).toMatchObject({ + provisional: true, + streaming: true, + chunk: { type: "text", text: "final" }, + }); + }); + + it("reconciles the same step identity even when provisional payload differs", () => { + let s = applyHistory(initialState(), committedStep0); + s = markHistoryRecovering(s, T); + s = fold(s, [userMsg, turnStart, think("DIFFERENT"), call("a"), result("a")]); + const rendered = selectChunks(s); + // Authoritative identity wins; payload equality cannot identify omitted/truncated content. + expect(rendered.filter((r) => r.provisional)).toHaveLength(0); + expect(rendered.map((r) => r.chunk)).toEqual(committedStep0.map((r) => r.chunk)); + }); + it("does nothing when the committed tail belongs to another turn", () => { const other = committedStep0.map((c) => ({ ...c, turnId: "older" })); let s = applyHistory(initialState(), other); diff --git a/src/core/chunks/selectors.ts b/src/core/chunks/selectors.ts index d32a628..b507cf7 100644 --- a/src/core/chunks/selectors.ts +++ b/src/core/chunks/selectors.ts @@ -1,4 +1,5 @@ import type { ChatMessage, Chunk, StoredChunk, TurnProviderRetryEvent } from "@dispatch/wire"; +import { historyBackedProvisional } from "./reconcile"; import type { ProvisionalChunk, RenderedChunk, TranscriptState } from "./types"; /** @@ -42,7 +43,8 @@ function sameChunk(p: ProvisionalChunk, c: StoredChunk): boolean { function committedOverlap(state: TranscriptState): { skip: number; hideAccumulating: boolean } { const first = state.provisional[0]; const turnId = first?.turnId ?? state.accumulating?.turnId; - if (turnId === undefined) return { skip: 0, hideAccumulating: false }; + if (turnId === undefined || state.historyRecoveryTurnIds?.includes(turnId)) + return { skip: 0, hideAccumulating: false }; // The committed run for this turn (contiguous at the tail of `committed`). let start = state.committed.length; while (start > 0 && state.committed[start - 1]?.turnId === turnId) start--; @@ -59,7 +61,7 @@ function committedOverlap(state: TranscriptState): { skip: number; hideAccumulat let hideAccumulating = false; const acc = state.accumulating; - if (acc !== null && skip === state.provisional.length && skip < run) { + if (acc !== null && acc.turnId === turnId && skip === state.provisional.length && skip < run) { const c = state.committed[start + skip]; hideAccumulating = c !== undefined && @@ -72,11 +74,12 @@ function committedOverlap(state: TranscriptState): { skip: number; hideAccumulat /** * Select all chunks for rendering: committed first (seq order), - * then provisional (seq: null) — minus provisional chunks already backed by - * committed history (see `committedOverlap`). + * then provisional (seq: null), excluding history-backed identities. Unknown + * accumulating content stays live, even if its text equals an earlier step. */ export function selectChunks(state: TranscriptState): readonly RenderedChunk[] { const result: RenderedChunk[] = []; + const backed = historyBackedProvisional(state); const { skip, hideAccumulating } = committedOverlap(state); for (const c of state.committed) { result.push( @@ -87,7 +90,7 @@ export function selectChunks(state: TranscriptState): readonly RenderedChunk[] { } for (let i = skip; i < state.provisional.length; i++) { const p = state.provisional[i]; - if (p === undefined) continue; + if (p === undefined || backed.has(p)) continue; result.push( p.turnId !== undefined ? { seq: null, role: p.role, chunk: p.chunk, provisional: true, turnId: p.turnId } diff --git a/src/core/chunks/types.ts b/src/core/chunks/types.ts index f7bfe01..2e2ee0c 100644 --- a/src/core/chunks/types.ts +++ b/src/core/chunks/types.ts @@ -2,6 +2,7 @@ import type { Chunk, Role, StoredChunk, TurnProviderRetryEvent, Usage } from "@d /** A chunk being accumulated from streaming deltas (text or thinking). */ export interface AccumulatingChunk { + readonly stepId?: string; readonly kind: "text" | "thinking"; readonly text: string; /** The turn generating this chunk (from the live event's `turnId`). */ @@ -10,6 +11,11 @@ export interface AccumulatingChunk { /** A provisional chunk that has no authoritative seq yet. */ export interface ProvisionalChunk { + /** Local provenance learned from tool-call/step-complete, not content. */ + readonly stepId?: string; + readonly opening?: boolean; + /** Steering is persisted immediately after this completed tool batch. */ + readonly afterToolCallId?: string; /** Local send identity only; turn-start tagging does NOT confirm this content. */ readonly localRequestId?: string; readonly role: Role; @@ -24,6 +30,8 @@ export interface ProvisionalChunk { /** The transcript reducer state. Holds committed history + live in-flight turn. */ export interface TranscriptState { + /** Turns with omitted delivery: never use payload equality as replay identity. */ + readonly historyRecoveryTurnIds?: readonly string[]; readonly committed: readonly StoredChunk[]; readonly provisional: readonly ProvisionalChunk[]; readonly accumulating: AccumulatingChunk | null; diff --git a/src/core/wire/conformance.test.ts b/src/core/wire/conformance.test.ts index 1179fbc..b3c774f 100644 --- a/src/core/wire/conformance.test.ts +++ b/src/core/wire/conformance.test.ts @@ -1,4 +1,8 @@ -import type { ChatSendMessage, ConversationHistoryResponse } from "@dispatch/transport-contract"; +import type { + ChatSendMessage, + ConversationHistoryResponse, + WsServerMessage, +} from "@dispatch/transport-contract"; import type { AgentEvent, StepId, StoredChunk } from "@dispatch/wire"; import { describe, expect, it } from "vitest"; import { @@ -180,6 +184,11 @@ describe("classifies every WsServerMessage type", () => { reason: "compaction" as const, messageId: "queued-1", }, + { + type: "chat.resync-required" as const, + conversationId: "c1", + reason: "replay-limit" as const, + }, { type: "conversation.open" as const, conversationId: "c1", workspaceId: "w1" }, { type: "conversation.statusChanged" as const, @@ -195,7 +204,29 @@ describe("classifies every WsServerMessage type", () => { historyEpoch: 0, messagesKept: 5, }, + { + type: "conversation.tabGroupChanged" as const, + conversationId: "c1", + workspaceId: "w1", + tabGroupParentId: null, + }, + { type: "extensions-changed" as const, action: "scan" as const }, + { + type: "layout" as const, + workspaceId: "w1", + tree: { version: 1 as const, root: { kind: "leaf" as const, id: "leaf1", view: "chat" } }, + }, + { + type: "tile-layout" as const, + mode: "wide" as const, + data: { version: 2 as const, tiles: [] }, + }, ]; + // Runtime samples stay exhaustive too, not just the switch guard. + const covered: Exclude extends never + ? true + : false = true; + expect(covered).toBe(true); const labels = msgs.map(assertWsServerMessageExhaustive); expect(labels).toEqual([ "catalog", @@ -205,9 +236,14 @@ describe("classifies every WsServerMessage type", () => { "chat.delta", "chat.error", "chat.queued", + "chat.resync-required", "conversation.open", "conversation.statusChanged", "conversation.compacted", + "conversation.tabGroupChanged", + "extensions-changed", + "layout", + "tile-layout", ]); }); }); diff --git a/src/core/wire/conformance.ts b/src/core/wire/conformance.ts index 78313f9..3429ed0 100644 --- a/src/core/wire/conformance.ts +++ b/src/core/wire/conformance.ts @@ -85,6 +85,8 @@ export function assertWsServerMessageExhaustive(msg: WsServerMessage): string { return "error"; case "chat.delta": return "chat.delta"; + case "chat.resync-required": + return "chat.resync-required"; case "chat.queued": return "chat.queued"; case "chat.error": diff --git a/src/features/chat/logic/tool-summary.ts b/src/features/chat/logic/tool-summary.ts index 566e40f..f6d1569 100644 --- a/src/features/chat/logic/tool-summary.ts +++ b/src/features/chat/logic/tool-summary.ts @@ -381,7 +381,9 @@ export function viewToolEntry(entry: ToolEntry): ToolCardView { const builtInSummary = summarize?.(args).trim(); const callSummary = entry.call === null - ? "Call in earlier history" + ? entry.recovering + ? "Recovering tool input…" + : "Call in earlier history" : isEmptyInput(input) ? "" : oneLine( @@ -416,7 +418,9 @@ export function viewToolEntry(entry: ToolEntry): ToolCardView { entry.call === null ? { kind: "unavailable", - text: "This call is earlier in the conversation. Show earlier messages to see it.", + text: entry.recovering + ? "Tool input is unavailable until it is persisted. Recovering from conversation history…" + : "This call is earlier in the conversation. Show earlier messages to see it.", } : detail(entry.toolName, input, args), resultSummary, diff --git a/src/features/chat/recovery.test.ts b/src/features/chat/recovery.test.ts new file mode 100644 index 0000000..4b46288 --- /dev/null +++ b/src/features/chat/recovery.test.ts @@ -0,0 +1,529 @@ +import type { + ChatResyncRequiredMessage, + ConversationHistoryResponse, +} from "@dispatch/transport-contract"; +import type { AgentEvent, StepId, StoredChunk, TurnEndedEvent } from "@dispatch/wire"; +import { IDBFactory, IDBKeyRange } from "fake-indexeddb"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import { createIdbChunkStore } from "../../adapters/idb"; +import { createConversationCache } from "../conversation-cache"; +import type { HistorySync } from "./ports"; +import { createChatStore } from "./store.svelte"; +import { + createFakeCache, + createFakeHistorySync, + createFakeMetricsSync, + createFakeTransport, +} from "./test-helpers"; + +const step = "t1#0" as StepId; +const opening: StoredChunk = { + seq: 1, + role: "user", + turnId: "t1", + chunk: { type: "text", text: "prompt" }, +}; +const call: StoredChunk = { + seq: 3, + role: "assistant", + turnId: "t1", + chunk: { + type: "tool-call", + toolCallId: "a", + toolName: "run_shell", + stepId: step, + input: { command: "full input" }, + }, +}; +const result: StoredChunk = { + seq: 4, + role: "tool", + turnId: "t1", + chunk: { + type: "tool-result", + toolCallId: "a", + toolName: "run_shell", + stepId: step, + content: "full output", + isError: false, + }, +}; +const committed: StoredChunk[] = [ + opening, + { seq: 2, role: "assistant", turnId: "t1", chunk: { type: "text", text: "complete text" } }, + call, + result, +]; + +function fixture(historySync?: HistorySync, batched = false) { + const history = createFakeHistorySync(); + const transport = createFakeTransport(); + const cache = createFakeCache(); + const scheduled: (() => void)[] = []; + const store = createChatStore({ + conversationId: "c1", + createRequestId: () => "r1", + historySync: historySync ?? history.impl, + metricsSync: createFakeMetricsSync().impl, + cache: cache.impl, + transport: transport.impl, + ...(batched ? { scheduleFlush: (fn: () => void) => scheduled.push(fn) } : {}), + }); + const event = (event: AgentEvent) => store.handleDelta({ type: "chat.delta", event }); + const notice = ( + turnId: string | undefined = "t1", + reason: ChatResyncRequiredMessage["reason"] = "replay-limit", + ) => + store.handleDelta({ + type: "chat.resync-required", + conversationId: "c1", + reason, + ...(turnId !== undefined ? { turnId } : {}), + }); + const end = (reason: TurnEndedEvent["reason"] = "sealed", turnId = "t1") => + event({ type: "turn-ended", conversationId: "c1", turnId, reason }); + const text = (delta: string, turnId = "t1") => + event({ type: "text-delta", conversationId: "c1", turnId, delta }); + return { store, history, transport, cache, scheduled, event, notice, end, text }; +} + +async function settled(store: ReturnType) { + for (let i = 0; i < 80 && store.pendingSync; i++) await Promise.resolve(); + expect(store.pendingSync).toBe(false); +} +function response(chunks: readonly StoredChunk[], historyEpoch = 0): ConversationHistoryResponse { + return { chunks, historyEpoch, latestSeq: chunks.at(-1)?.seq ?? 0 }; +} +function deferred() { + let resolve!: (value: ConversationHistoryResponse) => void; + const promise = new Promise((r) => { + resolve = r; + }); + return { promise, resolve }; +} + +afterEach(() => { + vi.useRealTimers(); + vi.unstubAllGlobals(); +}); + +describe("omitted chat recovery", () => { + it.each([ + false, + true, + ])("marks omitted replay active without content and recovers full HTTP payload (batched=%s)", async (batched) => { + vi.useFakeTimers(); + const f = fixture(undefined, batched); + f.text("partial"); + f.notice(); + expect(f.store.generating).toBe(true); + expect(f.store.recovering).toBe(true); + expect(f.store.chunks.map((r) => r.chunk)).toEqual([{ type: "text", text: "partial" }]); + await settled(f.store); + // No initial commit yet. A tool result follows an OMITTED call input. + f.event({ + type: "tool-result", + conversationId: "c1", + turnId: "t1", + stepId: step, + toolCallId: "a", + toolName: "run_shell", + content: "short replay", + isError: false, + }); + f.history.returnChunks = committed; + await vi.advanceTimersByTimeAsync(1000); + expect(f.store.chunks.filter((r) => r.chunk.type === "tool-call")).toEqual([ + expect.objectContaining({ chunk: call.chunk, provisional: false }), + ]); + expect(f.store.chunks.filter((r) => r.chunk.type === "tool-result")).toHaveLength(1); + expect(f.store.chunks.find((r) => r.chunk.type === "tool-result")?.chunk).toEqual(result.chunk); + expect(f.store.generating).toBe(true); + expect(f.store.error).toBeNull(); + expect(f.transport.sent).toEqual([]); + f.store.dispose(); + expect(vi.getTimerCount()).toBe(0); + }); + + it("coalesces repeat notices and polls after step notification races persistence, without parallel requests", async () => { + vi.useFakeTimers(); + const pending = deferred(); + let calls = 0; + const f = fixture(async () => { + calls++; + return calls === 1 ? pending.promise : response(committed); + }); + f.notice(); + for (let i = 0; i < 20; i++) f.notice(); + for (let i = 0; i < 15; i++) await Promise.resolve(); + expect(calls).toBe(1); + f.event({ type: "step-complete", conversationId: "c1", turnId: "t1", stepId: step }); + await vi.advanceTimersByTimeAsync(10000); + expect(calls).toBe(1); + pending.resolve(response([])); + await settled(f.store); + expect(calls).toBe(1); + await vi.advanceTimersByTimeAsync(999); + expect(calls).toBe(1); + await vi.advanceTimersByTimeAsync(1); + expect(calls).toBe(2); + expect(f.store.chunks.map((r) => r.chunk)).toEqual(committed.map((r) => r.chunk)); + f.store.dispose(); + }); + + it.each([ + "sealed", + "aborted", + "error", + "rejected", + ] as const)("final-syncs %s behind an in-flight fetch and preserves immediate queued carry", async (reason) => { + vi.useFakeTimers(); + const pending = deferred(); + let calls = 0; + const f = fixture(async () => (++calls === 1 ? pending.promise : response(committed))); + f.notice(); + for (let i = 0; i < 10; i++) await Promise.resolve(); + f.text("old tail"); + f.end(reason); + f.event({ type: "user-message", conversationId: "c1", turnId: "t2", text: "queued input" }); + f.event({ type: "turn-start", conversationId: "c1", turnId: "t2" }); + f.text("carried tail", "t2"); + pending.resolve(response([opening])); + await settled(f.store); + expect(calls).toBe(2); + expect(f.store.recovering).toBe(false); + expect(f.store.generating).toBe(true); + expect(f.store.chunks.filter((r) => r.provisional).map((r) => r.chunk)).toEqual([ + { type: "text", text: "queued input" }, + { type: "text", text: "carried tail" }, + ]); + expect(f.store.chunks.filter((r) => r.turnId === "t1").every((r) => !r.provisional)).toBe(true); + await vi.advanceTimersByTimeAsync(5000); + expect(calls).toBe(2); + f.end("sealed", "t2"); + expect(f.store.generating).toBe(false); + f.store.dispose(); + }); + + it.each([ + false, + true, + ])("notice during %s-cache load requests one rerun, preserving current window", async (warm) => { + vi.useFakeTimers(); + const pending = deferred(); + let calls = 0; + const f = fixture(async (_id, _since, window) => { + calls++; + if (!warm) expect(window?.limit).toBe(192); + return calls === 1 ? pending.promise : response([opening]); + }); + if (warm) f.cache.store.set("c1", [opening]); + const load = f.store.load(); + for (let i = 0; i < 15; i++) await Promise.resolve(); + f.notice(); + pending.resolve(response([opening])); + await load; + await settled(f.store); + expect(calls).toBe(2); + expect(f.store.chunks).toHaveLength(1); + expect(f.store.recovering).toBe(true); + expect(f.store.generating).toBe(true); + f.store.dispose(); + }); + + it("payload notice after a completed HTTP load preserves live tail, unrelated turn and tool progress", async () => { + vi.useFakeTimers(); + const f = fixture(); + await f.store.load(); + f.event({ type: "turn-start", conversationId: "c1", turnId: "other" }); + f.text("unrelated", "other"); + f.event({ type: "turn-start", conversationId: "c1", turnId: "t1" }); + f.text("partial text"); + f.event({ + type: "tool-call", + conversationId: "c1", + turnId: "t1", + stepId: step, + toolCallId: "a", + toolName: "run_shell", + input: { command: "short" }, + }); + f.event({ + type: "tool-result", + conversationId: "c1", + turnId: "t1", + stepId: step, + toolCallId: "a", + toolName: "run_shell", + content: "short", + isError: false, + }); + f.text("complete text"); // same payload as earlier step; NOT the same identity + f.event({ + type: "tool-output", + conversationId: "c1", + turnId: "t1", + toolCallId: "next", + data: "progress", + stream: "stdout", + }); + f.history.returnChunks = committed; + f.notice("t1", "payload-limit"); + await settled(f.store); + expect(f.store.chunks.filter((r) => r.provisional).map((r) => r.chunk)).toEqual([ + { type: "text", text: "unrelated" }, + { type: "text", text: "complete text" }, + ]); + expect(f.store.chunks.filter((r) => r.chunk.type === "tool-call")).toHaveLength(1); + expect(f.store.chunks.filter((r) => r.chunk.type === "tool-result")).toHaveLength(1); + f.store.dispose(); + }); + + it("bounds epoch churn during recovery and cancels its retry timer on disposal", async () => { + vi.useFakeTimers(); + let calls = 0; + const f = fixture(async () => { + f.store.invalidateHistory(++calls); + return response([], calls - 1); + }); + f.notice(); + await settled(f.store); + expect(calls).toBe(3); + expect(f.store.error).toContain("sync paused"); + await vi.advanceTimersByTimeAsync(10000); + expect(calls).toBe(3); + f.store.dispose(); + await vi.advanceTimersByTimeAsync(10000); + expect(calls).toBe(3); + expect(vi.getTimerCount()).toBe(0); + }); + + it("ignores other conversations and disposed notices; accepts a notice without turnId", async () => { + vi.useFakeTimers(); + const f = fixture(); + f.store.handleDelta({ + type: "chat.resync-required", + conversationId: "other", + reason: "payload-limit", + }); + expect(f.store.recovering).toBe(false); + f.store.handleDelta({ + type: "chat.resync-required", + conversationId: "c1", + reason: "payload-limit", + }); + await settled(f.store); + expect(f.store.generating).toBe(false); + expect(f.store.recovering).toBe(false); + expect(vi.getTimerCount()).toBe(0); + f.event({ type: "turn-start", conversationId: "c1", turnId: "t1" }); + f.store.handleDelta({ + type: "chat.resync-required", + conversationId: "c1", + reason: "payload-limit", + }); + expect(f.store.generating).toBe(true); + expect(f.store.recovering).toBe(true); + f.end("rejected"); + await settled(f.store); + expect(f.store.recovering).toBe(false); + expect(f.store.generating).toBe(false); + f.store.dispose(); + f.notice(); + expect(vi.getTimerCount()).toBe(0); + }); +}); + +it("recovers a windowed rollback with real IndexedDB cache while preserving lifecycle evidence", async () => { + vi.stubGlobal("IDBKeyRange", IDBKeyRange); + const cache = createConversationCache(createIdbChunkStore({ indexedDB: new IDBFactory() })); + await cache.commit( + "c1", + [{ seq: 1, role: "assistant", chunk: { type: "text", text: "old epoch" } }], + 5, + ); + const history = createFakeHistorySync(); + history.historyEpoch = 5; + const timers = new Map void>(); + const store = createChatStore({ + conversationId: "c1", + createRequestId: () => "r", + cache, + historySync: history.impl, + metricsSync: createFakeMetricsSync().impl, + transport: createFakeTransport().impl, + chatLimit: 10, + recoveryTimer: { + schedule(fn) { + const key = {}; + timers.set(key, fn); + return key; + }, + cancel(key) { + timers.delete(key as object); + }, + }, + }); + try { + await store.load(); + history.historyEpoch = 0; + history.returnChunks = Array.from({ length: 20 }, (_, i) => ({ + seq: i + 1, + role: "assistant" as const, + turnId: "t1", + chunk: { type: "text" as const, text: `restored-${i}` }, + })); + const before = history.calls.length; + store.handleDelta({ + type: "chat.resync-required", + conversationId: "c1", + turnId: "t1", + reason: "replay-limit", + }); + await vi.waitFor(() => expect(store.pendingSync).toBe(false)); + expect(history.calls.length - before).toBe(2); + expect(history.calls.at(-1)).toMatchObject({ sinceSeq: 0, window: { limit: 7 } }); + expect(await cache.historyEpoch("c1")).toBe(0); + expect(store.chunks.map((c) => c.seq)).toEqual([14, 15, 16, 17, 18, 19, 20]); + expect(store.hasEarlier).toBe(true); + expect(store.generating).toBe(true); + expect(timers.size).toBe(1); + store.handleDelta({ + type: "chat.delta", + event: { type: "turn-ended", conversationId: "c1", turnId: "t1", reason: "aborted" }, + }); + await vi.waitFor(() => expect(store.pendingSync).toBe(false)); + expect(store.generating).toBe(false); + expect(timers.size).toBe(0); + expect(store.hasEarlier).toBe(true); + } finally { + store.dispose(); + } +}); + +it("ignores an unrelated terminal and stays recovering through done/seal until its actual terminal", async () => { + vi.useFakeTimers(); + const f = fixture(); + f.notice(); + await settled(f.store); + f.end("aborted", "unrelated"); + f.event({ type: "done", conversationId: "c1", turnId: "t1", reason: "aborted" }); + expect(f.store.recovering).toBe(true); + expect(f.store.generating).toBe(true); + f.event({ type: "turn-sealed", conversationId: "c1", turnId: "t1" }); + await settled(f.store); + expect(f.store.recovering).toBe(true); + expect(f.store.generating).toBe(true); + f.end("aborted"); + await settled(f.store); + expect(f.store.recovering).toBe(false); + expect(f.store.generating).toBe(false); + expect(vi.getTimerCount()).toBe(0); + f.store.dispose(); +}); + +it("makes a post-terminal request even when the terminal races the third bounded epoch attempt", async () => { + vi.useFakeTimers(); + const third = deferred(); + let calls = 0; + const f = fixture(async () => { + calls++; + if (calls < 3) { + f.store.invalidateHistory(calls); + return response([], calls - 1); + } + return calls === 3 ? third.promise : response(committed, 2); + }); + f.notice(); + for (let i = 0; i < 60 && calls < 3; i++) await Promise.resolve(); + expect(calls).toBe(3); + f.end("error"); + third.resolve(response([], 2)); + await settled(f.store); + expect(calls).toBe(4); + expect(f.store.chunks.map((r) => r.chunk)).toEqual(committed.map((r) => r.chunk)); + expect(f.store.recovering).toBe(false); + expect(vi.getTimerCount()).toBe(0); + f.store.dispose(); +}); + +it("retains active recovery across a compaction epoch refresh and cancels it on reconnect/dispose", async () => { + vi.useFakeTimers(); + const f = fixture(); + f.notice(); + await settled(f.store); + f.history.historyEpoch = 1; + f.store.resync(1); + await settled(f.store); + expect(f.store.recovering).toBe(true); + expect(f.store.generating).toBe(true); + expect(vi.getTimerCount()).toBe(1); + f.store.resync(); + await settled(f.store); + expect(f.store.recovering).toBe(false); + expect(f.store.generating).toBe(false); + expect(vi.getTimerCount()).toBe(0); + f.store.dispose(); +}); + +it("flushes a batched aborted terminal plus carried turn before applying an older HTTP response", async () => { + vi.useFakeTimers(); + const pending = deferred(); + let calls = 0; + const f = fixture(async () => (++calls === 1 ? pending.promise : response(committed)), true); + f.notice(); + for (let i = 0; i < 15; i++) await Promise.resolve(); + f.text("old partial"); + f.end("aborted"); + f.event({ type: "user-message", conversationId: "c1", turnId: "t2", text: "queued" }); + f.event({ type: "turn-start", conversationId: "c1", turnId: "t2" }); + f.text("live carried output", "t2"); + pending.resolve(response([opening])); + await settled(f.store); + expect(calls).toBe(2); + expect(f.store.chunks.filter((r) => r.provisional).map((r) => r.chunk)).toEqual([ + { type: "text", text: "queued" }, + { type: "text", text: "live carried output" }, + ]); + expect(f.store.generating).toBe(true); + expect(f.store.recovering).toBe(false); + expect(vi.getTimerCount()).toBe(0); + f.store.dispose(); +}); + +it("does not truncate a payload larger than the WS budget recovered over HTTP", async () => { + vi.useFakeTimers(); + const large = "x".repeat(4 * 1024 * 1024 + 1); + const full: StoredChunk[] = [ + { + ...call, + chunk: { + ...call.chunk, + type: "tool-call", + toolCallId: "a", + toolName: "run_shell", + input: { command: large }, + stepId: step, + }, + }, + { + ...result, + chunk: { + type: "tool-result", + toolCallId: "a", + toolName: "run_shell", + content: large, + isError: false, + stepId: step, + }, + }, + ]; + const f = fixture(); + f.history.returnChunks = full; + f.notice(); + await settled(f.store); + expect(f.store.chunks).toHaveLength(2); + expect(f.store.chunks.map((r) => r.chunk)).toEqual(full.map((r) => r.chunk)); + expect(f.transport.sent).toEqual([]); + f.store.dispose(); +}); diff --git a/src/features/chat/store.svelte.ts b/src/features/chat/store.svelte.ts index 66ee5ee..bb0b218 100644 --- a/src/features/chat/store.svelte.ts +++ b/src/features/chat/store.svelte.ts @@ -4,6 +4,7 @@ import type { ChatQueueCancelMessage, ChatQueuedMessage, ChatQueueMessage, + ChatResyncRequiredMessage, ChatSendMessage, } from "@dispatch/transport-contract"; import type { ChatMessage, ImageInput, StoredChunk, TurnProviderRetryEvent } from "@dispatch/wire"; @@ -15,6 +16,7 @@ import { foldEvent, initialState, initialWindowSize, + markHistoryRecovering, normalizeChatLimit, removeLocalEcho, restoreEarlier, @@ -88,6 +90,11 @@ export interface ChatStoreDependencies { readonly scheduleFlush?: (flush: () => void) => void; /** Wall clock (epoch ms) — times the cache-timer gaps. Absent → `Date.now`. */ readonly now?: () => number; + /** Recovery polling clock; browser effects are replaceable by deterministic tests. */ + readonly recoveryTimer?: { + schedule(callback: () => void, delayMs: number): unknown; + cancel(handle: unknown): void; + }; } export interface ChatStore { @@ -117,6 +124,8 @@ export interface ChatStore { * to the newest attempt + delay; cleared when content resumes or the turn ends. */ readonly providerRetry: TurnProviderRetryEvent | null; + /** Omitted live payloads may not exist in HTTP history until step persistence. */ + readonly recovering: boolean; readonly pendingSync: boolean; readonly error: string | null; readonly model: string | undefined; @@ -146,7 +155,9 @@ export interface ChatStore { * Ignored while a turn is generating; a running turn gap keeps the later start. */ noteLastActivity(at: number): void; - handleDelta(msg: ChatDeltaMessage | ChatErrorMessage | ChatQueuedMessage): void; + handleDelta( + msg: ChatDeltaMessage | ChatErrorMessage | ChatQueuedMessage | ChatResyncRequiredMessage, + ): void; /** * Send a user message (start a turn via `chat.send`). Optimistically echoes * the text + any `images` as provisional user chunks (`[text, image, …]` in @@ -235,12 +246,38 @@ export function createChatStore(deps: ChatStoreDependencies): ChatStore { let revision = 0; let reloadWindow = false; let earliestSeq = $state(null); + let recoveringTurnId: string | undefined; + let recovering = $state(false); + const finishedRecoveryTurns = new Set(); + let finalSyncRequested = false; + let recoveryPaused = false; + const timer = deps.recoveryTimer ?? { + schedule: (callback: () => void, delay: number) => setTimeout(callback, delay), + cancel: (handle: unknown) => clearTimeout(handle as ReturnType), + }; + let recoveryTimer: unknown; + + function cancelRecoveryTimer(): void { + if (recoveryTimer !== undefined) timer.cancel(recoveryTimer); + recoveryTimer = undefined; + } + + function scheduleRecovery(): void { + if (disposed || !recovering || recoveryPaused || recoveryTimer !== undefined) return; + recoveryTimer = timer.schedule(() => { + recoveryTimer = undefined; + flushEvents(); + void syncTail(initialWindowSize(chatLimit)); + }, 1000); + } function resetTranscript(epoch: number): void { historyEpoch = epoch; revision++; reloadWindow = true; transcript = initialState(); + if (recoveringTurnId !== undefined) + transcript = markHistoryRecovering(transcript, recoveringTurnId); earliestSeq = null; pendingEvents = []; } @@ -294,6 +331,7 @@ export function createChatStore(deps: ChatStoreDependencies): ChatStore { let m = metrics; let sealed = false; let failed = false; + let recoveryEnded = false; let c = cacheTimer; const at = now(); for (const event of events) { @@ -301,14 +339,34 @@ export function createChatStore(deps: ChatStoreDependencies): ChatStore { m = foldMetricsEvent(m, event); const mNow = m; c = foldCacheTimerEvent(c, event, at, () => lastTurnIdOf(mNow)); - if (t.sealedTurnId !== null) sealed = true; + if ( + event.type === "turn-ended" && + recovering && + (recoveringTurnId === undefined || event.turnId === recoveringTurnId) + ) { + finishedRecoveryTurns.add(event.turnId); + finalSyncRequested = true; + recoveringTurnId = undefined; + recovering = false; + recoveryEnded = true; + cancelRecoveryTimer(); + } + if (event.type === "turn-sealed") sealed = true; else if (event.type === "error") failed = true; } transcript = t; metrics = m; cacheTimer = c; maybeTrim(); - if (sealed) { + if (recovering && recoveringTurnId !== undefined) { + // done/error/seal can precede persistence and the real terminal. Keep + // lifecycle evidence until turn-ended, not merely until a content event. + transcript = markHistoryRecovering(transcript, recoveringTurnId); + } + if (recoveryEnded) { + void syncTail(initialWindowSize(chatLimit)); + void syncMetrics(); + } else if (sealed) { void syncTail(); void syncMetrics(); } else if (failed) { @@ -366,7 +424,13 @@ export function createChatStore(deps: ChatStoreDependencies): ChatStore { continue; } const limit = - since === 0 ? (reloadWindow ? initialWindowSize(chatLimit) : coldLimit) : undefined; + since === 0 + ? reloadWindow || recovering || finishedRecoveryTurns.size > 0 + ? initialWindowSize(chatLimit) + : coldLimit + : undefined; + const finishedAtRequest = [...finishedRecoveryTurns]; + finalSyncRequested = false; const res = await deps.historySync( deps.conversationId, since, @@ -416,7 +480,9 @@ export function createChatStore(deps: ChatStoreDependencies): ChatStore { historyEpoch = res.historyEpoch; if (since === 0 && (limit === undefined || res.chunks.length < limit)) earliestSeq = merged[0]?.seq ?? 0; - transcript = applyHistory(transcript, merged); + flushEvents(); + transcript = applyHistory(transcript, merged, finishedAtRequest); + for (const turnId of finishedAtRequest) finishedRecoveryTurns.delete(turnId); maybeTrim(); _error = null; } catch (err) { @@ -424,9 +490,21 @@ export function createChatStore(deps: ChatStoreDependencies): ChatStore { } } while (rerunRequested && !disposed && attempts < 3); if (rerunRequested && !disposed) { + recoveryPaused = true; + cancelRecoveryTimer(); _error = "Conversation history changed repeatedly; sync paused until the next refresh."; } _pendingSync = false; + // Schedule AFTER the bounded sync burst. No interval can queue unbounded + // reruns behind a slow request; repeated notices coalesce on the same clock. + scheduleRecovery(); + // A terminal landing during the third (bounded epoch) attempt still owns + // one post-terminal request. It is a new external trigger, not an epoch + // retry loop; consume the flag before starting its bounded burst. + if (finalSyncRequested && !disposed) { + finalSyncRequested = false; + void syncTail(initialWindowSize(chatLimit)); + } } async function syncMetrics(): Promise { @@ -520,6 +598,9 @@ export function createChatStore(deps: ChatStoreDependencies): ChatStore { get providerRetry(): TurnProviderRetryEvent | null { return selectProviderRetry(transcript); }, + get recovering(): boolean { + return recovering; + }, get pendingSync(): boolean { return _pendingSync; }, @@ -544,7 +625,32 @@ export function createChatStore(deps: ChatStoreDependencies): ChatStore { cacheTimer = seedTurnGap(cacheTimer, at); }, - handleDelta(msg: ChatDeltaMessage | ChatErrorMessage | ChatQueuedMessage): void { + handleDelta( + msg: ChatDeltaMessage | ChatErrorMessage | ChatQueuedMessage | ChatResyncRequiredMessage, + ): void { + if (disposed) return; + if (msg.type === "chat.resync-required") { + if (msg.conversationId !== deps.conversationId) return; + flushEvents(); + const target = + msg.turnId ?? (transcript.generating ? transcript.currentTurnId : null) ?? undefined; + // Oversized status packets need not identify an active turn. Refresh + // once, but do not invent generation or poll an idle conversation. + if (target === undefined) { + void syncTail(initialWindowSize(chatLimit)); + return; + } + const first = !recovering || recoveryPaused; + recoveryPaused = false; + recovering = true; + recoveringTurnId = target; + if (recoveringTurnId !== undefined) + transcript = markHistoryRecovering(transcript, recoveringTurnId); + // Immediate first handoff, including a coalesced rerun behind a cold + // load. Repeats wait for the existing fetch/one-second timer. + if (first) void syncTail(initialWindowSize(chatLimit)); + return; + } if (msg.type === "chat.error") { if (msg.conversationId !== undefined && msg.conversationId !== deps.conversationId) { return; @@ -685,13 +791,21 @@ export function createChatStore(deps: ChatStoreDependencies): ChatStore { // 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. flushEvents(); - transcript = clearGenerating(transcript); + recoveryPaused = false; + if (epoch === undefined) { + recovering = false; + recoveringTurnId = undefined; + cancelRecoveryTimer(); + transcript = clearGenerating(transcript); + } void syncTail(); void syncMetrics(); }, dispose(): void { disposed = true; + cancelRecoveryTimer(); + finishedRecoveryTurns.clear(); pendingEvents = []; }, }; diff --git a/src/features/chat/ui/ChatView.svelte b/src/features/chat/ui/ChatView.svelte index b473766..c7f566a 100644 --- a/src/features/chat/ui/ChatView.svelte +++ b/src/features/chat/ui/ChatView.svelte @@ -24,6 +24,7 @@ let { chunks, + recovering = false, turnMetrics = [], hasEarlier = false, onShowEarlier, @@ -34,6 +35,7 @@ cacheTimer = null, }: { chunks: readonly RenderedChunk[]; + recovering?: boolean; turnMetrics?: readonly TurnMetricsEntry[]; /** Earlier history is unloaded (chat limit) and can be paged back in. */ hasEarlier?: boolean; @@ -90,7 +92,7 @@ } } - const groups = $derived(groupRenderedChunks(chunks)); + const groups = $derived(groupRenderedChunks(chunks, recovering)); const rows = $derived(interleaveTurnMetrics(groups, turnMetrics)); @@ -150,6 +152,12 @@ }); +{#if recovering} +

+ Recovering conversation history… Omitted tool input/output is unavailable until persisted. +

+{/if} + {#snippet chunkRow(rendered: RenderedChunk)} {#if rendered.role === "user"}