diff --git a/agent-chat/adapters/codex.ts b/agent-chat/adapters/codex.ts index 6cc494701b5d..6a5cf5b0018c 100644 --- a/agent-chat/adapters/codex.ts +++ b/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 | 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 { 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 { if (st.currentTurnId) return Promise.resolve(st.currentTurnId); return new Promise((resolve) => { diff --git a/agent-chat/test/codex-recovery.test.ts b/agent-chat/test/codex-recovery.test.ts new file mode 100644 index 000000000000..de7276c68bf8 --- /dev/null +++ b/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) { + 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 {}; diff --git a/agent-chat/test/fake-codex-recovery.ts b/agent-chat/test/fake-codex-recovery.ts new file mode 100644 index 000000000000..5fe6300b6d3f --- /dev/null +++ b/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(); +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}` } }); + } +}