From bdbed8f8067f7067457d7f587ab418d644bf0580 Mon Sep 17 00:00:00 2001 From: Patrick Lee Date: Thu, 6 Aug 2026 12:52:51 -0400 Subject: [PATCH 1/4] fix(codex): auto-unarchive archived sessions before retry --- .../agent-runtime/src/codex/adapter.test.ts | 3 + packages/agent-runtime/src/codex/adapter.ts | 12 +- .../src/runtime.command-contract.test.ts | 161 ++++++++++++++++++ packages/agent-runtime/src/runtime.ts | 133 ++++++++++++--- 4 files changed, 273 insertions(+), 36 deletions(-) diff --git a/packages/agent-runtime/src/codex/adapter.test.ts b/packages/agent-runtime/src/codex/adapter.test.ts index b491d1fce6..99360a963b 100644 --- a/packages/agent-runtime/src/codex/adapter.test.ts +++ b/packages/agent-runtime/src/codex/adapter.test.ts @@ -592,9 +592,11 @@ describe("codex provider adapter", () => { approvalsReviewer: "user", sandbox: "danger-full-access", cwd: "/tmp/worktree", + ephemeral: false, experimentalRawEvents: true, }, }); + expect(JSON.stringify(cmd)).not.toContain("persistExtendedHistory"); expect(JSON.stringify(cmd)).not.toContain("baseInstructions"); expect(JSON.stringify(cmd)).not.toContain("developerInstructions"); }); @@ -1871,6 +1873,7 @@ describe("codex provider adapter", () => { cwd: "/tmp/worktree", }, }); + expect(JSON.stringify(cmd)).not.toContain("persistExtendedHistory"); expect(JSON.stringify(cmd)).not.toContain("baseInstructions"); expect(JSON.stringify(cmd)).not.toContain("developerInstructions"); }); diff --git a/packages/agent-runtime/src/codex/adapter.ts b/packages/agent-runtime/src/codex/adapter.ts index ea4d9a0999..5f20b14b0c 100644 --- a/packages/agent-runtime/src/codex/adapter.ts +++ b/packages/agent-runtime/src/codex/adapter.ts @@ -94,11 +94,6 @@ interface CodexThreadPermissionSettings { type BbThreadStartParams = ThreadStartParams & { experimentalRawEvents?: boolean; - persistExtendedHistory?: boolean; -}; - -type BbThreadResumeParams = ThreadResumeParams & { - persistExtendedHistory?: boolean; }; type BbThreadForkParams = { @@ -1899,10 +1894,12 @@ export function createCodexProviderAdapter( ...resolveCodexInstructionOverrides(command), model: command.options?.model ?? undefined, serviceTier: toCodexServiceTier(command.options?.serviceTier), + // bb reaps idle thread-scoped Codex processes and later resumes by + // provider thread id, so Codex must materialize a rollout on disk. + ephemeral: false, config: preparedGitRoots.config ?? undefined, // Codex only exposes raw Responses items as a thread/start opt-in. experimentalRawEvents: true, - persistExtendedHistory: false, ...(dynamicTools && dynamicTools.length > 0 ? { dynamicTools } : {}), @@ -1916,7 +1913,7 @@ export function createCodexProviderAdapter( case "thread/resume": { const dynamicTools = toCodexDynamicTools(command.dynamicTools); const preparedGitRoots = prepareWorkspaceWriteGitRoots({ command }); - const params: BbThreadResumeParams = { + const params: ThreadResumeParams = { threadId: command.providerThreadId, approvalPolicy: preparedGitRoots.permissionSettings.approvalPolicy, approvalsReviewer: @@ -1927,7 +1924,6 @@ export function createCodexProviderAdapter( model: command.options?.model ?? undefined, serviceTier: toCodexServiceTier(command.options?.serviceTier), config: preparedGitRoots.config ?? undefined, - persistExtendedHistory: false, ...(dynamicTools && dynamicTools.length > 0 ? { dynamicTools } : {}), diff --git a/packages/agent-runtime/src/runtime.command-contract.test.ts b/packages/agent-runtime/src/runtime.command-contract.test.ts index a2f5a1edb7..93af079d95 100644 --- a/packages/agent-runtime/src/runtime.command-contract.test.ts +++ b/packages/agent-runtime/src/runtime.command-contract.test.ts @@ -168,6 +168,92 @@ rl.on("line", (line) => { ); } +function writeArchivedCodexProviderScript( + scriptPath: string, + commandLogPath: string, +): void { + writeFileSync( + scriptPath, + ` +const fs = require("node:fs"); +const readline = require("node:readline"); +const commandLogPath = ${JSON.stringify(commandLogPath)}; +const archivedCommands = new Set(["thread/resume", "turn/start", "turn/steer"]); + +function send(message) { + process.stdout.write(JSON.stringify(message) + "\\n"); +} + +const rl = readline.createInterface({ input: process.stdin }); +rl.on("line", (line) => { + const message = JSON.parse(line); + fs.appendFileSync(commandLogPath, message.method + "\\n", "utf8"); + + if (message.method === "initialize") { + send({ jsonrpc: "2.0", id: message.id, result: {} }); + return; + } + if (message.method === "thread/start") { + send({ + jsonrpc: "2.0", + id: message.id, + result: { thread: { id: "codex-thread-archived" } }, + }); + return; + } + if (archivedCommands.delete(message.method)) { + send({ + jsonrpc: "2.0", + id: message.id, + error: { + code: -32000, + message: + "session codex-thread-archived is archived. Run codex unarchive codex-thread-archived to unarchive it first.", + }, + }); + return; + } + if (message.method === "thread/unarchive") { + send({ jsonrpc: "2.0", id: message.id, result: {} }); + return; + } + if (message.method === "thread/resume") { + send({ + jsonrpc: "2.0", + id: message.id, + result: { thread: { id: "codex-thread-archived" } }, + }); + return; + } + if (message.method === "turn/start" || message.method === "turn/steer") { + send({ jsonrpc: "2.0", id: message.id, result: {} }); + if (message.method === "turn/start") { + send({ + jsonrpc: "2.0", + method: "turn/started", + params: { + threadId: "codex-thread-archived", + turn: { + id: "turn-archived", + status: "inProgress", + error: null, + }, + }, + }); + } + return; + } + send({ + jsonrpc: "2.0", + id: message.id, + error: { code: -32601, message: "Method not found: " + message.method }, + }); +}); +`, + "utf8", + ); +} + function createRuntimeLinkedWorktreeFixture( args: CreateRuntimeLinkedWorktreeFixtureArgs, ): RuntimeLinkedWorktreeFixture { @@ -992,6 +1078,81 @@ rl.on("line", (line) => { await runtime.shutdown(); }); + it("unarchives Codex sessions before retrying archived turns and resumes", async () => { + const providerScriptPath = join(tmpDir, "codex-archived-provider.cjs"); + const commandLogPath = join(tmpDir, "codex-archived-commands.log"); + writeArchivedCodexProviderScript(providerScriptPath, commandLogPath); + const events: ThreadEvent[] = []; + const runtime = createAgentRuntimeWithAdapters({ + workspacePath: tmpDir, + onEvent: (event) => events.push(event), + onToolCall: async () => ({ + contentItems: [{ type: "inputText", text: "ok" }], + success: true, + }), + adapterFactory: () => + createCodexProviderAdapter({ + additionalWorkspaceWriteRoots: [], + processArgs: [providerScriptPath], + processCommand: "node", + }), + }); + + try { + await runtime.startThread({ + environmentId: "env-1", + projectId: "p1", + providerId: "codex", + threadId: "t-archived", + options: fullRuntimeOptions, + }); + await runtime.runTurn({ + clientRequestId: "creq_222222224u", + input: [promptTextInput({ text: "continue" })], + options: fullRuntimeOptions, + threadId: "t-archived", + }); + await waitForThreadTurnStarted({ + events, + providerId: "codex", + runtime, + threadId: "t-archived", + turnId: "turn-archived", + }); + await runtime.steerTurn({ + clientRequestId: "creq_222222224v", + expectedTurnId: "turn-archived", + input: [promptTextInput({ text: "keep going" })], + options: fullRuntimeOptions, + threadId: "t-archived", + }); + await runtime.resumeThread({ + environmentId: "env-1", + projectId: "p1", + providerId: "codex", + providerThreadId: "codex-thread-archived", + threadId: "t-archived", + options: fullRuntimeOptions, + }); + + expect(readFileSync(commandLogPath, "utf8").trim().split("\n")).toEqual([ + "initialize", + "thread/start", + "turn/start", + "thread/unarchive", + "turn/start", + "turn/steer", + "thread/unarchive", + "turn/steer", + "thread/resume", + "thread/unarchive", + "thread/resume", + ]); + } finally { + await runtime.shutdown(); + } + }); + it("rejects turn steer when providerThreadId cannot be resolved", async () => { const events: ThreadEvent[] = []; const activeTurnScriptPath = join(tmpDir, "active-turn-provider.cjs"); diff --git a/packages/agent-runtime/src/runtime.ts b/packages/agent-runtime/src/runtime.ts index 017a72fd96..c63405fb43 100644 --- a/packages/agent-runtime/src/runtime.ts +++ b/packages/agent-runtime/src/runtime.ts @@ -186,6 +186,8 @@ interface RequireProviderRequestPlanArgs { providerId: string; } +type TurnStartRuntimeCommand = Extract; + const CODEX_PROVIDER_ID = "codex"; const CODEX_THREAD_PROCESS_KEY_PREFIX = `${CODEX_PROVIDER_ID}\0thread:`; const THREAD_CREATION_REQUEST_TIMEOUT_MS = 2 * 60_000; @@ -193,6 +195,8 @@ const CODEX_ACCOUNT_RESTART_PROVIDER_ERROR_CATEGORIES = new Set(["rate-limit", "unauthorized"]); const CODEX_ACCOUNT_RESTART_PROVIDER_ERROR_TEXT_PATTERN = /\b(?:40[19]|429|auth(?:entication|orization)?|credits?|quota|rate[-\s]?limit(?:ed)?|unauthori[sz]ed|usage limit)\b/i; +const CODEX_ARCHIVED_SESSION_ERROR_PATTERN = + /\b(?:session|thread)\s+\S+\s+is archived\b/i; function resolveThreadStoragePath( args: ResolveThreadStoragePathArgs, @@ -741,6 +745,66 @@ function createAgentRuntimeInternal( await shutdownThreadScopedCodexProcessIfIdle(proc); } + function isCodexArchivedSessionError( + providerId: string, + error: unknown, + ): error is Error { + return ( + providerId === CODEX_PROVIDER_ID && + error instanceof Error && + CODEX_ARCHIVED_SESSION_ERROR_PATTERN.test(error.message) + ); + } + + async function runWithCodexArchivedSessionRecovery(args: { + operation: () => Promise; + providerId: string; + providerThreadId: string; + threadId: string; + }): Promise { + try { + return await args.operation(); + } catch (error) { + if (!isCodexArchivedSessionError(args.providerId, error)) { + throw error; + } + + options.onStderr?.( + `Codex session "${args.providerThreadId}" is archived; unarchiving before retrying thread "${args.threadId}".`, + ); + await archiveOrUnarchiveThread({ + commandType: "thread/unarchive", + providerId: args.providerId, + providerThreadId: args.providerThreadId, + threadId: args.threadId, + }); + return args.operation(); + } + } + + async function sendTurnStartCommand(args: { + command: TurnStartRuntimeCommand; + cmd: ProviderRequestCommandPlan; + proc: ProviderProcess; + }): Promise { + const { command, proc } = args; + const preparedTurnStart = proc.adapter.prepareTurnStart(command); + pendingTurnStartThreadIds.add(command.threadId); + markProviderSessionNotIdle(command.threadId); + try { + await sendCommand({ + proc, + message: args.cmd, + resultSchema: ignoredJsonRpcResultSchema, + }); + } catch (error) { + pendingTurnStartThreadIds.delete(command.threadId); + markHostedProviderSessionIdle(command.threadId); + preparedTurnStart?.rollback(); + throw error; + } + } + async function reconfigureThreadIfNeeded( args: ReconfigureThreadIfNeededArgs, ): Promise { @@ -795,10 +859,16 @@ function createAgentRuntimeInternal( }; const plan = proc.adapter.buildCommandPlan(adapterCommand); if (plan.kind === "request") { - const result = await sendCommand({ - proc, - message: plan, - resultSchema: threadIdentityResultSchema, + const result = await runWithCodexArchivedSessionRecovery({ + operation: () => + sendCommand({ + proc, + message: plan, + resultSchema: threadIdentityResultSchema, + }), + providerId: currentConfig.providerId, + providerThreadId: adapterCommand.providerThreadId, + threadId: args.threadId, }); const providerThreadId = resolveThreadIdentityResult({ result, @@ -1252,10 +1322,16 @@ function createAgentRuntimeInternal( } const cmd = plan; - const result = await sendCommand({ - proc, - message: cmd, - resultSchema: threadIdentityResultSchema, + const result = await runWithCodexArchivedSessionRecovery({ + operation: () => + sendCommand({ + proc, + message: cmd, + resultSchema: threadIdentityResultSchema, + }), + providerId, + providerThreadId: adapterCommand.providerThreadId, + threadId, }); const resolvedId = resolveThreadIdentityResult({ result, threadId }) ?? @@ -1327,22 +1403,17 @@ function createAgentRuntimeInternal( plan: proc.adapter.buildCommandPlan(adapterCommand), providerId: pid, }); - const preparedTurnStart = - proc.adapter.prepareTurnStart(adapterCommand); - pendingTurnStartThreadIds.add(threadId); - markProviderSessionNotIdle(threadId); - try { - await sendCommand({ - proc, - message: cmd, - resultSchema: ignoredJsonRpcResultSchema, - }); - } catch (error) { - pendingTurnStartThreadIds.delete(threadId); - markHostedProviderSessionIdle(threadId); - preparedTurnStart?.rollback(); - throw error; - } + await runWithCodexArchivedSessionRecovery({ + operation: () => + sendTurnStartCommand({ + cmd, + command: adapterCommand, + proc, + }), + providerId: pid, + providerThreadId: adapterCommand.providerThreadId, + threadId, + }); emitAcceptedCommandEvents({ command: adapterCommand, proc, @@ -1413,10 +1484,16 @@ function createAgentRuntimeInternal( plan: proc.adapter.buildCommandPlan(adapterCommand), providerId: pid, }); - await sendCommand({ - proc, - message: cmd, - resultSchema: ignoredJsonRpcResultSchema, + await runWithCodexArchivedSessionRecovery({ + operation: () => + sendCommand({ + proc, + message: cmd, + resultSchema: ignoredJsonRpcResultSchema, + }), + providerId: pid, + providerThreadId: adapterCommand.providerThreadId, + threadId, }); emitAcceptedCommandEvents({ command: adapterCommand, From 61f27decac3215db5b6704b68eaa0a4dce5cd588 Mon Sep 17 00:00:00 2001 From: Patrick Lee Date: Thu, 6 Aug 2026 17:13:04 -0400 Subject: [PATCH 2/4] test(runtime): reuse fake provider for archived sessions --- .../src/runtime.command-contract.test.ts | 147 ++------------- packages/agent-runtime/src/runtime.ts | 174 ++++++++---------- .../src/test/fake-provider-script.ts | 52 ++++++ 3 files changed, 144 insertions(+), 229 deletions(-) diff --git a/packages/agent-runtime/src/runtime.command-contract.test.ts b/packages/agent-runtime/src/runtime.command-contract.test.ts index 93af079d95..7f41d8a79c 100644 --- a/packages/agent-runtime/src/runtime.command-contract.test.ts +++ b/packages/agent-runtime/src/runtime.command-contract.test.ts @@ -168,92 +168,6 @@ rl.on("line", (line) => { ); } -function writeArchivedCodexProviderScript( - scriptPath: string, - commandLogPath: string, -): void { - writeFileSync( - scriptPath, - ` -const fs = require("node:fs"); -const readline = require("node:readline"); -const commandLogPath = ${JSON.stringify(commandLogPath)}; -const archivedCommands = new Set(["thread/resume", "turn/start", "turn/steer"]); - -function send(message) { - process.stdout.write(JSON.stringify(message) + "\\n"); -} - -const rl = readline.createInterface({ input: process.stdin }); -rl.on("line", (line) => { - const message = JSON.parse(line); - fs.appendFileSync(commandLogPath, message.method + "\\n", "utf8"); - - if (message.method === "initialize") { - send({ jsonrpc: "2.0", id: message.id, result: {} }); - return; - } - if (message.method === "thread/start") { - send({ - jsonrpc: "2.0", - id: message.id, - result: { thread: { id: "codex-thread-archived" } }, - }); - return; - } - if (archivedCommands.delete(message.method)) { - send({ - jsonrpc: "2.0", - id: message.id, - error: { - code: -32000, - message: - "session codex-thread-archived is archived. Run codex unarchive codex-thread-archived to unarchive it first.", - }, - }); - return; - } - if (message.method === "thread/unarchive") { - send({ jsonrpc: "2.0", id: message.id, result: {} }); - return; - } - if (message.method === "thread/resume") { - send({ - jsonrpc: "2.0", - id: message.id, - result: { thread: { id: "codex-thread-archived" } }, - }); - return; - } - if (message.method === "turn/start" || message.method === "turn/steer") { - send({ jsonrpc: "2.0", id: message.id, result: {} }); - if (message.method === "turn/start") { - send({ - jsonrpc: "2.0", - method: "turn/started", - params: { - threadId: "codex-thread-archived", - turn: { - id: "turn-archived", - status: "inProgress", - error: null, - }, - }, - }); - } - return; - } - send({ - jsonrpc: "2.0", - id: message.id, - error: { code: -32601, message: "Method not found: " + message.method }, - }); -}); -`, - "utf8", - ); -} - function createRuntimeLinkedWorktreeFixture( args: CreateRuntimeLinkedWorktreeFixtureArgs, ): RuntimeLinkedWorktreeFixture { @@ -1078,24 +992,25 @@ rl.on("line", (line) => { await runtime.shutdown(); }); - it("unarchives Codex sessions before retrying archived turns and resumes", async () => { - const providerScriptPath = join(tmpDir, "codex-archived-provider.cjs"); - const commandLogPath = join(tmpDir, "codex-archived-commands.log"); - writeArchivedCodexProviderScript(providerScriptPath, commandLogPath); - const events: ThreadEvent[] = []; + it("unarchives Codex sessions before retrying a turn", async () => { const runtime = createAgentRuntimeWithAdapters({ workspacePath: tmpDir, - onEvent: (event) => events.push(event), + onEvent: () => {}, onToolCall: async () => ({ contentItems: [{ type: "inputText", text: "ok" }], success: true, }), - adapterFactory: () => - createCodexProviderAdapter({ - additionalWorkspaceWriteRoots: [], - processArgs: [providerScriptPath], - processCommand: "node", - }), + adapterFactory: () => { + const adapter = createFakeAdapter(scriptPath); + return { + ...adapter, + id: "codex", + process: { + ...adapter.process, + args: [...adapter.process.args, "--archived-session"], + }, + }; + }, }); try { @@ -1112,42 +1027,6 @@ rl.on("line", (line) => { options: fullRuntimeOptions, threadId: "t-archived", }); - await waitForThreadTurnStarted({ - events, - providerId: "codex", - runtime, - threadId: "t-archived", - turnId: "turn-archived", - }); - await runtime.steerTurn({ - clientRequestId: "creq_222222224v", - expectedTurnId: "turn-archived", - input: [promptTextInput({ text: "keep going" })], - options: fullRuntimeOptions, - threadId: "t-archived", - }); - await runtime.resumeThread({ - environmentId: "env-1", - projectId: "p1", - providerId: "codex", - providerThreadId: "codex-thread-archived", - threadId: "t-archived", - options: fullRuntimeOptions, - }); - - expect(readFileSync(commandLogPath, "utf8").trim().split("\n")).toEqual([ - "initialize", - "thread/start", - "turn/start", - "thread/unarchive", - "turn/start", - "turn/steer", - "thread/unarchive", - "turn/steer", - "thread/resume", - "thread/unarchive", - "thread/resume", - ]); } finally { await runtime.shutdown(); } diff --git a/packages/agent-runtime/src/runtime.ts b/packages/agent-runtime/src/runtime.ts index c63405fb43..002f8de86a 100644 --- a/packages/agent-runtime/src/runtime.ts +++ b/packages/agent-runtime/src/runtime.ts @@ -115,6 +115,12 @@ interface ArchiveOrUnarchiveThreadArgs { threadId: string; } +interface CodexArchivedSessionRecoveryArgs { + providerId: string; + providerThreadId: string; + threadId: string; +} + interface AgentRuntimeInternalOptions extends AgentRuntimeOptions { adapterFactory?: ProviderAdapterFactory; } @@ -186,8 +192,6 @@ interface RequireProviderRequestPlanArgs { providerId: string; } -type TurnStartRuntimeCommand = Extract; - const CODEX_PROVIDER_ID = "codex"; const CODEX_THREAD_PROCESS_KEY_PREFIX = `${CODEX_PROVIDER_ID}\0thread:`; const THREAD_CREATION_REQUEST_TIMEOUT_MS = 2 * 60_000; @@ -338,20 +342,42 @@ function createAgentRuntimeInternal( }); } - function sendCommand(args: { + async function sendCommand(args: { proc: ProviderProcess; message: SendJsonRpcRequestArgs["message"]; resultSchema: SendJsonRpcRequestArgs["resultSchema"]; timeoutMs?: number; + recovery?: CodexArchivedSessionRecoveryArgs; }): Promise { - return sendJsonRpcRequest({ + const request = { child: args.proc.child, getNextId: () => nextRequestId++, message: args.message, pending: args.proc.pending, resultSchema: args.resultSchema, ...(args.timeoutMs !== undefined ? { timeoutMs: args.timeoutMs } : {}), - }); + }; + + try { + return await sendJsonRpcRequest(request); + } catch (error) { + const recovery = args.recovery; + if ( + !recovery || + !isCodexArchivedSessionError(recovery.providerId, error) + ) { + throw error; + } + + options.onStderr?.( + `Codex session "${recovery.providerThreadId}" is archived; unarchiving before retrying thread "${recovery.threadId}".`, + ); + await archiveOrUnarchiveThread({ + commandType: "thread/unarchive", + ...recovery, + }); + return sendJsonRpcRequest(request); + } } function resolveProviderForThread(threadId: string): string { @@ -756,55 +782,6 @@ function createAgentRuntimeInternal( ); } - async function runWithCodexArchivedSessionRecovery(args: { - operation: () => Promise; - providerId: string; - providerThreadId: string; - threadId: string; - }): Promise { - try { - return await args.operation(); - } catch (error) { - if (!isCodexArchivedSessionError(args.providerId, error)) { - throw error; - } - - options.onStderr?.( - `Codex session "${args.providerThreadId}" is archived; unarchiving before retrying thread "${args.threadId}".`, - ); - await archiveOrUnarchiveThread({ - commandType: "thread/unarchive", - providerId: args.providerId, - providerThreadId: args.providerThreadId, - threadId: args.threadId, - }); - return args.operation(); - } - } - - async function sendTurnStartCommand(args: { - command: TurnStartRuntimeCommand; - cmd: ProviderRequestCommandPlan; - proc: ProviderProcess; - }): Promise { - const { command, proc } = args; - const preparedTurnStart = proc.adapter.prepareTurnStart(command); - pendingTurnStartThreadIds.add(command.threadId); - markProviderSessionNotIdle(command.threadId); - try { - await sendCommand({ - proc, - message: args.cmd, - resultSchema: ignoredJsonRpcResultSchema, - }); - } catch (error) { - pendingTurnStartThreadIds.delete(command.threadId); - markHostedProviderSessionIdle(command.threadId); - preparedTurnStart?.rollback(); - throw error; - } - } - async function reconfigureThreadIfNeeded( args: ReconfigureThreadIfNeededArgs, ): Promise { @@ -859,16 +836,15 @@ function createAgentRuntimeInternal( }; const plan = proc.adapter.buildCommandPlan(adapterCommand); if (plan.kind === "request") { - const result = await runWithCodexArchivedSessionRecovery({ - operation: () => - sendCommand({ - proc, - message: plan, - resultSchema: threadIdentityResultSchema, - }), - providerId: currentConfig.providerId, - providerThreadId: adapterCommand.providerThreadId, - threadId: args.threadId, + const result = await sendCommand({ + proc, + message: plan, + resultSchema: threadIdentityResultSchema, + recovery: { + providerId: currentConfig.providerId, + providerThreadId: adapterCommand.providerThreadId, + threadId: args.threadId, + }, }); const providerThreadId = resolveThreadIdentityResult({ result, @@ -1322,16 +1298,15 @@ function createAgentRuntimeInternal( } const cmd = plan; - const result = await runWithCodexArchivedSessionRecovery({ - operation: () => - sendCommand({ - proc, - message: cmd, - resultSchema: threadIdentityResultSchema, - }), - providerId, - providerThreadId: adapterCommand.providerThreadId, - threadId, + const result = await sendCommand({ + proc, + message: cmd, + resultSchema: threadIdentityResultSchema, + recovery: { + providerId, + providerThreadId: adapterCommand.providerThreadId, + threadId, + }, }); const resolvedId = resolveThreadIdentityResult({ result, threadId }) ?? @@ -1403,17 +1378,27 @@ function createAgentRuntimeInternal( plan: proc.adapter.buildCommandPlan(adapterCommand), providerId: pid, }); - await runWithCodexArchivedSessionRecovery({ - operation: () => - sendTurnStartCommand({ - cmd, - command: adapterCommand, - proc, - }), - providerId: pid, - providerThreadId: adapterCommand.providerThreadId, - threadId, - }); + const preparedTurnStart = + proc.adapter.prepareTurnStart(adapterCommand); + pendingTurnStartThreadIds.add(threadId); + markProviderSessionNotIdle(threadId); + try { + await sendCommand({ + proc, + message: cmd, + resultSchema: ignoredJsonRpcResultSchema, + recovery: { + providerId: pid, + providerThreadId: adapterCommand.providerThreadId, + threadId, + }, + }); + } catch (error) { + pendingTurnStartThreadIds.delete(threadId); + markHostedProviderSessionIdle(threadId); + preparedTurnStart?.rollback(); + throw error; + } emitAcceptedCommandEvents({ command: adapterCommand, proc, @@ -1484,16 +1469,15 @@ function createAgentRuntimeInternal( plan: proc.adapter.buildCommandPlan(adapterCommand), providerId: pid, }); - await runWithCodexArchivedSessionRecovery({ - operation: () => - sendCommand({ - proc, - message: cmd, - resultSchema: ignoredJsonRpcResultSchema, - }), - providerId: pid, - providerThreadId: adapterCommand.providerThreadId, - threadId, + await sendCommand({ + proc, + message: cmd, + resultSchema: ignoredJsonRpcResultSchema, + recovery: { + providerId: pid, + providerThreadId: adapterCommand.providerThreadId, + threadId, + }, }); emitAcceptedCommandEvents({ command: adapterCommand, diff --git a/packages/agent-runtime/src/test/fake-provider-script.ts b/packages/agent-runtime/src/test/fake-provider-script.ts index 6ed41cb3d0..3dcf89f54b 100644 --- a/packages/agent-runtime/src/test/fake-provider-script.ts +++ b/packages/agent-runtime/src/test/fake-provider-script.ts @@ -66,6 +66,15 @@ const defaultModelList = { selectedOnlyModels: [], }; +// Test-only mode used by runtime command-contract coverage. +const simulateArchivedSession = process.argv.includes("--archived-session"); +const archivedSessionMethods = new Set([ + "thread/resume", + "turn/start", + "turn/steer", +]); +const unarchivedProviderThreadIds = new Set(); + function isJsonRecord(value: unknown): value is JsonRecord { return typeof value === "object" && value !== null; } @@ -84,6 +93,32 @@ function getParams(message: JsonRecord): JsonRecord { return isJsonRecord(message.params) ? message.params : {}; } +function rejectArchivedSession(message: JsonRecord): boolean { + const method = getString(message.method); + if (!simulateArchivedSession || !archivedSessionMethods.has(method)) { + return false; + } + + const params = getParams(message); + const providerThreadId = getString( + params.providerThreadId, + getString(params.threadId, "unknown"), + ); + if (unarchivedProviderThreadIds.has(providerThreadId)) { + return false; + } + + send({ + jsonrpc: "2.0", + id: getJsonRpcId(message.id) ?? 0, + error: { + code: -32000, + message: `session ${providerThreadId} is archived. Run codex unarchive ${providerThreadId} to unarchive it first.`, + }, + }); + return true; +} + function send(message: JsonRecord): void { process.stdout.write(`${JSON.stringify(message)}\n`); } @@ -523,6 +558,10 @@ function handleMessage(message: JsonRecord): void { return; } + if (rejectArchivedSession(message)) { + return; + } + if (method === "thread/start") { startOrResumeThread(message, "start"); return; @@ -533,6 +572,19 @@ function handleMessage(message: JsonRecord): void { return; } + if (method === "thread/unarchive") { + const params = getParams(message); + unarchivedProviderThreadIds.add( + getString(params.providerThreadId, getString(params.threadId, "unknown")), + ); + send({ + jsonrpc: "2.0", + id: getJsonRpcId(message.id) ?? 0, + result: { ok: true }, + }); + return; + } + if (method === "turn/start") { startTurn(message); return; From b069e6df8d3be2eb983939468db61523f194f5bb Mon Sep 17 00:00:00 2001 From: Sawyer Hood Date: Sun, 9 Aug 2026 17:38:35 +0000 Subject: [PATCH 3/4] fix(codex): cover fork and process restart in archived-session recovery Address review feedback on the archived-session recovery. - Recover `thread/fork` too. A fork reads its source session, so an archived source failed the same way a resume did. - Resolve the provider process again before the retry. Unarchiving can replace a dead process, and the retry wrote to the old child's stdin. - Keep the archived-session error when unarchiving fails. That message names the session and the CLI command that fixes it. - Drop the stale `persistExtendedHistory` field from `thread/fork`. Codex does not know the field, and start and resume already dropped it. - Correct the `ephemeral` comment. Codex already defaults to non-ephemeral, so the field pins the value; it does not change it. Add runtime tests for the resume, fork, and failed-unarchive paths, and assert the fork adapter sends no `persistExtendedHistory`. Co-Authored-By: Claude Opus 5 (1M context) --- .../agent-runtime/src/codex/adapter.test.ts | 1 + packages/agent-runtime/src/codex/adapter.ts | 6 +- .../src/runtime.command-contract.test.ts | 66 ++++++++++++++++++- packages/agent-runtime/src/runtime.ts | 37 +++++++++-- .../src/test/fake-provider-script.ts | 39 +++++++---- 5 files changed, 128 insertions(+), 21 deletions(-) diff --git a/packages/agent-runtime/src/codex/adapter.test.ts b/packages/agent-runtime/src/codex/adapter.test.ts index 99360a963b..cc654b42a2 100644 --- a/packages/agent-runtime/src/codex/adapter.test.ts +++ b/packages/agent-runtime/src/codex/adapter.test.ts @@ -1758,6 +1758,7 @@ describe("codex provider adapter", () => { ], }, }); + expect(JSON.stringify(cmd)).not.toContain("persistExtendedHistory"); }); it("buildCommand maps max reasoning level through to Codex", () => { diff --git a/packages/agent-runtime/src/codex/adapter.ts b/packages/agent-runtime/src/codex/adapter.ts index 5f20b14b0c..895a5df4bc 100644 --- a/packages/agent-runtime/src/codex/adapter.ts +++ b/packages/agent-runtime/src/codex/adapter.ts @@ -108,7 +108,6 @@ type BbThreadForkParams = { baseInstructions?: string | null; developerInstructions?: string | null; dynamicTools?: DynamicToolSpec[]; - persistExtendedHistory?: boolean; }; interface ToCodexPermissionSettingsArgs { @@ -1895,7 +1894,9 @@ export function createCodexProviderAdapter( model: command.options?.model ?? undefined, serviceTier: toCodexServiceTier(command.options?.serviceTier), // bb reaps idle thread-scoped Codex processes and later resumes by - // provider thread id, so Codex must materialize a rollout on disk. + // provider thread id, so the rollout must exist on disk. Codex + // already defaults to non-ephemeral; pin the value so a future + // default flip cannot silently break resume. ephemeral: false, config: preparedGitRoots.config ?? undefined, // Codex only exposes raw Responses items as a thread/start opt-in. @@ -1948,7 +1949,6 @@ export function createCodexProviderAdapter( model: command.options?.model ?? undefined, serviceTier: toCodexServiceTier(command.options?.serviceTier), config: preparedGitRoots.config ?? undefined, - persistExtendedHistory: false, ...(dynamicTools && dynamicTools.length > 0 ? { dynamicTools } : {}), diff --git a/packages/agent-runtime/src/runtime.command-contract.test.ts b/packages/agent-runtime/src/runtime.command-contract.test.ts index 7f41d8a79c..fecc19af18 100644 --- a/packages/agent-runtime/src/runtime.command-contract.test.ts +++ b/packages/agent-runtime/src/runtime.command-contract.test.ts @@ -992,8 +992,8 @@ rl.on("line", (line) => { await runtime.shutdown(); }); - it("unarchives Codex sessions before retrying a turn", async () => { - const runtime = createAgentRuntimeWithAdapters({ + function createArchivedSessionRuntime(extraArgs: string[] = []) { + return createAgentRuntimeWithAdapters({ workspacePath: tmpDir, onEvent: () => {}, onToolCall: async () => ({ @@ -1007,11 +1007,15 @@ rl.on("line", (line) => { id: "codex", process: { ...adapter.process, - args: [...adapter.process.args, "--archived-session"], + args: [...adapter.process.args, "--archived-session", ...extraArgs], }, }; }, }); + } + + it("unarchives Codex sessions before retrying a turn", async () => { + const runtime = createArchivedSessionRuntime(); try { await runtime.startThread({ @@ -1032,6 +1036,62 @@ rl.on("line", (line) => { } }); + // The fake keys its archived set on the exact provider thread id it was + // asked to unarchive, so a call that succeeds proves bb unarchived the + // right session before it retried. + it("unarchives Codex sessions before retrying a resume", async () => { + const runtime = createArchivedSessionRuntime(); + + try { + await runtime.resumeThread({ + environmentId: "env-1", + projectId: "p1", + providerId: "codex", + providerThreadId: "prov-archived-resume", + threadId: "t-archived-resume", + options: fullRuntimeOptions, + }); + } finally { + await runtime.shutdown(); + } + }); + + it("unarchives an archived Codex source session before retrying a fork", async () => { + const runtime = createArchivedSessionRuntime(); + + try { + await runtime.startThread({ + environmentId: "env-1", + fork: { sourceProviderThreadId: "prov-archived-source" }, + projectId: "p1", + providerId: "codex", + threadId: "t-archived-fork", + options: fullRuntimeOptions, + }); + } finally { + await runtime.shutdown(); + } + }); + + it("reports the archived-session error when unarchiving fails", async () => { + const runtime = createArchivedSessionRuntime(["--unarchive-fails"]); + + try { + await expect( + runtime.resumeThread({ + environmentId: "env-1", + projectId: "p1", + providerId: "codex", + providerThreadId: "prov-unarchive-fails", + threadId: "t-unarchive-fails", + options: fullRuntimeOptions, + }), + ).rejects.toThrow(/is archived/); + } finally { + await runtime.shutdown(); + } + }); + it("rejects turn steer when providerThreadId cannot be resolved", async () => { const events: ThreadEvent[] = []; const activeTurnScriptPath = join(tmpDir, "active-turn-provider.cjs"); diff --git a/packages/agent-runtime/src/runtime.ts b/packages/agent-runtime/src/runtime.ts index 002f8de86a..2edc0ba3cc 100644 --- a/packages/agent-runtime/src/runtime.ts +++ b/packages/agent-runtime/src/runtime.ts @@ -372,11 +372,29 @@ function createAgentRuntimeInternal( options.onStderr?.( `Codex session "${recovery.providerThreadId}" is archived; unarchiving before retrying thread "${recovery.threadId}".`, ); - await archiveOrUnarchiveThread({ - commandType: "thread/unarchive", - ...recovery, + try { + await archiveOrUnarchiveThread({ + commandType: "thread/unarchive", + ...recovery, + }); + } catch (unarchiveError) { + // The archived-session error names the session and the CLI command + // that fixes it, so keep it as the reported failure and attach the + // recovery failure as the cause. + throw new Error(error.message, { cause: unarchiveError }); + } + + // Unarchiving can replace a dead provider process, so resolve the + // process again instead of writing to the captured child's stdin. + const retryProc = requireProviderProcess({ + processKey: args.proc.processKey, + providerId: args.proc.providerId, + }); + return sendJsonRpcRequest({ + ...request, + child: retryProc.child, + pending: retryProc.pending, }); - return sendJsonRpcRequest(request); } } @@ -1151,6 +1169,17 @@ function createAgentRuntimeInternal( message: cmd, resultSchema: threadIdentityResultSchema, timeoutMs: THREAD_CREATION_REQUEST_TIMEOUT_MS, + // A fork reads the source session, so an archived source fails the + // same way a resume does. A plain start has no session to unarchive. + ...(fork + ? { + recovery: { + providerId, + providerThreadId: fork.sourceProviderThreadId, + threadId, + }, + } + : {}), }); const providerThreadId = resolveThreadIdentityResult({ result, diff --git a/packages/agent-runtime/src/test/fake-provider-script.ts b/packages/agent-runtime/src/test/fake-provider-script.ts index 3dcf89f54b..cbb3be4785 100644 --- a/packages/agent-runtime/src/test/fake-provider-script.ts +++ b/packages/agent-runtime/src/test/fake-provider-script.ts @@ -68,7 +68,9 @@ const defaultModelList = { // Test-only mode used by runtime command-contract coverage. const simulateArchivedSession = process.argv.includes("--archived-session"); +const failUnarchive = process.argv.includes("--unarchive-fails"); const archivedSessionMethods = new Set([ + "thread/fork", "thread/resume", "turn/start", "turn/steer", @@ -93,17 +95,22 @@ function getParams(message: JsonRecord): JsonRecord { return isJsonRecord(message.params) ? message.params : {}; } +// A fork reads its source session, so the source id is the one that can be +// archived. Every other method acts on the thread's own provider session. +function archivedSessionKey(params: JsonRecord): string { + return getString( + params.sourceProviderThreadId, + getString(params.providerThreadId, getString(params.threadId, "unknown")), + ); +} + function rejectArchivedSession(message: JsonRecord): boolean { const method = getString(message.method); if (!simulateArchivedSession || !archivedSessionMethods.has(method)) { return false; } - const params = getParams(message); - const providerThreadId = getString( - params.providerThreadId, - getString(params.threadId, "unknown"), - ); + const providerThreadId = archivedSessionKey(getParams(message)); if (unarchivedProviderThreadIds.has(providerThreadId)) { return false; } @@ -421,7 +428,7 @@ function startTurn(message: JsonRecord): void { function startOrResumeThread( message: JsonRecord, - mode: "resume" | "start", + mode: "fork" | "resume" | "start", ): void { const params = getParams(message); const threadId = getString(params.threadId, "unknown"); @@ -444,7 +451,7 @@ function startOrResumeThread( result: { providerThreadId }, }); - if (mode === "start") { + if (mode === "start" || mode === "fork") { send({ jsonrpc: "2.0", method: "thread/identity", @@ -572,11 +579,21 @@ function handleMessage(message: JsonRecord): void { return; } + if (method === "thread/fork") { + startOrResumeThread(message, "fork"); + return; + } + if (method === "thread/unarchive") { - const params = getParams(message); - unarchivedProviderThreadIds.add( - getString(params.providerThreadId, getString(params.threadId, "unknown")), - ); + if (failUnarchive) { + send({ + jsonrpc: "2.0", + id: getJsonRpcId(message.id) ?? 0, + error: { code: -32000, message: "unarchive is unavailable" }, + }); + return; + } + unarchivedProviderThreadIds.add(archivedSessionKey(getParams(message))); send({ jsonrpc: "2.0", id: getJsonRpcId(message.id) ?? 0, From 0fc1ddf48833fb07e89b9253807887686d57ad56 Mon Sep 17 00:00:00 2001 From: Sawyer Hood Date: Sun, 9 Aug 2026 18:02:38 +0000 Subject: [PATCH 4/4] fix(codex): keep the archived-session error when recovery cannot run Address the SlopCop finding about a provider exit during recovery. A provider that exits while bb unarchives cannot be unarchived or retried. Report the archived-session error in that case, and attach the recovery failure as the cause. That message names the session and the CLI command that fixes it. A process-level error such as `Provider "codex" has exited` tells the user nothing actionable. Add a test where the fake provider reports the archived error and then exits. Co-Authored-By: Claude Opus 5 (1M context) --- .../src/runtime.command-contract.test.ts | 23 +++++++++++++++++++ packages/agent-runtime/src/runtime.ts | 21 +++++++++-------- .../src/test/fake-provider-script.ts | 6 +++++ 3 files changed, 40 insertions(+), 10 deletions(-) diff --git a/packages/agent-runtime/src/runtime.command-contract.test.ts b/packages/agent-runtime/src/runtime.command-contract.test.ts index fecc19af18..2caa576e68 100644 --- a/packages/agent-runtime/src/runtime.command-contract.test.ts +++ b/packages/agent-runtime/src/runtime.command-contract.test.ts @@ -1092,6 +1092,29 @@ rl.on("line", (line) => { } }); + // A provider that dies while bb recovers cannot be unarchived or retried. + // The caller must still get the archived-session error, because it names the + // session and the CLI command that fixes it. A process-level error such as + // `Provider "codex" has exited` tells the user nothing actionable. + it("keeps the archived-session error when the provider exits mid-recovery", async () => { + const runtime = createArchivedSessionRuntime(["--exit-after-archived"]); + + try { + await expect( + runtime.resumeThread({ + environmentId: "env-1", + projectId: "p1", + providerId: "codex", + providerThreadId: "prov-exit-recovery", + threadId: "t-exit-recovery", + options: fullRuntimeOptions, + }), + ).rejects.toThrow(/session prov-exit-recovery is archived/); + } finally { + await runtime.shutdown(); + } + }); + it("rejects turn steer when providerThreadId cannot be resolved", async () => { const events: ThreadEvent[] = []; const activeTurnScriptPath = join(tmpDir, "active-turn-provider.cjs"); diff --git a/packages/agent-runtime/src/runtime.ts b/packages/agent-runtime/src/runtime.ts index 2edc0ba3cc..a925322043 100644 --- a/packages/agent-runtime/src/runtime.ts +++ b/packages/agent-runtime/src/runtime.ts @@ -372,24 +372,25 @@ function createAgentRuntimeInternal( options.onStderr?.( `Codex session "${recovery.providerThreadId}" is archived; unarchiving before retrying thread "${recovery.threadId}".`, ); + let retryProc: ProviderProcess; try { await archiveOrUnarchiveThread({ commandType: "thread/unarchive", ...recovery, }); - } catch (unarchiveError) { + // Unarchiving can replace an exited provider process, so resolve the + // process again instead of writing to the captured child's stdin. + retryProc = requireProviderProcess({ + processKey: args.proc.processKey, + providerId: args.proc.providerId, + }); + } catch (recoveryError) { // The archived-session error names the session and the CLI command - // that fixes it, so keep it as the reported failure and attach the - // recovery failure as the cause. - throw new Error(error.message, { cause: unarchiveError }); + // that fixes it, so keep it as the reported failure whenever the + // recovery itself could not run. + throw new Error(error.message, { cause: recoveryError }); } - // Unarchiving can replace a dead provider process, so resolve the - // process again instead of writing to the captured child's stdin. - const retryProc = requireProviderProcess({ - processKey: args.proc.processKey, - providerId: args.proc.providerId, - }); return sendJsonRpcRequest({ ...request, child: retryProc.child, diff --git a/packages/agent-runtime/src/test/fake-provider-script.ts b/packages/agent-runtime/src/test/fake-provider-script.ts index cbb3be4785..adbce8b793 100644 --- a/packages/agent-runtime/src/test/fake-provider-script.ts +++ b/packages/agent-runtime/src/test/fake-provider-script.ts @@ -69,6 +69,9 @@ const defaultModelList = { // Test-only mode used by runtime command-contract coverage. const simulateArchivedSession = process.argv.includes("--archived-session"); const failUnarchive = process.argv.includes("--unarchive-fails"); +// Exit right after reporting the archived error, so recovery has to work +// against a replacement process. +const exitAfterArchivedError = process.argv.includes("--exit-after-archived"); const archivedSessionMethods = new Set([ "thread/fork", "thread/resume", @@ -123,6 +126,9 @@ function rejectArchivedSession(message: JsonRecord): boolean { message: `session ${providerThreadId} is archived. Run codex unarchive ${providerThreadId} to unarchive it first.`, }, }); + if (exitAfterArchivedError) { + process.exit(0); + } return true; }