Merge local dev into feature/close-tab-cascade-subagents

3 files changed, 127 insertions, 9 deletions

diff --git a/src/features/conversation-cache/cache.test.ts b/src/features/conversation-cache/cache.test.ts
index ce1c60c..72b4e20 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,48 @@ 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);
+});
+
+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 4953944..600a6e6 100644
--- a/src/features/conversation-cache/cache.ts
+++ b/src/features/conversation-cache/cache.ts
@@ -3,6 +3,22 @@ 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>;
+
+  /**
+   * 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<boolean>;
+
   /** Load all cached chunks for a conversation. */
   load(conversationId: string): Promise<readonly StoredChunk[]>;
 
@@ -10,7 +26,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 +63,61 @@ 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);
+      }),
+    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);
     },
 
-    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);
+          // 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);
+        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 +132,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[]>;