Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
50 changes: 50 additions & 0 deletions packages/agent/src/adapters/codex-app-server/thread-state.test.ts
Original file line number Diff line number Diff line change
@@ -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);
});
});
25 changes: 25 additions & 0 deletions packages/agent/src/adapters/codex-app-server/thread-state.ts
Original file line number Diff line number Diff line change
@@ -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);
});
Comment thread
k11kirky marked this conversation as resolved.
} catch {
return false;
}
}
73 changes: 72 additions & 1 deletion packages/agent/src/server/agent-server.test.ts
Original file line number Diff line number Diff line change
@@ -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";
Expand Down Expand Up @@ -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", () => {
Expand Down
59 changes: 42 additions & 17 deletions packages/agent/src/server/agent-server.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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;

Expand All @@ -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));
Expand Down Expand Up @@ -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 ?? [],
Expand Down
Loading