fix(review-followup-fec): reconcile epoch rollback and retain reconnect backoff

7 files changed, 383 insertions, 20 deletions

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<typeof setTimeout> | null = null;
+  let healthyTimer: ReturnType<typeof setTimeout> | 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<typeof createChatStore>;
+    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<Response>((r) => {
+    resolve = r;
+  });
+  let delayNext = false;
+  let epoch = 5;
+  const sync = vi.fn<import("./ports").HistorySync>().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<void>;
+    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<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[]>;
 
@@ -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);