Skip to content

Commit 58b9c19

Browse files
rwspatinclaude
rwspatin
and
claude
committed
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 <[email protected]>
1 parent 47f2c8d commit 58b9c19

2 files changed

Lines changed: 132 additions & 46 deletions

File tree

‎apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts‎

Lines changed: 54 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -722,6 +722,23 @@ function makeCodexReplayTranscript(input: {
722722
};
723723
}
724724

725+
function codexRateLimitsReadEntries(
726+
response: { readonly result: unknown } | { readonly error: unknown },
727+
): Array<CodexReplay.CodexAppServerReplayEntry> {
728+
return [
729+
{
730+
type: "expect_outbound",
731+
label: "account/rateLimits/read",
732+
frame: { id: 4, method: "account/rateLimits/read", params: null },
733+
},
734+
{ type: "emit_inbound", label: "account/rateLimits/read", frame: { id: 4, ...response } },
735+
];
736+
}
737+
738+
const CODEX_RATE_LIMITS_READ_FAILURE = {
739+
error: { code: -32603, message: "Usage is unavailable." },
740+
} as const;
741+
725742
function withReplayRequestId(
726743
entry: CodexReplay.CodexAppServerReplayEntry,
727744
id: number,
@@ -5256,6 +5273,12 @@ describe("CodexAdapterV2 post-settle continuation", () => {
52565273
expectedClass: "usage_limit",
52575274
},
52585275
{ name: "retry", code: "usageLimitExceeded", notification: true, expectedClass: "usage_limit" },
5276+
{
5277+
name: "read-reset",
5278+
code: "usageLimitExceeded",
5279+
notification: true,
5280+
expectedClass: "usage_limit",
5281+
},
52595282
{
52605283
name: "exhausted-429",
52615284
code: "responseTooManyFailedAttempts",
@@ -5386,6 +5409,31 @@ describe("CodexAdapterV2 post-settle continuation", () => {
53865409
},
53875410
},
53885411
},
5412+
...(scenario.expectedClass === "usage_limit" && scenario.name !== "deferred-reset"
5413+
? codexRateLimitsReadEntries(
5414+
scenario.name === "read-reset"
5415+
? {
5416+
result: {
5417+
rateLimits: {
5418+
limitId: "codex",
5419+
primary: {
5420+
usedPercent: 0,
5421+
windowDurationMins: 300,
5422+
resetsAt: 1999990000,
5423+
},
5424+
secondary: {
5425+
usedPercent: 100,
5426+
windowDurationMins: 10080,
5427+
resetsAt: 2000100000,
5428+
},
5429+
planType: "plus",
5430+
rateLimitReachedType: "rate_limit_reached",
5431+
},
5432+
},
5433+
}
5434+
: CODEX_RATE_LIMITS_READ_FAILURE,
5435+
)
5436+
: []),
53895437
...(scenario.name === "late-reset" ? [snapshot] : []),
53905438
...(scenario.name === "deferred-reset"
53915439
? [
@@ -5446,7 +5494,9 @@ describe("CodexAdapterV2 post-settle continuation", () => {
54465494
if (terminal?.status !== "failed") return;
54475495
assert.equal(terminal.failure.class, scenario.expectedClass);
54485496
assert.equal(terminal.threadDisposition, "reusable");
5449-
if (scenario.name === "deferred-reset") assert.equal(terminal.failure.resetAt, resetAt);
5497+
if (scenario.name === "deferred-reset" || scenario.name === "read-reset")
5498+
assert.equal(terminal.failure.resetAt, resetAt);
5499+
if (scenario.name === "usage") assert.isNull(terminal.failure.resetAt);
54505500
if (scenario.name === "matching-details")
54515501
assert.equal(terminal.failure.message, "Detailed provider allowance explanation.");
54525502
if (scenario.name === "late-reset") {
@@ -5550,6 +5600,9 @@ describe("CodexAdapterV2 post-settle continuation", () => {
55505600
},
55515601
},
55525602
},
5603+
...(expectedClass === "usage_limit"
5604+
? codexRateLimitsReadEntries(CODEX_RATE_LIMITS_READ_FAILURE)
5605+
: []),
55535606
],
55545607
});
55555608
const harness = yield* makeCodexReplayHarness(transcript);

‎apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts‎

Lines changed: 78 additions & 45 deletions
Original file line numberDiff line numberDiff line change
@@ -927,6 +927,7 @@ function describeCodexGoal(goal: OrchestrationV2ProviderGoal): string {
927927

928928
// Codex starts the next goal turn milliseconds after the last one completes.
929929
const CODEX_GOAL_CONTINUATION_GRACE = "5 seconds";
930+
const CODEX_RATE_LIMITS_READ_TIMEOUT = "3 seconds";
930931

931932
const providerTurnsForThread = (
932933
providerTurns: ReadonlyArray<OrchestrationV2ProviderTurn>,
@@ -4429,34 +4430,37 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* (
44294430
}).pipe(Effect.orDie),
44304431
);
44314432

4432-
yield* client.handleServerNotification("account/rateLimits/updated", (payload) =>
4433-
Effect.gen(function* () {
4434-
const previous = yield* Ref.getAndUpdate(rateLimitSnapshot, (previous) =>
4435-
mergeCodexRateLimits(previous, payload.rateLimits),
4433+
// Untraced: rate-limit notifications repeat on every token tick.
4434+
const applyRateLimits = Effect.fnUntraced(function* (rateLimits: CodexRateLimitSnapshot) {
4435+
const previous = yield* Ref.getAndUpdate(rateLimitSnapshot, (previous) =>
4436+
mergeCodexRateLimits(previous, rateLimits),
4437+
);
4438+
const resetAt = codexUsageLimitResetAt(yield* Ref.get(rateLimitSnapshot));
4439+
for (const item of (yield* Ref.get(limitedTurnItems)).values()) {
4440+
if (resetAt === null || item.failure.resetAt != null) continue;
4441+
const updated = {
4442+
...item,
4443+
updatedAt: yield* DateTime.now,
4444+
failure: { ...item.failure, resetAt },
4445+
};
4446+
yield* Ref.update(limitedTurnItems, (current) =>
4447+
new Map(current).set(item.providerThreadId!, updated),
44364448
);
4437-
const resetAt = codexUsageLimitResetAt(yield* Ref.get(rateLimitSnapshot));
4438-
for (const item of (yield* Ref.get(limitedTurnItems)).values()) {
4439-
if (resetAt === null || item.failure.resetAt != null) continue;
4440-
const updated = {
4441-
...item,
4442-
updatedAt: yield* DateTime.now,
4443-
failure: { ...item.failure, resetAt },
4444-
};
4445-
yield* Ref.update(limitedTurnItems, (current) =>
4446-
new Map(current).set(item.providerThreadId!, updated),
4447-
);
4448-
yield* emitProviderEvent({
4449-
type: "turn_item.updated",
4450-
driver: CODEX_PROVIDER,
4451-
turnItem: updated,
4452-
});
4453-
}
4454-
const update = codexRateLimitsToUpdate(payload.rateLimits, previous);
4455-
if (update && adapterOptions.onUsageLimits) {
4456-
const checkedAt = DateTime.formatIso(yield* DateTime.now);
4457-
yield* adapterOptions.onUsageLimits({ ...update, checkedAt });
4458-
}
4459-
}).pipe(Effect.orDie),
4449+
yield* emitProviderEvent({
4450+
type: "turn_item.updated",
4451+
driver: CODEX_PROVIDER,
4452+
turnItem: updated,
4453+
});
4454+
}
4455+
const update = codexRateLimitsToUpdate(rateLimits, previous);
4456+
if (update && adapterOptions.onUsageLimits) {
4457+
const checkedAt = DateTime.formatIso(yield* DateTime.now);
4458+
yield* adapterOptions.onUsageLimits({ ...update, checkedAt });
4459+
}
4460+
});
4461+
4462+
yield* client.handleServerNotification("account/rateLimits/updated", (payload) =>
4463+
applyRateLimits(payload.rateLimits).pipe(Effect.orDie),
44604464
);
44614465

44624466
yield* client.handleServerNotification("thread/tokenUsage/updated", (payload) =>
@@ -5808,7 +5812,7 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* (
58085812
},
58095813
);
58105814

5811-
const emitRootTerminal = Effect.fnUntraced(function* (
5815+
const publishRootTerminal = Effect.fnUntraced(function* (
58125816
context: ActiveCodexTurnContext,
58135817
event: CodexRootTerminalEvent,
58145818
) {
@@ -5825,23 +5829,6 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* (
58255829
}
58265830
: event;
58275831
yield* emitProviderEvent(current);
5828-
const continued = continuedRuns.get(context.providerTurnId) ?? [context];
5829-
if (current.status !== "completed") {
5830-
yield* Ref.update(
5831-
terminalizedNonCompletedNativeTurns,
5832-
(nativeTurns) =>
5833-
new Set([...nativeTurns, ...continued.map((turn) => turn.nativeTurnId)]),
5834-
);
5835-
}
5836-
for (const turn of continued) {
5837-
continuedRuns.delete(turn.providerTurnId);
5838-
}
5839-
const nativeThreadId = context.providerThread.nativeThreadRef?.nativeId;
5840-
const rootProviderThread =
5841-
nativeThreadId == null ? undefined : rootProviderThreads.get(nativeThreadId);
5842-
if (nativeThreadId != null && rootProviderThread !== undefined) {
5843-
rootProviderThreads.set(nativeThreadId, { ...rootProviderThread, status: "idle" });
5844-
}
58455832
if (current.status === "failed" && current.failure.class === "usage_limit") {
58465833
const item = makeProviderFailureTurnItem({
58475834
driver: CODEX_PROVIDER,
@@ -5860,6 +5847,52 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* (
58605847
}
58615848
});
58625849

5850+
const emitRootTerminal = Effect.fnUntraced(function* (
5851+
context: ActiveCodexTurnContext,
5852+
event: CodexRootTerminalEvent,
5853+
) {
5854+
// Runs stop ingesting at a failed terminal, so a later reset time is never stored. A session
5855+
// that starts limited gets no rateLimits/updated, so read them before publishing. Forked:
5856+
// notification handlers run on the reader fiber, which must stay free for the reply.
5857+
if (
5858+
event.status === "failed" &&
5859+
event.failure.class === "usage_limit" &&
5860+
event.failure.resetAt == null &&
5861+
codexUsageLimitResetAt(yield* Ref.get(rateLimitSnapshot)) === null
5862+
) {
5863+
yield* client.request("account/rateLimits/read", null).pipe(
5864+
Effect.timeout(CODEX_RATE_LIMITS_READ_TIMEOUT),
5865+
Effect.flatMap((response) =>
5866+
applyRateLimits(response.rateLimitsByLimitId?.codex ?? response.rateLimits),
5867+
),
5868+
Effect.catch((cause) =>
5869+
Effect.logDebug("orchestration-v2.codex-rate-limits-read-failed", { cause }),
5870+
),
5871+
Effect.andThen(publishRootTerminal(context, event)),
5872+
Effect.forkIn(scope),
5873+
);
5874+
} else {
5875+
yield* publishRootTerminal(context, event);
5876+
}
5877+
const continued = continuedRuns.get(context.providerTurnId) ?? [context];
5878+
if (event.status !== "completed") {
5879+
yield* Ref.update(
5880+
terminalizedNonCompletedNativeTurns,
5881+
(nativeTurns) =>
5882+
new Set([...nativeTurns, ...continued.map((turn) => turn.nativeTurnId)]),
5883+
);
5884+
}
5885+
for (const turn of continued) {
5886+
continuedRuns.delete(turn.providerTurnId);
5887+
}
5888+
const nativeThreadId = context.providerThread.nativeThreadRef?.nativeId;
5889+
const rootProviderThread =
5890+
nativeThreadId == null ? undefined : rootProviderThreads.get(nativeThreadId);
5891+
if (nativeThreadId != null && rootProviderThread !== undefined) {
5892+
rootProviderThreads.set(nativeThreadId, { ...rootProviderThread, status: "idle" });
5893+
}
5894+
});
5895+
58635896
const emitOrDeferRootTerminal = Effect.fn("CodexAdapterV2.emitOrDeferRootTerminal")(
58645897
function* (input: {
58655898
readonly context: ActiveCodexTurnContext;

0 commit comments

Comments
 (0)