From bdbed8f8067f7067457d7f587ab418d644bf0580 Mon Sep 17 00:00:00 2001 From: Patrick Lee Date: Thu, 6 Aug 2026 12:52:51 -0400 Subject: [PATCH 1/2] 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/2] 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;