From c663630bbd78fe629c04439063d9b7f8c5f44b7e Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Fri, 30 Oct 2026 12:57:21 +0700 Subject: [PATCH] fix(server): read Codex history in pages instead of one whole-thread reply (#8171) Opening a Codex agent whose timeline the daemon doesn't hold asked Codex for the whole thread in one thread/read reply. Past the 128 MiB provider line cap that reply killed the session; before the cap it crashed the daemon. Codex 0.145 and newer reject that read on paginated threads, so importing them failed. Paginated threads now load through thread/items/list pages. A legacy thread keeps its single read. A reply over the line cap fails only its own request: for a legacy read, a warning row names the archive, `codex migrate-rollouts`, unarchive steps; for a page, a warning row ends the history there. Items are capped as they are projected, and rewind's fork no longer returns the thread's turns. Co-authored-by: hunterricks Co-authored-by: dundunHa --- .../reproduce-codex-history-string-limit.mts | 108 +++++++++ .../providers/codex-app-server-agent.test.ts | 23 +- .../agent/providers/codex-app-server-agent.ts | 162 ++++++------- ...odex-history-string-limit.real.e2e.test.ts | 53 +++++ .../codex/app-server-transport.test.ts | 31 ++- .../providers/codex/app-server-transport.ts | 32 ++- .../agent/providers/codex/rewind.test.ts | 4 +- .../server/agent/providers/codex/rewind.ts | 24 +- .../codex/test-utils/fake-app-server.ts | 104 ++++++++- .../providers/codex/thread-history.test.ts | 220 ++++++++++++++++++ .../agent/providers/codex/thread-history.ts | 168 +++++++++++++ .../agent/providers/jsonl-frame-decoder.ts | 47 +++- 12 files changed, 839 insertions(+), 137 deletions(-) create mode 100644 packages/server/scripts/reproduce-codex-history-string-limit.mts create mode 100644 packages/server/src/server/agent/providers/codex-history-string-limit.real.e2e.test.ts create mode 100644 packages/server/src/server/agent/providers/codex/thread-history.test.ts create mode 100644 packages/server/src/server/agent/providers/codex/thread-history.ts diff --git a/packages/server/scripts/reproduce-codex-history-string-limit.mts b/packages/server/scripts/reproduce-codex-history-string-limit.mts new file mode 100644 index 0000000000..b0f49e4faf --- /dev/null +++ b/packages/server/scripts/reproduce-codex-history-string-limit.mts @@ -0,0 +1,108 @@ +// Opt-in diagnostic for #3905. Run from the repository root: +// node --import tsx packages/server/scripts/reproduce-codex-history-string-limit.mts 520 [legacy|paginated] +// Requires Codex 0.148+ on PATH and approximately 2 GiB of available memory. +// `paginated` migrates the synthetic rollout with `codex migrate-rollouts` first. +import { execFileSync, spawn } from "node:child_process"; +import { randomUUID } from "node:crypto"; +import { mkdtemp, mkdir, open, rm } from "node:fs/promises"; +import path from "node:path"; +import pino from "pino"; +import { CodexAppServerAgentSession } from "../src/server/agent/providers/codex-app-server-agent.js"; + +const megabytes = Number(process.argv[2] ?? "520"); +if (!Number.isSafeInteger(megabytes) || megabytes < 1 || megabytes > 600) { + throw new Error("Expected an integer size between 1 and 600 MiB"); +} +const historyMode = process.argv[3] ?? "legacy"; +if (historyMode !== "legacy" && historyMode !== "paginated") { + throw new Error("Expected history mode legacy or paginated"); +} +const root = await mkdtemp(path.join(process.cwd(), ".codex-history-repro-")); +const home = path.join(root, "codex-home"); +const now = new Date(); +const timestamp = now.toISOString(); +const threadId = randomUUID(); +const turnId = randomUUID(); +const sessions = path.join(home, "sessions", ...timestamp.slice(0, 10).split("-")); +await mkdir(sessions, { recursive: true }); +const rollout = await open( + path.join(sessions, `rollout-${timestamp.slice(0, 19).replaceAll(":", "-")}-${threadId}.jsonl`), + "w", +); +async function write(type: string, payload: unknown) { + await rollout.write(`${JSON.stringify({ timestamp, type, payload })}\n`); +} +await write("session_meta", { + id: threadId, + timestamp, + cwd: root, + originator: "codex_cli_rs", + cli_version: "0.159.0", + source: "cli", + model_provider: "openai", + history_mode: "legacy", +}); +await write("event_msg", { type: "task_started", turn_id: turnId, model_context_window: 200000 }); +await write("event_msg", { + type: "user_message", + message: "Synthetic oversized history", + images: [], + local_images: [], + text_elements: [], +}); +for (let index = 0; index < megabytes; index++) { + await write("event_msg", { + type: "agent_message", + message: "x".repeat(1024 * 1024), + phase: "commentary", + }); +} +await write("event_msg", { type: "task_complete", turn_id: turnId, last_agent_message: "done" }); +await rollout.close(); +if (historyMode === "paginated") { + execFileSync("codex", ["migrate-rollouts", "--apply", "--thread", threadId], { + env: { ...process.env, CODEX_HOME: home }, + stdio: "ignore", + }); +} +console.log( + `Opening synthetic ${megabytes} MiB ${historyMode} history through the real Codex app-server`, +); +const session = new CodexAppServerAgentSession( + { provider: "codex", cwd: root }, + { sessionId: threadId }, + pino({ level: "silent" }), + async () => + spawn( + "codex", + ["app-server", "--disable", "plugins", "--disable", "apps", "--disable", "remote_plugin"], + { + stdio: ["pipe", "pipe", "pipe"], + env: { ...process.env, CODEX_HOME: home }, + }, + ), + { codexHome: home }, + false, + false, + false, + undefined, + "history", +); +try { + await session.connect(); + let count = 0; + for await (const event of session.streamHistory()) { + if (event.type !== "timeline") continue; + count++; + if (event.item.type === "notification") console.log(event.item.message); + } + console.log(`History loaded: ${count} timeline rows`); +} catch (error) { + // A recoverable session error is acceptable for the crash-isolation assertion. + console.log( + `History rejected without crashing: ${error instanceof Error ? error.message : String(error)}`, + ); +} finally { + await session.close(); + await rm(root, { recursive: true, force: true, maxRetries: 10, retryDelay: 100 }); +} diff --git a/packages/server/src/server/agent/providers/codex-app-server-agent.test.ts b/packages/server/src/server/agent/providers/codex-app-server-agent.test.ts index e709fd490c..cb8cb2684b 100644 --- a/packages/server/src/server/agent/providers/codex-app-server-agent.test.ts +++ b/packages/server/src/server/agent/providers/codex-app-server-agent.test.ts @@ -88,6 +88,7 @@ describe("Codex executable discovery", () => { import { CodexAppServerClient } from "./codex/app-server-transport.js"; import { createFakeCodexAppServer, + fakeCodexThreadHistory, type FakeCodexAppServer, waitForNextPermission, waitForNextTimelineItem, @@ -1991,7 +1992,7 @@ describe("Codex app-server provider", () => { purpose: "history", }); - expect(threadRequests).toEqual(["thread/read"]); + expect(threadRequests).toEqual(["thread/read", "thread/read"]); await session.close(); appServer.assertNoErrors(); }); @@ -2068,6 +2069,7 @@ describe("Codex app-server provider", () => { "thread/loaded/list", "thread/resume", "thread/read", + "thread/read", ]); expect(codex.threadMethods(1)).toEqual(["thread/loaded/list", "thread/resume"]); expect(codex.manager.getTimeline(codex.agentId)).toEqual(projected); @@ -2427,7 +2429,7 @@ describe("Codex app-server provider", () => { cwd: "/workspace/project", model: "gpt-5.4", serviceTier: null, - excludeTurns: false, + excludeTurns: true, persistExtendedHistory: true, }, ]); @@ -2441,8 +2443,9 @@ describe("Codex app-server provider", () => { test("rewinds a paginated conversation through the public session capability", async () => { const appServer = createFakeCodexAppServer({ - "thread/read": () => ({ - thread: { id: "thread-1", historyMode: "paginated", turns: [] }, + ...fakeCodexThreadHistory({ + "thread-1": { historyMode: "paginated", turns: [] }, + "forked-thread": { historyMode: "paginated", turns: [] }, }), "thread/rollback": () => { throw new Error("paginated threads do not support thread/rollback"); @@ -2487,7 +2490,7 @@ describe("Codex app-server provider", () => { cwd: "/workspace/project", model: "gpt-5.4", serviceTier: null, - excludeTurns: false, + excludeTurns: true, persistExtendedHistory: true, }, ]); @@ -2504,11 +2507,10 @@ describe("Codex app-server provider", () => { async (historyMode) => { const appServer = createFakeCodexAppServer( historyMode === "paginated" - ? { - "thread/read": () => ({ - thread: { id: "thread-1", historyMode: "paginated", turns: [] }, - }), - } + ? fakeCodexThreadHistory({ + "thread-1": { historyMode: "paginated", turns: [] }, + "forked-thread": { historyMode: "paginated", turns: [] }, + }) : undefined, ); const customCodexConfig = { @@ -4947,6 +4949,7 @@ describe("Codex app-server provider", () => { } expect(requests.map((request) => [request.method, request.params])).toEqual([ + ["thread/read", { threadId: "test-thread", includeTurns: false }], ["thread/read", { threadId: "test-thread", includeTurns: true }], ]); expect(history).toEqual([ diff --git a/packages/server/src/server/agent/providers/codex-app-server-agent.ts b/packages/server/src/server/agent/providers/codex-app-server-agent.ts index 5e43f582c5..f7d4f50215 100644 --- a/packages/server/src/server/agent/providers/codex-app-server-agent.ts +++ b/packages/server/src/server/agent/providers/codex-app-server-agent.ts @@ -92,12 +92,18 @@ import { type CodexAppServerTraceContext, } from "./codex/app-server-transport.js"; import { type CodexUserMessageTurnIndex, revertCodexConversation } from "./codex/rewind.js"; +import { + type CodexHistoryRow, + CodexThreadHistoryTooLargeError, + readCodexThreadHistory, +} from "./codex/thread-history.js"; import { materializeProviderImage, renderProviderImageOutputAsAssistantMarkdown, type ProviderImageOutput, } from "./provider-image-output.js"; import { normalizeProviderReplayTimestamp } from "../provider-history-timestamps.js"; +import { limitAgentTimelineItemContent } from "../agent-timeline-content.js"; import { formatProviderDiagnostic, formatProviderDiagnosticError, @@ -1804,22 +1810,13 @@ function readCodexHistoryTimestamp(item: unknown): string | null { ); } -function readCodexTurnHistoryTimestamp( - turn: unknown, +/** Falls back to the row's start time for a user message and its end time otherwise. */ +function readCodexHistoryRowTimestamp( + row: CodexHistoryRow, timelineItem: AgentTimelineItem, ): string | null { - const record = toObjectRecord(turn); - if (!record) { - return null; - } - - const startedAt = - normalizeProviderReplayTimestamp(record.startedAt) ?? - normalizeProviderReplayTimestamp(record.started_at); - const completedAt = - normalizeProviderReplayTimestamp(record.completedAt) ?? - normalizeProviderReplayTimestamp(record.completed_at); - + const startedAt = normalizeProviderReplayTimestamp(row.startedAt); + const completedAt = normalizeProviderReplayTimestamp(row.completedAt); if (timelineItem.type === "user_message") { return startedAt ?? completedAt; } @@ -2063,84 +2060,19 @@ async function threadItemToTimelineEntries( return [timelineItem, ...(await mcpToolResultImagesToTimeline(item))]; } -const CodexThreadReadResponseSchema = z - .object({ - thread: z - .object({ - turns: z - .array( - z - .object({ - items: z.array(z.unknown()).default([]), - }) - .passthrough(), - ) - .default([]), - }) - .passthrough() - .default({ turns: [] }), - }) - .passthrough(); - -type CodexThreadReadResponse = z.infer; -type CodexThreadReadRequest = (threadId: string) => Promise; - -async function requestCodexThreadHistory( - requestThread: CodexThreadReadRequest, - threadId: string, -): Promise { - const response = await requestThread(threadId); - return CodexThreadReadResponseSchema.parse(response); -} - async function loadCodexThreadHistoryTimeline(params: { + client: CodexAppServerClientLike; threadId: string; cwd: string | null; - requestThread: CodexThreadReadRequest; }): Promise { - const response = await requestCodexThreadHistory(params.requestThread, params.threadId); const timeline: PersistedTimelineEntry[] = []; const subAgentTimelineIndexByThreadId = new Map(); - for (const turn of response.thread.turns) { - for (const item of turn.items) { - const historicalSubAgentActivity = readCodexSubAgentActivity(item); - if (historicalSubAgentActivity) { - const existingIndex = subAgentTimelineIndexByThreadId.get( - historicalSubAgentActivity.agentThreadId, - ); - if (existingIndex !== undefined) { - const activityTimelineItem = await threadItemToTimeline(item, { cwd: params.cwd }); - updateHistoricalSubAgentActivity( - timeline, - existingIndex, - historicalSubAgentActivity.kind, - activityTimelineItem?.type === "tool_call" && - activityTimelineItem.detail.type === "sub_agent" - ? activityTimelineItem.detail.subAgentType - : undefined, - ); - continue; - } - } - for (const timelineItem of await threadItemToTimelineEntries(item, { cwd: params.cwd })) { - const timestamp = - readCodexHistoryTimestamp(item) ?? readCodexTurnHistoryTimestamp(turn, timelineItem); - const settledTimelineItem = - historicalSubAgentActivity && timelineItem.type === "tool_call" - ? settleHistoricalSubAgentActivity(timelineItem, historicalSubAgentActivity.kind) - : timelineItem; - timeline.push({ - item: settledTimelineItem, - timestamp: timestamp ?? undefined, - ...(timelineItem.type === "user_message" && typeof turn.id === "string" - ? { providerTurnId: turn.id } - : {}), - }); - for (const childThreadId of readCodexHistoricalSubAgentThreadIds(item)) { - subAgentTimelineIndexByThreadId.set(childThreadId, timeline.length - 1); - } - } - } + try { + await projectCodexThreadHistory(params, { timeline, subAgentTimelineIndexByThreadId }); + } catch (error) { + if (!(error instanceof CodexThreadHistoryTooLargeError)) throw error; + // The session stays usable; the row says why the history stops here. + timeline.push({ item: { type: "notification", level: "warning", message: error.message } }); } const subAgentRoutes = Array.from(subAgentTimelineIndexByThreadId.entries()).flatMap( ([childThreadId, timelineIndex]): PersistedSubAgentRoute[] => { @@ -2153,11 +2085,55 @@ async function loadCodexThreadHistoryTimeline(params: { return { timeline, subAgentRoutes }; } -function readCodexThread(client: CodexAppServerClientLike, threadId: string): Promise { - return client.request("thread/read", { - threadId, - includeTurns: true, - }); +async function projectCodexThreadHistory( + params: { client: CodexAppServerClientLike; threadId: string; cwd: string | null }, + projection: { + timeline: PersistedTimelineEntry[]; + subAgentTimelineIndexByThreadId: Map; + }, +): Promise { + const { timeline, subAgentTimelineIndexByThreadId } = projection; + for await (const row of readCodexThreadHistory(params.client, params.threadId)) { + const { item } = row; + const historicalSubAgentActivity = readCodexSubAgentActivity(item); + if (historicalSubAgentActivity) { + const existingIndex = subAgentTimelineIndexByThreadId.get( + historicalSubAgentActivity.agentThreadId, + ); + if (existingIndex !== undefined) { + const activityTimelineItem = await threadItemToTimeline(item, { cwd: params.cwd }); + updateHistoricalSubAgentActivity( + timeline, + existingIndex, + historicalSubAgentActivity.kind, + activityTimelineItem?.type === "tool_call" && + activityTimelineItem.detail.type === "sub_agent" + ? activityTimelineItem.detail.subAgentType + : undefined, + ); + continue; + } + } + for (const timelineItem of await threadItemToTimelineEntries(item, { cwd: params.cwd })) { + const timestamp = + readCodexHistoryTimestamp(item) ?? readCodexHistoryRowTimestamp(row, timelineItem); + const settledTimelineItem = + historicalSubAgentActivity && timelineItem.type === "tool_call" + ? settleHistoricalSubAgentActivity(timelineItem, historicalSubAgentActivity.kind) + : timelineItem; + timeline.push({ + // Cap each item as it is projected so a large history is never held uncapped. + item: limitAgentTimelineItemContent(settledTimelineItem), + timestamp: timestamp ?? undefined, + ...(timelineItem.type === "user_message" && row.turnId !== undefined + ? { providerTurnId: row.turnId } + : {}), + }); + for (const childThreadId of readCodexHistoricalSubAgentThreadIds(item)) { + subAgentTimelineIndexByThreadId.set(childThreadId, timeline.length - 1); + } + } + } } export async function forkCodexThread( @@ -3997,9 +3973,9 @@ export class CodexAppServerAgentSession implements AgentSession { return { timeline: [], providerSubagentEvents: [] }; } const { timeline, subAgentRoutes } = await loadCodexThreadHistoryTimeline({ + client, threadId: this.currentThreadId, cwd: this.config.cwd ?? null, - requestThread: (threadIdToRead) => readCodexThread(client, threadIdToRead), }); this.subAgentCallsByCallId.clear(); this.subAgentCallIdByChildThreadId.clear(); @@ -4055,9 +4031,9 @@ export class CodexAppServerAgentSession implements AgentSession { }); try { const childHistory = await loadCodexThreadHistoryTimeline({ + client, threadId: next.route.childThreadId, cwd: this.config.cwd ?? null, - requestThread: (childThreadId) => readCodexThread(client, childThreadId), }); for (const entry of childHistory.timeline) { this.emitProviderSubagentTimeline(next.route.childThreadId, entry.item, entry.timestamp); diff --git a/packages/server/src/server/agent/providers/codex-history-string-limit.real.e2e.test.ts b/packages/server/src/server/agent/providers/codex-history-string-limit.real.e2e.test.ts new file mode 100644 index 0000000000..fba51b9e1d --- /dev/null +++ b/packages/server/src/server/agent/providers/codex-history-string-limit.real.e2e.test.ts @@ -0,0 +1,53 @@ +import { execFile } from "node:child_process"; +import { mkdtemp, rm } from "node:fs/promises"; +import path from "node:path"; +import { fileURLToPath } from "node:url"; +import { expect, test } from "vitest"; + +// Regression for #3905. Real Codex, synthetic on-disk history, no model turns. +// Opt in because the oversized cases need approximately 2 GiB of memory and over +// 1 GiB of temporary storage. +const enabled = process.env.PASEO_CODEX_HISTORY_STRING_LIMIT_REPRO === "1"; +const script = fileURLToPath( + new URL("../../../../scripts/reproduce-codex-history-string-limit.mts", import.meta.url), +); + +test.runIf(enabled).each([ + { megabytes: 2, historyMode: "legacy", expected: "History loaded: 3 timeline rows" }, + { megabytes: 520, historyMode: "legacy", expected: "codex migrate-rollouts --apply --thread" }, + { megabytes: 520, historyMode: "paginated", expected: "History loaded: 521 timeline rows" }, +])( + "opening $megabytes MiB of $historyMode Codex history does not crash its hosting process", + async ({ megabytes, historyMode, expected }) => { + const root = await mkdtemp(path.join(process.cwd(), ".codex-history-test-")); + try { + const result = await new Promise<{ code: number; stdout: string; stderr: string }>( + (resolve) => { + const child = execFile( + process.execPath, + ["--import", "tsx", script, String(megabytes), historyMode], + { + cwd: root, + detached: true, + timeout: 120_000, + maxBuffer: 1024 * 1024, + }, + (error, stdout, stderr) => { + try { + if (child.pid) process.kill(-child.pid, "SIGKILL"); + } catch (killError) { + if ((killError as NodeJS.ErrnoException).code !== "ESRCH") throw killError; + } + resolve({ code: error ? Number(error.code ?? 1) : 0, stdout, stderr }); + }, + ); + }, + ); + expect(result.code, result.stderr).toBe(0); + expect(result.stdout).toContain(expected); + } finally { + await rm(root, { recursive: true, force: true, maxRetries: 10, retryDelay: 100 }); + } + }, + 180_000, +); diff --git a/packages/server/src/server/agent/providers/codex/app-server-transport.test.ts b/packages/server/src/server/agent/providers/codex/app-server-transport.test.ts index 3018461ba9..84b82b431b 100644 --- a/packages/server/src/server/agent/providers/codex/app-server-transport.test.ts +++ b/packages/server/src/server/agent/providers/codex/app-server-transport.test.ts @@ -185,14 +185,41 @@ describe("Codex app-server transport", () => { }); describe("Codex stdout byte admission", () => { - test("fails pending requests and the session before joining an oversized line", async () => { + test("an oversized response fails only its own request", async () => { + const child = createCodexAppServerChildProcess(); + const client = new CodexAppServerClient(child, createTestLogger()); + const terminations: Error[] = []; + client.setUnexpectedTerminationHandler((error) => terminations.push(error)); + const oversized = client.request("thread/items/list", {}).catch((error: Error) => error); + const other = client.request("model/list", {}); + const chunk = Buffer.alloc(64 * 1024, "x"); + child.stdout.write('{"id":1,"result":"'); + for (let index = 0; index < 2049; index++) child.stdout.write(chunk); + child.stdout.write('"}\n{"id":2,"result":{"data":[]}}\n'); + + expect(await oversized).toMatchObject({ + name: "CodexAppServerResponseTooLargeError", + method: "thread/items/list", + limitBytes: 128 * 1024 * 1024, + }); + await expect(other).resolves.toEqual({ data: [] }); + const next = client.request("model/list", {}); + child.stdout.write('{"id":3,"result":{"data":["next"]}}\n'); + await expect(next).resolves.toEqual({ data: ["next"] }); + expect(terminations).toEqual([]); + await client.dispose(); + child.stdout.end(); + child.stderr.end(); + }); + + test("an oversized notification fails pending requests and the session", async () => { const child = createCodexAppServerChildProcess(); const client = new CodexAppServerClient(child, createTestLogger()); const terminations: Error[] = []; client.setUnexpectedTerminationHandler((error) => terminations.push(error)); const request = client.request("thread/resume", {}, 1000).catch((error: Error) => error); const chunk = Buffer.alloc(64 * 1024, "x"); - child.stdout.write('{"id":1,"result":"'); + child.stdout.write('{"method":"item/completed","params":"'); for (let index = 0; index < 2048; index++) child.stdout.write(chunk); const error = await request; expect(error).toMatchObject({ diff --git a/packages/server/src/server/agent/providers/codex/app-server-transport.ts b/packages/server/src/server/agent/providers/codex/app-server-transport.ts index dfd4adfac4..5a218e2edb 100644 --- a/packages/server/src/server/agent/providers/codex/app-server-transport.ts +++ b/packages/server/src/server/agent/providers/codex/app-server-transport.ts @@ -33,11 +33,26 @@ interface JsonRpcNotification { } interface PendingRequest { + method: string; resolve: (value: unknown) => void; reject: (error: Error) => void; timer: NodeJS.Timeout; } +/** A response past the provider line cap. Only its request fails; the session stays up. */ +export class CodexAppServerResponseTooLargeError extends Error { + constructor( + readonly method: string, + readonly limitBytes: number, + ) { + super(`Codex app-server response to ${method} exceeds ${limitBytes} bytes`); + this.name = "CodexAppServerResponseTooLargeError"; + } +} + +// Codex serializes a response with `id` first, so its head names the request. +const OVERSIZED_RESPONSE_HEAD = /^\{"id":(\d+),"result":/; + export class CodexAppServerRpcError extends Error { constructor( message: string, @@ -258,7 +273,7 @@ export class CodexAppServerClient { this.pending.delete(id); reject(new Error(`Codex app-server request timed out for ${method}`)); }, timeoutMs); - this.pending.set(id, { resolve, reject, timer }); + this.pending.set(id, { method, resolve, reject, timer }); }); } @@ -304,8 +319,21 @@ export class CodexAppServerClient { } private handleStreamLimit(failure: ProviderStreamLimitError): void { - // A rejected response/notification cannot be replayed safely. Fail the session, + const responseId = OVERSIZED_RESPONSE_HEAD.exec(failure.lineHead)?.[1]; + const pending = responseId === undefined ? undefined : this.pending.get(Number(responseId)); + if (pending) { + clearTimeout(pending.timer); + this.pending.delete(Number(responseId)); + this.logger.warn( + { method: pending.method, limitBytes: failure.limitBytes }, + "Codex app-server response exceeded the line cap", + ); + pending.reject(new CodexAppServerResponseTooLargeError(pending.method, failure.limitBytes)); + return; + } + // A lost notification or server request cannot be replayed. Fail the session, // including live turns with no pending RPC, rather than dropping a stream unit. + this.stdoutLineReader.close(); this.handleUnexpectedTermination(failure); void this.dispose().catch((err) => { this.logger.warn({ err }, "Failed to terminate oversized Codex stdout producer"); diff --git a/packages/server/src/server/agent/providers/codex/rewind.test.ts b/packages/server/src/server/agent/providers/codex/rewind.test.ts index da7fe6a8f5..7a0b2dc0f9 100644 --- a/packages/server/src/server/agent/providers/codex/rewind.test.ts +++ b/packages/server/src/server/agent/providers/codex/rewind.test.ts @@ -108,7 +108,7 @@ describe("Codex Rewind", () => { cwd: "/workspace/project", model: "gpt-5.4-mini", serviceTier: null, - excludeTurns: false, + excludeTurns: true, persistExtendedHistory: true, }, ]); @@ -190,7 +190,7 @@ describe("Codex Rewind", () => { cwd: "/workspace/project", model: "gpt-5.4-mini", serviceTier: null, - excludeTurns: false, + excludeTurns: true, persistExtendedHistory: true, }, ]); diff --git a/packages/server/src/server/agent/providers/codex/rewind.ts b/packages/server/src/server/agent/providers/codex/rewind.ts index 021d57645c..a22931b92b 100644 --- a/packages/server/src/server/agent/providers/codex/rewind.ts +++ b/packages/server/src/server/agent/providers/codex/rewind.ts @@ -8,6 +8,7 @@ import { parseCodexThreadForkResponse, parseCodexThreadRollbackResponse, } from "./app-server-transport.js"; +import { readCodexThreadHistoryMode } from "./thread-history.js"; export interface CodexRewindClient { forkThread?(params: CodexThreadForkParams): Promise; @@ -20,26 +21,6 @@ export interface CodexUserMessageTurnIndex { count(): number; } -type CodexThreadHistoryMode = "legacy" | "paginated"; - -async function readCodexThreadHistoryMode( - client: CodexRewindClient, - threadId: string, -): Promise { - const response = await client.request("thread/read", { threadId, includeTurns: false }); - if (typeof response !== "object" || response === null || !("thread" in response)) { - throw new Error("Codex thread/read did not return thread metadata"); - } - const thread = response.thread; - if (typeof thread !== "object" || thread === null || !("historyMode" in thread)) { - return "legacy"; - } - if (thread.historyMode === "legacy" || thread.historyMode === "paginated") { - return thread.historyMode; - } - throw new Error(`Codex thread/read returned unknown history mode ${String(thread.historyMode)}`); -} - async function forkCodexThread( client: CodexRewindClient, params: CodexThreadForkParams, @@ -95,7 +76,8 @@ export async function revertCodexConversation(input: { model: input.model ?? null, serviceTier: input.serviceTier ?? null, ...(input.config ? { config: input.config } : {}), - excludeTurns: false, + // Paseo re-reads the rewound thread through paged history, not the fork reply. + excludeTurns: true, persistExtendedHistory: true, }; diff --git a/packages/server/src/server/agent/providers/codex/test-utils/fake-app-server.ts b/packages/server/src/server/agent/providers/codex/test-utils/fake-app-server.ts index 226eeb9738..99e6a73050 100644 --- a/packages/server/src/server/agent/providers/codex/test-utils/fake-app-server.ts +++ b/packages/server/src/server/agent/providers/codex/test-utils/fake-app-server.ts @@ -3,6 +3,10 @@ import { EventEmitter } from "node:events"; import { PassThrough } from "node:stream"; import type { AgentSession, AgentStreamEvent } from "../../../agent-sdk-types.js"; +import { CODEX_PROVIDER_LINE_BYTES } from "../app-server-transport.js"; + +/** A handler result that the fake writes as a response line past the provider line cap. */ +export const OVERSIZED_RESPONSE = Symbol("oversized Codex response"); type JsonObject = Record; type FakeCodexAppServerHandler = (params: unknown) => unknown; @@ -229,8 +233,18 @@ export function createFakeCodexAppServer( return; } - Promise.resolve(handler(message.params)) + let response: unknown; + try { + response = handler(message.params); + } catch (error) { + response = Promise.reject(error); + } + Promise.resolve(response) .then((result) => { + if (result === OVERSIZED_RESPONSE) { + writeOversizedResponse(message.id as number); + return undefined; + } const rpcError = toJsonObject(result).__jsonRpcError; if (rpcError) { child.stdout.write(`${JSON.stringify({ id: message.id, error: rpcError })}\n`); @@ -313,6 +327,16 @@ export function createFakeCodexAppServer( }); } + // Codex writes `id` first, then a result past the provider line cap. + function writeOversizedResponse(id: number): void { + child.stdout.write(`{"id":${id},"result":{"data":[{"item":{"text":"`); + const chunk = Buffer.alloc(1024 * 1024, "x"); + for (let written = 0; written <= CODEX_PROVIDER_LINE_BYTES; written += chunk.byteLength) { + child.stdout.write(chunk); + } + child.stdout.write('"}}]}}\n'); + } + function writeNotification(method: string, params: JsonObject): void { child.stdout.write(`${JSON.stringify({ method, params })}\n`); } @@ -696,3 +720,81 @@ export function waitForProviderSubagent( ): Promise { return waitForNextEvent(session, "provider_subagent", (event) => event.event.id === id); } + +export interface FakeCodexHistoryTurn { + id: string; + startedAt?: number; + completedAt?: number; + items: JsonObject[]; +} + +export interface FakeCodexHistoryThread { + /** Omitted on Codex builds older than 0.143, which only have legacy history. */ + historyMode?: "legacy" | "paginated"; + turns: FakeCodexHistoryTurn[]; +} + +export interface FakeCodexHistoryOptions { + /** Any response carrying one of these items is written past the line cap. */ + oversizedItemIds?: string[]; +} + +/** + * Codex's persisted thread history: `thread/read` for metadata and legacy full + * reads, `thread/items/list` for paginated threads. Paginated threads reject + * the full read as Codex 0.145 and 0.146 do. + */ +export function fakeCodexThreadHistory( + threads: Record, + options: FakeCodexHistoryOptions = {}, +): Record { + const oversizedItemIds = new Set(options.oversizedItemIds ?? []); + const isOversized = (items: JsonObject[]) => + items.some((item) => oversizedItemIds.has(String(item.id))); + + function readThread(params: unknown): { id: string; thread: FakeCodexHistoryThread } { + const id = String(toJsonObject(params).threadId); + const thread = threads[id]; + if (!thread) throw new Error(`thread not found: ${id}`); + return { id, thread }; + } + + return { + "thread/read": (params) => { + const { id, thread } = readThread(params); + if (!toJsonObject(params).includeTurns) { + return { + thread: { id, ...(thread.historyMode ? { historyMode: thread.historyMode } : {}) }, + }; + } + if (thread.historyMode === "paginated") { + throw new Error("paginated threads do not support thread/read(includeTurns=true)"); + } + if (isOversized(thread.turns.flatMap((turn) => turn.items))) return OVERSIZED_RESPONSE; + return { thread: { id, turns: thread.turns } }; + }, + "thread/items/list": (params) => { + const { thread } = readThread(params); + if (thread.historyMode !== "paginated") { + throw new Error("thread/items/list is not supported for legacy threads"); + } + const request = toJsonObject(params); + // Codex records each item's own times; the fake gives every item its turn's. + const rows = thread.turns.flatMap((turn) => + turn.items.map((item) => ({ + turnId: turn.id, + item, + startedAtMs: turn.startedAt === undefined ? null : turn.startedAt * 1000, + completedAtMs: turn.completedAt === undefined ? null : turn.completedAt * 1000, + })), + ); + const limit = Math.min(typeof request.limit === "number" ? request.limit : 25, 100); + const start = + typeof request.cursor === "string" ? Number(request.cursor.slice("cursor-".length)) : 0; + const end = Math.min(rows.length, start + limit); + const data = rows.slice(start, end); + if (isOversized(data.map((row) => row.item))) return OVERSIZED_RESPONSE; + return { data, nextCursor: end < rows.length ? `cursor-${end}` : null }; + }, + }; +} diff --git a/packages/server/src/server/agent/providers/codex/thread-history.test.ts b/packages/server/src/server/agent/providers/codex/thread-history.test.ts new file mode 100644 index 0000000000..8d3d8fce3d --- /dev/null +++ b/packages/server/src/server/agent/providers/codex/thread-history.test.ts @@ -0,0 +1,220 @@ +import type { ChildProcessWithoutNullStreams } from "node:child_process"; +import { describe, expect, test } from "vitest"; + +import { createTestLogger } from "../../../../test-utils/test-logger.js"; +import type { AgentSession, AgentStreamEvent } from "../../agent-sdk-types.js"; +import { CodexAppServerAgentClient } from "../codex-app-server-agent.js"; +import { + createFakeCodexAppServer, + type FakeCodexAppServer, + type FakeCodexHistoryOptions, + type FakeCodexHistoryThread, + fakeCodexThreadHistory, +} from "./test-utils/fake-app-server.js"; + +function userItem(id: string, text: string) { + return { type: "userMessage", id, content: [{ type: "text", text }] }; +} + +function agentItem(id: string, text: string) { + return { type: "agentMessage", id, text }; +} + +async function openCodexThread( + thread: FakeCodexHistoryThread, + options?: FakeCodexHistoryOptions, +): Promise<{ appServer: FakeCodexAppServer; session: AgentSession }> { + const appServer = createFakeCodexAppServer( + fakeCodexThreadHistory({ "saved-thread": thread }, options), + ); + const provider = new CodexAppServerAgentClient(createTestLogger()); + const internals = provider as unknown as { + goalsEnabledPromise: Promise | null; + autoReviewEnabledPromise: Promise | null; + spawnAppServer: () => Promise; + }; + internals.goalsEnabledPromise = Promise.resolve(false); + internals.autoReviewEnabledPromise = Promise.resolve(false); + internals.spawnAppServer = async () => appServer.child; + const session = await provider.resumeSession({ sessionId: "saved-thread" }); + return { appServer, session }; +} + +async function readHistory(session: AgentSession): Promise { + const events: AgentStreamEvent[] = []; + for await (const event of session.streamHistory()) events.push(event); + return events; +} + +function timelineItems(events: AgentStreamEvent[]) { + return events.flatMap((event) => (event.type === "timeline" ? [event.item] : [])); +} + +function historyRequests(appServer: FakeCodexAppServer) { + return appServer + .requests() + .filter( + (request) => + request.method === "thread/read" || + request.method === "thread/turns/list" || + request.method === "thread/items/list", + ) + .map((request) => ({ method: request.method, params: request.params })); +} + +describe("Codex thread history", () => { + test("loads a paginated thread through pages", async () => { + const answers = Array.from({ length: 150 }, (_, index) => + agentItem(`agent-${index}`, `Answer ${index}`), + ); + const { appServer, session } = await openCodexThread({ + historyMode: "paginated", + turns: [ + { + id: "turn-1", + startedAt: 1_778_832_941, + completedAt: 1_778_833_000, + items: [userItem("user-1", "First"), ...answers.slice(0, 120)], + }, + { + id: "turn-2", + startedAt: 1_778_833_094, + completedAt: 1_778_833_100, + items: [userItem("user-2", "Second"), ...answers.slice(120)], + }, + ], + }); + try { + const events = await readHistory(session); + + expect(timelineItems(events)).toEqual([ + { type: "user_message", text: "First", messageId: "user-1" }, + ...answers.slice(0, 120).map((answer) => ({ + type: "assistant_message", + text: answer.text, + messageId: answer.id, + })), + { type: "user_message", text: "Second", messageId: "user-2" }, + ...answers.slice(120).map((answer) => ({ + type: "assistant_message", + text: answer.text, + messageId: answer.id, + })), + ]); + const timestamps = events.flatMap((event) => + event.type === "timeline" ? [event.timestamp] : [], + ); + expect(timestamps[0]).toBe("2026-05-15T08:15:41.000Z"); + expect(timestamps[1]).toBe("2026-05-15T08:16:40.000Z"); + expect(timestamps[121]).toBe("2026-05-15T08:18:14.000Z"); + const requests = historyRequests(appServer); + expect(requests).not.toContainEqual( + expect.objectContaining({ params: expect.objectContaining({ includeTurns: true }) }), + ); + const itemPages = requests.filter((request) => request.method === "thread/items/list"); + expect(itemPages.length).toBeGreaterThan(1); + for (const page of itemPages) { + expect(page.params).toMatchObject({ threadId: "saved-thread", sortDirection: "asc" }); + expect((page.params as { limit: number }).limit).toBeLessThanOrEqual(100); + } + } finally { + await session.close(); + } + appServer.assertNoErrors(); + }); + + test("loads a paginated thread whose full read Codex 0.145 rejects", async () => { + const { appServer, session } = await openCodexThread({ + historyMode: "paginated", + turns: [{ id: "turn-1", items: [userItem("user-1", "Hello"), agentItem("agent-1", "Hi")] }], + }); + try { + expect(timelineItems(await readHistory(session))).toEqual([ + { type: "user_message", text: "Hello", messageId: "user-1" }, + { type: "assistant_message", text: "Hi", messageId: "agent-1" }, + ]); + } finally { + await session.close(); + } + appServer.assertNoErrors(); + }); + + test("a page too big to send shows a warning row and stops loading", async () => { + const answers = Array.from({ length: 120 }, (_, index) => + agentItem(`agent-${index}`, `Answer ${index}`), + ); + const { appServer, session } = await openCodexThread( + { + historyMode: "paginated", + turns: [ + { id: "turn-1", items: [userItem("user-1", "One"), ...answers] }, + { id: "turn-2", items: [userItem("user-2", "Two"), agentItem("huge", "never sent")] }, + ], + }, + { oversizedItemIds: ["huge"] }, + ); + const loaded = [ + { type: "user_message", text: "One", messageId: "user-1" }, + ...answers.slice(0, 99).map((answer) => ({ + type: "assistant_message", + text: answer.text, + messageId: answer.id, + })), + { + type: "notification", + level: "warning", + message: "The rest of this Codex thread's history is too large to load.", + }, + ]; + try { + expect(timelineItems(await readHistory(session))).toEqual(loaded); + // The oversized reply failed only its own request; the session still answers. + expect(timelineItems(await readHistory(session))).toEqual(loaded); + } finally { + await session.close(); + } + appServer.assertNoErrors(); + }); + + test("loads a legacy thread that fits in one read", async () => { + const { appServer, session } = await openCodexThread({ + historyMode: "legacy", + turns: [{ id: "turn-1", items: [userItem("user-1", "Hello"), agentItem("agent-1", "Hi")] }], + }); + try { + expect(timelineItems(await readHistory(session))).toEqual([ + { type: "user_message", text: "Hello", messageId: "user-1" }, + { type: "assistant_message", text: "Hi", messageId: "agent-1" }, + ]); + expect(historyRequests(appServer).map((request) => request.method)).not.toContain( + "thread/items/list", + ); + } finally { + await session.close(); + } + appServer.assertNoErrors(); + }); + + test("a legacy thread too big for one read shows the migration command", async () => { + const { appServer, session } = await openCodexThread( + { + historyMode: "legacy", + turns: [{ id: "turn-1", items: [userItem("user-1", "Hello"), agentItem("huge", "x")] }], + }, + { oversizedItemIds: ["huge"] }, + ); + const migrationRow = { + type: "notification", + level: "warning", + message: expect.stringContaining("codex migrate-rollouts --apply --thread saved-thread"), + }; + try { + expect(timelineItems(await readHistory(session))).toEqual([migrationRow]); + // The session stays up for the user to prompt until they archive it to migrate. + expect(timelineItems(await readHistory(session))).toEqual([migrationRow]); + } finally { + await session.close(); + } + appServer.assertNoErrors(); + }); +}); diff --git a/packages/server/src/server/agent/providers/codex/thread-history.ts b/packages/server/src/server/agent/providers/codex/thread-history.ts new file mode 100644 index 0000000000..bf7461deb6 --- /dev/null +++ b/packages/server/src/server/agent/providers/codex/thread-history.ts @@ -0,0 +1,168 @@ +import { z } from "zod"; + +import { CodexAppServerResponseTooLargeError } from "./app-server-transport.js"; + +// Codex clamps history pages to 100 rows. +const PAGE_SIZE = 100; + +export interface CodexThreadHistoryClient { + request(method: string, params?: unknown): Promise; +} + +export type CodexThreadHistoryMode = "legacy" | "paginated"; + +export interface CodexHistoryRow { + item: unknown; + turnId: string | undefined; + /** The item's times on a paginated thread; its turn's on a legacy thread. */ + startedAt: unknown; + completedAt: unknown; +} + +/** A history reply over the line cap; the rows before it were already read. */ +export class CodexThreadHistoryTooLargeError extends Error { + constructor(message: string) { + super(message); + this.name = "CodexThreadHistoryTooLargeError"; + } +} + +const CodexThreadMetadataSchema = z + .object({ + thread: z.object({ historyMode: z.enum(["legacy", "paginated"]).optional() }).passthrough(), + }) + .passthrough(); + +const CodexLegacyThreadReadSchema = z + .object({ + thread: z + .object({ + turns: z + .array( + z + .object({ + id: z.string().optional(), + startedAt: z.unknown().optional(), + completedAt: z.unknown().optional(), + items: z.array(z.unknown()).default([]), + }) + .passthrough(), + ) + .default([]), + }) + .passthrough() + .default({ turns: [] }), + }) + .passthrough(); + +const CodexItemsPageSchema = z + .object({ + data: z.array( + z + .object({ + turnId: z.string(), + item: z.unknown(), + startedAtMs: z.number().nullable().optional(), + completedAtMs: z.number().nullable().optional(), + }) + .passthrough(), + ), + nextCursor: z.string().nullable().optional(), + }) + .passthrough(); + +/** Codex builds before 0.143 report no history mode; their threads are legacy. */ +export async function readCodexThreadHistoryMode( + client: CodexThreadHistoryClient, + threadId: string, +): Promise { + const response = CodexThreadMetadataSchema.parse( + await client.request("thread/read", { threadId, includeTurns: false }), + ); + return response.thread.historyMode ?? "legacy"; +} + +/** + * Reads a thread's persisted items in order. Paginated threads are read in + * pages; a legacy thread arrives in one `thread/read` reply, as Codex offers + * no pages for it. + */ +export async function* readCodexThreadHistory( + client: CodexThreadHistoryClient, + threadId: string, +): AsyncGenerator { + if ((await readCodexThreadHistoryMode(client, threadId)) === "paginated") { + yield* readPaginatedHistory(client, threadId); + } else { + yield* readLegacyHistory(client, threadId); + } +} + +async function* readLegacyHistory( + client: CodexThreadHistoryClient, + threadId: string, +): AsyncGenerator { + const response = await requestHistory( + client, + "thread/read", + { threadId, includeTurns: true }, + `This Codex thread's history is too large to load in one read. ` + + `To move it to paginated history, archive this agent, ` + + `run \`codex migrate-rollouts --apply --thread ${threadId}\` (Codex 0.148 or newer), ` + + `then unarchive the agent.`, + ); + for (const turn of CodexLegacyThreadReadSchema.parse(response).thread.turns) { + for (const item of turn.items) { + yield { item, turnId: turn.id, startedAt: turn.startedAt, completedAt: turn.completedAt }; + } + } +} + +async function* readPaginatedHistory( + client: CodexThreadHistoryClient, + threadId: string, +): AsyncGenerator { + const seenCursors = new Set(); + let cursor: string | undefined; + for (;;) { + const page = CodexItemsPageSchema.parse( + await requestHistory( + client, + "thread/items/list", + { threadId, limit: PAGE_SIZE, sortDirection: "asc", cursor }, + "The rest of this Codex thread's history is too large to load.", + ), + ); + for (const row of page.data) { + yield { + item: row.item, + turnId: row.turnId, + startedAt: row.startedAtMs, + completedAt: row.completedAtMs, + }; + } + if (page.nextCursor == null) return; + // Codex's own client guards against a cursor that does not advance. + if (seenCursors.has(page.nextCursor)) { + throw new Error(`Codex repeated history cursor for thread ${threadId}`); + } + seenCursors.add(page.nextCursor); + cursor = page.nextCursor; + } +} + +async function requestHistory( + client: CodexThreadHistoryClient, + method: string, + params: unknown, + tooLargeMessage: string, +): Promise { + try { + return await client.request(method, params); + } catch (error) { + if (error instanceof CodexAppServerResponseTooLargeError) { + throw new CodexThreadHistoryTooLargeError(tooLargeMessage); + } + throw error; + } +} diff --git a/packages/server/src/server/agent/providers/jsonl-frame-decoder.ts b/packages/server/src/server/agent/providers/jsonl-frame-decoder.ts index 8b1b9767de..ff8fa83489 100644 --- a/packages/server/src/server/agent/providers/jsonl-frame-decoder.ts +++ b/packages/server/src/server/agent/providers/jsonl-frame-decoder.ts @@ -1,10 +1,17 @@ // Provider-process stdout only; these are not app/daemon wire limits. export const JSONL_PROVIDER_LINE_BYTES = 64 * 1024 * 1024; +// Enough of an oversized line to read which response it was, e.g. `{"id":12,"result":`. +const OVERSIZED_LINE_HEAD_BYTES = 64; + export class ProviderStreamLimitError extends Error { readonly code = "PROVIDER_STREAM_LINE_TOO_LARGE"; - constructor(readonly limitBytes: number) { + constructor( + readonly limitBytes: number, + /** The first bytes of the oversized line; the rest is discarded unread. */ + readonly lineHead = "", + ) { super( `Provider stdout line exceeds ${limitBytes} bytes. Start a new thread or reduce the provider response size.`, ); @@ -12,10 +19,26 @@ export class ProviderStreamLimitError extends Error { } } -/** Admits raw UTF-8 bytes before retaining, decoding or joining a complete line. */ +function readLineHead(parts: Buffer[]): string { + const head: Buffer[] = []; + let bytes = 0; + for (const part of parts) { + if (bytes >= OVERSIZED_LINE_HEAD_BYTES) break; + head.push(part.subarray(0, OVERSIZED_LINE_HEAD_BYTES - bytes)); + bytes += head[head.length - 1].byteLength; + } + return Buffer.concat(head).toString("utf8"); +} + +/** + * Admits raw UTF-8 bytes before retaining, decoding or joining a complete line. + * An oversized line is reported with its head and skipped up to its newline; + * the owner closes the reader if the stream cannot continue without it. + */ export class ProviderLineReader { private parts: Buffer[] = []; private bytes = 0; + private skippingOversizedLine = false; private stopped = false; constructor( @@ -31,11 +54,20 @@ export class ProviderLineReader { while (start < buffer.byteLength && !this.stopped) { const newline = buffer.indexOf(10, start); const end = newline === -1 ? buffer.byteLength : newline; + if (this.skippingOversizedLine) { + if (newline === -1) return; + this.skippingOversizedLine = false; + start = newline + 1; + continue; + } const length = end - start; if (length > this.limitBytes - this.bytes) { - this.close(); - this.onLimit(new ProviderStreamLimitError(this.limitBytes)); - return; + const head = readLineHead([...this.parts, buffer.subarray(start, end)]); + this.parts = []; + this.bytes = 0; + this.skippingOversizedLine = true; + this.onLimit(new ProviderStreamLimitError(this.limitBytes, head)); + continue; } if (length) this.parts.push(buffer.subarray(start, end)); this.bytes += length; @@ -150,7 +182,10 @@ export class JsonlFrameDecoder { (line) => { if (line.trim()) this.consumeLine(line); }, - (error) => this.drop("line-too-large", error), + (error) => { + this.lines.close(); + this.drop("line-too-large", error); + }, ); }