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..7f41d8a79c 100644 --- a/packages/agent-runtime/src/runtime.command-contract.test.ts +++ b/packages/agent-runtime/src/runtime.command-contract.test.ts @@ -992,6 +992,46 @@ rl.on("line", (line) => { await runtime.shutdown(); }); + it("unarchives Codex sessions before retrying a turn", async () => { + const runtime = createAgentRuntimeWithAdapters({ + workspacePath: tmpDir, + onEvent: () => {}, + onToolCall: async () => ({ + contentItems: [{ type: "inputText", text: "ok" }], + success: true, + }), + adapterFactory: () => { + const adapter = createFakeAdapter(scriptPath); + return { + ...adapter, + id: "codex", + process: { + ...adapter.process, + args: [...adapter.process.args, "--archived-session"], + }, + }; + }, + }); + + 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", + }); + } 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..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; } @@ -193,6 +199,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, @@ -334,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 { @@ -741,6 +771,17 @@ 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 reconfigureThreadIfNeeded( args: ReconfigureThreadIfNeededArgs, ): Promise { @@ -799,6 +840,11 @@ function createAgentRuntimeInternal( proc, message: plan, resultSchema: threadIdentityResultSchema, + recovery: { + providerId: currentConfig.providerId, + providerThreadId: adapterCommand.providerThreadId, + threadId: args.threadId, + }, }); const providerThreadId = resolveThreadIdentityResult({ result, @@ -1256,6 +1302,11 @@ function createAgentRuntimeInternal( proc, message: cmd, resultSchema: threadIdentityResultSchema, + recovery: { + providerId, + providerThreadId: adapterCommand.providerThreadId, + threadId, + }, }); const resolvedId = resolveThreadIdentityResult({ result, threadId }) ?? @@ -1336,6 +1387,11 @@ function createAgentRuntimeInternal( proc, message: cmd, resultSchema: ignoredJsonRpcResultSchema, + recovery: { + providerId: pid, + providerThreadId: adapterCommand.providerThreadId, + threadId, + }, }); } catch (error) { pendingTurnStartThreadIds.delete(threadId); @@ -1417,6 +1473,11 @@ function createAgentRuntimeInternal( 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;