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

6 files changed, 351 insertions, 9 deletions

diff --git a/src/adapters/idb/index.test.ts b/src/adapters/idb/index.test.ts
index 04e2177..aa56f15 100644
--- a/src/adapters/idb/index.test.ts
+++ b/src/adapters/idb/index.test.ts
@@ -128,3 +128,18 @@ describe("createIdbChunkStore", () => {
     expect(loaded2[0]?.chunk).toEqual(textChunk("b1"));
   });
 });
+
+it("persists epoch across adapter recreation and append, including empty histories", async () => {
+  const indexedDB = new IDBFactory();
+  const store = createIdbChunkStore({ indexedDB });
+  await store.append("c", [makeChunk(1, "legacy")]);
+  expect(await store.historyEpoch("c")).toBeUndefined();
+  await store.reset("c", 2);
+  const reopened = createIdbChunkStore({ indexedDB });
+  expect(await reopened.load("c")).toEqual([]);
+  expect(await reopened.historyEpoch("c")).toBe(2);
+  await reopened.append("c", [makeChunk(101, "new")]);
+  expect(await reopened.historyEpoch("c")).toBe(2);
+  await reopened.delete("c");
+  expect(await reopened.historyEpoch("c")).toBeUndefined();
+});
diff --git a/src/adapters/idb/index.ts b/src/adapters/idb/index.ts
index 3dffef9..8b1848a 100644
--- a/src/adapters/idb/index.ts
+++ b/src/adapters/idb/index.ts
@@ -20,6 +20,7 @@ interface ChunkRecord {
 interface MetaRecord {
   conversationId: string;
   lastAccess: number;
+  historyEpoch?: number;
 }
 
 export interface CreateIdbChunkStoreOptions {
@@ -92,6 +93,26 @@ export function createIdbChunkStore(opts?: CreateIdbChunkStoreOptions): Conversa
   }
 
   return {
+    async historyEpoch(conversationId) {
+      const db = await getDb();
+      const tx = db.transaction(META_STORE, "readonly");
+      const meta = await requestToPromise<MetaRecord | undefined>(
+        tx.objectStore(META_STORE).get(conversationId),
+      );
+      await txComplete(tx);
+      return meta?.historyEpoch;
+    },
+    async reset(conversationId, historyEpoch) {
+      const db = await getDb();
+      const tx = db.transaction([CHUNKS_STORE, META_STORE], "readwrite");
+      tx.objectStore(CHUNKS_STORE).delete(keyRangeFor(conversationId));
+      tx.objectStore(META_STORE).put({
+        conversationId,
+        historyEpoch,
+        lastAccess: Date.now(),
+      } satisfies MetaRecord);
+      await txComplete(tx);
+    },
     async load(conversationId: string): Promise<readonly StoredChunk[]> {
       const db = await getDb();
       const tx = db.transaction(CHUNKS_STORE, "readonly");
@@ -122,7 +143,9 @@ export function createIdbChunkStore(opts?: CreateIdbChunkStoreOptions): Conversa
         } satisfies ChunkRecord);
       }
 
+      const meta = await requestToPromise<MetaRecord | undefined>(metaStore.get(conversationId));
       metaStore.put({
+        ...meta,
         conversationId,
         lastAccess: Date.now(),
       } satisfies MetaRecord);
