Resume Cloud Codex chats after app-server restart

agent-chat/adapters/codex.ts

@@ -74,18 +74,25 @@ export const codexAdapter: Adapter = {
       const srv = await ensureServer();
       const st = await ensureCodexState(sess);
       let threadId = sess.internal.threadId as string | undefined;
-      if (!threadId) {
-        // Single-flight: concurrent first sends must share one thread/start or
-        // each spawns its own thread and the UI tracks only one of them.
+      if (!threadId || srv.sessionsByThread.get(threadId) !== sess) {
+        // Single-flight: concurrent first sends and post-crash resumes must
+        // share one thread/start or resume, rather than creating duplicates.
         let starting = sess.internal.threadStarting as Promise<string> | undefined;
         if (!starting) {
           starting = (async () => {
-            const res = await srv.request("thread/start", { cwd: sess.cwd });
+            const savedThreadId = sess.internal.threadId as string | undefined;
+            const res = savedThreadId
+              ? await srv.request("thread/resume", { threadId: savedThreadId, cwd: sess.cwd })
+              : await srv.request("thread/start", { cwd: sess.cwd });
             const id: string | undefined = res.thread?.id;
-            if (!id) throw new Error("codex thread/start returned no thread id");
+            if (!id) {
+              throw new Error(
+                `codex ${savedThreadId ? "thread/resume" : "thread/start"} returned no thread id`,
+              );
+            }
             sess.internal.threadId = id;
             srv.sessionsByThread.set(id, sess);
-            sess.emit({ kind: "meta", providerSessionId: id });
+            if (id !== savedThreadId) sess.emit({ kind: "meta", providerSessionId: id });
             emitOptions(sess);
             await refreshCommands(sess);
             return id;
@@ -293,7 +300,6 @@ async function startServer(): Promise<AppServer> {
         sess.emit({ kind: "done", generation } as any);
         sess.setStatus("idle");
       }
-      sess.internal.threadId = undefined;
     }
     if (shared === srv) shared = null;
   });
@@ -553,6 +559,13 @@ export function codexSetSharedServerForTest(srv: unknown): void {
   shared = (srv as AppServer | null) ?? null;
 }
 
+/** Stops the shared child used by an isolated adapter lifecycle test. */
+export function codexStopSharedServerForTest(): void {
+  const srv = shared;
+  shared = null;
+  srv?.proc.kill();
+}
+
 function waitForTurnId(st: CodexState): Promise<string | null> {
   if (st.currentTurnId) return Promise.resolve(st.currentTurnId);
   return new Promise((resolve) => {

agent-chat/test/codex-recovery.test.ts

@@ -0,0 +1,100 @@
+// Run the real adapter in a child with a fake codex executable on PATH. This
+// exercises spawn, stdout EOF, resume and turn dispatch without user sessions.
+import { chmod, mkdtemp, readFile, rm, writeFile } from "node:fs/promises";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import { codexAdapter, codexStopSharedServerForTest } from "../adapters/codex";
+import type { AgentEvent, SessionCtx } from "../types";
+
+async function waitUntil(check: () => boolean | Promise<boolean>) {
+  const deadline = Date.now() + 5_000;
+  while (Date.now() < deadline) {
+    if (await check()) return;
+    await Bun.sleep(10);
+  }
+  throw new Error("timed out waiting for fake Codex lifecycle");
+}
+
+if (!process.env.CMUX_TEST_CODEX_RECOVERY_DIR) {
+  const root = await mkdtemp(join(tmpdir(), "cmux-codex-recovery-"));
+  const quote = (value: string) => `'${value.replaceAll("'", "'\\''")}'`;
+  await writeFile(join(root, "codex"), `#!/bin/sh\nexec ${quote(process.execPath)} ${quote(join(import.meta.dir, "fake-codex-recovery.ts"))}\n`);
+  await chmod(join(root, "codex"), 0o700);
+  const child = Bun.spawn([process.execPath, import.meta.path], {
+    env: { ...process.env, PATH: `${root}:${process.env.PATH}`, CMUX_TEST_CODEX_RECOVERY_DIR: root },
+    stdio: ["ignore", "inherit", "inherit"],
+  });
+  try {
+    const result = await child.exited;
+    if (result !== 0) throw new Error(`Codex recovery regression failed (${result})`);
+  } finally {
+    child.kill();
+    const pids = (await readFile(join(root, "pids"), "utf8").catch(() => "")).trim().split(/\s+/).filter(Boolean);
+    for (const pid of pids) { try { process.kill(Number(pid), "SIGKILL"); } catch {} }
+    await rm(root, { recursive: true, force: true });
+  }
+} else {
+  const root = process.env.CMUX_TEST_CODEX_RECOVERY_DIR;
+  const requests = async () => (await readFile(join(root, "requests"), "utf8")).trim().split("\n").map((line) => JSON.parse(line));
+  const latestPid = async () => Number((await readFile(join(root, "pids"), "utf8")).trim().split("\n").at(-1));
+  function session(id: string): SessionCtx {
+    return {
+      id, provider: "codex", cwd: root, title: id, autoApprove: false,
+      startOptions: {}, status: "idle", events: [], internal: {},
+      emit(event: AgentEvent) { this.events.push(event); },
+      setStatus(status) { this.status = status; },
+    };
+  }
+  try {
+    const idle = session("idle");
+    const active = session("active");
+    await codexAdapter.send(idle, "first prompt");
+    await waitUntil(() => idle.events.some((event) => event.kind === "done"));
+    await codexAdapter.send(active, "stay active");
+    await waitUntil(() => active.status === "running");
+    const idleThread = idle.internal.threadId;
+    const activeThread = active.internal.threadId;
+    process.kill(await latestPid(), "SIGKILL");
+    await waitUntil(() => active.events.some((event) => event.kind === "error"));
+
+    await codexAdapter.send(idle, "continue after crash");
+    await codexAdapter.send(active, "continue after crash");
+    await waitUntil(async () => (await requests()).filter((entry) => entry.method === "turn/start" && entry.params.input[0].text === "continue after crash").length === 2);
+    const recorded = await requests();
+    const starts = recorded.filter((entry) => entry.method === "thread/start");
+    if (starts.length !== 2) {
+      throw new Error(`crash recovery started ${starts.length} threads instead of preserving the original two`);
+    }
+    for (const threadId of [idleThread, activeThread]) {
+      if (!recorded.some((entry) => entry.method === "thread/resume" && entry.params.threadId === threadId)) {
+        throw new Error(`crash recovery did not resume ${threadId}`);
+      }
+      if (!recorded.some((entry) => entry.method === "turn/start" && entry.params.threadId === threadId && entry.params.input[0].text === "continue after crash")) {
+        throw new Error(`recovered prompt was not delivered to ${threadId}`);
+      }
+    }
+
+    // A failed resume must preserve the ID for retry, never fall back to a blank thread.
+    await writeFile(join(root, "refuse-resume"), "1");
+    const errorCount = idle.events.filter((event) => event.kind === "error").length;
+    await codexAdapter.send(active, "stay active");
+    process.kill(await latestPid(), "SIGKILL");
+    const activeErrors = active.events.filter((event) => event.kind === "error").length;
+    await waitUntil(() => active.events.filter((event) => event.kind === "error").length > activeErrors);
+    await codexAdapter.send(idle, "retry saved conversation");
+    if (idle.events.filter((event) => event.kind === "error").length <= errorCount) {
+      throw new Error("failed resume did not surface a recoverable error");
+    }
+    if ((await requests()).filter((entry) => entry.method === "thread/start").length !== 2) {
+      throw new Error("failed resume silently started a new conversation");
+    }
+    await rm(join(root, "refuse-resume"));
+    await codexAdapter.send(idle, "retry saved conversation");
+    if (idle.internal.threadId !== idleThread) throw new Error("resume retry lost the saved conversation ID");
+    console.log("codex crash recovery assertions passed");
+  } finally {
+    codexStopSharedServerForTest();
+  }
+}
+
+export {};

agent-chat/test/fake-codex-recovery.ts

@@ -0,0 +1,53 @@
+// A protocol peer for crash recovery tests; never contacts a provider.
+import { appendFileSync, existsSync } from "node:fs";
+import { createInterface } from "node:readline";
+
+const root = process.env.CMUX_TEST_CODEX_RECOVERY_DIR!;
+appendFileSync(`${root}/pids`, `${process.pid}\n`);
+const threads = new Set<string>();
+let nextThread = 0;
+let nextTurn = 0;
+const write = (message: unknown) => process.stdout.write(JSON.stringify(message) + "\n");
+for await (const line of createInterface({ input: process.stdin })) {
+  const message = JSON.parse(line);
+  const { id, method, params } = message;
+  appendFileSync(`${root}/requests`, JSON.stringify({ pid: process.pid, method, params }) + "\n");
+  const answer = (result: unknown) => write({ id, result });
+  switch (method) {
+    case "initialize": answer({}); break;
+    case "model/list":
+      answer({ data: [{ id: "test-model", model: "test-model", isDefault: true, supportedReasoningEfforts: [] }] });
+      break;
+    case "collaborationMode/list": answer({ data: [] }); break;
+    case "skills/list": answer({ data: [] }); break;
+    case "thread/start": {
+      const threadId = `thread-${process.pid}-${++nextThread}`;
+      threads.add(threadId);
+      answer({ thread: { id: threadId } });
+      break;
+    }
+    case "thread/resume":
+      if (existsSync(`${root}/refuse-resume`)) {
+        write({ id, error: { message: "saved thread is unavailable" } });
+      } else {
+        threads.add(params.threadId);
+        answer({ thread: { id: params.threadId } });
+      }
+      break;
+    case "turn/start": {
+      if (!threads.has(params.threadId)) {
+        write({ id, error: { message: "thread must be resumed first" } });
+        break;
+      }
+      const turn = { id: `turn-${++nextTurn}` };
+      answer({ turn });
+      write({ method: "turn/started", params: { threadId: params.threadId, turn } });
+      if (params.input[0].text !== "stay active") {
+        write({ method: "turn/completed", params: { threadId: params.threadId, turn } });
+      }
+      break;
+    }
+    case "turn/steer": answer({}); break;
+    default: write({ id, error: { message: `unexpected request ${method}` } });
+  }
+}