feat(agent): resume codex cloud threads natively on restarts (#3216)

Author: k11kirky

packages/agent/src/adapters/codex-app-server/thread-state.test.ts

@@ -0,0 +1,50 @@
+import { mkdir, mkdtemp, rm, writeFile } from "node:fs/promises";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
+import { hasCodexThreadState } from "./thread-state";
+
+const THREAD_ID = "0199a5c3-2f60-7b21-9c39-1d2e3f4a5b6c";
+
+describe("hasCodexThreadState", () => {
+  let codexHome: string;
+
+  beforeEach(async () => {
+    codexHome = await mkdtemp(join(tmpdir(), "codex-home-"));
+    vi.stubEnv("CODEX_HOME", codexHome);
+  });
+
+  afterEach(async () => {
+    vi.unstubAllEnvs();
+    await rm(codexHome, { recursive: true, force: true });
+  });
+
+  const writeRollout = async (threadId: string) => {
+    const dir = join(codexHome, "sessions", "2026", "07", "07");
+    await mkdir(dir, { recursive: true });
+    await writeFile(
+      join(dir, `rollout-2026-07-07T10-00-00-${threadId}.jsonl`),
+      "",
+    );
+  };
+
+  it.each([
+    [true, "the persisted thread id", THREAD_ID],
+    [false, "a different thread id", "11111111-2222-3333-4444-555555555555"],
+    [false, "an empty thread id", ""],
+  ])("returns %s querying %s", async (expected, _case, queriedId) => {
+    await writeRollout(THREAD_ID);
+    await expect(hasCodexThreadState(queriedId)).resolves.toBe(expected);
+  });
+
+  it("returns false when there is no sessions directory", async () => {
+    await expect(hasCodexThreadState(THREAD_ID)).resolves.toBe(false);
+  });
+
+  it("ignores files that are not rollouts", async () => {
+    const dir = join(codexHome, "sessions", "2026", "07", "07");
+    await mkdir(dir, { recursive: true });
+    await writeFile(join(dir, `notes-${THREAD_ID}.jsonl`), "");
+    await expect(hasCodexThreadState(THREAD_ID)).resolves.toBe(false);
+  });
+});

packages/agent/src/adapters/codex-app-server/thread-state.ts

@@ -0,0 +1,25 @@
+import { readdir } from "node:fs/promises";
+import os from "node:os";
+import path from "node:path";
+
+/**
+ * Whether codex still holds the persisted rollout for `threadId` — written as
+ * `CODEX_HOME/sessions/YYYY/MM/DD/rollout-<timestamp>-<threadId>.jsonl` — i.e.
+ * whether `thread/resume` can restore the thread natively. Mirrors codex's own
+ * CODEX_HOME resolution (env override, else ~/.codex).
+ */
+export async function hasCodexThreadState(threadId: string): Promise<boolean> {
+  if (!threadId) return false;
+  const codexHome = process.env.CODEX_HOME || path.join(os.homedir(), ".codex");
+  const sessionsDir = path.join(codexHome, "sessions");
+  const suffix = `-${threadId}.jsonl`;
+  try {
+    const entries = await readdir(sessionsDir, { recursive: true });
+    return entries.some((entry) => {
+      const name = path.basename(entry);
+      return name.startsWith("rollout-") && name.endsWith(suffix);
+    });
+  } catch {
+    return false;
+  }
+}

packages/agent/src/server/agent-server.test.ts

@@ -1,5 +1,5 @@
 import { createHash } from "node:crypto";
-import { mkdtemp, readFile, rm } from "node:fs/promises";
+import { mkdir, mkdtemp, readFile, rm, writeFile } from "node:fs/promises";
 import { tmpdir } from "node:os";
 import { join } from "node:path";
 import type { ContentBlock } from "@agentclientprotocol/sdk";
@@ -1766,6 +1766,77 @@ describe("AgentServer HTTP Mode", () => {
         }
       }
     });