diff --git a/src/adapters/ws/index.test.ts b/src/adapters/ws/index.test.ts
index 9d821e2..e4e4ba4 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,134 @@ 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();
+    }
+  });
+});
+
+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();
+});
+
+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 76d9e75..e1d7d22 100644
--- a/src/adapters/ws/index.ts
+++ b/src/adapters/ws/index.ts
@@ -1,6 +1,10 @@
 import type {
+  ChatAcceptedMessage,
   ChatDeltaMessage,
   ChatErrorMessage,
+  ChatQueuedMessage,
+  ChatResyncRequiredMessage,
+  ChatSubscriptionRejectedMessage,
   ConversationCompactedMessage,
   ConversationOpenMessage,
   ConversationStatusChangedMessage,
@@ -24,7 +28,15 @@ export interface WebSocketLike {
 export interface SurfaceSocketOptions {
   url: string;
   onMessage: (msg: SurfaceServerMessage) => void;
-  onChat?: (msg: ChatDeltaMessage | ChatErrorMessage) => void;
+  onChat?: (
+    msg:
+      | ChatAcceptedMessage
+      | ChatSubscriptionRejectedMessage
+      | 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). */
@@ -67,7 +79,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 +94,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);
       }
@@ -90,8 +112,23 @@ 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.accepted" ||
+          msg.type === "chat.subscription-rejected" ||
+          msg.type === "chat.delta" ||
+          msg.type === "chat.error" ||
+          msg.type === "chat.queued" ||
+          msg.type === "chat.resync-required"
+        ) {
+          opts.onChat?.(
+            msg as
+              | ChatAcceptedMessage
+              | ChatSubscriptionRejectedMessage
+              | ChatDeltaMessage
+              | ChatErrorMessage
+              | ChatQueuedMessage
+              | ChatResyncRequiredMessage,
+          );
         } else if (msg.type === "conversation.open") {
           opts.onConversationOpen?.(msg as ConversationOpenMessage);
         } else if (msg.type === "conversation.statusChanged") {
@@ -112,8 +149,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 +182,7 @@ export function createSurfaceSocket(opts: SurfaceSocketOptions): SurfaceSocketHa
     },
     close(): void {
       disposed = true;
+      clearHealthyTimer();
       if (reconnectTimer !== null) {
         clearTimeout(reconnectTimer);
         reconnectTimer = null;
diff --git a/src/adapters/ws/logic.test.ts b/src/adapters/ws/logic.test.ts
index 2179ad0..4bc2aad 100644
--- a/src/adapters/ws/logic.test.ts
+++ b/src/adapters/ws/logic.test.ts
@@ -562,3 +562,79 @@ 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();
+  }
+});
+
+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);
+  });
+});
+
+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 cecc66c..d167669 100644
--- a/src/adapters/ws/logic.ts
+++ b/src/adapters/ws/logic.ts
@@ -1,6 +1,8 @@
 import type {
   ChatDeltaMessage,
   ChatErrorMessage,
+  ChatQueuedMessage,
+  ChatResyncRequiredMessage,
   ConversationCompactedMessage,
   ConversationOpenMessage,
   ConversationStatusChangedMessage,
@@ -23,8 +25,12 @@ const VALID_SERVER_TYPES = new Set([
   "surface",
   "update",
   "error",
+  "chat.accepted",
+  "chat.subscription-rejected",
   "chat.delta",
   "chat.error",
+  "chat.queued",
+  "chat.resync-required",
   "conversation.open",
   "conversation.statusChanged",
   "conversation.compacted",
@@ -105,12 +111,56 @@ export function parseServerMessage(data: string): WsServerMessage | null {
           : { type: "error", message: parsed.message };
       return msg;
     }
+    case "chat.accepted": {
+      if (typeof parsed.conversationId !== "string" || typeof parsed.requestId !== "string")
+        return null;
+      return {
+        type: "chat.accepted",
+        conversationId: parsed.conversationId,
+        requestId: parsed.requestId,
+      };
+    }
+    case "chat.subscription-rejected": {
+      if (typeof parsed.conversationId !== "string" || parsed.reason !== "subscription-limit")
+        return null;
+      return {
+        type: "chat.subscription-rejected",
+        conversationId: parsed.conversationId,
+        reason: parsed.reason,
+      };
+    }
     case "chat.delta": {
       const event = parsed.event;
       if (!isRecord(event)) return 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;
+      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;
@@ -119,7 +169,11 @@ export function parseServerMessage(data: string): WsServerMessage | null {
         conversationId !== undefined
           ? { type: "chat.error", conversationId, message: parsed.message }
           : { type: "chat.error", message: parsed.message };
-      return msg;
+      if (parsed.requestId !== undefined && typeof parsed.requestId !== "string") return null;
+      return {
+        ...msg,
+        ...(typeof parsed.requestId === "string" ? { requestId: parsed.requestId } : {}),
+      };
     }
     case "conversation.open": {
       if (typeof parsed.conversationId !== "string") return null;
@@ -180,7 +234,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,