diff --git a/packages/agent-runtime/src/codex/adapter.test.ts b/packages/agent-runtime/src/codex/adapter.test.ts index b491d1fce6..cc654b42a2 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"); }); @@ -1756,6 +1758,7 @@ describe("codex provider adapter", () => { ], }, }); + expect(JSON.stringify(cmd)).not.toContain("persistExtendedHistory"); }); it("buildCommand maps max reasoning level through to Codex", () => { @@ -1871,6 +1874,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..895a5df4bc 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 = { @@ -113,7 +108,6 @@ type BbThreadForkParams = { baseInstructions?: string | null; developerInstructions?: string | null; dynamicTools?: DynamicToolSpec[]; - persistExtendedHistory?: boolean; }; interface ToCodexPermissionSettingsArgs { @@ -1899,10 +1893,14 @@ 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 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. experimentalRawEvents: true, - persistExtendedHistory: false, ...(dynamicTools && dynamicTools.length > 0 ? { dynamicTools } : {}), @@ -1916,7 +1914,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 +1925,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 } : {}), @@ -1952,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 a2f5a1edb7..2caa576e68 100644 --- a/packages/agent-runtime/src/runtime.command-contract.test.ts +++ b/packages/agent-runtime/src/runtime.command-contract.test.ts @@ -992,6 +992,129 @@ rl.on("line", (line) => { await runtime.shutdown(); }); + function createArchivedSessionRuntime(extraArgs: string[] = []) { + return 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", ...extraArgs], + }, + }; + }, + }); + } + + it("unarchives Codex sessions before retrying a turn", async () => { + const runtime = createArchivedSessionRuntime(); + + 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(); + } + }); + + // 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(); + } + }); + + // 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 017a72fd96..a925322043 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,61 @@ 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}".`, + ); + let retryProc: ProviderProcess; + try { + await archiveOrUnarchiveThread({ + commandType: "thread/unarchive", + ...recovery, + }); + // 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 whenever the + // recovery itself could not run. + throw new Error(error.message, { cause: recoveryError }); + } + + return sendJsonRpcRequest({ + ...request, + child: retryProc.child, + pending: retryProc.pending, + }); + } } function resolveProviderForThread(threadId: string): string { @@ -741,6 +790,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 +859,11 @@ function createAgentRuntimeInternal( proc, message: plan, resultSchema: threadIdentityResultSchema, + recovery: { + providerId: currentConfig.providerId, + providerThreadId: adapterCommand.providerThreadId, + threadId: args.threadId, + }, }); const providerThreadId = resolveThreadIdentityResult({ result, @@ -1105,6 +1170,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, @@ -1256,6 +1332,11 @@ function createAgentRuntimeInternal( proc, message: cmd, resultSchema: threadIdentityResultSchema, + recovery: { + providerId, + providerThreadId: adapterCommand.providerThreadId, + threadId, + }, }); const resolvedId = resolveThreadIdentityResult({ result, threadId }) ?? @@ -1336,6 +1417,11 @@ function createAgentRuntimeInternal( proc, message: cmd, resultSchema: ignoredJsonRpcResultSchema, + recovery: { + providerId: pid, + providerThreadId: adapterCommand.providerThreadId, + threadId, + }, }); } catch (error) { pendingTurnStartThreadIds.delete(threadId); @@ -1417,6 +1503,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..adbce8b793 100644 --- a/packages/agent-runtime/src/test/fake-provider-script.ts +++ b/packages/agent-runtime/src/test/fake-provider-script.ts @@ -66,6 +66,20 @@ const defaultModelList = { selectedOnlyModels: [], }; +// 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", + "turn/start", + "turn/steer", +]); +const unarchivedProviderThreadIds = new Set(); + function isJsonRecord(value: unknown): value is JsonRecord { return typeof value === "object" && value !== null; } @@ -84,6 +98,40 @@ 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 providerThreadId = archivedSessionKey(getParams(message)); + 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.`, + }, + }); + if (exitAfterArchivedError) { + process.exit(0); + } + return true; +} + function send(message: JsonRecord): void { process.stdout.write(`${JSON.stringify(message)}\n`); } @@ -386,7 +434,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"); @@ -409,7 +457,7 @@ function startOrResumeThread( result: { providerThreadId }, }); - if (mode === "start") { + if (mode === "start" || mode === "fork") { send({ jsonrpc: "2.0", method: "thread/identity", @@ -523,6 +571,10 @@ function handleMessage(message: JsonRecord): void { return; } + if (rejectArchivedSession(message)) { + return; + } + if (method === "thread/start") { startOrResumeThread(message, "start"); return; @@ -533,6 +585,29 @@ function handleMessage(message: JsonRecord): void { return; } + if (method === "thread/fork") { + startOrResumeThread(message, "fork"); + return; + } + + if (method === "thread/unarchive") { + 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, + result: { ok: true }, + }); + return; + } + if (method === "turn/start") { startTurn(message); return;