fix(server): read Codex history in pages instead of one whole-thread reply

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 <hr@savagestrategic.io> Co-authored-by: dundunHa <lixinpengde@gmail.com>
2026-10-29 20:52:51 +00:00 · 2026-10-29 10:56:25 +01:00
parent ac6e4e9f96
commit 4c4bb45761
12 changed files with 839 additions and 137 deletions
@@ -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 });
+}
@@ -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([
@@ -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<typeof CodexThreadReadResponseSchema>;
-type CodexThreadReadRequest = (threadId: string) => Promise<unknown>;
-
-async function requestCodexThreadHistory(
-  requestThread: CodexThreadReadRequest,
-  threadId: string,
-): Promise<CodexThreadReadResponse> {
-  const response = await requestThread(threadId);
-  return CodexThreadReadResponseSchema.parse(response);
-}
-
 async function loadCodexThreadHistoryTimeline(params: {
+  client: CodexAppServerClientLike;
  threadId: string;
  cwd: string | null;
-  requestThread: CodexThreadReadRequest;
 }): Promise<CodexThreadHistoryProjection> {
-  const response = await requestCodexThreadHistory(params.requestThread, params.threadId);
  const timeline: PersistedTimelineEntry[] = [];
  const subAgentTimelineIndexByThreadId = new Map<string, number>();
-  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<unknown> {
-  return client.request("thread/read", {
-    threadId,
-    includeTurns: true,
-  });
+async function projectCodexThreadHistory(
+  params: { client: CodexAppServerClientLike; threadId: string; cwd: string | null },
+  projection: {
+    timeline: PersistedTimelineEntry[];
+    subAgentTimelineIndexByThreadId: Map<string, number>;
+  },
+): Promise<void> {
+  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);
@@ -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,
+);
@@ -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({
@@ -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");
@@ -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,
      },
    ]);
@@ -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<CodexThreadForkResponse>;
@@ -20,26 +21,6 @@ export interface CodexUserMessageTurnIndex {
  count(): number;
 }

-type CodexThreadHistoryMode = "legacy" | "paginated";
-
-async function readCodexThreadHistoryMode(
-  client: CodexRewindClient,
-  threadId: string,
-): Promise<CodexThreadHistoryMode> {
-  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,
  };

@@ -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<string, unknown>;
 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<ProviderSubagentEvent> {
  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<string, FakeCodexHistoryThread>,
+  options: FakeCodexHistoryOptions = {},
+): Record<string, FakeCodexAppServerHandler> {
+  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 };
+    },
+  };
+}
@@ -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<boolean> | null;
+    autoReviewEnabledPromise: Promise<boolean> | null;
+    spawnAppServer: () => Promise<ChildProcessWithoutNullStreams>;
+  };
+  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<AgentStreamEvent[]> {
+  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();
+  });
+});
@@ -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<unknown>;
+}
+
+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<CodexThreadHistoryMode> {
+  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<CodexHistoryRow> {
+  if ((await readCodexThreadHistoryMode(client, threadId)) === "paginated") {
+    yield* readPaginatedHistory(client, threadId);
+  } else {
+    yield* readLegacyHistory(client, threadId);
+  }
+}
+
+async function* readLegacyHistory(
+  client: CodexThreadHistoryClient,
+  threadId: string,
+): AsyncGenerator<CodexHistoryRow> {
+  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<CodexHistoryRow> {
+  const seenCursors = new Set<string>();
+  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<unknown> {
+  try {
+    return await client.request(method, params);
+  } catch (error) {
+    if (error instanceof CodexAppServerResponseTooLargeError) {
+      throw new CodexThreadHistoryTooLargeError(tooLargeMessage);
+    }
+    throw error;
+  }
+}
@@ -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);
+      },
    );
  }