From 58b9c1902b4bb10ba815f5d2bcf5c18d9839cd0e Mon Sep 17 00:00:00 2001 From: Renan Spatin Date: Mon, 2 Nov 2026 09:15:03 -0300 Subject: [PATCH] fix(server): Codex usage-limit failures on a fresh session carry their reset time A Codex session that starts already limited never receives account/rateLimits/updated, so the usage-limit failure was published with resetAt null and auto-resume never armed. A later back-fill could not help: runs stop ingesting at a failed terminal. When the failure has no reset time and the snapshot cannot provide one, read account/rateLimits/read before publishing the terminal (forked, 3s timeout, falls back to null on error). Fixes #29277 Co-Authored-By: Claude Opus 5.5 --- .../Adapters/CodexAdapterV2.test.ts | 55 +++++++- .../Adapters/CodexAdapterV2.ts | 123 +++++++++++------- 2 files changed, 132 insertions(+), 46 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts index 2d14fa9518de6..c40a336808298 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts @@ -722,6 +722,23 @@ function makeCodexReplayTranscript(input: { }; } +function codexRateLimitsReadEntries( + response: { readonly result: unknown } | { readonly error: unknown }, +): Array { + return [ + { + type: "expect_outbound", + label: "account/rateLimits/read", + frame: { id: 4, method: "account/rateLimits/read", params: null }, + }, + { type: "emit_inbound", label: "account/rateLimits/read", frame: { id: 4, ...response } }, + ]; +} + +const CODEX_RATE_LIMITS_READ_FAILURE = { + error: { code: -32603, message: "Usage is unavailable." }, +} as const; + function withReplayRequestId( entry: CodexReplay.CodexAppServerReplayEntry, id: number, @@ -5256,6 +5273,12 @@ describe("CodexAdapterV2 post-settle continuation", () => { expectedClass: "usage_limit", }, { name: "retry", code: "usageLimitExceeded", notification: true, expectedClass: "usage_limit" }, + { + name: "read-reset", + code: "usageLimitExceeded", + notification: true, + expectedClass: "usage_limit", + }, { name: "exhausted-429", code: "responseTooManyFailedAttempts", @@ -5386,6 +5409,31 @@ describe("CodexAdapterV2 post-settle continuation", () => { }, }, }, + ...(scenario.expectedClass === "usage_limit" && scenario.name !== "deferred-reset" + ? codexRateLimitsReadEntries( + scenario.name === "read-reset" + ? { + result: { + rateLimits: { + limitId: "codex", + primary: { + usedPercent: 0, + windowDurationMins: 300, + resetsAt: 1999990000, + }, + secondary: { + usedPercent: 100, + windowDurationMins: 10080, + resetsAt: 2000100000, + }, + planType: "plus", + rateLimitReachedType: "rate_limit_reached", + }, + }, + } + : CODEX_RATE_LIMITS_READ_FAILURE, + ) + : []), ...(scenario.name === "late-reset" ? [snapshot] : []), ...(scenario.name === "deferred-reset" ? [ @@ -5446,7 +5494,9 @@ describe("CodexAdapterV2 post-settle continuation", () => { if (terminal?.status !== "failed") return; assert.equal(terminal.failure.class, scenario.expectedClass); assert.equal(terminal.threadDisposition, "reusable"); - if (scenario.name === "deferred-reset") assert.equal(terminal.failure.resetAt, resetAt); + if (scenario.name === "deferred-reset" || scenario.name === "read-reset") + assert.equal(terminal.failure.resetAt, resetAt); + if (scenario.name === "usage") assert.isNull(terminal.failure.resetAt); if (scenario.name === "matching-details") assert.equal(terminal.failure.message, "Detailed provider allowance explanation."); if (scenario.name === "late-reset") { @@ -5550,6 +5600,9 @@ describe("CodexAdapterV2 post-settle continuation", () => { }, }, }, + ...(expectedClass === "usage_limit" + ? codexRateLimitsReadEntries(CODEX_RATE_LIMITS_READ_FAILURE) + : []), ], }); const harness = yield* makeCodexReplayHarness(transcript); diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts index f81a3809b35fe..ca5d7bd160ce3 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -927,6 +927,7 @@ function describeCodexGoal(goal: OrchestrationV2ProviderGoal): string { // Codex starts the next goal turn milliseconds after the last one completes. const CODEX_GOAL_CONTINUATION_GRACE = "5 seconds"; +const CODEX_RATE_LIMITS_READ_TIMEOUT = "3 seconds"; const providerTurnsForThread = ( providerTurns: ReadonlyArray, @@ -4429,34 +4430,37 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( }).pipe(Effect.orDie), ); - yield* client.handleServerNotification("account/rateLimits/updated", (payload) => - Effect.gen(function* () { - const previous = yield* Ref.getAndUpdate(rateLimitSnapshot, (previous) => - mergeCodexRateLimits(previous, payload.rateLimits), + // Untraced: rate-limit notifications repeat on every token tick. + const applyRateLimits = Effect.fnUntraced(function* (rateLimits: CodexRateLimitSnapshot) { + const previous = yield* Ref.getAndUpdate(rateLimitSnapshot, (previous) => + mergeCodexRateLimits(previous, rateLimits), + ); + const resetAt = codexUsageLimitResetAt(yield* Ref.get(rateLimitSnapshot)); + for (const item of (yield* Ref.get(limitedTurnItems)).values()) { + if (resetAt === null || item.failure.resetAt != null) continue; + const updated = { + ...item, + updatedAt: yield* DateTime.now, + failure: { ...item.failure, resetAt }, + }; + yield* Ref.update(limitedTurnItems, (current) => + new Map(current).set(item.providerThreadId!, updated), ); - const resetAt = codexUsageLimitResetAt(yield* Ref.get(rateLimitSnapshot)); - for (const item of (yield* Ref.get(limitedTurnItems)).values()) { - if (resetAt === null || item.failure.resetAt != null) continue; - const updated = { - ...item, - updatedAt: yield* DateTime.now, - failure: { ...item.failure, resetAt }, - }; - yield* Ref.update(limitedTurnItems, (current) => - new Map(current).set(item.providerThreadId!, updated), - ); - yield* emitProviderEvent({ - type: "turn_item.updated", - driver: CODEX_PROVIDER, - turnItem: updated, - }); - } - const update = codexRateLimitsToUpdate(payload.rateLimits, previous); - if (update && adapterOptions.onUsageLimits) { - const checkedAt = DateTime.formatIso(yield* DateTime.now); - yield* adapterOptions.onUsageLimits({ ...update, checkedAt }); - } - }).pipe(Effect.orDie), + yield* emitProviderEvent({ + type: "turn_item.updated", + driver: CODEX_PROVIDER, + turnItem: updated, + }); + } + const update = codexRateLimitsToUpdate(rateLimits, previous); + if (update && adapterOptions.onUsageLimits) { + const checkedAt = DateTime.formatIso(yield* DateTime.now); + yield* adapterOptions.onUsageLimits({ ...update, checkedAt }); + } + }); + + yield* client.handleServerNotification("account/rateLimits/updated", (payload) => + applyRateLimits(payload.rateLimits).pipe(Effect.orDie), ); yield* client.handleServerNotification("thread/tokenUsage/updated", (payload) => @@ -5808,7 +5812,7 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( }, ); - const emitRootTerminal = Effect.fnUntraced(function* ( + const publishRootTerminal = Effect.fnUntraced(function* ( context: ActiveCodexTurnContext, event: CodexRootTerminalEvent, ) { @@ -5825,23 +5829,6 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( } : event; yield* emitProviderEvent(current); - const continued = continuedRuns.get(context.providerTurnId) ?? [context]; - if (current.status !== "completed") { - yield* Ref.update( - terminalizedNonCompletedNativeTurns, - (nativeTurns) => - new Set([...nativeTurns, ...continued.map((turn) => turn.nativeTurnId)]), - ); - } - for (const turn of continued) { - continuedRuns.delete(turn.providerTurnId); - } - const nativeThreadId = context.providerThread.nativeThreadRef?.nativeId; - const rootProviderThread = - nativeThreadId == null ? undefined : rootProviderThreads.get(nativeThreadId); - if (nativeThreadId != null && rootProviderThread !== undefined) { - rootProviderThreads.set(nativeThreadId, { ...rootProviderThread, status: "idle" }); - } if (current.status === "failed" && current.failure.class === "usage_limit") { const item = makeProviderFailureTurnItem({ driver: CODEX_PROVIDER, @@ -5860,6 +5847,52 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( } }); + const emitRootTerminal = Effect.fnUntraced(function* ( + context: ActiveCodexTurnContext, + event: CodexRootTerminalEvent, + ) { + // Runs stop ingesting at a failed terminal, so a later reset time is never stored. A session + // that starts limited gets no rateLimits/updated, so read them before publishing. Forked: + // notification handlers run on the reader fiber, which must stay free for the reply. + if ( + event.status === "failed" && + event.failure.class === "usage_limit" && + event.failure.resetAt == null && + codexUsageLimitResetAt(yield* Ref.get(rateLimitSnapshot)) === null + ) { + yield* client.request("account/rateLimits/read", null).pipe( + Effect.timeout(CODEX_RATE_LIMITS_READ_TIMEOUT), + Effect.flatMap((response) => + applyRateLimits(response.rateLimitsByLimitId?.codex ?? response.rateLimits), + ), + Effect.catch((cause) => + Effect.logDebug("orchestration-v2.codex-rate-limits-read-failed", { cause }), + ), + Effect.andThen(publishRootTerminal(context, event)), + Effect.forkIn(scope), + ); + } else { + yield* publishRootTerminal(context, event); + } + const continued = continuedRuns.get(context.providerTurnId) ?? [context]; + if (event.status !== "completed") { + yield* Ref.update( + terminalizedNonCompletedNativeTurns, + (nativeTurns) => + new Set([...nativeTurns, ...continued.map((turn) => turn.nativeTurnId)]), + ); + } + for (const turn of continued) { + continuedRuns.delete(turn.providerTurnId); + } + const nativeThreadId = context.providerThread.nativeThreadRef?.nativeId; + const rootProviderThread = + nativeThreadId == null ? undefined : rootProviderThreads.get(nativeThreadId); + if (nativeThreadId != null && rootProviderThread !== undefined) { + rootProviderThreads.set(nativeThreadId, { ...rootProviderThread, status: "idle" }); + } + }); + const emitOrDeferRootTerminal = Effect.fn("CodexAdapterV2.emitOrDeferRootTerminal")( function* (input: { readonly context: ActiveCodexTurnContext;