diff --git a/apps/mobile/src/features/settings/SettingsServerControlsRouteScreen.tsx b/apps/mobile/src/features/settings/SettingsServerControlsRouteScreen.tsx index 885f269251c8f..b6cf1020c7f50 100644 --- a/apps/mobile/src/features/settings/SettingsServerControlsRouteScreen.tsx +++ b/apps/mobile/src/features/settings/SettingsServerControlsRouteScreen.tsx @@ -390,6 +390,16 @@ function ServerSettingsDetail(props: { readonly page: SettingsPage }) { /> ))} + + write({ autoRetryCapacityErrors: value })} + /> + { onEvent: (event: ProviderAdapterV2Event) => Effect.Effect = () => Effect.void, onRequest: (method: string, params: unknown) => Effect.Effect = () => Effect.void, readChildMetadata?: Parameters[2], + autoRetryCapacityErrors = false, ) => Effect.gen(function* () { const fileSystem = yield* FileSystem.FileSystem; @@ -1912,6 +1913,7 @@ describe("CodexAdapterV2 post-settle continuation", () => { const adapter = CodexAdapterV2.makeCodexAdapterV2({ instanceId: CodexAdapterV2.CODEX_DEFAULT_INSTANCE_ID, settings: DEFAULT_CODEX_SETTINGS, + autoRetryCapacityErrors: Effect.succeed(autoRetryCapacityErrors), environment: {}, clientFactory, crypto: yield* Crypto.Crypto, @@ -3093,6 +3095,167 @@ describe("CodexAdapterV2 post-settle continuation", () => { ), ); + it.effect.each(["recovery", "exhaustion", "stop", "disabled", "unrelated", "parallel"] as const)( + "handles capacity retries: %s", + (mode) => + Effect.scoped( + Effect.gen(function* () { + const makeScenario = (suffix: string) => + Effect.gen(function* () { + const scenario = `codex-capacity-${mode}-${suffix}`; + const nativeThreadId = `native-${scenario}-thread`; + const nativeTurnId = `native-${scenario}-turn`; + const failures = mode === "exhaustion" ? 6 : 1; + const retryCount = + mode === "exhaustion" ? 5 : mode === "recovery" || mode === "parallel" ? 1 : 0; + const waits = yield* Effect.forEach([0, 1, 2, 3, 4], () => Deferred.make()); + const entries: Array = codexReplayPreamble({ + nativeThreadId, + nativeTurnId, + prompt: "Continue work.", + }); + const terminal = ( + id: string, + status: "failed" | "completed", + ): CodexReplay.CodexAppServerReplayEntry => ({ + type: "emit_inbound", + label: `terminal-${id}`, + frame: { + method: "turn/completed", + params: { + threadId: nativeThreadId, + turn: { + ...makeCodexReplayTurn({ id, status }), + error: + status === "completed" + ? null + : { + message: + mode === "unrelated" + ? "Please sign in." + : "Selected model is at capacity. Please try a different model.", + codexErrorInfo: mode === "unrelated" ? "unauthorized" : "other", + }, + }, + }, + }, + }); + entries.push(terminal(nativeTurnId, "failed")); + for (let index = 1; index <= retryCount; index++) { + const retryId = `${nativeTurnId}-${index}`; + entries.push( + ...codexReplayPreamble({ nativeThreadId, nativeTurnId: retryId, prompt: "" }) + .slice(5) + .map((entry) => { + const updated = + entry.type !== "runtime_exit" && + Predicate.isObject(entry.frame) && + "id" in entry.frame + ? withReplayRequestId(entry, 3 + index) + : entry; + if ( + updated.type !== "expect_outbound" || + !Predicate.isObject(updated.frame) || + !Predicate.isObject(updated.frame.params) + ) + return updated; + return { + ...updated, + frame: { ...updated.frame, params: { ...updated.frame.params, input: [] } }, + }; + }), + terminal(retryId, failures > 1 ? "failed" : "completed"), + ); + } + let startRequests = 0; + const harness = yield* makeCodexReplayHarness( + makeCodexReplayTranscript({ scenario, entries }), + (event) => + event.type === "turn_item.updated" && + event.turnItem.type === "error" && + event.turnItem.status === "running" && + event.turnItem.retry != null + ? Deferred.succeed(waits[event.turnItem.retry.attempt - 1]!, undefined) + : Effect.void, + (method) => + Effect.sync(() => { + if (method === "turn/start") startRequests++; + }), + undefined, + mode !== "disabled", + ); + yield* harness.runtime.startTurn( + makeCodexTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now: yield* DateTime.now, + attemptId: RunAttemptId.make(`attempt-${scenario}`), + text: "Continue work.", + }), + ); + return { harness, waits, startRequests: () => startRequests, retryCount }; + }); + const scenarios = yield* Effect.all( + mode === "parallel" ? [makeScenario("a"), makeScenario("b")] : [makeScenario("a")], + { concurrency: "unbounded" }, + ); + if (mode === "disabled" || mode === "unrelated") { + for (const { harness, startRequests } of scenarios) { + yield* harness.firstTerminal; + assert.equal(startRequests(), 1); + assert.equal(harness.terminalEvents()[0]?.status, "failed"); + } + return; + } + for (let attempt = 0; attempt < (mode === "exhaustion" ? 5 : 1); attempt++) { + for (const { harness, waits } of scenarios) { + const result = yield* Effect.race( + Deferred.await(waits[attempt]!).pipe(Effect.as("retry")), + harness.firstTerminal.pipe(Effect.as("terminal")), + ); + assert.equal(result, "retry"); + assert.equal(harness.terminalEvents().length, 0); + } + if (mode === "stop") { + const { harness } = scenarios[0]!; + const running = harness.events.find( + (event) => event.type === "provider_turn.updated", + ); + assert(running?.type === "provider_turn.updated"); + yield* harness.runtime.interruptTurn({ + providerThread: harness.providerThread, + providerTurnId: running.providerTurn.id, + }); + } + yield* TestClock.adjust(mode === "exhaustion" ? "380 seconds" : "38 seconds"); + } + for (const { harness, startRequests, retryCount } of scenarios) { + yield* harness.firstTerminal; + assert.equal(harness.terminalEvents().length, 1); + assert.equal( + harness.terminalEvents()[0]?.status, + mode === "exhaustion" ? "failed" : mode === "stop" ? "interrupted" : "completed", + ); + assert.equal(startRequests(), retryCount + 1); + const retries = harness.events.flatMap((event) => + event.type === "turn_item.updated" && + event.turnItem.type === "error" && + event.turnItem.retry != null && + event.turnItem.status === "running" + ? [event.turnItem.retry] + : [], + ); + for (const [index, retry] of retries.entries()) { + assert.equal(retry.attempt, index + 1); + const baseDelay = Math.min(30_000 * 2 ** index, 300_000); + assert.isAtLeast(retry.retryDelayMs!, baseDelay); + assert.isAtMost(retry.retryDelayMs!, baseDelay * 1.25); + } + } + }).pipe(Effect.provide(Layer.merge(IdAllocator.layer, NodeServices.layer))), + ), + ); + it.effect("resolves retryable app-server errors on resumed provider activity", () => Effect.scoped( Effect.gen(function* () { diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts index 19517c3491161..108dd8d5026bc 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -76,6 +76,7 @@ import * as Option from "effect/Option"; import * as Path from "effect/Path"; import * as Queue from "effect/Queue"; import * as Ref from "effect/Ref"; +import * as Random from "effect/Random"; import * as Schema from "effect/Schema"; import * as Scope from "effect/Scope"; import * as Semaphore from "effect/Semaphore"; @@ -1085,10 +1086,13 @@ function codexErrorInfoCode(value: unknown): string | null { } interface ActiveCodexTurnContext { - latestProviderFailure?: { - readonly nativeMessage: string; - readonly failure: OrchestrationV2ProviderFailure; - }; + capacityRetryAttempt?: number; + latestProviderFailure?: + | { + readonly nativeMessage: string; + readonly failure: OrchestrationV2ProviderFailure; + } + | undefined; readonly nativeStartReady?: Deferred.Deferred; readonly input: ProviderAdapterV2TurnInput; readonly projectionAppThread: OrchestrationV2AppThread; @@ -1534,7 +1538,10 @@ export type CodexAdapterV2DriverEnv = export const createCodexAdapterV2 = ( { instanceId, environment, enabled, config }: ProviderAdapterDriverCreateInput, - hooks: Pick = {}, + hooks: Pick< + CodexAdapterV2Options, + "onUsageLimits" | "resolveRuntime" | "autoRetryCapacityErrors" + > = {}, ) => Effect.gen(function* () { const clientFactory = yield* CodexAppServerClientFactory; @@ -1618,6 +1625,7 @@ const layer: Layer.Layer< export interface CodexAdapterV2Options { readonly instanceId: ProviderInstanceId; readonly settings: CodexSettings; + readonly autoRetryCapacityErrors?: Effect.Effect; readonly environment: NodeJS.ProcessEnv; readonly clientFactory: CodexAppServerClientFactoryShape; readonly onUsageLimits?: ServerProviderShape["applyUsageLimits"]; @@ -1777,6 +1785,8 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi context.nativeTurnId, ).hasSubagents = true; }; + const capacityRetryWaiting = new Set(); + const pendingCapacityRetries = new Map(); const pendingRootTurns = yield* Ref.make(new Map()); const turnWaiters = yield* Ref.make(new Map>()); const subagentThreads = yield* Ref.make(new Map()); @@ -1918,27 +1928,51 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi driver: CODEX_PROVIDER, nativeTurnId: input.nativeTurnId, }); - const context: ActiveCodexTurnContext = { - ...(input.waitForNativeStart - ? { nativeStartReady: yield* Deferred.make() } - : {}), - input: input.turnInput, - projectionAppThread: input.turnInput.appThread, - projectionThreadId: input.turnInput.threadId, - projectionRunId: input.turnInput.runId, - nativeTurnId: input.nativeTurnId, - providerThread: input.turnInput.providerThread, - providerTurnId, - providerTurnOrdinal: input.turnInput.providerTurnOrdinal, - providerNodeId: input.turnInput.rootNodeId, - providerNodeKind: "root_turn", - providerNodeStartedAt: input.startedAt, - itemParentNodeId: input.turnInput.rootNodeId, - rootNodeId: input.turnInput.rootNodeId, - subagent: null, - startedAt: input.startedAt, - itemPositions: new Map(), - }; + const retryContext = pendingCapacityRetries.get( + input.turnInput.providerThread.nativeThreadRef?.nativeId ?? "", + ); + const context: ActiveCodexTurnContext = + retryContext === undefined + ? { + ...(input.waitForNativeStart + ? { nativeStartReady: yield* Deferred.make() } + : {}), + input: input.turnInput, + projectionAppThread: input.turnInput.appThread, + projectionThreadId: input.turnInput.threadId, + projectionRunId: input.turnInput.runId, + nativeTurnId: input.nativeTurnId, + providerThread: input.turnInput.providerThread, + providerTurnId, + providerTurnOrdinal: input.turnInput.providerTurnOrdinal, + providerNodeId: input.turnInput.rootNodeId, + providerNodeKind: "root_turn", + providerNodeStartedAt: input.startedAt, + itemParentNodeId: input.turnInput.rootNodeId, + rootNodeId: input.turnInput.rootNodeId, + subagent: null, + startedAt: input.startedAt, + itemPositions: new Map(), + } + : { + ...retryContext, + nativeTurnId: input.nativeTurnId, + latestProviderFailure: undefined, + }; + if (retryContext !== undefined) { + pendingCapacityRetries.delete( + input.turnInput.providerThread.nativeThreadRef?.nativeId ?? "", + ); + capacityRetryWaiting.delete(retryContext.nativeTurnId); + yield* Ref.update(activeTurns, (turns) => { + const next = new Map(turns); + next.delete(retryContext.nativeTurnId); + return next; + }); + yield* Ref.update(terminalizedNonCompletedNativeTurns, (turns) => + new Set(turns).add(retryContext.nativeTurnId), + ); + } yield* Ref.update(limitedTurnItems, (current) => { const next = new Map(current); next.delete(context.providerThread.id); @@ -1955,7 +1989,7 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi driver: CODEX_PROVIDER, threadId: input.turnInput.threadId, providerTurn: { - id: providerTurnId, + id: context.providerTurnId, providerThreadId: input.turnInput.providerThread.id, nodeId: input.turnInput.rootNodeId, runAttemptId: input.turnInput.attemptId, @@ -1966,7 +2000,7 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi }, ordinal: input.turnInput.providerTurnOrdinal, status: "running", - startedAt: input.startedAt, + startedAt: context.startedAt, completedAt: null, }, }); @@ -5410,6 +5444,7 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi if (current !== input.context) { return false; } + capacityRetryWaiting.delete(input.nativeTurnId); const providerRetry = yield* Ref.modify(providerRetries, (current) => { const retry = current.get(input.context.providerTurnId); if (retry === undefined) { @@ -5657,12 +5692,125 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi if (context === undefined) { return; } + if (capacityRetryWaiting.has(context.nativeTurnId)) return; const nativeStatus = mapCodexTurnStatus(payload.turn.status); const status = nativeStatus === "completed" && (yield* Ref.get(interruptingNativeTurns)).has(payload.turn.id) ? "interrupted" : nativeStatus; + const message = + payload.turn.error?.message ?? context.latestProviderFailure?.nativeMessage ?? ""; + const code = + codexErrorInfoCode(payload.turn.error?.codexErrorInfo) ?? + context.latestProviderFailure?.failure.code; + const attempt = (context.capacityRetryAttempt ?? 0) + 1; + const capacityFailure = + code === "serverOverloaded" || + ((code == null || code === "other") && + /\b(?:model|server) (?:is |is currently )?at capacity\b/i.test(message)); + if ( + status === "failed" && + context.subagent === null && + capacityFailure && + attempt <= 5 && + !capacityRetryWaiting.has(context.nativeTurnId) && + (yield* adapterOptions.autoRetryCapacityErrors ?? Effect.succeed(false)) + ) { + context.capacityRetryAttempt = attempt; + capacityRetryWaiting.add(context.nativeTurnId); + const now = yield* DateTime.now; + yield* terminalizeRunningCommandItems(context, context.nativeTurnId, "failed", now); + yield* terminalizeRunningDynamicTools( + context, + context.nativeTurnId, + "failed", + now, + true, + ); + // Independent jitter keeps a capacity outage from waking every thread together. + const retryDelayMs = Math.round( + Math.min(30_000 * 2 ** (attempt - 1), 300_000) * (1 + (yield* Random.next) * 0.25), + ); + const failure = makeProviderFailure({ + message, + code, + class: "provider_error", + retryable: true, + }); + const itemOrdinal = yield* resolveItemOrdinal( + context, + `terminal-failure:${context.providerTurnId}`, + ); + const retry = { attempt, maxAttempts: 5, retryDelayMs }; + yield* Ref.update(providerRetries, (current) => + new Map(current).set(context.providerTurnId, { + nativeMessage: message, + failure, + retry, + startedAt: now, + itemOrdinal, + }), + ); + yield* emitProviderEvent({ + type: "turn_item.updated", + driver: CODEX_PROVIDER, + turnItem: makeProviderRetryTurnItem({ + idAllocator, + driver: CODEX_PROVIDER, + threadId: context.projectionThreadId, + runId: context.projectionRunId, + nodeId: context.providerNodeId, + providerThreadId: context.providerThread.id, + providerTurnId: context.providerTurnId, + itemOrdinal, + failure, + retry, + status: "running", + startedAt: now, + updatedAt: now, + }), + }); + yield* Effect.sleep(retryDelayMs).pipe( + Effect.andThen( + turnTerminalizationPermit.withPermits(1)( + Effect.gen(function* () { + if ( + !capacityRetryWaiting.has(context.nativeTurnId) || + (yield* Ref.get(activeTurns)).get(context.nativeTurnId) !== context + ) + return; + if (!(yield* adapterOptions.autoRetryCapacityErrors ?? Effect.succeed(false))) + return false; + const threadId = yield* getNativeThreadId(context.providerThread); + pendingCapacityRetries.set(threadId, context); + // The original prompt and any completed work are already in native history. + yield* startNativeTurn(context.input, []).pipe( + Effect.ensuring(Effect.sync(() => pendingCapacityRetries.delete(threadId))), + ); + return true; + }), + ), + ), + Effect.catchCause(() => Effect.succeed(false)), + Effect.flatMap((started) => + started === false + ? Effect.flatMap(DateTime.now, (completedAt) => + finalizeCodexTurn({ + context, + nativeTurnId: context.nativeTurnId, + status: "failed", + completedAt, + failureMessage: message, + failureCode: code ?? null, + }), + ) + : Effect.void, + ), + Effect.forkIn(scope), + ); + return; + } yield* finalizeCodexTurn({ context, nativeTurnId: payload.turn.id, @@ -6325,6 +6473,16 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi (candidate) => candidate.providerThread.id === turnInput.providerThread.id, ) : undefined); + const waitingForCapacity = + activeTurn !== undefined && capacityRetryWaiting.has(activeTurn.nativeTurnId); + if (activeTurn !== undefined && waitingForCapacity) { + yield* finalizeCodexTurn({ + context: activeTurn, + nativeTurnId: activeTurn.nativeTurnId, + status: "interrupted", + completedAt: yield* DateTime.now, + }); + } // A goal run spans several root turns, and work an earlier one // started can outlive it. const lineageRoots = Array.from( @@ -6335,7 +6493,7 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi (root) => context === root || isDescendantCodexTurn(context, root), ); const interruptTargetContexts = [ - ...(activeTurn === undefined ? [] : [activeTurn]), + ...(activeTurn === undefined || waitingForCapacity ? [] : [activeTurn]), ...activeTurnContexts.filter( (candidate) => candidate !== activeTurn && inLineage(candidate), ), @@ -6348,6 +6506,7 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi ), ]; if (interruptTargetContexts.length === 0) { + if (waitingForCapacity) return; // Stop on a settled turn this process retains nothing for // (released, restarted, or every command already reported). if (turnInput.requestRuntimeRestart === true) return; diff --git a/apps/server/src/provider/Drivers/CodexDriver.ts b/apps/server/src/provider/Drivers/CodexDriver.ts index 0b958e2978a5f..d7272eabcfed2 100644 --- a/apps/server/src/provider/Drivers/CodexDriver.ts +++ b/apps/server/src/provider/Drivers/CodexDriver.ts @@ -200,7 +200,13 @@ export const CodexDriver: ProviderDriver = { enabled, config, }, - { onUsageLimits: (update) => snapshot.applyUsageLimits(update) }, + { + onUsageLimits: (update) => snapshot.applyUsageLimits(update), + autoRetryCapacityErrors: serverSettings.getSettings.pipe( + Effect.map((settings) => settings.autoRetryCapacityErrors), + Effect.orElseSucceed(() => false), + ), + }, ).pipe( Effect.mapError( (cause) => diff --git a/apps/server/src/provider/Drivers/CodexManagedProvider.ts b/apps/server/src/provider/Drivers/CodexManagedProvider.ts index b75e1edd5414e..95a47db50e3c0 100644 --- a/apps/server/src/provider/Drivers/CodexManagedProvider.ts +++ b/apps/server/src/provider/Drivers/CodexManagedProvider.ts @@ -220,6 +220,10 @@ export const makeManagedCodexProvider = Effect.fn("makeManagedCodexProvider")(fu const orchestrationAdapter = yield* createCodexAdapterV2(input, { onUsageLimits: (update) => snapshot.applyUsageLimits(update), resolveRuntime: runtime.resolve, + autoRetryCapacityErrors: settings.getSettings.pipe( + Effect.map((settings) => settings.autoRetryCapacityErrors), + Effect.orElseSucceed(() => false), + ), }).pipe( Effect.mapError( (cause) => diff --git a/apps/web/src/components/settings/SettingsPanels.tsx b/apps/web/src/components/settings/SettingsPanels.tsx index a805b7b129b56..9d411f7d7a32c 100644 --- a/apps/web/src/components/settings/SettingsPanels.tsx +++ b/apps/web/src/components/settings/SettingsPanels.tsx @@ -626,6 +626,9 @@ export function useSettingsRestore(onRestored?: () => void) { ...(settings.responseStreamingMode !== DEFAULT_UNIFIED_SETTINGS.responseStreamingMode ? ["Response streaming"] : []), + ...(settings.autoRetryCapacityErrors !== DEFAULT_UNIFIED_SETTINGS.autoRetryCapacityErrors + ? ["Auto-retry capacity errors"] + : []), ...(settings.enableProviderUpdateChecks !== DEFAULT_UNIFIED_SETTINGS.enableProviderUpdateChecks ? ["Provider update checks"] @@ -705,6 +708,7 @@ export function useSettingsRestore(onRestored?: () => void) { settings.panelAnimationDurationMs, settings.responseStreamingMode, settings.persistComposerContextStrip, + settings.autoRetryCapacityErrors, settings.enableProviderUpdateChecks, settings.continueThreadsAfterServerUpdate, settings.sidebarAutoSettleAfterDays, @@ -821,6 +825,7 @@ export function useSettingsRestore(onRestored?: () => void) { autoResumeLimitedThreads: DEFAULT_UNIFIED_SETTINGS.autoResumeLimitedThreads, snoozeLimitedThreads: DEFAULT_UNIFIED_SETTINGS.snoozeLimitedThreads, responseStreamingMode: DEFAULT_UNIFIED_SETTINGS.responseStreamingMode, + autoRetryCapacityErrors: DEFAULT_UNIFIED_SETTINGS.autoRetryCapacityErrors, enableProviderUpdateChecks: DEFAULT_UNIFIED_SETTINGS.enableProviderUpdateChecks, continueThreadsAfterServerUpdate: DEFAULT_UNIFIED_SETTINGS.continueThreadsAfterServerUpdate, backgroundActivity: DEFAULT_UNIFIED_SETTINGS.backgroundActivity, @@ -2533,6 +2538,30 @@ export function GeneralSettingsPanel() { } /> + updateSettings({ autoRetryCapacityErrors: false })} + /> + ) : null + } + control={ + + updateSettings({ autoRetryCapacityErrors: Boolean(checked) }) + } + aria-label="Auto-retry capacity errors" + /> + } + />