Skip to content

Commit b895c86

Browse files
authored
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
1 parent d9ec3c8 commit b895c86

4 files changed

Lines changed: 198 additions & 18 deletions

File tree

Lines changed: 58 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,58 @@
1+
import { mkdir, mkdtemp, rm, writeFile } from "node:fs/promises";
2+
import { tmpdir } from "node:os";
3+
import { join } from "node:path";
4+
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
5+
import { hasCodexThreadState } from "./thread-state";
6+
7+
const THREAD_ID = "0199a5c3-2f60-7b21-9c39-1d2e3f4a5b6c";
8+
9+
describe("hasCodexThreadState", () => {
10+
let codexHome: string;
11+
12+
beforeEach(async () => {
13+
codexHome = await mkdtemp(join(tmpdir(), "codex-home-"));
14+
vi.stubEnv("CODEX_HOME", codexHome);
15+
});
16+
17+
afterEach(async () => {
18+
vi.unstubAllEnvs();
19+
await rm(codexHome, { recursive: true, force: true });
20+
});
21+
22+
const writeRollout = async (threadId: string) => {
23+
const dir = join(codexHome, "sessions", "2026", "07", "07");
24+
await mkdir(dir, { recursive: true });
25+
await writeFile(
26+
join(dir, `rollout-2026-07-07T10-00-00-${threadId}.jsonl`),
27+
"",
28+
);
29+
};
30+
31+
it("finds a persisted rollout for the thread id", async () => {
32+
await writeRollout(THREAD_ID);
33+
await expect(hasCodexThreadState(THREAD_ID)).resolves.toBe(true);
34+
});
35+
36+
it("returns false when only a different thread's rollout exists", async () => {
37+
await writeRollout(THREAD_ID);
38+
await expect(
39+
hasCodexThreadState("11111111-2222-3333-4444-555555555555"),
40+
).resolves.toBe(false);
41+
});
42+
43+
it("returns false when there is no sessions directory", async () => {
44+
await expect(hasCodexThreadState(THREAD_ID)).resolves.toBe(false);
45+
});
46+
47+
it("returns false for an empty thread id", async () => {
48+
await writeRollout(THREAD_ID);
49+
await expect(hasCodexThreadState("")).resolves.toBe(false);
50+
});
51+
52+
it("ignores files that are not rollouts", async () => {
53+
const dir = join(codexHome, "sessions", "2026", "07", "07");
54+
await mkdir(dir, { recursive: true });
55+
await writeFile(join(dir, `notes-${THREAD_ID}.jsonl`), "");
56+
await expect(hasCodexThreadState(THREAD_ID)).resolves.toBe(false);
57+
});
58+
});
Lines changed: 25 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,25 @@
1+
import { readdir } from "node:fs/promises";
2+
import os from "node:os";
3+
import path from "node:path";
4+
5+
/**
6+
* Whether codex still holds the persisted rollout for `threadId` — written as
7+
* `CODEX_HOME/sessions/YYYY/MM/DD/rollout-<timestamp>-<threadId>.jsonl` — i.e.
8+
* whether `thread/resume` can restore the thread natively. Mirrors codex's own
9+
* CODEX_HOME resolution (env override, else ~/.codex).
10+
*/
11+
export async function hasCodexThreadState(threadId: string): Promise<boolean> {
12+
if (!threadId) return false;
13+
const codexHome = process.env.CODEX_HOME || path.join(os.homedir(), ".codex");
14+
const sessionsDir = path.join(codexHome, "sessions");
15+
const suffix = `-${threadId}.jsonl`;
16+
try {
17+
const entries = await readdir(sessionsDir, { recursive: true });
18+
return entries.some((entry) => {
19+
const name = path.basename(entry.toString());
20+
return name.startsWith("rollout-") && name.endsWith(suffix);
21+
});
22+
} catch {
23+
return false;
24+
}
25+
}

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

Lines changed: 73 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
import { createHash } from "node:crypto";
2-
import { mkdtemp, readFile, rm } from "node:fs/promises";
2+
import { mkdir, mkdtemp, readFile, rm, writeFile } from "node:fs/promises";
33
import { tmpdir } from "node:os";
44
import { join } from "node:path";
55
import type { ContentBlock } from "@agentclientprotocol/sdk";
@@ -1766,6 +1766,78 @@ describe("AgentServer HTTP Mode", () => {
17661766
}
17671767
}
17681768
});
1769+
1770+
describe("codex", () => {
1771+
const THREAD_ID = "0199a5c3-2f60-7b21-9c39-1d2e3f4a5b6c";
1772+
let codexHome: string;
1773+
1774+
const payload: JwtPayload = {
1775+
task_id: "test-task-id",
1776+
run_id: "test-run-id",
1777+
team_id: 1,
1778+
user_id: 1,
1779+
distinct_id: "test-distinct-id",
1780+
mode: "interactive",
1781+
};
1782+
1783+
const codexServer = (sessionId: string | null) => {
1784+
const s = createServer() as unknown as NativeResumeTestServer;
1785+
s.resumeState = {
1786+
conversation: [
1787+
{ role: "user", content: [{ type: "text", text: "continue" }] },
1788+
],
1789+
latestGitCheckpoint: null,
1790+
interrupted: false,
1791+
logEntryCount: 1,
1792+
sessionId,
1793+
};
1794+
return s;
1795+
};
1796+
1797+
const prepare = (s: NativeResumeTestServer) =>
1798+
s.prepareNativeResume(
1799+
payload,
1800+
createMockApiClient(),
1801+
createTaskRun({
1802+
id: "test-run-id",
1803+
state: { resume_from_run_id: "previous-run" },
1804+
}),
1805+
"codex",
1806+
repo.path,
1807+
"auto",
1808+
);
1809+
1810+
beforeEach(() => {
1811+
codexHome = join(repo.path, ".codex-test");
1812+
vi.stubEnv("CODEX_HOME", codexHome);
1813+
});
1814+
1815+
afterEach(() => {
1816+
vi.unstubAllEnvs();
1817+
});
1818+
1819+
it("resumes natively when the thread rollout survived in CODEX_HOME", async () => {
1820+
const dir = join(codexHome, "sessions", "2026", "07", "07");
1821+
await mkdir(dir, { recursive: true });
1822+
await writeFile(
1823+
join(dir, `rollout-2026-07-07T10-00-00-${THREAD_ID}.jsonl`),
1824+
"",
1825+
);
1826+
1827+
await expect(prepare(codexServer(THREAD_ID))).resolves.toEqual({
1828+
sessionId: THREAD_ID,
1829+
warm: true,
1830+
});
1831+
});
1832+
1833+
it("falls back to summary resume when the thread state is gone", async () => {
1834+
await expect(prepare(codexServer(THREAD_ID))).resolves.toBeNull();
1835+
});
1836+
1837+
it("falls back to summary resume without a prior session id", async () => {
1838+
await expect(prepare(codexServer(null))).resolves.toBeNull();
1839+
});
1840+
});
17691841
});
17701842

