fix(review-3.6): invalidate history caches by epoch and coalesce syncs

7 files changed, 289 insertions, 38 deletions

diff --git a/src/features/chat/store.svelte.ts b/src/features/chat/store.svelte.ts
index 01ad9f9..dab1ba2 100644
--- a/src/features/chat/store.svelte.ts
+++ b/src/features/chat/store.svelte.ts
@@ -209,7 +209,9 @@ export interface ChatStore {
    * running, the server's post-subscribe replay re-asserts `generating`. The
    * app store pairs this with a `chat.subscribe` for the conversation.
    */
-  resync(): void;
+  resync(historyEpoch?: number): void;
+  /** Discard pre-replacement transcript state without loading an unseen conversation. */
+  invalidateHistory(historyEpoch: number): void;
   dispose(): void;
 }
 
@@ -223,6 +225,23 @@ export function createChatStore(deps: ChatStoreDependencies): ChatStore {
   let cacheTimer = $state<CacheTimerState>(initialCacheTimerState());
   const now = deps.now ?? Date.now;
   let disposed = false;
+  let rerunRequested = false;
+  let historyEpoch: number | undefined;
+  let revision = 0;
+  let reloadWindow = false;
+  let earliestSeq = $state<number | null>(null);
+
+  function resetTranscript(epoch: number): void {
+    historyEpoch = epoch;
+    revision++;
+    reloadWindow = true;
+    transcript = initialState();
+    earliestSeq = null;
+    pendingEvents = [];
+  }
+  function hasEarlier(): boolean {
+    return selectHasEarlier(transcript, earliestSeq);
+  }
 
   /** The most recent turn's id (resolves a seeded turn gap when it is stamped). */
   function lastTurnIdOf(m: MetricsState): string | undefined {
@@ -313,21 +332,58 @@ export function createChatStore(deps: ChatStoreDependencies): ChatStore {
    * between the cache and the fetched window.
    */
   async function syncTail(coldLimit?: number): Promise<void> {
-    if (disposed || _pendingSync) return;
-    _pendingSync = true;
-    try {
-      const since = await deps.cache.sinceSeq(deps.conversationId);
-      const window = since === 0 && coldLimit !== undefined ? { limit: coldLimit } : undefined;
-      const res = await deps.historySync(deps.conversationId, since, window);
-      const merged = await deps.cache.commit(deps.conversationId, res.chunks);
-      transcript = applyHistory(transcript, merged);
-      maybeTrim();
-      _error = null;
-    } catch (err) {
-      _error = err instanceof Error ? err.message : String(err);
-    } finally {
-      _pendingSync = false;
+    if (disposed) return;
+    if (_pendingSync) {
+      rerunRequested = true;
+      return;
     }
+    _pendingSync = true;
+    do {
+      rerunRequested = false;
+      try {
+        const currentRevision = revision;
+        const cachedEpoch = await deps.cache.historyEpoch(deps.conversationId);
+        if (historyEpoch !== undefined && cachedEpoch !== historyEpoch)
+          await deps.cache.reset(deps.conversationId, historyEpoch);
+        const since =
+          cachedEpoch === undefined ? 0 : await deps.cache.sinceSeq(deps.conversationId);
+        const limit =
+          since === 0 ? (reloadWindow ? initialWindowSize(chatLimit) : coldLimit) : undefined;
+        const res = await deps.historySync(
+          deps.conversationId,
+          since,
+          limit === undefined ? undefined : { limit },
+        );
+        if (disposed) break;
+        if (revision !== currentRevision) {
+          rerunRequested = true;
+          continue;
+        }
+        if (
+          (cachedEpoch !== undefined && cachedEpoch !== res.historyEpoch) ||
+          (historyEpoch !== undefined && historyEpoch !== res.historyEpoch)
+        ) {
+          resetTranscript(res.historyEpoch);
+          await deps.cache.reset(deps.conversationId, res.historyEpoch);
+          if (since !== 0) {
+            rerunRequested = true;
+            continue;
+          }
+        }
+        reloadWindow = false;
+        historyEpoch = res.historyEpoch;
+        const merged = await deps.cache.commit(deps.conversationId, res.chunks, res.historyEpoch);
+        if (disposed || historyEpoch !== res.historyEpoch) continue;
+        if (since === 0 && (limit === undefined || res.chunks.length < limit))
+          earliestSeq = merged[0]?.seq ?? 0;
+        transcript = applyHistory(transcript, merged);
+        maybeTrim();
+        _error = null;
+      } catch (err) {
+        _error = err instanceof Error ? err.message : String(err);
+      }
+    } while (rerunRequested && !disposed);
+    _pendingSync = false;
   }
 
   async function syncMetrics(): Promise<void> {
@@ -351,14 +407,25 @@ export function createChatStore(deps: ChatStoreDependencies): ChatStore {
    * newest `count` of them). Shared by `showEarlier` and the raise-refill.
    */
   async function backfillOlder(oldest: number, want: number): Promise<readonly StoredChunk[]> {
+    const currentRevision = revision;
     let earlier = (await deps.cache.load(deps.conversationId)).filter((c) => c.seq < oldest);
+    if (disposed || revision !== currentRevision) return [];
     const oldestKnown = earlier[0]?.seq ?? oldest;
-    if (earlier.length < want && oldestKnown > 1) {
+    if (earlier.length < want && (earliestSeq === null || oldestKnown > earliestSeq)) {
       const res = await deps.historySync(deps.conversationId, 0, {
         beforeSeq: oldestKnown,
         limit: want - earlier.length,
       });
-      const merged = await deps.cache.commit(deps.conversationId, res.chunks);
+      if (disposed || revision !== currentRevision) return [];
+      if (res.historyEpoch !== historyEpoch) {
+        resetTranscript(res.historyEpoch);
+        await deps.cache.reset(deps.conversationId, res.historyEpoch);
+        await syncTail(initialWindowSize(chatLimit));
+        return [];
+      }
+      const merged = await deps.cache.commit(deps.conversationId, res.chunks, res.historyEpoch);
+      if (disposed || revision !== currentRevision) return [];
+      if (res.chunks.length < want - earlier.length) earliestSeq = merged[0]?.seq ?? 0;
       earlier = merged.filter((c) => c.seq < oldest);
     }
     return earlier;
@@ -368,15 +435,15 @@ export function createChatStore(deps: ChatStoreDependencies): ChatStore {
    * Refill toward the fresh-load window after a limit RAISE: pull older
    * history (cache first, then server) so the loaded set grows to match what a
    * fresh `load()` would show at the new limit. No-op when already at the
-   * origin (seq 1) or already within the window. `restoreEarlier` re-derives
+   * known history boundary or already within the window. `restoreEarlier` re-derives
    * the window start at apply time, so a delta landing during the await can't
    * corrupt the merge. NOT gated (refilling prepends above the viewport; the
    * caller preserves scroll position).
    */
   async function refill(): Promise<void> {
     if (disposed) return;
+    if (!hasEarlier()) return;
     const oldest = transcript.committed[0]?.seq ?? transcript.hiddenBeforeSeq;
-    if (oldest <= 1) return;
     const want = initialWindowSize(chatLimit) - transcript.committed.length;
     if (want <= 0) return;
     try {
@@ -418,7 +485,7 @@ export function createChatStore(deps: ChatStoreDependencies): ChatStore {
       return _model;
     },
     get hasEarlier(): boolean {
-      return selectHasEarlier(transcript);
+      return hasEarlier();
     },
     get thinkingKeyBase(): number {
       return transcript.hiddenThinkingCount;
@@ -524,8 +591,11 @@ export function createChatStore(deps: ChatStoreDependencies): ChatStore {
       // server instead (CR-5 `?limit=`), so a huge conversation never ships
       // whole. The post-sync window re-asserts the cap either way.
       const windowSize = initialWindowSize(chatLimit);
+      const currentRevision = revision;
+      const cachedEpoch = await deps.cache.historyEpoch(deps.conversationId);
       const cached = await deps.cache.load(deps.conversationId);
-      if (cached.length > 0) {
+      if (!disposed && revision === currentRevision && cachedEpoch !== undefined) {
+        historyEpoch = cachedEpoch;
         transcript = windowTranscript(applyHistory(transcript, cached), windowSize);
       }
       await syncTail(windowSize);
@@ -536,7 +606,7 @@ export function createChatStore(deps: ChatStoreDependencies): ChatStore {
     async showEarlier(): Promise<void> {
       if (disposed) return;
       const oldest = transcript.committed[0]?.seq ?? transcript.hiddenBeforeSeq;
-      if (oldest <= 1) return;
+      if (!hasEarlier()) return;
       const want = unloadCount(chatLimit);
       try {
         const earlier = await backfillOlder(oldest, want);
@@ -547,8 +617,14 @@ export function createChatStore(deps: ChatStoreDependencies): ChatStore {
       }
     },
 
-    resync(): void {
+    invalidateHistory(epoch): void {
+      if (!disposed && (historyEpoch === undefined || epoch > historyEpoch)) resetTranscript(epoch);
+    },
+
+    resync(epoch?: number): void {
       if (disposed) return;
+      if (epoch !== undefined && (historyEpoch === undefined || epoch > historyEpoch))
+        resetTranscript(epoch);
       // A turn may have sealed while we were disconnected (missed `turn-sealed`):
       // clear the now-stale spinner BEFORE re-subscribing, so a finished turn
       // doesn't spin forever. A still-running turn's replay re-asserts it.
diff --git a/src/features/chat/store.test.ts b/src/features/chat/store.test.ts
index 8604107..bb1e304 100644
--- a/src/features/chat/store.test.ts
+++ b/src/features/chat/store.test.ts
@@ -1,5 +1,8 @@
 import type { AgentEvent, ImageInput, StepId, StoredChunk } from "@dispatch/wire";
+import { IDBFactory, IDBKeyRange } from "fake-indexeddb";
 import { describe, expect, it, vi } from "vitest";
+import { createIdbChunkStore } from "../../adapters/idb";
+import { createConversationCache } from "../conversation-cache";
 import { createChatStore } from "./store.svelte";
 import {
   createFakeCache,
@@ -528,7 +531,7 @@ describe("createChatStore", () => {
     store.dispose();
   });
 
-  it("overlapping syncs are guarded", async () => {
+  it("overlapping syncs coalesce into a sequential rerun", async () => {
     const transport = createFakeTransport();
     const _historySync = createFakeHistorySync();
     const metricsSync = createFakeMetricsSync();
@@ -546,7 +549,7 @@ describe("createChatStore", () => {
       if (callCount === 1) {
         await firstSyncPromise;
       }
-      return { chunks: [], latestSeq: sinceSeq };
+      return { chunks: [], latestSeq: sinceSeq, historyEpoch: 0 };
     };
 
     const store = createChatStore({
@@ -573,7 +576,7 @@ describe("createChatStore", () => {
 
     // Release the first sync
     resolveFirstSync?.();
-    await new Promise((r) => setTimeout(r, 10));
+    await vi.waitFor(() => expect(callCount).toBe(2));
 
     store.dispose();
   });
@@ -1645,6 +1648,7 @@ describe("createChatStore", () => {
     });
     await store.load(); // only 50 chunks → all loaded, window starts at seq 1
     expect(store.chunks).toHaveLength(50);
+    await store.showEarlier(); // Probe server exhaustion; seq 1 alone is not a boundary.
     expect(store.hasEarlier).toBe(false);
     const callsAfterLoad = historySync.calls.length;
 
@@ -1892,3 +1896,96 @@ describe("createChatStore — cacheTimer", () => {
     busy.dispose();
   });
 });
+
+describe("history epochs", () => {
+  it.each([
+    "tail",
+    "backfill",
+  ])("recovers missed compaction on %s and exhausts non-1 history", async (path) => {
+    const history = createFakeHistorySync();
+    const cache = createFakeCache();
+    history.returnChunks = Array.from({ length: 20 }, (_, i) => makeStoredChunk(i + 1));
+    const store = createChatStore({
+      conversationId: CONV_ID,
+      transport: createFakeTransport().impl,
+      historySync: history.impl,
+      metricsSync: createFakeMetricsSync().impl,
+      cache: cache.impl,
+      chatLimit: 10,
+    });
+    await store.load();
+    history.historyEpoch = 1;
+    history.returnChunks = [makeStoredChunk(101), makeStoredChunk(102)];
+    if (path === "tail") {
+      store.resync();
+      await vi.waitFor(() => expect(store.chunks[0]?.seq).toBe(101));
+    } else await store.showEarlier();
+    expect(store.chunks.map((c) => c.seq)).toEqual([101, 102]);
+    expect((await cache.impl.load(CONV_ID)).map((c) => c.seq)).toEqual([101, 102]);
+    expect(history.calls.at(-1)).toMatchObject({ sinceSeq: 0, window: { limit: 7 } });
+    expect(store.hasEarlier).toBe(false);
+    const calls = history.calls.length;
+    await store.showEarlier();
+    expect(history.calls).toHaveLength(calls);
+    store.dispose();
+  });
+});
+
+it("a compaction broadcast during an in-flight fetch discards the old response", async () => {
+  const history = createFakeHistorySync();
+  history.returnChunks = [makeStoredChunk(101)];
+  let resolve!: (res: import("@dispatch/transport-contract").ConversationHistoryResponse) => void;
+  const first = new Promise<import("@dispatch/transport-contract").ConversationHistoryResponse>(
+    (r) => {
+      resolve = r;
+    },
+  );
+  const sync = vi
+    .fn<import("./ports").HistorySync>()
+    .mockImplementationOnce(() => first)
+    .mockImplementation(history.impl);
+  const cache = createFakeCache();
+  const store = createChatStore({
+    conversationId: CONV_ID,
+    transport: createFakeTransport().impl,
+    historySync: sync,
+    metricsSync: createFakeMetricsSync().impl,
+    cache: cache.impl,
+  });
+  const loading = store.load();
+  await vi.waitFor(() => expect(sync).toHaveBeenCalledTimes(1));
+  history.historyEpoch = 1;
+  store.resync(1);
+  resolve({ chunks: [makeStoredChunk(1)], latestSeq: 1, historyEpoch: 0 });
+  await loading;
+  expect(sync).toHaveBeenCalledTimes(2);
+  expect(store.chunks.map((c) => c.seq)).toEqual([101]);
+  expect((await cache.impl.load(CONV_ID)).map((c) => c.seq)).toEqual([101]);
+  store.dispose();
+});
+
+it("reloads unknown legacy cache epochs rather than merging pre-upgrade rows", async () => {
+  vi.stubGlobal("IDBKeyRange", IDBKeyRange);
+  const storage = createIdbChunkStore({ indexedDB: new IDBFactory() });
+  const cache = createConversationCache(storage);
+  await storage.append(CONV_ID, [makeStoredChunk(1)]);
+  const history = createFakeHistorySync();
+  history.returnChunks = [makeStoredChunk(101)];
+  const store = createChatStore({
+    conversationId: CONV_ID,
+    transport: createFakeTransport().impl,
+    historySync: history.impl,
+    metricsSync: createFakeMetricsSync().impl,
+    cache,
+  });
+  try {
+    await store.load();
+    expect(history.calls[0]).toMatchObject({ sinceSeq: 0 });
+    expect(store.chunks.map((c) => c.seq)).toEqual([101]);
+    expect(await cache.historyEpoch(CONV_ID)).toBe(0);
+    expect((await cache.load(CONV_ID)).map((c) => c.seq)).toEqual([101]);
+  } finally {
+    store.dispose();
+    vi.unstubAllGlobals();
+  }
+});
diff --git a/src/features/chat/test-helpers.ts b/src/features/chat/test-helpers.ts
index c99d1f4..c1b020b 100644
--- a/src/features/chat/test-helpers.ts
+++ b/src/features/chat/test-helpers.ts
@@ -43,14 +43,22 @@ export interface FakeHistorySync {
   readonly calls: Array<{ conversationId: string; sinceSeq: number; window?: HistoryWindow }>;
   /** Set the chunks to return on the next call. */
   returnChunks: readonly StoredChunk[];
+  historyEpoch: number;
   readonly impl: HistorySync;
 }
 
 export function createFakeHistorySync(): FakeHistorySync {
   const calls: Array<{ conversationId: string; sinceSeq: number; window?: HistoryWindow }> = [];
   let returnChunks: readonly StoredChunk[] = [];
+  let historyEpoch = 0;
   return {
     calls,
+    get historyEpoch() {
+      return historyEpoch;
+    },
+    set historyEpoch(v: number) {
+      historyEpoch = v;
+    },
     get returnChunks() {
       return returnChunks;
     },
@@ -72,7 +80,7 @@ export function createFakeHistorySync(): FakeHistorySync {
         chunks = chunks.slice(-window.limit);
       }
       const latestSeq = chunks.length > 0 ? Math.max(...chunks.map((c) => c.seq)) : sinceSeq;
-      return { chunks, latestSeq };
+      return { chunks, latestSeq, historyEpoch };
     },
   };
 }
@@ -122,13 +130,22 @@ export interface FakeCache {
 
 export function createFakeCache(): FakeCache {
   const store = new Map<string, StoredChunk[]>();
+  const epochs = new Map<string, number>();
   return {
     store,
     impl: {
+      async historyEpoch(id) {
+        return epochs.get(id) ?? 0;
+      },
+      async reset(id, epoch) {
+        store.delete(id);
+        epochs.set(id, epoch);
+      },
       async load(conversationId) {
         return store.get(conversationId) ?? [];
       },
-      async commit(conversationId, incoming) {
+      async commit(conversationId, incoming, epoch) {
+        if (epoch !== undefined) epochs.set(conversationId, epoch);
         const existing = store.get(conversationId) ?? [];
         const seen = new Set(existing.map((c) => c.seq));
         const toAppend = incoming.filter((c) => !seen.has(c.seq));
diff --git a/src/features/conversation-cache/cache.test.ts b/src/features/conversation-cache/cache.test.ts
index ce1c60c..4306d51 100644
--- a/src/features/conversation-cache/cache.test.ts
+++ b/src/features/conversation-cache/cache.test.ts
@@ -15,8 +15,16 @@ const chunk = (seq: number, role: "user" | "assistant" = "user"): StoredChunk =>
  */
 function createFakeStore(): ConversationChunkStore {
   const store = new Map<string, StoredChunk[]>();
+  const epochs = new Map<string, number>();
 
   return {
+    async historyEpoch(id) {
+      return epochs.get(id);
+    },
+    async reset(id, epoch) {
+      store.delete(id);
+      epochs.set(id, epoch);
+    },
     async load(conversationId) {
       return store.get(conversationId) ?? [];
     },
@@ -33,6 +41,7 @@ function createFakeStore(): ConversationChunkStore {
 
     async delete(conversationId) {
       store.delete(conversationId);
+      epochs.delete(conversationId);
     },
 
     async index() {
@@ -195,3 +204,13 @@ describe("cache.delete", () => {
     expect(result).toEqual([]);
   });
 });
+
+it("serializes invalidation and rejects stale epoch commits", async () => {
+  const store = createFakeStore();
+  const cache = createConversationCache(store);
+  await store.append("c", [chunk(1)]); // Legacy rows have no trusted epoch.
+  expect(await cache.commit("c", [chunk(101)], 1)).toEqual([chunk(101)]);
+  await Promise.all([cache.commit("c", [chunk(102)], 1), cache.reset("c", 2)]);
+  expect(await cache.commit("c", [chunk(103)], 1)).toEqual([]);
+  expect(await store.historyEpoch("c")).toBe(2);
+});
diff --git a/src/features/conversation-cache/cache.ts b/src/features/conversation-cache/cache.ts
index 4953944..431a2a7 100644
--- a/src/features/conversation-cache/cache.ts
+++ b/src/features/conversation-cache/cache.ts
@@ -3,6 +3,9 @@ import { nextSinceSeq, reconcileCache, selectEvictions } from "./logic";
 import type { ConversationChunkStore } from "./types";
 
 export interface ConversationCache {
+  historyEpoch(conversationId: string): Promise<number | undefined>;
+  reset(conversationId: string, historyEpoch: number): Promise<void>;
+
   /** Load all cached chunks for a conversation. */
   load(conversationId: string): Promise<readonly StoredChunk[]>;
 
@@ -10,7 +13,11 @@ export interface ConversationCache {
    * Load + reconcile + append new chunks.
    * Returns the merged cache (the new authoritative cache for this conversation).
    */
-  commit(conversationId: string, incoming: readonly StoredChunk[]): Promise<readonly StoredChunk[]>;
+  commit(
+    conversationId: string,
+    incoming: readonly StoredChunk[],
+    historyEpoch?: number,
+  ): Promise<readonly StoredChunk[]>;
 
   /** Return the `?sinceSeq=` cursor for the next incremental sync. */
   sinceSeq(conversationId: string): Promise<number>;
@@ -43,21 +50,52 @@ export function createConversationCache(
 ): ConversationCache {
   const maxChunks = opts?.maxChunks ?? DEFAULT_MAX_CHUNKS;
 
+  // Serialize mutation per conversation, including invalidation, so an old fetch
+  // cannot resurrect rows after a compaction broadcast has invalidated them.
+  const pending = new Map<string, Promise<unknown>>();
+  function exclusive<T>(id: string, run: () => Promise<T>): Promise<T> {
+    const next = (pending.get(id) ?? Promise.resolve()).catch(() => {}).then(run);
+    pending.set(id, next);
+    void next
+      .finally(() => {
+        if (pending.get(id) === next) pending.delete(id);
+      })
+      .catch(() => {});
+    return next;
+  }
   return {
+    historyEpoch: async (id) => {
+      await pending.get(id);
+      return store.historyEpoch(id);
+    },
+    reset: (id, epoch) =>
+      exclusive(id, async () => {
+        const current = await store.historyEpoch(id);
+        if (current === undefined || current < epoch) await store.reset(id, epoch);
+      }),
     async load(conversationId) {
+      await pending.get(conversationId);
       return store.load(conversationId);
     },
 
-    async commit(conversationId, incoming) {
-      const cached = await store.load(conversationId);
-      const { merged, toAppend } = reconcileCache(cached, incoming);
-      if (toAppend.length > 0) {
-        await store.append(conversationId, toAppend);
-      }
-      return merged;
+    commit(conversationId, incoming, epoch) {
+      return exclusive(conversationId, async () => {
+        if (epoch !== undefined) {
+          const current = await store.historyEpoch(conversationId);
+          if (current !== undefined && current > epoch) return store.load(conversationId);
+          if (current !== epoch) await store.reset(conversationId, epoch);
+        }
+        const cached = await store.load(conversationId);
+        const { merged, toAppend } = reconcileCache(cached, incoming);
+        if (toAppend.length > 0) {
+          await store.append(conversationId, toAppend);
+        }
+        return merged;
+      });
     },
 
     async sinceSeq(conversationId) {
+      await pending.get(conversationId);
       const cached = await store.load(conversationId);
       return nextSinceSeq(cached);
     },
@@ -72,7 +110,7 @@ export function createConversationCache(
     },
 
     async delete(conversationId) {
-      await store.delete(conversationId);
+      await exclusive(conversationId, () => store.delete(conversationId));
     },
   };
 }
diff --git a/src/features/conversation-cache/types.ts b/src/features/conversation-cache/types.ts
index 345ab71..a273f34 100644
--- a/src/features/conversation-cache/types.ts
+++ b/src/features/conversation-cache/types.ts
@@ -17,6 +17,10 @@ export interface ConversationCacheIndexEntry {
  * All methods MUST be idempotent on `seq`: re-appending an existing seq is a no-op.
  */
 export interface ConversationChunkStore {
+  historyEpoch(conversationId: string): Promise<number | undefined>;
+  /** Atomically replace cached rows with empty history at this epoch. */
+  reset(conversationId: string, historyEpoch: number): Promise<void>;
+
   /** Load all cached chunks for a conversation, seq-ordered. Returns [] if absent. */
   load(conversationId: string): Promise<readonly StoredChunk[]>;
 
diff --git a/src/features/layout/adapter/http.test.ts b/src/features/layout/adapter/http.test.ts
index cc7d7e9..30e194a 100644
--- a/src/features/layout/adapter/http.test.ts
+++ b/src/features/layout/adapter/http.test.ts
@@ -77,7 +77,7 @@ describe("createLayoutHttp — getTree", () => {
   it("a malformed 200 (SPA-fallback JSON without a tree) degrades to notFound", async () => {
     const http = createLayoutHttp(
       "http://backend.test",
-      fetchReturning(200, { chunks: [], latestSeq: 0 }),
+      fetchReturning(200, { chunks: [], historyEpoch: 0, latestSeq: 0 }),
     );
     const result = await http.getTree("ws-a");
     expect(result).toMatchObject({ ok: false, notFound: true });