From 9c0b1f29ca2f53991e4796360c27088cf15edf78 Mon Sep 17 00:00:00 2001 From: Lakshmi Tanmay Date: Mon, 12 Oct 2026 22:40:33 +0530 Subject: [PATCH 1/2] fix(server): Codex usage-limit stops keep their reset time A Codex turn refused while its last rate-limit update read 99%, or before the session saw any update, recorded no reset, so the composer hid Resume at reset and Snooze until reset. Treat a 99% window as exhausted when none reads 100%, ignore windows that have already renewed, and fall back to the instance's published limits when the session snapshot has no reset. Fixes #14169 Co-Authored-By: Claude Opus 5.5 --- .../Adapters/CodexAdapterV2.test.ts | 43 +++++++++- .../Adapters/CodexAdapterV2.ts | 32 +++---- .../src/provider/Drivers/CodexDriver.ts | 5 +- .../src/provider/codexUsageLimits.test.ts | 86 +++++++++++++++++-- apps/server/src/provider/codexUsageLimits.ts | 40 +++++++-- 5 files changed, 171 insertions(+), 35 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts index 84c78b91453c2..3c0392985bfff 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts @@ -1237,6 +1237,7 @@ describe("CodexAdapterV2 post-settle continuation", () => { onRequest: (method: string, params: unknown) => Effect.Effect = () => Effect.void, readChildMetadata?: Parameters[2], beforeEmitInbound?: CodexReplay.CodexAppServerReplayDriver["beforeEmitInbound"], + getUsageLimits?: CodexAdapterV2.CodexAdapterV2Options["getUsageLimits"], onUsageLimits?: CodexAdapterV2.CodexAdapterV2Options["onUsageLimits"], ) => Effect.gen(function* () { @@ -1306,6 +1307,7 @@ describe("CodexAdapterV2 post-settle continuation", () => { continuationRequests.push(request); }).pipe(Effect.andThen(changes.notify)), }, + ...(getUsageLimits ? { getUsageLimits } : {}), }); const threadId = ThreadId.make(`thread-${transcript.scenario}`); const runtime = yield* adapter.openSession({ @@ -5662,6 +5664,18 @@ describe("CodexAdapterV2 post-settle continuation", () => { willRetry: true, expectedClass: "transport_error", }, + { + name: "near-full-reset", + code: "usageLimitExceeded", + notification: true, + expectedClass: "usage_limit", + }, + { + name: "published-reset", + code: "usageLimitExceeded", + notification: false, + expectedClass: "usage_limit", + }, ] as const)("classifies Codex terminal failures from $name evidence", (scenario) => Effect.scoped( Effect.gen(function* () { @@ -5687,7 +5701,7 @@ describe("CodexAdapterV2 post-settle continuation", () => { scenario: `codex-limit-${scenario.name}`, entries: [ ...codexReplayPreamble({ nativeThreadId, nativeTurnId, prompt: "Continue." }), - ...("usedBefore" in scenario + ...("usedBefore" in scenario || scenario.name === "near-full-reset" ? [ { ...snapshot, @@ -5697,10 +5711,11 @@ describe("CodexAdapterV2 post-settle continuation", () => { rateLimits: { limitId: "codex", primary: { - usedPercent: scenario.usedBefore, + usedPercent: "usedBefore" in scenario ? scenario.usedBefore : 99, resetsAt: 2000100000, windowDurationMins: 300, }, + secondary: { usedPercent: 73, resetsAt: 2000500000, windowDurationMins: 10080 }, }, }, }, @@ -5841,6 +5856,22 @@ describe("CodexAdapterV2 post-settle continuation", () => { undefined, undefined, undefined, + // The session got no rate-limit notification; Usage → Limits still knows the reset. + scenario.name === "published-reset" + ? Effect.succeed({ + checkedAt: "2033-05-19T03:00:00.000Z", + windows: [ + { + id: "primary", + kind: "session", + label: "Session", + usedPercent: 100, + resetsAt: resetAt, + }, + { id: "secondary", kind: "weekly", label: "Weekly", usedPercent: 40 }, + ], + }) + : undefined, () => Deferred.succeed(snapshotReceipt, undefined), ); yield* harness.runtime.startTurn( @@ -5858,7 +5889,13 @@ 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 === "known-reset" || + scenario.name === "deferred-reset" || + scenario.name === "near-full-reset" || + scenario.name === "published-reset" + ) + assert.equal(terminal.failure.resetAt, resetAt); if (scenario.name === "matching-details") assert.equal(terminal.failure.message, "Detailed provider allowance explanation."); if ("resetAfter" in scenario) { diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts index 369608a289d2e..9175bca5b3c2f 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -25,6 +25,7 @@ import { import { codexRateLimitsToUpdate, mergeCodexRateLimits, + codexPublishedUsageLimitResetAt, codexUsageLimitResetAt, type CodexRateLimitSnapshot, } from "../../provider/codexUsageLimits.ts"; @@ -38,6 +39,7 @@ import { ProviderDriverKind, ProviderSetupError, type ServerProviderCompatibilityAdvisory, + type ServerProviderUsageLimits, } from "@t3tools/contracts"; import { SKILL_MENTION_PATTERN } from "@t3tools/shared/composerInlineTokens"; import { AgentScope } from "@t3tools/shared/AgentScope"; @@ -1166,15 +1168,6 @@ function withCodexManagedFailure( }; } -function failureResetAt( - failure: OrchestrationV2ProviderFailure, - snapshot: CodexRateLimitSnapshot | undefined, -) { - // Sharing caps have no reset timestamp in the ordinary Codex allowance snapshot. - return classifyCodexManagedError(failure.code)?.usageLimit - ? null - : codexUsageLimitResetAt(snapshot); -} interface ActiveCodexTurnContext { hookStopReason?: string; @@ -1646,7 +1639,7 @@ export type CodexAdapterV2DriverEnv = export const createCodexAdapterV2 = Effect.fn("CodexAdapterV2.createCodexAdapterV2")(function* ( { instanceId, environment, enabled, config }: ProviderAdapterDriverCreateInput, - hooks: Pick = {}, + hooks: Pick = {}, ) { const clientFactory = yield* CodexAppServerClientFactory; const continuationRequests = yield* ProviderContinuationRequests.ProviderContinuationRequests; @@ -1696,6 +1689,7 @@ export interface CodexAdapterV2Options { readonly environment: NodeJS.ProcessEnv; readonly clientFactory: CodexAppServerClientFactoryShape; readonly onUsageLimits?: ManagedServerProvider["applyUsageLimits"]; + readonly getUsageLimits?: Effect.Effect; readonly resolveRuntime?: Effect.Effect; readonly compatibility?: ( version: string, @@ -1868,6 +1862,15 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( Effect.forkScoped, ); const rateLimitSnapshot = yield* Ref.make(undefined); + const usageLimitResetAt = Effect.fnUntraced(function* (failure: OrchestrationV2ProviderFailure) { + // Sharing caps have no reset in the ordinary allowance snapshot. + if (classifyCodexManagedError(failure.code)?.usageLimit) return null; + const atMs = DateTime.toEpochMillis(yield* DateTime.now); + const resetAt = codexUsageLimitResetAt(yield* Ref.get(rateLimitSnapshot), atMs); + if (resetAt !== null || adapterOptions.getUsageLimits === undefined) return resetAt; + // A turn refused at its first request may never get a rate-limit notification. + return codexPublishedUsageLimitResetAt(yield* adapterOptions.getUsageLimits, atMs); + }); const limitedTurnItems = yield* Ref.make( new Map>(), ); @@ -4496,9 +4499,8 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( const previous = yield* Ref.getAndUpdate(rateLimitSnapshot, (previous) => mergeCodexRateLimits(previous, payload.rateLimits), ); - const snapshot = yield* Ref.get(rateLimitSnapshot); for (const item of (yield* Ref.get(limitedTurnItems)).values()) { - const resetAt = failureResetAt(item.failure, snapshot); + const resetAt = yield* usageLimitResetAt(item.failure); if (resetAt === null || item.failure.resetAt != null) continue; const updated = { ...item, @@ -5843,7 +5845,7 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( failure.class === "usage_limit" ? { ...failure, - resetAt: failureResetAt(failure, yield* Ref.get(rateLimitSnapshot)), + resetAt: yield* usageLimitResetAt(failure), } : failure, ...(input.providerRetry === undefined @@ -5881,9 +5883,7 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( ...event, failure: { ...event.failure, - resetAt: - event.failure.resetAt ?? - failureResetAt(event.failure, yield* Ref.get(rateLimitSnapshot)), + resetAt: event.failure.resetAt ?? (yield* usageLimitResetAt(event.failure)), }, } : event; diff --git a/apps/server/src/provider/Drivers/CodexDriver.ts b/apps/server/src/provider/Drivers/CodexDriver.ts index 871ce9ed6e84f..0b2e7384bad45 100644 --- a/apps/server/src/provider/Drivers/CodexDriver.ts +++ b/apps/server/src/provider/Drivers/CodexDriver.ts @@ -212,10 +212,7 @@ export const CodexDriver: ProviderDriver< enabled, config, }, - { - onUsageLimits: (update) => snapshot.applyUsageLimits(update), - compatibility: installation.compatibility, - }, + { compatibility: installation.compatibility, onUsageLimits: (update) => snapshot.applyUsageLimits(update), getUsageLimits: Effect.suspend(() => snapshot.getSnapshot).pipe(Effect.map((provider) => provider.usageLimits)) }, ).pipe( Effect.mapError( (cause) => diff --git a/apps/server/src/provider/codexUsageLimits.test.ts b/apps/server/src/provider/codexUsageLimits.test.ts index 9c863f54c0867..c872cad76c24c 100644 --- a/apps/server/src/provider/codexUsageLimits.test.ts +++ b/apps/server/src/provider/codexUsageLimits.test.ts @@ -261,19 +261,91 @@ describe("mergeCodexRateLimits", () => { }); }); -import { codexUsageLimitResetAt } from "./codexUsageLimits.ts"; +import { codexPublishedUsageLimitResetAt, codexUsageLimitResetAt } from "./codexUsageLimits.ts"; describe("codexUsageLimitResetAt", () => { + const stoppedAtMs = Date.parse("2033-05-18T03:00:00.000Z"); + it("waits for every exhausted window and never invents an unknown reset", () => { expect( - codexUsageLimitResetAt({ - primary: { usedPercent: 100, resetsAt: 2000000000 }, - secondary: { usedPercent: 100, resetsAt: 2000100000 }, - }), + codexUsageLimitResetAt( + { + primary: { usedPercent: 100, resetsAt: 2000000000 }, + secondary: { usedPercent: 100, resetsAt: 2000100000 }, + }, + stoppedAtMs, + ), ).toBe("2033-05-19T07:20:00.000Z"); - expect(codexUsageLimitResetAt({ primary: { usedPercent: 100 } })).toBeNull(); + expect(codexUsageLimitResetAt({ primary: { usedPercent: 100 } }, stoppedAtMs)).toBeNull(); expect( - codexUsageLimitResetAt({ primary: { usedPercent: 50, resetsAt: 2000000000 } }), + codexUsageLimitResetAt({ primary: { usedPercent: 50, resetsAt: 2000000000 } }, stoppedAtMs), ).toBeNull(); }); + + it("takes a window at 99% when Codex refuses the turn before any reads 100%", () => { + // Codex sends whole percentages; this is the last update seen before a real refusal. + expect( + codexUsageLimitResetAt( + { + limitId: "codex", + primary: { usedPercent: 99, resetsAt: 2000000000, windowDurationMins: 300 }, + secondary: { usedPercent: 73, resetsAt: 2000100000, windowDurationMins: 10080 }, + }, + stoppedAtMs, + ), + ).toBe("2033-05-18T03:33:20.000Z"); + }); + + it("waits only for the windows at 100% when another reads 99%", () => { + expect( + codexUsageLimitResetAt( + { + primary: { usedPercent: 100, resetsAt: 2000000000 }, + secondary: { usedPercent: 99, resetsAt: 2000100000 }, + }, + stoppedAtMs, + ), + ).toBe("2033-05-18T03:33:20.000Z"); + }); + + it("ignores a window that has renewed since it was observed", () => { + const afterSessionResetMs = Date.parse("2033-05-18T04:00:00.000Z"); + expect( + codexUsageLimitResetAt( + { + primary: { usedPercent: 100, resetsAt: 2000000000 }, + secondary: { usedPercent: 40, resetsAt: 2000100000 }, + }, + afterSessionResetMs, + ), + ).toBeNull(); + }); +}); + +describe("codexPublishedUsageLimitResetAt", () => { + it("reads the same reset from the published limits", () => { + const limits = { + checkedAt, + windows: [ + { + id: "primary", + kind: "session" as const, + label: "Session", + usedPercent: 100, + resetsAt: "2033-05-18T03:33:20.000Z", + }, + { + id: "secondary", + kind: "weekly" as const, + label: "Weekly", + usedPercent: 40, + resetsAt: "2033-05-19T07:20:00.000Z", + }, + ], + }; + expect(codexPublishedUsageLimitResetAt(limits, Date.parse("2033-05-18T03:00:00.000Z"))).toBe( + "2033-05-18T03:33:20.000Z", + ); + expect(codexPublishedUsageLimitResetAt(undefined, 0)).toBeNull(); + }); }); diff --git a/apps/server/src/provider/codexUsageLimits.ts b/apps/server/src/provider/codexUsageLimits.ts index d4f9c621119df..f02032285e9c0 100644 --- a/apps/server/src/provider/codexUsageLimits.ts +++ b/apps/server/src/provider/codexUsageLimits.ts @@ -186,12 +186,22 @@ export function mergeCodexRateLimits( }; } -/** All exhausted windows must reset before a continuation can run. */ -export function codexUsageLimitResetAt( - snapshot: CodexRateLimitSnapshot | undefined, +/** + * The reset a usage-limit stop at `atMs` waits for. All exhausted windows must + * reset before a continuation can run. Codex reports whole percentages and can + * refuse a turn while its last update still reads 99%, so a window at 99% + * counts as exhausted when none reads 100%. A window whose reset has already + * passed has renewed since it was observed, so its percentage no longer applies. + */ +function usageLimitStopResetAt( + allWindows: ReadonlyArray, + atMs: number, ): string | null { - if (!snapshot) return null; - const windows = codexRateLimitsToWindows(snapshot).filter((window) => window.usedPercent >= 100); + const current = allWindows.filter( + (window) => window.resetsAt === undefined || Date.parse(window.resetsAt) > atMs, + ); + const full = current.filter((window) => window.usedPercent >= 100); + const windows = full.length > 0 ? full : current.filter((window) => window.usedPercent >= 99); if (windows.length === 0 || windows.some((window) => !window.resetsAt)) return null; return windows.reduce( (latest, window) => @@ -201,3 +211,23 @@ export function codexUsageLimitResetAt( null, ); } + +/** The stop's reset from the session's own rate-limit notifications. */ +export function codexUsageLimitResetAt( + snapshot: CodexRateLimitSnapshot | undefined, + atMs: number, +): string | null { + return snapshot ? usageLimitStopResetAt(codexRateLimitsToWindows(snapshot), atMs) : null; +} + +/** + * The stop's reset from the instance's published limits, which the status + * probe fills even when a session is refused before Codex sends it any + * rate-limit notification. + */ +export function codexPublishedUsageLimitResetAt( + limits: ServerProviderUsageLimits | undefined, + atMs: number, +): string | null { + return limits ? usageLimitStopResetAt(limits.windows, atMs) : null; +} From 35edccb9e754d1259b1d07c5d82d2e9311cc80ea Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 19 Oct 2026 15:27:48 +0000 Subject: [PATCH 2/2] fix(server): preserve authoritative Codex usage reset times --- .../Adapters/CodexAdapterV2.test.ts | 223 ++++++++++++++++-- .../Adapters/CodexAdapterV2.ts | 95 +++++--- .../src/provider/Drivers/CodexDriver.ts | 8 +- .../provider/Drivers/CodexManagedProvider.ts | 1 + .../src/provider/codexUsageLimits.test.ts | 55 +---- apps/server/src/provider/codexUsageLimits.ts | 16 +- 6 files changed, 273 insertions(+), 125 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts index 3c0392985bfff..c6ec99dd15fc5 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts @@ -1,3 +1,4 @@ +import { applyUsageLimitsUpdate } from "@t3tools/provider-core/server/usageLimits"; import { historyResponseItems } from "@t3tools/provider-core/server/handoffBudget"; import { makeProviderFailure } from "@t3tools/provider-core/server/failure"; import { @@ -29,6 +30,7 @@ import { ProviderTurnId, RunAttemptId, RunId, + type ServerProviderUsageLimits, ThreadId, TurnItemId, } from "@t3tools/contracts"; @@ -1236,9 +1238,11 @@ describe("CodexAdapterV2 post-settle continuation", () => { Effect.void, onRequest: (method: string, params: unknown) => Effect.Effect = () => Effect.void, readChildMetadata?: Parameters[2], + usageLimits: Pick< + CodexAdapterV2.CodexAdapterV2Options, + "getUsageLimits" | "onUsageLimits" + > = {}, beforeEmitInbound?: CodexReplay.CodexAppServerReplayDriver["beforeEmitInbound"], - getUsageLimits?: CodexAdapterV2.CodexAdapterV2Options["getUsageLimits"], - onUsageLimits?: CodexAdapterV2.CodexAdapterV2Options["onUsageLimits"], ) => Effect.gen(function* () { const fileSystem = yield* FileSystem.FileSystem; @@ -1249,13 +1253,13 @@ describe("CodexAdapterV2 post-settle continuation", () => { const changes = yield* makeChangeSignal; const continuationRequests: Array = []; - const driver = yield* CodexReplay.makeReplayDriver( + const replayDriver = yield* CodexReplay.makeReplayDriver( transcript, beforeEmitInbound === undefined ? {} : { beforeEmitInbound }, ); const clientFactory: CodexAdapterV2.CodexAppServerClientFactoryShape = { open: (openInput) => - Layer.build(CodexReplay.layerReplayWithDriver(driver)).pipe( + Layer.build(CodexReplay.layerReplayWithDriver(replayDriver)).pipe( Effect.mapError( (cause) => new ProviderAdapter.ProviderAdapterOpenSessionError({ @@ -1296,7 +1300,6 @@ describe("CodexAdapterV2 post-settle continuation", () => { settings: DEFAULT_CODEX_SETTINGS, environment: {}, clientFactory, - ...(onUsageLimits ? { onUsageLimits } : {}), digest: yield* Digest.Digest, fileSystem, idAllocator, @@ -1307,7 +1310,7 @@ describe("CodexAdapterV2 post-settle continuation", () => { continuationRequests.push(request); }).pipe(Effect.andThen(changes.notify)), }, - ...(getUsageLimits ? { getUsageLimits } : {}), + ...usageLimits, }); const threadId = ThreadId.make(`thread-${transcript.scenario}`); const runtime = yield* adapter.openSession({ @@ -4000,6 +4003,7 @@ describe("CodexAdapterV2 post-settle continuation", () => { : Effect.void, undefined, undefined, + {}, (entry) => entry.label === "first-completed" ? Deferred.await(completions.first) @@ -5715,13 +5719,30 @@ describe("CodexAdapterV2 post-settle continuation", () => { resetsAt: 2000100000, windowDurationMins: 300, }, - secondary: { usedPercent: 73, resetsAt: 2000500000, windowDurationMins: 10080 }, + secondary: { + usedPercent: 73, + resetsAt: 2000500000, + windowDurationMins: 10080, + }, }, }, }, }, ] : []), + ...(scenario.name === "near-full-reset" || scenario.name === "published-reset" + ? [ + { + ...snapshot, + frame: { + method: "account/rateLimits/updated", + params: { + rateLimits: { limitId: "premium", primary: null, secondary: null }, + }, + }, + }, + ] + : []), ...(scenario.name === "deferred-reset" ? [snapshot] : []), ...(scenario.name === "deferred-reset" ? [ @@ -5855,24 +5876,31 @@ describe("CodexAdapterV2 post-settle continuation", () => { : Effect.void, undefined, undefined, - undefined, - // The session got no rate-limit notification; Usage → Limits still knows the reset. - scenario.name === "published-reset" - ? Effect.succeed({ - checkedAt: "2033-05-19T03:00:00.000Z", - windows: [ - { - id: "primary", - kind: "session", - label: "Session", - usedPercent: 100, - resetsAt: resetAt, - }, - { id: "secondary", kind: "weekly", label: "Weekly", usedPercent: 40 }, - ], - }) - : undefined, - () => Deferred.succeed(snapshotReceipt, undefined), + { + onUsageLimits: () => Deferred.succeed(snapshotReceipt, undefined), + ...(scenario.name === "published-reset" || + scenario.name === "near-full-reset" || + "expectedCode" in scenario + ? { + getUsageLimits: Effect.succeed({ + checkedAt: "2033-05-19T03:00:00.000Z", + windows: [ + { + id: "primary", + kind: "session", + label: "Session", + usedPercent: 100, + resetsAt: + scenario.name === "near-full-reset" + ? "2033-05-23T22:26:40.000Z" + : resetAt, + }, + { id: "secondary", kind: "weekly", label: "Weekly", usedPercent: 40 }, + ], + }), + } + : {}), + }, ); yield* harness.runtime.startTurn( makeCodexTestTurnInput({ @@ -5890,7 +5918,6 @@ describe("CodexAdapterV2 post-settle continuation", () => { assert.equal(terminal.failure.class, scenario.expectedClass); assert.equal(terminal.threadDisposition, "reusable"); if ( - scenario.name === "known-reset" || scenario.name === "deferred-reset" || scenario.name === "near-full-reset" || scenario.name === "published-reset" @@ -5939,6 +5966,150 @@ describe("CodexAdapterV2 post-settle continuation", () => { ), ); + it.effect.each(["before", "after"] as const)( + "does not lock a stopped turn to stale published limits published %s stop while fresh notifications arrive", + (publishedWhen) => + Effect.scoped( + Effect.gen(function* () { + yield* TestClock.setTime(Date.parse("2033-05-18T03:00:00.000Z")); + const nativeThreadId = "native-stale-published-limit"; + const nativeTurnId = "turn-stale-published-limit"; + const resetAt = "2033-05-19T07:20:00.000Z"; + const staleResetAt = "2033-05-23T22:26:40.000Z"; + const stalePublished: ServerProviderUsageLimits = { + checkedAt: "2033-05-18T02:00:00.000Z", + windows: [ + { + id: "primary", + kind: "session", + label: "Session", + usedPercent: 100, + resetsAt: staleResetAt, + }, + ], + }; + const published = yield* Ref.make( + publishedWhen === "before" ? stalePublished : undefined, + ); + const stopped = yield* Deferred.make(); + const freshLimitsPublished = yield* Deferred.make(); + const transcript = makeCodexReplayTranscript({ + scenario: "codex-stale-published-limit", + entries: [ + ...codexReplayPreamble({ nativeThreadId, nativeTurnId, prompt: "Continue." }), + { + type: "emit_inbound", + label: "turn/completed", + frame: { + method: "turn/completed", + params: { + threadId: nativeThreadId, + turn: { + ...makeCodexReplayTurn({ id: nativeTurnId, status: "failed" }), + error: { + message: "Usage limit reached.", + codexErrorInfo: "usageLimitExceeded", + }, + }, + }, + }, + }, + ...[ + { label: "renewed-window", primary: { usedPercent: 50, windowDurationMins: 300 } }, + { + label: "fresh-reset", + primary: { usedPercent: 100, resetsAt: 2000100000, windowDurationMins: 300 }, + }, + ].map(({ label, primary }) => ({ + type: "emit_inbound" as const, + label, + frame: { + method: "account/rateLimits/updated", + params: { rateLimits: { limitId: "codex", primary } }, + }, + })), + ], + }); + const harness = yield* makeCodexReplayHarness( + transcript, + (event) => { + if (event.type === "turn.terminal") return Deferred.succeed(stopped, undefined); + return Effect.void; + }, + undefined, + undefined, + { + getUsageLimits: Ref.get(published), + onUsageLimits: (update) => + Ref.update(published, (previous) => + applyUsageLimitsUpdate({ previous, update, checkedAt: update.checkedAt }), + ).pipe( + Effect.andThen( + update.windows.some((window) => window.resetsAt === resetAt) + ? Deferred.succeed(freshLimitsPublished, undefined) + : Effect.void, + ), + ), + }, + (entry) => + entry.label === "renewed-window" + ? Deferred.await(stopped).pipe( + Effect.andThen( + publishedWhen === "after" ? Ref.set(published, stalePublished) : Effect.void, + ), + ) + : Effect.void, + ); + yield* harness.runtime.startTurn( + makeCodexTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now: yield* DateTime.now, + text: "Continue.", + attemptId: RunAttemptId.make("attempt-stale-published-limit"), + }), + ); + yield* harness.firstTerminal; + const terminal = harness.terminalEvents()[0]; + assert.equal(terminal?.status, "failed"); + if (terminal?.status !== "failed") return; + assert.equal(terminal.failure.class, "usage_limit"); + assert.equal(terminal.failure.resetAt, publishedWhen === "before" ? staleResetAt : null); + yield* Deferred.await(freshLimitsPublished); + yield* harness.until(() => + harness.events.some( + (event) => + event.type === "turn_item.updated" && + event.turnItem.type === "error" && + event.turnItem.failure.resetAt === resetAt, + ), + ); + const resets = harness.events.flatMap((event) => + event.type === "turn_item.updated" && event.turnItem.type === "error" + ? [event.turnItem.failure.resetAt] + : [], + ); + const latestResetAt = resets.length > 0 ? resets.at(-1) : terminal.failure.resetAt; + assert.deepEqual( + { resets, latestResetAt }, + { + resets: publishedWhen === "before" ? [null, resetAt] : [resetAt], + latestResetAt: resetAt, + }, + ); + }).pipe( + Effect.provide( + Layer.mergeAll( + IdAllocator.layer, + McpProviderSessions.layer, + NodeServices.layer, + NodeCrypto.layer, + ), + ), + ), + ), + ); + // regression: #17501 Codex 429 retry exhaustion classifies as a usage limit it.effect.each([ { diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts index 9175bca5b3c2f..b982ebeca296f 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -1168,7 +1168,6 @@ function withCodexManagedFailure( }; } - interface ActiveCodexTurnContext { hookStopReason?: string; capacityRetryAttempt?: number; @@ -1639,7 +1638,10 @@ export type CodexAdapterV2DriverEnv = export const createCodexAdapterV2 = Effect.fn("CodexAdapterV2.createCodexAdapterV2")(function* ( { instanceId, environment, enabled, config }: ProviderAdapterDriverCreateInput, - hooks: Pick = {}, + hooks: Pick< + CodexAdapterV2Options, + "onUsageLimits" | "getUsageLimits" | "resolveRuntime" | "compatibility" + > = {}, ) { const clientFactory = yield* CodexAppServerClientFactory; const continuationRequests = yield* ProviderContinuationRequests.ProviderContinuationRequests; @@ -1862,17 +1864,35 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( Effect.forkScoped, ); const rateLimitSnapshot = yield* Ref.make(undefined); - const usageLimitResetAt = Effect.fnUntraced(function* (failure: OrchestrationV2ProviderFailure) { + const usageLimitReset = Effect.fn("CodexAdapterV2.usageLimitReset")(function* ( + failure: OrchestrationV2ProviderFailure, + ) { // Sharing caps have no reset in the ordinary allowance snapshot. - if (classifyCodexManagedError(failure.code)?.usageLimit) return null; + if (classifyCodexManagedError(failure.code)?.usageLimit) { + return { resetAt: null, source: "session" as const }; + } const atMs = DateTime.toEpochMillis(yield* DateTime.now); - const resetAt = codexUsageLimitResetAt(yield* Ref.get(rateLimitSnapshot), atMs); - if (resetAt !== null || adapterOptions.getUsageLimits === undefined) return resetAt; - // A turn refused at its first request may never get a rate-limit notification. - return codexPublishedUsageLimitResetAt(yield* adapterOptions.getUsageLimits, atMs); + const snapshot = yield* Ref.get(rateLimitSnapshot); + // Session notifications outrank probes, even when they lack a reset. + if (snapshot !== undefined) { + return { resetAt: codexUsageLimitResetAt(snapshot, atMs), source: "session" as const }; + } + return { + resetAt: + adapterOptions.getUsageLimits === undefined + ? null + : codexPublishedUsageLimitResetAt(yield* adapterOptions.getUsageLimits, atMs), + source: "published" as const, + }; }); const limitedTurnItems = yield* Ref.make( - new Map>(), + new Map< + ProviderThreadId, + { + readonly item: Extract; + readonly resetSource: "session" | "published"; + } + >(), ); const activeTurns = yield* Ref.make(new Map()); const turnTokenUsageByThread = new Map(); @@ -4496,25 +4516,38 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( yield* client.handleServerNotification("account/rateLimits/updated", (payload) => Effect.gen(function* () { + if (payload.rateLimits.limitId && payload.rateLimits.limitId !== "codex") return; const previous = yield* Ref.getAndUpdate(rateLimitSnapshot, (previous) => mergeCodexRateLimits(previous, payload.rateLimits), ); - for (const item of (yield* Ref.get(limitedTurnItems)).values()) { - const resetAt = yield* usageLimitResetAt(item.failure); - if (resetAt === null || item.failure.resetAt != null) continue; + const resetAt = codexUsageLimitResetAt( + yield* Ref.get(rateLimitSnapshot), + DateTime.toEpochMillis(yield* DateTime.now), + ); + for (const { item, resetSource } of (yield* Ref.get(limitedTurnItems)).values()) { + if (classifyCodexManagedError(item.failure.code)?.usageLimit) continue; + // Published resets stay provisional until the session reports its allowance. + if (resetSource === "session" && (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), + new Map(current).set(item.providerThreadId!, { + item: updated, + resetSource: "session", + }), ); - yield* emitProviderEvent({ - type: "turn_item.updated", - driver: CODEX_PROVIDER, - turnItem: updated, - }); + if (item.failure.resetAt !== resetAt) { + yield* emitProviderEvent({ + type: "turn_item.updated", + driver: CODEX_PROVIDER, + turnItem: updated, + }); + } } const update = codexRateLimitsToUpdate(payload.rateLimits, previous); if (update && adapterOptions.onUsageLimits) { @@ -5841,13 +5874,7 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( `terminal-failure:${input.context.providerTurnId}`, ), status: terminalStatus, - failure: - failure.class === "usage_limit" - ? { - ...failure, - resetAt: yield* usageLimitResetAt(failure), - } - : failure, + failure, ...(input.providerRetry === undefined ? {} : { @@ -5877,15 +5904,13 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( context: ActiveCodexTurnContext, event: CodexRootTerminalEvent, ) { - const current = + const reset = event.status === "failed" && event.failure.class === "usage_limit" - ? { - ...event, - failure: { - ...event.failure, - resetAt: event.failure.resetAt ?? (yield* usageLimitResetAt(event.failure)), - }, - } + ? yield* usageLimitReset(event.failure) + : undefined; + const current = + event.status === "failed" && reset !== undefined + ? { ...event, failure: { ...event.failure, resetAt: reset.resetAt } } : event; yield* emitProviderEvent(current); const continued = continuedRuns.get(context.providerTurnId) ?? [context]; @@ -5905,7 +5930,7 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( if (nativeThreadId != null && rootProviderThread !== undefined) { rootProviderThreads.set(nativeThreadId, { ...rootProviderThread, status: "idle" }); } - if (current.status === "failed" && current.failure.class === "usage_limit") { + if (current.status === "failed" && reset !== undefined) { const item = makeProviderFailureTurnItem({ driver: CODEX_PROVIDER, threadId: context.input.threadId, @@ -5918,7 +5943,7 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( occurredAt: yield* DateTime.now, }); yield* Ref.update(limitedTurnItems, (items) => - new Map(items).set(context.providerThread.id, item), + new Map(items).set(context.providerThread.id, { item, resetSource: reset.source }), ); } }); diff --git a/apps/server/src/provider/Drivers/CodexDriver.ts b/apps/server/src/provider/Drivers/CodexDriver.ts index 0b2e7384bad45..89a733638d4f3 100644 --- a/apps/server/src/provider/Drivers/CodexDriver.ts +++ b/apps/server/src/provider/Drivers/CodexDriver.ts @@ -212,7 +212,13 @@ export const CodexDriver: ProviderDriver< enabled, config, }, - { compatibility: installation.compatibility, onUsageLimits: (update) => snapshot.applyUsageLimits(update), getUsageLimits: Effect.suspend(() => snapshot.getSnapshot).pipe(Effect.map((provider) => provider.usageLimits)) }, + { + compatibility: installation.compatibility, + onUsageLimits: (update) => snapshot.applyUsageLimits(update), + getUsageLimits: Effect.suspend(() => snapshot.getSnapshot).pipe( + Effect.map((provider) => provider.usageLimits), + ), + }, ).pipe( Effect.mapError( (cause) => diff --git a/apps/server/src/provider/Drivers/CodexManagedProvider.ts b/apps/server/src/provider/Drivers/CodexManagedProvider.ts index ab5e7add556a9..eed5d98e71967 100644 --- a/apps/server/src/provider/Drivers/CodexManagedProvider.ts +++ b/apps/server/src/provider/Drivers/CodexManagedProvider.ts @@ -244,6 +244,7 @@ export const makeManagedCodexProvider = Effect.fn("makeManagedCodexProvider")(fu // already wraps openSession in withAccess, so resolve without re-entering it. const orchestrationAdapter = yield* createCodexAdapterV2(input, { onUsageLimits: (update) => snapshot.applyUsageLimits(update), + getUsageLimits: snapshot.getSnapshot.pipe(Effect.map((provider) => provider.usageLimits)), resolveRuntime: runtime.resolve, }).pipe( Effect.mapError( diff --git a/apps/server/src/provider/codexUsageLimits.test.ts b/apps/server/src/provider/codexUsageLimits.test.ts index c872cad76c24c..0467f85035ab2 100644 --- a/apps/server/src/provider/codexUsageLimits.test.ts +++ b/apps/server/src/provider/codexUsageLimits.test.ts @@ -261,7 +261,7 @@ describe("mergeCodexRateLimits", () => { }); }); -import { codexPublishedUsageLimitResetAt, codexUsageLimitResetAt } from "./codexUsageLimits.ts"; +import { codexUsageLimitResetAt } from "./codexUsageLimits.ts"; describe("codexUsageLimitResetAt", () => { const stoppedAtMs = Date.parse("2033-05-18T03:00:00.000Z"); @@ -276,27 +276,6 @@ describe("codexUsageLimitResetAt", () => { stoppedAtMs, ), ).toBe("2033-05-19T07:20:00.000Z"); - expect(codexUsageLimitResetAt({ primary: { usedPercent: 100 } }, stoppedAtMs)).toBeNull(); - expect( - codexUsageLimitResetAt({ primary: { usedPercent: 50, resetsAt: 2000000000 } }, stoppedAtMs), - ).toBeNull(); - }); - - it("takes a window at 99% when Codex refuses the turn before any reads 100%", () => { - // Codex sends whole percentages; this is the last update seen before a real refusal. - expect( - codexUsageLimitResetAt( - { - limitId: "codex", - primary: { usedPercent: 99, resetsAt: 2000000000, windowDurationMins: 300 }, - secondary: { usedPercent: 73, resetsAt: 2000100000, windowDurationMins: 10080 }, - }, - stoppedAtMs, - ), - ).toBe("2033-05-18T03:33:20.000Z"); - }); - - it("waits only for the windows at 100% when another reads 99%", () => { expect( codexUsageLimitResetAt( { @@ -306,6 +285,10 @@ describe("codexUsageLimitResetAt", () => { stoppedAtMs, ), ).toBe("2033-05-18T03:33:20.000Z"); + expect(codexUsageLimitResetAt({ primary: { usedPercent: 100 } }, stoppedAtMs)).toBeNull(); + expect( + codexUsageLimitResetAt({ primary: { usedPercent: 50, resetsAt: 2000000000 } }, stoppedAtMs), + ).toBeNull(); }); it("ignores a window that has renewed since it was observed", () => { @@ -321,31 +304,3 @@ describe("codexUsageLimitResetAt", () => { ).toBeNull(); }); }); - -describe("codexPublishedUsageLimitResetAt", () => { - it("reads the same reset from the published limits", () => { - const limits = { - checkedAt, - windows: [ - { - id: "primary", - kind: "session" as const, - label: "Session", - usedPercent: 100, - resetsAt: "2033-05-18T03:33:20.000Z", - }, - { - id: "secondary", - kind: "weekly" as const, - label: "Weekly", - usedPercent: 40, - resetsAt: "2033-05-19T07:20:00.000Z", - }, - ], - }; - expect(codexPublishedUsageLimitResetAt(limits, Date.parse("2033-05-18T03:00:00.000Z"))).toBe( - "2033-05-18T03:33:20.000Z", - ); - expect(codexPublishedUsageLimitResetAt(undefined, 0)).toBeNull(); - }); -}); diff --git a/apps/server/src/provider/codexUsageLimits.ts b/apps/server/src/provider/codexUsageLimits.ts index f02032285e9c0..21a045969a063 100644 --- a/apps/server/src/provider/codexUsageLimits.ts +++ b/apps/server/src/provider/codexUsageLimits.ts @@ -186,13 +186,8 @@ export function mergeCodexRateLimits( }; } -/** - * The reset a usage-limit stop at `atMs` waits for. All exhausted windows must - * reset before a continuation can run. Codex reports whole percentages and can - * refuse a turn while its last update still reads 99%, so a window at 99% - * counts as exhausted when none reads 100%. A window whose reset has already - * passed has renewed since it was observed, so its percentage no longer applies. - */ +// Codex can refuse a turn at 99% because it reports whole percentages. +// Prefer fully exhausted windows and discard observations whose reset has passed. function usageLimitStopResetAt( allWindows: ReadonlyArray, atMs: number, @@ -212,7 +207,6 @@ function usageLimitStopResetAt( ); } -/** The stop's reset from the session's own rate-limit notifications. */ export function codexUsageLimitResetAt( snapshot: CodexRateLimitSnapshot | undefined, atMs: number, @@ -220,11 +214,7 @@ export function codexUsageLimitResetAt( return snapshot ? usageLimitStopResetAt(codexRateLimitsToWindows(snapshot), atMs) : null; } -/** - * The stop's reset from the instance's published limits, which the status - * probe fills even when a session is refused before Codex sends it any - * rate-limit notification. - */ +// A refused first request may never receive a session rate-limit notification. export function codexPublishedUsageLimitResetAt( limits: ServerProviderUsageLimits | undefined, atMs: number,