+
+    describe("codex", () => {
+      const THREAD_ID = "0199a5c3-2f60-7b21-9c39-1d2e3f4a5b6c";
+      let codexHome: string;
+
+      const payload: JwtPayload = {
+        task_id: "test-task-id",
+        run_id: "test-run-id",
+        team_id: 1,
+        user_id: 1,
+        distinct_id: "test-distinct-id",
+        mode: "interactive",
+      };
+
+      const codexServer = (sessionId: string | null) => {
+        const s = createServer() as unknown as NativeResumeTestServer;
+        s.resumeState = {
+          conversation: [
+            { role: "user", content: [{ type: "text", text: "continue" }] },
+          ],
+          latestGitCheckpoint: null,
+          interrupted: false,
+          logEntryCount: 1,
+          sessionId,
+        };
+        return s;
+      };
+
+      const prepare = (s: NativeResumeTestServer) =>
+        s.prepareNativeResume(
+          payload,
+          createMockApiClient(),
+          createTaskRun({
+            id: "test-run-id",
+            state: { resume_from_run_id: "previous-run" },
+          }),
+          "codex",
+          repo.path,
+          "auto",
+        );
+
+      beforeEach(() => {
+        codexHome = join(repo.path, ".codex-test");
+        vi.stubEnv("CODEX_HOME", codexHome);
+      });
+
+      afterEach(() => {
+        vi.unstubAllEnvs();
+      });
+
+      it("resumes natively when the thread rollout survived in CODEX_HOME", async () => {
+        const dir = join(codexHome, "sessions", "2026", "07", "07");
+        await mkdir(dir, { recursive: true });
+        await writeFile(
+          join(dir, `rollout-2026-07-07T10-00-00-${THREAD_ID}.jsonl`),
+          "",
+        );
+
+        await expect(prepare(codexServer(THREAD_ID))).resolves.toEqual({
+          sessionId: THREAD_ID,
+          warm: true,
+        });
+      });
+
+      it.each([
+        ["the thread state is gone", THREAD_ID],
+        ["there is no prior session id", null],
+      ])("falls back to summary resume when %s", async (_case, sessionId) => {
+        await expect(prepare(codexServer(sessionId))).resolves.toBeNull();
+      });
+    });
   });
 
   describe("PR attribution", () => {

packages/agent/src/server/agent-server.ts

@@ -30,6 +30,7 @@ import {
   hydrateSessionJsonl,
 } from "../adapters/claude/session/jsonl-hydration";
 import type { GatewayEnv } from "../adapters/claude/session/options";
+import { hasCodexThreadState } from "../adapters/codex-app-server/thread-state";
 import {
   type AgentErrorClassification,
   classifyAgentError,
@@ -672,8 +673,6 @@ export class AgentServer {
     cwd: string,
     permissionMode: PermissionMode,
   ): Promise<{ sessionId: string; warm: boolean } | null> {
-    if (runtimeAdapter !== "claude") return null;
-
     const resumeRunId = this.getResumeRunId(preTaskRun);
     if (!resumeRunId) return null;
 
@@ -689,6 +688,22 @@ export class AgentServer {
       return null;
     }
 
+    if (runtimeAdapter === "codex") {
+      // Codex owns thread persistence in CODEX_HOME (the ACP sessionId is the
+      // codex thread id). The rollout only survives a snapshot restart — there
+      // is no cold hydration equivalent, so a fresh sandbox keeps the summary
+      // fallback while a warm one resumes the thread natively via thread/resume.
+      if (!(await hasCodexThreadState(priorSessionId))) {
+        this.logger.debug(
+          "No codex thread state on disk; using summary resume fallback",
+          { resumeRunId, priorSessionId },
+        );
+        return null;
+      }
+      this.logger.debug("Native codex resume prepared", { priorSessionId });
+      return { sessionId: priorSessionId, warm: true };
+    }
+
     let warm = false;
     try {
       await access(getSessionJsonlPath(priorSessionId, cwd));
@@ -1309,22 +1324,32 @@ export class AgentServer {
       initialPermissionMode,
     );
 
-    let acpSessionId: string;
+    let acpSessionId: string | null = null;
     if (nativeResume) {
-      await clientConnection.resumeSession({
-        sessionId: nativeResume.sessionId,
-        cwd: sessionCwd,
-        mcpServers: this.config.mcpServers ?? [],
-        _meta: { ...sessionMeta, sessionId: nativeResume.sessionId },
-      });
-      acpSessionId = nativeResume.sessionId;
-      this.nativeResume = nativeResume;
-      this.logger.debug("ACP session resumed", {
-        acpSessionId,
-        runId: payload.run_id,
-        warm: nativeResume.warm,
-      });
-    } else {
+      try {
+        await clientConnection.resumeSession({
+          sessionId: nativeResume.sessionId,
+          cwd: sessionCwd,
+          mcpServers: this.config.mcpServers ?? [],
+          _meta: { ...sessionMeta, sessionId: nativeResume.sessionId },
+        });
+        acpSessionId = nativeResume.sessionId;
+        this.nativeResume = nativeResume;
+        this.logger.debug("ACP session resumed", {
+          acpSessionId,
+          runId: payload.run_id,
+          warm: nativeResume.warm,
+        });
+      } catch (error) {
+        // resumeState is still loaded, so the summary resume path takes over
+        // on the fresh session below.
+        this.logger.warn("Native resume failed; starting a fresh session", {
+          sessionId: nativeResume.sessionId,
+          error: error instanceof Error ? error.message : String(error),
+        });
+      }
+    }
+    if (!acpSessionId) {
       const sessionResponse = await clientConnection.newSession({
         cwd: sessionCwd,
         mcpServers: this.config.mcpServers ?? [],