From b895c86c2d1a0eb2b4e61efcbd885c5ac426c3c5 Mon Sep 17 00:00:00 2001 From: Peter Kirkham Date: Tue, 7 Jul 2026 10:35:29 +0100 Subject: [PATCH 1/2] feat(agent): resume codex cloud threads natively on warm restarts prepareNativeResume was Claude-only, so every codex cloud resume started a fresh thread and injected the prior conversation as a hidden markdown blob, dropping reasoning context, compaction state, and cache continuity. Now, when the thread's rollout file survived in CODEX_HOME (snapshot/warm sandbox restart), the server resumes the thread via the adapter's native thread/resume path. Cold restores keep the summary fallback (rollouts hold encrypted reasoning that cannot be rebuilt from logs), and a failed native resume now falls back to a fresh session instead of failing the run. Generated-By: PostHog Code Task-Id: 958c74c2-c281-438d-b112-f74dab2d1e48 --- .../codex-app-server/thread-state.test.ts | 58 +++++++++++++++ .../adapters/codex-app-server/thread-state.ts | 25 +++++++ .../agent/src/server/agent-server.test.ts | 74 ++++++++++++++++++- packages/agent/src/server/agent-server.ts | 59 ++++++++++----- 4 files changed, 198 insertions(+), 18 deletions(-) create mode 100644 packages/agent/src/adapters/codex-app-server/thread-state.test.ts create mode 100644 packages/agent/src/adapters/codex-app-server/thread-state.ts diff --git a/packages/agent/src/adapters/codex-app-server/thread-state.test.ts b/packages/agent/src/adapters/codex-app-server/thread-state.test.ts new file mode 100644 index 0000000000..77bf625be5 --- /dev/null +++ b/packages/agent/src/adapters/codex-app-server/thread-state.test.ts @@ -0,0 +1,58 @@ +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("finds a persisted rollout for the thread id", async () => { + await writeRollout(THREAD_ID); + await expect(hasCodexThreadState(THREAD_ID)).resolves.toBe(true); + }); + + it("returns false when only a different thread's rollout exists", async () => { + await writeRollout(THREAD_ID); + await expect( + hasCodexThreadState("11111111-2222-3333-4444-555555555555"), + ).resolves.toBe(false); + }); + + it("returns false when there is no sessions directory", async () => { + await expect(hasCodexThreadState(THREAD_ID)).resolves.toBe(false); + }); + + it("returns false for an empty thread id", async () => { + await writeRollout(THREAD_ID); + await expect(hasCodexThreadState("")).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); + }); +}); diff --git a/packages/agent/src/adapters/codex-app-server/thread-state.ts b/packages/agent/src/adapters/codex-app-server/thread-state.ts new file mode 100644 index 0000000000..0963405aca --- /dev/null +++ b/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--.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 { + 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.toString()); + return name.startsWith("rollout-") && name.endsWith(suffix); + }); + } catch { + return false; + } +} diff --git a/packages/agent/src/server/agent-server.test.ts b/packages/agent/src/server/agent-server.test.ts index 615bd854c1..d8ed1252a7 100644 --- a/packages/agent/src/server/agent-server.test.ts +++ b/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,78 @@ 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("falls back to summary resume when the thread state is gone", async () => { + await expect(prepare(codexServer(THREAD_ID))).resolves.toBeNull(); + }); + + it("falls back to summary resume without a prior session id", async () => { + await expect(prepare(codexServer(null))).resolves.toBeNull(); + }); + }); }); describe("PR attribution", () => { diff --git a/packages/agent/src/server/agent-server.ts b/packages/agent/src/server/agent-server.ts index 172e5e666f..f1d4385ee7 100644 --- a/packages/agent/src/server/agent-server.ts +++ b/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 ?? [], From 30042b76be70326e1e5ac5d309f08b8bd477df86 Mon Sep 17 00:00:00 2001 From: Peter Kirkham Date: Tue, 7 Jul 2026 11:01:28 +0100 Subject: [PATCH 2/2] refactor(agent): address review comments on codex thread-state - drop the no-op .toString() on readdir entries (string[] in this form) - parameterise the identical-setup thread-state and codex fallback tests with it.each per repo convention Generated-By: PostHog Code Task-Id: 958c74c2-c281-438d-b112-f74dab2d1e48 --- .../codex-app-server/thread-state.test.ts | 20 ++++++------------- .../adapters/codex-app-server/thread-state.ts | 2 +- .../agent/src/server/agent-server.test.ts | 11 +++++----- 3 files changed, 12 insertions(+), 21 deletions(-) diff --git a/packages/agent/src/adapters/codex-app-server/thread-state.test.ts b/packages/agent/src/adapters/codex-app-server/thread-state.test.ts index 77bf625be5..ed56d8a965 100644 --- a/packages/agent/src/adapters/codex-app-server/thread-state.test.ts +++ b/packages/agent/src/adapters/codex-app-server/thread-state.test.ts @@ -28,27 +28,19 @@ describe("hasCodexThreadState", () => { ); }; - it("finds a persisted rollout for the thread id", async () => { + 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(THREAD_ID)).resolves.toBe(true); - }); - - it("returns false when only a different thread's rollout exists", async () => { - await writeRollout(THREAD_ID); - await expect( - hasCodexThreadState("11111111-2222-3333-4444-555555555555"), - ).resolves.toBe(false); + 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("returns false for an empty thread id", async () => { - await writeRollout(THREAD_ID); - await expect(hasCodexThreadState("")).resolves.toBe(false); - }); - it("ignores files that are not rollouts", async () => { const dir = join(codexHome, "sessions", "2026", "07", "07"); await mkdir(dir, { recursive: true }); diff --git a/packages/agent/src/adapters/codex-app-server/thread-state.ts b/packages/agent/src/adapters/codex-app-server/thread-state.ts index 0963405aca..6fe640c0a3 100644 --- a/packages/agent/src/adapters/codex-app-server/thread-state.ts +++ b/packages/agent/src/adapters/codex-app-server/thread-state.ts @@ -16,7 +16,7 @@ export async function hasCodexThreadState(threadId: string): Promise { try { const entries = await readdir(sessionsDir, { recursive: true }); return entries.some((entry) => { - const name = path.basename(entry.toString()); + const name = path.basename(entry); return name.startsWith("rollout-") && name.endsWith(suffix); }); } catch { diff --git a/packages/agent/src/server/agent-server.test.ts b/packages/agent/src/server/agent-server.test.ts index d8ed1252a7..28d8ee9e5d 100644 --- a/packages/agent/src/server/agent-server.test.ts +++ b/packages/agent/src/server/agent-server.test.ts @@ -1830,12 +1830,11 @@ describe("AgentServer HTTP Mode", () => { }); }); - it("falls back to summary resume when the thread state is gone", async () => { - await expect(prepare(codexServer(THREAD_ID))).resolves.toBeNull(); - }); - - it("falls back to summary resume without a prior session id", async () => { - await expect(prepare(codexServer(null))).resolves.toBeNull(); + 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(); }); }); });