From f036ed43a2bf78e177afce8ea59229f6eb1bbc09 Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Mon, 27 Jul 2026 23:56:44 -0400 Subject: [PATCH] fix(codex): prevent long conversations slowing on thread resume (#114882) * fix(codex): reduce native thread resume overhead * test(codex): register native attempt connection shard * fix(codex): keep oversized rollouts on bounded fast path --- extensions/codex/src/app-server/protocol.ts | 1 + .../app-server/run-attempt-connection.test.ts | 127 ++++++ .../src/app-server/run-attempt-connection.ts | 50 ++- .../codex/src/app-server/session-binding.ts | 1 + .../src/app-server/startup-binding.test.ts | 372 ++++++++++++++++++ .../codex/src/app-server/startup-binding.ts | 132 ++++++- .../src/app-server/thread-lifecycle-io.ts | 20 +- .../thread-lifecycle.binding.test.ts | 55 +++ ...e-extension-package-boundary-artifacts.mjs | 12 +- ...tension-package-boundary-artifacts.test.ts | 9 + test/scripts/test-projects.test.ts | 11 +- ...n-codex-app-server-attempt-light.config.ts | 1 + 12 files changed, 747 insertions(+), 44 deletions(-) create mode 100644 extensions/codex/src/app-server/run-attempt-connection.test.ts diff --git a/extensions/codex/src/app-server/protocol.ts b/extensions/codex/src/app-server/protocol.ts index f57c8d1a069c..76c34e6d37b1 100644 --- a/extensions/codex/src/app-server/protocol.ts +++ b/extensions/codex/src/app-server/protocol.ts @@ -368,6 +368,7 @@ export type CodexTurn = { export type CodexThread = { id: string; sessionId?: string; + path?: string | null; historyMode?: "legacy" | "paginated"; extra?: JsonObject | null; name?: string | null; diff --git a/extensions/codex/src/app-server/run-attempt-connection.test.ts b/extensions/codex/src/app-server/run-attempt-connection.test.ts new file mode 100644 index 000000000000..c45ad2eb749a --- /dev/null +++ b/extensions/codex/src/app-server/run-attempt-connection.test.ts @@ -0,0 +1,127 @@ +import fs from "node:fs/promises"; +import path from "node:path"; +import { initializeGlobalHookRunner } from "openclaw/plugin-sdk/hook-runtime"; +import { createMockPluginRegistry } from "openclaw/plugin-sdk/plugin-test-runtime"; +import { describe, expect, it, vi } from "vitest"; +import * as appServerPolicy from "./app-server-policy.js"; +import * as bindingConnection from "./binding-connection.js"; +import * as codexConfig from "./config.js"; +import { prepareCodexAttemptConnection } from "./run-attempt-connection.js"; +import { createParams, setupRunAttemptTestHooks, tempDir } from "./run-attempt-test-harness.js"; +import { + registerCodexTestSessionIdentity, + testCodexAppServerBindingStore, + writeCodexAppServerBinding, +} from "./session-binding.test-helpers.js"; + +setupRunAttemptTestHooks(); + +describe("prepareCodexAttemptConnection", () => { + it.each([ + { name: "fresh thread", existingThread: false }, + { name: "unchanged resumed thread", existingThread: true }, + ])("resolves a $name and its workspace only once", async ({ existingThread }) => { + const sessionFile = path.join(tempDir, "session.jsonl"); + const workspaceDir = path.join(tempDir, "workspace"); + const params = createParams(sessionFile, workspaceDir); + params.agentDir = path.join(tempDir, "agent"); + registerCodexTestSessionIdentity(sessionFile, params.sessionId, params.sessionKey); + if (existingThread) { + await writeCodexAppServerBinding(sessionFile, { + threadId: "thread-existing", + cwd: workspaceDir, + model: params.modelId, + modelProvider: "openai", + }); + } + + const resolveConnection = vi.spyOn(bindingConnection, "resolveCodexBindingAppServerConnection"); + const resolveModelPolicy = vi.spyOn(appServerPolicy, "resolveCodexAppServerForModelProvider"); + const readApprovalRequirements = vi.spyOn( + codexConfig, + "isCodexAppServerApprovalPolicyAllowedByRequirements", + ); + const stat = vi.spyOn(fs, "stat"); + + const connection = await prepareCodexAttemptConnection({ + params, + options: { bindingStore: testCodexAppServerBindingStore }, + }); + + expect(connection.effectiveWorkspace).toBe(workspaceDir); + expect(resolveConnection).toHaveBeenCalledTimes(1); + expect(resolveModelPolicy).toHaveBeenCalledTimes(1); + expect(readApprovalRequirements).not.toHaveBeenCalled(); + expect(stat.mock.calls.filter(([candidate]) => candidate === workspaceDir)).toHaveLength(0); + expect(connection.mutable.startupBinding?.threadId).toBe( + existingThread ? "thread-existing" : undefined, + ); + }); + + it("re-resolves model and connection policy when an oversized thread rotates", async () => { + const sessionFile = path.join(tempDir, "session.jsonl"); + const workspaceDir = path.join(tempDir, "workspace"); + const agentDir = path.join(tempDir, "agent"); + const params = createParams(sessionFile, workspaceDir); + params.agentDir = agentDir; + params.config = { + agents: { + defaults: { + compaction: { + truncateAfterCompaction: true, + maxActiveTranscriptBytes: "1mb", + }, + }, + }, + }; + registerCodexTestSessionIdentity(sessionFile, params.sessionId, params.sessionKey); + await writeCodexAppServerBinding(sessionFile, { + threadId: "thread-existing", + cwd: workspaceDir, + model: params.modelId, + modelProvider: "openai", + }); + const rolloutDir = path.join(agentDir, "codex-home", "sessions"); + await fs.mkdir(rolloutDir, { recursive: true }); + await fs.writeFile( + path.join(rolloutDir, "rollout-thread-existing.jsonl"), + "x".repeat(1_048_577), + ); + + const resolveConnection = vi.spyOn(bindingConnection, "resolveCodexBindingAppServerConnection"); + const resolveModelPolicy = vi.spyOn(appServerPolicy, "resolveCodexAppServerForModelProvider"); + + const connection = await prepareCodexAttemptConnection({ + params, + options: { bindingStore: testCodexAppServerBindingStore }, + }); + + expect(connection.mutable.startupBinding).toBeUndefined(); + expect(resolveConnection).toHaveBeenCalledTimes(2); + expect(resolveModelPolicy).toHaveBeenCalledTimes(2); + }); + + it("still checks enterprise requirements before promoting tool approvals", async () => { + initializeGlobalHookRunner( + createMockPluginRegistry([{ hookName: "before_tool_call", handler: vi.fn() }]), + ); + const sessionFile = path.join(tempDir, "session.jsonl"); + const workspaceDir = path.join(tempDir, "workspace"); + const params = createParams(sessionFile, workspaceDir); + params.agentDir = path.join(tempDir, "agent"); + registerCodexTestSessionIdentity(sessionFile, params.sessionId, params.sessionKey); + const readApprovalRequirements = vi.spyOn( + codexConfig, + "isCodexAppServerApprovalPolicyAllowedByRequirements", + ); + + const connection = await prepareCodexAttemptConnection({ + params, + options: { bindingStore: testCodexAppServerBindingStore }, + }); + + expect(readApprovalRequirements).toHaveBeenCalledOnce(); + expect(readApprovalRequirements).toHaveBeenCalledWith("untrusted"); + expect(connection.appServer.approvalPolicy).toBe("untrusted"); + }); +}); diff --git a/extensions/codex/src/app-server/run-attempt-connection.ts b/extensions/codex/src/app-server/run-attempt-connection.ts index dca4f046f447..1e2445cc6bdc 100644 --- a/extensions/codex/src/app-server/run-attempt-connection.ts +++ b/extensions/codex/src/app-server/run-attempt-connection.ts @@ -245,20 +245,25 @@ export async function prepareCodexAttemptConnection({ params, options }: CodexRu ); } const effectiveCwd = sandbox?.enabled ? effectiveWorkspace : (requestedCwd ?? effectiveWorkspace); - await ensureCodexWorkspaceDirOnce(effectiveWorkspace); + if (effectiveWorkspace !== resolvedWorkspace) { + await ensureCodexWorkspaceDirOnce(effectiveWorkspace); + } preDynamicStartupStages.mark("effective-workspace"); + const shouldPromoteApprovalPolicy = + beforeToolCallPolicy.hasBeforeToolCallHook || + beforeToolCallPolicy.trustedToolPolicies.length > 0; const resolvePolicyAppServer = () => resolveCodexAppServerForOpenClawToolPolicy({ appServer: configuredAppServer, pluginConfig, env: process.env, - shouldPromote: - beforeToolCallPolicy.hasBeforeToolCallHook || - beforeToolCallPolicy.trustedToolPolicies.length > 0, + shouldPromote: shouldPromoteApprovalPolicy, execPolicy, canUseUntrustedApprovalPolicy: - configuredAppServer.start.transport !== "stdio" || - isCodexAppServerApprovalPolicyAllowedByRequirements("untrusted"), + shouldPromoteApprovalPolicy && + configuredAppServer.approvalPolicy === "never" && + (configuredAppServer.start.transport !== "stdio" || + isCodexAppServerApprovalPolicyAllowedByRequirements("untrusted")), }); let policyAppServer = resolvePolicyAppServer(); let appServer = resolveCodexAppServerForModelProvider({ @@ -315,6 +320,7 @@ export async function prepareCodexAttemptConnection({ params, options }: CodexRu } else { params.abortSignal?.addEventListener("abort", abortFromUpstream, { once: true }); } + const startupBindingBeforeRotation = startupBinding; startupBinding = await rotateOversizedCodexAppServerStartupBinding({ binding: startupBinding, bindingStore, @@ -328,20 +334,24 @@ export async function prepareCodexAttemptConnection({ params, options }: CodexRu const initialInactiveThreadBootstrapBindingForcedFreshStart = initialStartupBindingHadInactiveThreadBootstrap && !startupBinding?.threadId; preDynamicStartupStages.mark("rotate-binding"); - reviewerPolicyContext = resolveReviewerPolicyContext(startupBinding); - configuredAppServer = resolveRuntimeOptionsForBinding({ - modelProvider: reviewerPolicyContext.modelProvider, - model: reviewerPolicyContext.model, - }); - policyAppServer = resolvePolicyAppServer(); - appServer = resolveCodexAppServerForModelProvider({ - appServer: policyAppServer, - provider: reviewerPolicyContext.modelProvider, - model: reviewerPolicyContext.model, - config: params.config, - env: process.env, - agentDir, - }); + // Rotation returns the original binding on the common resume path; only a + // cleared or replaced native thread changes its model, policy, or connection. + if (startupBinding !== startupBindingBeforeRotation) { + reviewerPolicyContext = resolveReviewerPolicyContext(startupBinding); + configuredAppServer = resolveRuntimeOptionsForBinding({ + modelProvider: reviewerPolicyContext.modelProvider, + model: reviewerPolicyContext.model, + }); + policyAppServer = resolvePolicyAppServer(); + appServer = resolveCodexAppServerForModelProvider({ + appServer: policyAppServer, + provider: reviewerPolicyContext.modelProvider, + model: reviewerPolicyContext.model, + config: params.config, + env: process.env, + agentDir, + }); + } const nativeHookRelayEvents = resolveCodexNativeHookRelayEvents({ configuredEvents: options.nativeHookRelay?.events, appServer, diff --git a/extensions/codex/src/app-server/session-binding.ts b/extensions/codex/src/app-server/session-binding.ts index 190fd698ec8d..dfab8d004b1a 100644 --- a/extensions/codex/src/app-server/session-binding.ts +++ b/extensions/codex/src/app-server/session-binding.ts @@ -175,6 +175,7 @@ const threadBindingSchema = z threadId: z.string().refine((value) => Boolean(value.trim())), clientId: optionalStringSchema, cwd: z.string(), + rolloutPath: optionalNonBlankStringSchema, // Private runtime ownership. Only the supervision catalog creates this // marker; public OpenClaw session metadata must never authorize user-home access. connectionScope: z.literal("supervision").optional(), diff --git a/extensions/codex/src/app-server/startup-binding.test.ts b/extensions/codex/src/app-server/startup-binding.test.ts index 66027b8a428f..bfa5371c1ade 100644 --- a/extensions/codex/src/app-server/startup-binding.test.ts +++ b/extensions/codex/src/app-server/startup-binding.test.ts @@ -240,6 +240,234 @@ describe("Codex app-server startup binding", () => { expect(savedBinding).toBeUndefined(); }); + it("reads the latest native token snapshot from one bounded rollout tail", async () => { + const sessionFile = path.join(tempDir, "session.jsonl"); + const workspaceDir = path.join(tempDir, "workspace"); + const agentDir = path.join(tempDir, "agent"); + const rolloutDir = path.join(agentDir, "codex-home", "sessions"); + const rolloutPath = path.join(rolloutDir, "rollout-thread-existing.jsonl"); + await fs.mkdir(rolloutDir, { recursive: true }); + await fs.writeFile( + rolloutPath, + [ + JSON.stringify({ payload: { type: "agent_message", text: "x".repeat(200_000) } }), + JSON.stringify({ + payload: { + type: "token_count", + info: { + last_token_usage: { total_tokens: 120_000 }, + model_context_window: 128_000, + }, + }, + }), + "", + ].join("\n"), + ); + await writeExistingBinding(sessionFile, workspaceDir, { rolloutPath }); + const openFile = fs.open.bind(fs); + const countFileReads: Array<() => number> = []; + vi.spyOn(fs, "open").mockImplementation(async (file, flags, mode) => { + const handle = await openFile(file, flags, mode); + const read = vi.spyOn(handle, "read"); + countFileReads.push(() => read.mock.calls.length); + return handle; + }); + + const binding = await rotateOversizedCodexAppServerStartupBinding({ + binding: await readCodexAppServerBinding(sessionFile), + sessionFile, + agentDir, + config: undefined, + }); + + expect(binding).toBeUndefined(); + expect(countFileReads.map((count) => count())).toEqual([1]); + }); + + it("keeps rollouts above the safe root default on the bounded direct path", async () => { + const sessionFile = path.join(tempDir, "session.jsonl"); + const workspaceDir = path.join(tempDir, "workspace"); + const agentDir = path.join(tempDir, "agent"); + const rolloutDir = path.join(agentDir, "codex-home", "sessions"); + const rolloutPath = path.join(rolloutDir, "rollout-thread-existing.jsonl"); + await fs.mkdir(rolloutDir, { recursive: true }); + const tokenSnapshot = Buffer.from( + `\n${JSON.stringify({ + payload: { + type: "token_count", + info: { + last_token_usage: { total_tokens: 120_000 }, + model_context_window: 128_000, + }, + }, + })}\n`, + ); + const oversizedOffset = 16 * 1024 * 1024 + 1; + const rolloutHandle = await fs.open(rolloutPath, "w"); + try { + await rolloutHandle.truncate(oversizedOffset); + await rolloutHandle.write(tokenSnapshot, 0, tokenSnapshot.byteLength, oversizedOffset); + } finally { + await rolloutHandle.close(); + } + await writeExistingBinding(sessionFile, workspaceDir, { rolloutPath }); + const readDirectories = vi.spyOn(fs, "readdir"); + const openFile = fs.open.bind(fs); + const countFileReads: Array<() => number> = []; + vi.spyOn(fs, "open").mockImplementation(async (file, flags, mode) => { + const handle = await openFile(file, flags, mode); + const read = vi.spyOn(handle, "read"); + countFileReads.push(() => read.mock.calls.length); + return handle; + }); + + const binding = await rotateOversizedCodexAppServerStartupBinding({ + binding: await readCodexAppServerBinding(sessionFile), + sessionFile, + agentDir, + config: undefined, + }); + + expect(binding).toBeUndefined(); + expect(readDirectories).not.toHaveBeenCalled(); + expect(countFileReads.map((count) => count())).toEqual([1]); + }); + + it("scans a large trailing transcript record without repeatedly copying it", async () => { + const sessionFile = path.join(tempDir, "session.jsonl"); + const workspaceDir = path.join(tempDir, "workspace"); + const agentDir = path.join(tempDir, "agent"); + const rolloutDir = path.join(agentDir, "codex-home", "sessions"); + const rolloutPath = path.join(rolloutDir, "rollout-thread-existing.jsonl"); + await fs.mkdir(rolloutDir, { recursive: true }); + await fs.writeFile( + rolloutPath, + [ + JSON.stringify({ + payload: { + type: "token_count", + info: { + last_token_usage: { total_tokens: 120_000 }, + model_context_window: 128_000, + }, + }, + }), + JSON.stringify({ payload: { type: "agent_message", text: "x".repeat(200_000) } }), + "", + ].join("\n"), + ); + await writeExistingBinding(sessionFile, workspaceDir, { rolloutPath }); + const concatenateBuffers = vi.spyOn(Buffer, "concat"); + + const binding = await rotateOversizedCodexAppServerStartupBinding({ + binding: await readCodexAppServerBinding(sessionFile), + sessionFile, + agentDir, + config: undefined, + }); + + expect(binding).toBeUndefined(); + const largeRecordCopies = concatenateBuffers.mock.calls.filter( + ([fragments]) => + fragments.reduce((total, fragment) => total + fragment.byteLength, 0) > 65_536, + ); + expect(largeRecordCopies).toHaveLength(1); + }); + + it("fills short filesystem reads before advancing the rollout tail cursor", async () => { + const sessionFile = path.join(tempDir, "session.jsonl"); + const workspaceDir = path.join(tempDir, "workspace"); + const agentDir = path.join(tempDir, "agent"); + const rolloutDir = path.join(agentDir, "codex-home", "sessions"); + const rolloutPath = path.join(rolloutDir, "rollout-thread-existing.jsonl"); + await fs.mkdir(rolloutDir, { recursive: true }); + await fs.writeFile( + rolloutPath, + [ + JSON.stringify({ payload: { type: "agent_message", text: "x".repeat(80_000) } }), + JSON.stringify({ + payload: { + type: "token_count", + info: { + last_token_usage: { total_tokens: 120_000 }, + model_context_window: 128_000, + }, + }, + }), + "", + ].join("\n"), + ); + await writeExistingBinding(sessionFile, workspaceDir, { rolloutPath }); + const openFile = fs.open.bind(fs); + let fileReads = 0; + vi.spyOn(fs, "open").mockImplementation(async (file, flags, mode) => { + const handle = await openFile(file, flags, mode); + return new Proxy(handle, { + get(target, property) { + if (property === "read") { + return (buffer: Uint8Array, offset: number, length: number, position: number) => { + fileReads += 1; + return target.read(buffer, offset, Math.min(length, 8_192), position); + }; + } + const value = Reflect.get(target, property, target) as unknown; + return typeof value === "function" ? value.bind(target) : value; + }, + }); + }); + + const binding = await rotateOversizedCodexAppServerStartupBinding({ + binding: await readCodexAppServerBinding(sessionFile), + sessionFile, + agentDir, + config: undefined, + }); + + expect(binding).toBeUndefined(); + expect(fileReads).toBeGreaterThan(1); + }); + + it("combines the latest usage with an older context window across rollout tail chunks", async () => { + const sessionFile = path.join(tempDir, "session.jsonl"); + const workspaceDir = path.join(tempDir, "workspace"); + const agentDir = path.join(tempDir, "agent"); + const rolloutDir = path.join(agentDir, "codex-home", "sessions"); + const rolloutPath = path.join(rolloutDir, "rollout-thread-existing.jsonl"); + await fs.mkdir(rolloutDir, { recursive: true }); + await fs.writeFile( + rolloutPath, + [ + JSON.stringify({ + payload: { + type: "token_count", + info: { + last_token_usage: { total_tokens: 1_000 }, + model_context_window: 128_000, + }, + }, + }), + JSON.stringify({ payload: { type: "agent_message", text: "x".repeat(80_000) } }), + JSON.stringify({ + payload: { + type: "token_count", + info: { last_token_usage: { total_tokens: 120_000 } }, + }, + }), + "", + ].join("\n"), + ); + await writeExistingBinding(sessionFile, workspaceDir, { rolloutPath }); + + const binding = await rotateOversizedCodexAppServerStartupBinding({ + binding: await readCodexAppServerBinding(sessionFile), + sessionFile, + agentDir, + config: undefined, + }); + + expect(binding).toBeUndefined(); + }); + it("caps the default native reserve so small context windows keep prompt budget", async () => { const sessionFile = path.join(tempDir, "session.jsonl"); const workspaceDir = path.join(tempDir, "workspace"); @@ -306,6 +534,150 @@ describe("Codex app-server startup binding", () => { expect(savedBinding).toBeUndefined(); }); + it("checks the native rollout path without walking the session directories", async () => { + const sessionFile = path.join(tempDir, "session.jsonl"); + const workspaceDir = path.join(tempDir, "workspace"); + const agentDir = path.join(tempDir, "agent"); + const rolloutDir = path.join(agentDir, "codex-home", "sessions", "2026", "07", "27"); + const rolloutPath = path.join(rolloutDir, "rollout-thread-existing.jsonl"); + await fs.mkdir(rolloutDir, { recursive: true }); + await fs.writeFile(rolloutPath, "x".repeat(2_000)); + await writeExistingBinding(sessionFile, workspaceDir, { rolloutPath }); + const readDirectories = vi.spyOn(fs, "readdir"); + + const binding = await rotateOversizedCodexAppServerStartupBinding({ + binding: await readCodexAppServerBinding(sessionFile), + sessionFile, + agentDir, + config: { + agents: { + defaults: { + compaction: { + truncateAfterCompaction: true, + maxActiveTranscriptBytes: "1k", + }, + }, + }, + } as never, + }); + + expect(binding).toBeUndefined(); + expect(readDirectories).not.toHaveBeenCalled(); + await expect(readCodexAppServerBinding(sessionFile)).resolves.toBeUndefined(); + }); + + it("never reads a stored rollout path outside Codex session roots", async () => { + const sessionFile = path.join(tempDir, "session.jsonl"); + const workspaceDir = path.join(tempDir, "workspace"); + const agentDir = path.join(tempDir, "agent"); + const outsideDir = path.join(tempDir, "outside"); + const outsideRolloutPath = path.join(outsideDir, "rollout-thread-existing.jsonl"); + const rolloutDir = path.join(agentDir, "codex-home", "sessions"); + const rolloutPath = path.join(rolloutDir, "rollout-thread-existing.jsonl"); + await fs.mkdir(outsideDir, { recursive: true }); + await fs.mkdir(rolloutDir, { recursive: true }); + await fs.writeFile(outsideRolloutPath, "x".repeat(2_000)); + await fs.writeFile(rolloutPath, "x".repeat(2_000)); + await writeExistingBinding(sessionFile, workspaceDir, { rolloutPath: outsideRolloutPath }); + const stat = vi.spyOn(fs, "stat"); + const readDirectories = vi.spyOn(fs, "readdir"); + + const binding = await rotateOversizedCodexAppServerStartupBinding({ + binding: await readCodexAppServerBinding(sessionFile), + sessionFile, + agentDir, + config: { + agents: { + defaults: { + compaction: { + truncateAfterCompaction: true, + maxActiveTranscriptBytes: "1k", + }, + }, + }, + } as never, + }); + + expect(binding).toBeUndefined(); + expect(stat.mock.calls.filter(([file]) => file === outsideRolloutPath)).toHaveLength(0); + expect(readDirectories).toHaveBeenCalled(); + }); + + it("does not follow a native rollout symlink outside Codex session roots", async () => { + const sessionFile = path.join(tempDir, "session.jsonl"); + const workspaceDir = path.join(tempDir, "workspace"); + const agentDir = path.join(tempDir, "agent"); + const outsideDir = path.join(tempDir, "outside"); + const outsideRolloutPath = path.join(outsideDir, "rollout-thread-existing.jsonl"); + const rolloutDir = path.join(agentDir, "codex-home", "sessions"); + const symlinkPath = path.join(rolloutDir, "symlink-rollout-thread-existing.jsonl"); + const rolloutPath = path.join(rolloutDir, "rollout-thread-existing.jsonl"); + await fs.mkdir(outsideDir, { recursive: true }); + await fs.mkdir(rolloutDir, { recursive: true }); + await fs.writeFile(outsideRolloutPath, "x".repeat(2_000)); + await fs.writeFile(rolloutPath, "x".repeat(2_000)); + await fs.symlink(outsideRolloutPath, symlinkPath); + await writeExistingBinding(sessionFile, workspaceDir, { rolloutPath: symlinkPath }); + const readDirectories = vi.spyOn(fs, "readdir"); + + const binding = await rotateOversizedCodexAppServerStartupBinding({ + binding: await readCodexAppServerBinding(sessionFile), + sessionFile, + agentDir, + config: { + agents: { + defaults: { + compaction: { + truncateAfterCompaction: true, + maxActiveTranscriptBytes: "1k", + }, + }, + }, + } as never, + }); + + expect(binding).toBeUndefined(); + expect(readDirectories).toHaveBeenCalled(); + }); + + it("does not follow a symlinked native rollout parent outside Codex session roots", async () => { + const sessionFile = path.join(tempDir, "session.jsonl"); + const workspaceDir = path.join(tempDir, "workspace"); + const agentDir = path.join(tempDir, "agent"); + const outsideDir = path.join(tempDir, "outside"); + const outsideRolloutPath = path.join(outsideDir, "rollout-thread-existing.jsonl"); + const rolloutDir = path.join(agentDir, "codex-home", "sessions"); + const redirectedDir = path.join(rolloutDir, "redirect"); + const redirectedRolloutPath = path.join(redirectedDir, "rollout-thread-existing.jsonl"); + const safeRolloutPath = path.join(rolloutDir, "rollout-thread-existing.jsonl"); + await fs.mkdir(outsideDir, { recursive: true }); + await fs.mkdir(rolloutDir, { recursive: true }); + await fs.writeFile(outsideRolloutPath, "x".repeat(2_000)); + await fs.writeFile(safeRolloutPath, "x".repeat(2_000)); + await fs.symlink(outsideDir, redirectedDir, "dir"); + await writeExistingBinding(sessionFile, workspaceDir, { rolloutPath: redirectedRolloutPath }); + const readDirectories = vi.spyOn(fs, "readdir"); + + const binding = await rotateOversizedCodexAppServerStartupBinding({ + binding: await readCodexAppServerBinding(sessionFile), + sessionFile, + agentDir, + config: { + agents: { + defaults: { + compaction: { + truncateAfterCompaction: true, + maxActiveTranscriptBytes: "1k", + }, + }, + }, + } as never, + }); + + expect(binding).toBeUndefined(); + expect(readDirectories).toHaveBeenCalled(); + }); + it("honors custom Codex home rollout files for native rollout limits", async () => { const sessionFile = path.join(tempDir, "session.jsonl"); const workspaceDir = path.join(tempDir, "workspace"); diff --git a/extensions/codex/src/app-server/startup-binding.ts b/extensions/codex/src/app-server/startup-binding.ts index e8c6ce33800c..209e8d740907 100644 --- a/extensions/codex/src/app-server/startup-binding.ts +++ b/extensions/codex/src/app-server/startup-binding.ts @@ -9,6 +9,7 @@ import { embeddedAgentLog, type EmbeddedRunAttemptParams, } from "openclaw/plugin-sdk/agent-harness-runtime"; +import { root as openSafeFilesystemRoot } from "openclaw/plugin-sdk/file-access-runtime"; import { parseSqliteSessionFileMarker } from "openclaw/plugin-sdk/session-store-runtime"; import { resolveCodexAppServerHomeDir } from "./auth-bridge.js"; import { isJsonObject, type JsonValue } from "./protocol.js"; @@ -24,6 +25,7 @@ const CODEX_APP_SERVER_NATIVE_THREAD_FALLBACK_MAX_TOKENS = 300_000; const CODEX_APP_SERVER_NATIVE_THREAD_DEFAULT_RESERVE_TOKENS = 20_000; const CODEX_APP_SERVER_NATIVE_THREAD_MIN_PROMPT_BUDGET_TOKENS = 8_000; const CODEX_APP_SERVER_NATIVE_THREAD_MIN_PROMPT_BUDGET_RATIO = 0.5; +const CODEX_APP_SERVER_ROLLOUT_TAIL_READ_BYTES = 64 * 1024; const CODEX_APP_SERVER_BYTE_UNITS: Record = { b: 1, k: 1024, @@ -46,6 +48,12 @@ type CodexSessionRecordCacheEntry = { record: (Record & { sessionKey: string }) | undefined; }; +type CodexAppServerRolloutFile = { + path: string; + bytes: number; + handle?: Awaited>; +}; + const codexSessionRecordCache = new Map(); function parseCodexAppServerByteLimit(value: unknown): number | undefined { @@ -75,7 +83,8 @@ async function listCodexAppServerRolloutFilesForThread( agentDir: string, threadId: string, codexHome?: string, -): Promise> { + rolloutPath?: string, +): Promise { const resolvedAgentDir = path.resolve(agentDir); const resolvedCodexHome = codexHome?.trim() ? path.resolve(codexHome) @@ -86,7 +95,41 @@ async function listCodexAppServerRolloutFilesForThread( path.join(resolvedAgentDir, "agent", "codex-home", "sessions"), path.join(path.dirname(resolvedAgentDir), "codex-home", "sessions"), ]; - const files: Array<{ path: string; bytes: number }> = []; + const rolloutRoot = rolloutPath + ? roots.find((root) => { + const relativePath = path.relative(root, rolloutPath); + return ( + relativePath !== "" && + relativePath !== ".." && + !relativePath.startsWith(`..${path.sep}`) && + !path.isAbsolute(relativePath) + ); + }) + : undefined; + if ( + rolloutPath && + rolloutRoot && + path.isAbsolute(rolloutPath) && + path.extname(rolloutPath) === ".jsonl" && + path.basename(rolloutPath).includes(threadId) + ) { + try { + // Pin the verified descriptor: lexical checks or realpath followed by + // another open would allow a symlinked parent to escape the native root. + const safeRoot = await openSafeFilesystemRoot(rolloutRoot, { + hardlinks: "reject", + // Tail reads stay bounded; the default 16 MiB whole-file cap would + // send large native rollouts back through recursive discovery. + maxBytes: Number.MAX_SAFE_INTEGER, + symlinks: "reject", + }); + const opened = await safeRoot.open(path.relative(rolloutRoot, rolloutPath)); + return [{ path: opened.realPath, bytes: opened.stat.size, handle: opened.handle }]; + } catch { + // Older Codex servers and moved rollouts still need root-scoped discovery. + } + } + const files: CodexAppServerRolloutFile[] = []; const visited = new Set(); for (const root of roots) { if (visited.has(root)) { @@ -190,26 +233,73 @@ type CodexAppServerRolloutTokenSnapshot = { async function readCodexAppServerRolloutTokenSnapshot( file: string, + openedHandle?: Awaited>, ): Promise { - let handle: Awaited>; - try { - handle = await fs.open(file, "r"); - } catch { - return undefined; + let handle = openedHandle; + if (!handle) { + try { + handle = await fs.open(file, "r"); + } catch { + return undefined; + } } let snapshot: CodexAppServerRolloutTokenSnapshot | undefined; try { - for await (const line of handle.readLines()) { + let position = (await handle.stat()).size; + const partialLineFragments: Buffer[] = []; + const applySnapshotLine = (line: string): boolean => { const lineSnapshot = readCodexAppServerRolloutTokenSnapshotLine(line); - if (lineSnapshot !== undefined) { - snapshot ??= {}; - if (lineSnapshot.totalTokens !== undefined) { - snapshot.totalTokens = lineSnapshot.totalTokens; - } - if (lineSnapshot.modelContextWindow !== undefined) { - snapshot.modelContextWindow = lineSnapshot.modelContextWindow; - } + if (lineSnapshot === undefined) { + return false; } + snapshot ??= {}; + snapshot.totalTokens ??= lineSnapshot.totalTokens; + snapshot.modelContextWindow ??= lineSnapshot.modelContextWindow; + return snapshot.totalTokens !== undefined && snapshot.modelContextWindow !== undefined; + }; + + // Codex appends token snapshots to its rollout. Walk backward so a growing + // thread does not reparse its entire conversation before every new turn. + while (position > 0) { + const bytesToRead = Math.min(position, CODEX_APP_SERVER_ROLLOUT_TAIL_READ_BYTES); + const nextPosition = position - bytesToRead; + const chunk = Buffer.allocUnsafe(bytesToRead); + let bytesRead = 0; + while (bytesRead < bytesToRead) { + const result = await handle.read( + chunk, + bytesRead, + bytesToRead - bytesRead, + nextPosition + bytesRead, + ); + if (result.bytesRead === 0) { + return snapshot; + } + bytesRead += result.bytesRead; + } + let lineEnd = bytesRead; + for (let index = bytesRead - 1; index >= 0; index -= 1) { + if (chunk[index] !== 0x0a) { + continue; + } + const lineFragment = chunk.subarray(index + 1, lineEnd); + const line = + partialLineFragments.length === 0 + ? lineFragment.toString("utf8") + : Buffer.concat([lineFragment, ...partialLineFragments.toReversed()]).toString("utf8"); + partialLineFragments.length = 0; + if (applySnapshotLine(line)) { + return snapshot; + } + lineEnd = index; + } + if (lineEnd > 0) { + partialLineFragments.push(chunk.subarray(0, lineEnd)); + } + position = nextPosition; + } + if (partialLineFragments.length > 0) { + applySnapshotLine(Buffer.concat(partialLineFragments.toReversed()).toString("utf8")); } } finally { await handle.close(); @@ -346,6 +436,7 @@ export async function rotateOversizedCodexAppServerStartupBinding(params: { params.agentDir, binding.threadId, params.codexHome, + binding.rolloutPath, ); const compaction = readCompactionConfig(params.config); const shouldDeferByteGuard = @@ -357,6 +448,11 @@ export async function rotateOversizedCodexAppServerStartupBinding(params: { if (maxBytes !== undefined) { const oversizedFiles = rolloutFiles.filter((file) => file.bytes >= maxBytes); if (oversizedFiles.length > 0) { + await Promise.all( + rolloutFiles.map(async (file) => { + await file.handle?.close(); + }), + ); embeddedAgentLog.warn( "codex app-server native transcript exceeded active byte limit; starting a fresh thread", { @@ -374,7 +470,9 @@ export async function rotateOversizedCodexAppServerStartupBinding(params: { } } const nativeTokenSnapshots = await Promise.all( - rolloutFiles.map(async (file) => readCodexAppServerRolloutTokenSnapshot(file.path)), + rolloutFiles.map(async (file) => + readCodexAppServerRolloutTokenSnapshot(file.path, file.handle), + ), ); const nativeTokens = maxFiniteNumber( nativeTokenSnapshots.map((snapshot) => snapshot?.totalTokens), diff --git a/extensions/codex/src/app-server/thread-lifecycle-io.ts b/extensions/codex/src/app-server/thread-lifecycle-io.ts index a663a287a41f..4c8451a7e1a8 100644 --- a/extensions/codex/src/app-server/thread-lifecycle-io.ts +++ b/extensions/codex/src/app-server/thread-lifecycle-io.ts @@ -1,3 +1,4 @@ +import path from "node:path"; import { embeddedAgentLog } from "openclaw/plugin-sdk/agent-harness-runtime"; import { CODEX_APP_SERVER_UNSUBSCRIBE_TIMEOUT_MS, @@ -18,7 +19,7 @@ import { type CodexPluginThreadConfig, } from "./plugin-thread-config.js"; import { assertCodexThreadStartResponse } from "./protocol-validators.js"; -import type { JsonObject } from "./protocol.js"; +import type { CodexThread, JsonObject } from "./protocol.js"; import type { CodexAppServerBindingIdentity, CodexAppServerContextEngineBinding, @@ -84,6 +85,19 @@ type StartThreadContext = ThreadRequestContext & { rotatedContextEngineBinding: boolean; }; +function resolveCodexThreadRolloutPath(thread: CodexThread): string | undefined { + const rolloutPath = thread.path?.trim(); + if ( + !rolloutPath || + !path.isAbsolute(rolloutPath) || + path.extname(rolloutPath) !== ".jsonl" || + !path.basename(rolloutPath).includes(thread.id) + ) { + return undefined; + } + return rolloutPath; +} + export async function resumeExistingCodexThread( params: CodexStartOrResumeThreadParams, context: ResumeThreadContext, @@ -209,6 +223,7 @@ export async function resumeExistingCodexThread( : resumeBinding.mcpServersFingerprint; const resumePatch = { cwd: params.cwd, + rolloutPath: resolveCodexThreadRolloutPath(response.thread) ?? resumeBinding.rolloutPath, authProfileId: boundAuthProfileId, model: response.model ?? resumeParams.model ?? params.params.modelId, preserveNativeModel: resumeBinding.preserveNativeModel === true ? true : undefined, @@ -409,6 +424,7 @@ export async function startFreshCodexThread( } }); const response = assertCodexThreadStartResponse(threadStartResponse); + const rolloutPath = resolveCodexThreadRolloutPath(response.thread); if (ringZeroActive) { try { await lifecycleTiming.measure("ring-zero-mcp-attestation", () => @@ -438,6 +454,7 @@ export async function startFreshCodexThread( threadId: response.thread.id, ...(clientId ? { clientId } : {}), cwd: params.cwd, + ...(rolloutPath ? { rolloutPath } : {}), authProfileId: params.params.authProfileId, model: response.model ?? startParams.model ?? params.params.modelId, modelProvider: normalizeBindingModelProvider( @@ -490,6 +507,7 @@ export async function startFreshCodexThread( threadId: response.thread.id, ...(clientId ? { clientId } : {}), cwd: params.cwd, + ...(rolloutPath ? { rolloutPath } : {}), authProfileId: params.params.authProfileId, model: response.model ?? startParams.model ?? params.params.modelId, modelProvider: diff --git a/extensions/codex/src/app-server/thread-lifecycle.binding.test.ts b/extensions/codex/src/app-server/thread-lifecycle.binding.test.ts index 6c11f82975ea..c22eb4ed017c 100644 --- a/extensions/codex/src/app-server/thread-lifecycle.binding.test.ts +++ b/extensions/codex/src/app-server/thread-lifecycle.binding.test.ts @@ -268,6 +268,61 @@ function createTwoCalendarAppPolicyContext() { setupRunAttemptTestHooks(); describe("Codex app-server thread lifecycle bindings", () => { + it("persists the native rollout path across thread start and resume", async () => { + const sessionFile = path.join(tempDir, "session.jsonl"); + const workspaceDir = path.join(tempDir, "workspace"); + const threadId = "thread-native-rollout"; + const rolloutPath = path.join( + tempDir, + "agent", + "codex-home", + "sessions", + `rollout-${threadId}.jsonl`, + ); + const params = createParams(sessionFile, workspaceDir); + const request = vi.fn(async (method: string) => { + if (method !== "thread/start" && method !== "thread/resume") { + throw new Error(`unexpected method: ${method}`); + } + const response = threadStartResult(threadId); + return { + ...response, + thread: { ...response.thread, path: rolloutPath }, + }; + }); + const common = { + client: { getInstanceId: () => "native-rollout-client", request } as never, + params, + cwd: workspaceDir, + dynamicTools: [], + appServer: createThreadLifecycleAppServerOptions(), + userMcpServersEnabled: false, + }; + + const started = await startOrResumeThread(common); + expect(started).toMatchObject({ + threadId, + rolloutPath, + lifecycle: { action: "started" }, + }); + await expect(readCodexAppServerBinding(sessionFile)).resolves.toMatchObject({ + threadId, + rolloutPath, + }); + + const resumed = await startOrResumeThread(common); + expect(resumed).toMatchObject({ + threadId, + rolloutPath, + lifecycle: { action: "resumed" }, + }); + expect(request.mock.calls.map(([method]) => method)).toEqual(["thread/start", "thread/resume"]); + await expect(readCodexAppServerBinding(sessionFile)).resolves.toMatchObject({ + threadId, + rolloutPath, + }); + }); + it("reuses one live ephemeral thread across two incognito turns", async () => { const sessionFile = path.join(tempDir, "incognito-session.jsonl"); const workspaceDir = path.join(tempDir, "incognito-workspace"); diff --git a/scripts/prepare-extension-package-boundary-artifacts.mjs b/scripts/prepare-extension-package-boundary-artifacts.mjs index 0986e477c921..c62a1a75597a 100644 --- a/scripts/prepare-extension-package-boundary-artifacts.mjs +++ b/scripts/prepare-extension-package-boundary-artifacts.mjs @@ -755,7 +755,10 @@ async function main(argv = process.argv.slice(2)) { ], outputPaths: [ "dist/plugin-sdk/.boundary-entry-shims.stamp", - ...resolveBoundaryEntryShimRequiredOutputs(), + ...resolveBoundaryEntryShimRequiredOutputs({ + ...process.env, + OPENCLAW_BUILD_PRIVATE_QA: "1", + }), ], }); const qaChannelDtsFresh = @@ -1036,7 +1039,12 @@ async function main(argv = process.argv.slice(2)) { resolve(repoRoot, "scripts/write-plugin-sdk-entry-dts.ts"), ], ROOT_SHIMS_TIMEOUT_MS, - { env: { NODE_OPTIONS: ROOT_SHIMS_NODE_OPTIONS } }, + { + env: { + NODE_OPTIONS: ROOT_SHIMS_NODE_OPTIONS, + OPENCLAW_BUILD_PRIVATE_QA: "1", + }, + }, ); } else if (mode === "all") { process.stdout.write("[plugin-sdk boundary root shims] fresh; skipping\n"); diff --git a/test/scripts/prepare-extension-package-boundary-artifacts.test.ts b/test/scripts/prepare-extension-package-boundary-artifacts.test.ts index b874a6d227e6..b7d78415e61a 100644 --- a/test/scripts/prepare-extension-package-boundary-artifacts.test.ts +++ b/test/scripts/prepare-extension-package-boundary-artifacts.test.ts @@ -630,6 +630,15 @@ describe("prepare-extension-package-boundary-artifacts", () => { expect(productionOutputs).not.toContain("dist/plugin-sdk/test-fixtures.d.ts"); expect(privateQaOutputs).toContain("dist/plugin-sdk/provider-auth-runtime.d.ts"); expect(privateQaOutputs).toContain("dist/plugin-sdk/test-fixtures.d.ts"); + for (const entry of [ + "channel-contract-testing", + "plugin-state-test-runtime", + "plugin-test-runtime", + ]) { + expect(productionOutputs).not.toContain(`dist/plugin-sdk/${entry}.d.ts`); + expect(privateQaOutputs).toContain(`dist/plugin-sdk/${entry}.d.ts`); + expect(privateQaOutputs).toContain(`packages/plugin-sdk/dist/src/plugin-sdk/${entry}.d.ts`); + } }); it("parses prep mode and rejects unknown values", () => { diff --git a/test/scripts/test-projects.test.ts b/test/scripts/test-projects.test.ts index 6ed2cfa70b5a..8cf0613c8204 100644 --- a/test/scripts/test-projects.test.ts +++ b/test/scripts/test-projects.test.ts @@ -4383,10 +4383,13 @@ describe("scripts/test-projects full-suite sharding", () => { ]); }); - it("covers Codex attempt client prewarming in full-suite routing", () => { - expect( - fullSuiteMatches.get("extensions/codex/src/app-server/run-attempt-client-prewarm.test.ts"), - ).toEqual(["test/vitest/vitest.extension-codex-app-server-attempt-light.config.ts"]); + it.each([ + "extensions/codex/src/app-server/run-attempt-client-prewarm.test.ts", + "extensions/codex/src/app-server/run-attempt-connection.test.ts", + ])("covers Codex attempt lifecycle test %s in full-suite routing", (file) => { + expect(fullSuiteMatches.get(file)).toEqual([ + "test/vitest/vitest.extension-codex-app-server-attempt-light.config.ts", + ]); }); it("uses the global host worker budget for roomy local hosts", () => { diff --git a/test/vitest/vitest.extension-codex-app-server-attempt-light.config.ts b/test/vitest/vitest.extension-codex-app-server-attempt-light.config.ts index 8b3e34ff1359..530ab505e2e8 100644 --- a/test/vitest/vitest.extension-codex-app-server-attempt-light.config.ts +++ b/test/vitest/vitest.extension-codex-app-server-attempt-light.config.ts @@ -10,6 +10,7 @@ function createExtensionCodexAppServerAttemptLightVitestConfig( "extensions/codex/src/app-server/attempt-diagnostics.test.ts", "extensions/codex/src/app-server/attempt-steering.test.ts", "extensions/codex/src/app-server/run-attempt-client-prewarm.test.ts", + "extensions/codex/src/app-server/run-attempt-connection.test.ts", ], { dir: "extensions",