17711843
describe("PR attribution", () => {

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

Lines changed: 42 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,7 @@ import {
3030
hydrateSessionJsonl,
3131
} from "../adapters/claude/session/jsonl-hydration";
3232
import type { GatewayEnv } from "../adapters/claude/session/options";
33+
import { hasCodexThreadState } from "../adapters/codex-app-server/thread-state";
3334
import {
3435
type AgentErrorClassification,
3536
classifyAgentError,
@@ -672,8 +673,6 @@ export class AgentServer {
672673
cwd: string,
673674
permissionMode: PermissionMode,
674675
): Promise<{ sessionId: string; warm: boolean } | null> {
675-
if (runtimeAdapter !== "claude") return null;
676-
677676
const resumeRunId = this.getResumeRunId(preTaskRun);
678677
if (!resumeRunId) return null;
679678

@@ -689,6 +688,22 @@ export class AgentServer {
689688
return null;
690689
}
691690

691+
if (runtimeAdapter === "codex") {
692+
// Codex owns thread persistence in CODEX_HOME (the ACP sessionId is the
693+
// codex thread id). The rollout only survives a snapshot restart — there
694+
// is no cold hydration equivalent, so a fresh sandbox keeps the summary
695+
// fallback while a warm one resumes the thread natively via thread/resume.
696+
if (!(await hasCodexThreadState(priorSessionId))) {
697+
this.logger.debug(
698+
"No codex thread state on disk; using summary resume fallback",
699+
{ resumeRunId, priorSessionId },
700+
);
701+
return null;
702+
}
703+
this.logger.debug("Native codex resume prepared", { priorSessionId });
704+
return { sessionId: priorSessionId, warm: true };
705+
}
706+
692707
let warm = false;
693708
try {
694709
await access(getSessionJsonlPath(priorSessionId, cwd));
@@ -1309,22 +1324,32 @@ export class AgentServer {
13091324
initialPermissionMode,
13101325
);
13111326

1312-
let acpSessionId: string;
1327+
let acpSessionId: string | null = null;
13131328
if (nativeResume) {
1314-
await clientConnection.resumeSession({
1315-
sessionId: nativeResume.sessionId,
1316-
cwd: sessionCwd,
1317-
mcpServers: this.config.mcpServers ?? [],
1318-
_meta: { ...sessionMeta, sessionId: nativeResume.sessionId },
1319-
});
1320-
acpSessionId = nativeResume.sessionId;
1321-
this.nativeResume = nativeResume;
1322-
this.logger.debug("ACP session resumed", {
1323-
acpSessionId,
1324-
runId: payload.run_id,
1325-
warm: nativeResume.warm,
1326-
});
1327-
} else {
1329+
try {
1330+
await clientConnection.resumeSession({
1331+
sessionId: nativeResume.sessionId,
1332+
cwd: sessionCwd,
1333+
mcpServers: this.config.mcpServers ?? [],
1334+
_meta: { ...sessionMeta, sessionId: nativeResume.sessionId },
1335+
});
1336+
acpSessionId = nativeResume.sessionId;
1337+
this.nativeResume = nativeResume;
1338+
this.logger.debug("ACP session resumed", {
1339+
acpSessionId,
1340+
runId: payload.run_id,
1341+
warm: nativeResume.warm,
1342+
});
1343+
} catch (error) {
1344+
// resumeState is still loaded, so the summary resume path takes over
1345+
// on the fresh session below.
1346+
this.logger.warn("Native resume failed; starting a fresh session", {
1347+
sessionId: nativeResume.sessionId,
1348+
error: error instanceof Error ? error.message : String(error),
1349+
});
1350+
}
1351+
}
1352+
if (!acpSessionId) {
13281353
const sessionResponse = await clientConnection.newSession({
13291354
cwd: sessionCwd,
13301355
mcpServers: this.config.mcpServers ?? [],

0 commit comments

Comments
 (0)