From 0cf8add51f78167b0007b2ef4f022bf2f4ba1f7b Mon Sep 17 00:00:00 2001 From: onerandomroamer Date: Sat, 31 Oct 2026 12:40:55 -0700 Subject: [PATCH 1/3] fix(server): refresh Codex reset times after usage limits --- .../Adapters/CodexAdapterV2.test.ts | 188 +++++++++++++++++- .../Adapters/CodexAdapterV2.ts | 44 +++- 2 files changed, 217 insertions(+), 15 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts index 90ee44f3d5b6f..5f3c0ef95eb65 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts @@ -5043,6 +5043,42 @@ describe("CodexAdapterV2 post-settle continuation", () => { notification: false, expectedClass: "usage_limit", }, + { + name: "stale-read", + code: "usageLimitExceeded", + notification: false, + expectedClass: "usage_limit", + }, + { + name: "empty-read", + code: "usageLimitExceeded", + notification: false, + expectedClass: "usage_limit", + }, + { + name: "failed-read", + code: "usageLimitExceeded", + notification: false, + expectedClass: "usage_limit", + }, + { + name: "timed-out-read", + code: "usageLimitExceeded", + notification: false, + expectedClass: "usage_limit", + }, + { + name: "unexhausted-read", + code: "usageLimitExceeded", + notification: false, + expectedClass: "usage_limit", + }, + { + name: "known-reset", + code: "usageLimitExceeded", + notification: false, + expectedClass: "usage_limit", + }, { name: "replacement", code: "usageLimitExceeded", @@ -5114,7 +5150,48 @@ describe("CodexAdapterV2 post-settle continuation", () => { scenario: `codex-limit-${scenario.name}`, entries: [ ...codexReplayPreamble({ nativeThreadId, nativeTurnId, prompt: "Continue." }), - ...(scenario.name === "deferred-reset" ? [snapshot] : []), + ...(scenario.name === "deferred-reset" || scenario.name === "known-reset" + ? [snapshot] + : scenario.name === "stale-read" || + scenario.name === "failed-read" || + scenario.name === "unexhausted-read" + ? [ + { + ...snapshot, + frame: { + method: "account/rateLimits/updated", + params: { + rateLimits: { + limitId: "codex", + primary: { + usedPercent: 99, + resetsAt: 2000000000, + windowDurationMins: 300, + }, + }, + }, + }, + }, + ] + : []), + ...(scenario.name === "stale-read" || scenario.name === "empty-read" + ? [ + { + type: "emit_inbound" as const, + label: "account/rateLimits/updated/premium", + frame: { + method: "account/rateLimits/updated", + params: { + rateLimits: { + limitId: "premium", + primary: null, + secondary: null, + }, + }, + }, + }, + ] + : []), ...(scenario.name === "deferred-reset" ? [ { @@ -5198,6 +5275,61 @@ describe("CodexAdapterV2 post-settle continuation", () => { }, }, }, + ...(scenario.expectedClass === "usage_limit" && + scenario.name !== "deferred-reset" && + scenario.name !== "known-reset" && + scenario.name !== "timed-out-read" + ? [ + { + type: "expect_outbound" as const, + label: "account/rateLimits/read", + frame: { + id: 4, + method: "account/rateLimits/read", + params: null, + }, + }, + { + type: "emit_inbound" as const, + label: "account/rateLimits/read", + frame: { + id: 4, + ...(scenario.name === "failed-read" + ? { error: { code: -32603, message: "Usage endpoint unavailable." } } + : { + result: { + rateLimits: { limitId: "premium", primary: null, secondary: null }, + rateLimitsByLimitId: { + codex: + scenario.name === "stale-read" || scenario.name === "empty-read" + ? { + limitId: "codex", + primary: { + usedPercent: 100, + resetsAt: 2000000000, + windowDurationMins: 300, + }, + secondary: { + usedPercent: 100, + resetsAt: 2000100000, + windowDurationMins: 10080, + }, + } + : { + limitId: "codex", + primary: { + usedPercent: 99, + resetsAt: 2000000000, + windowDurationMins: 300, + }, + }, + }, + }, + }), + }, + }, + ] + : []), ...(scenario.name === "late-reset" ? [snapshot] : []), ...(scenario.name === "deferred-reset" ? [ @@ -5236,12 +5368,29 @@ describe("CodexAdapterV2 post-settle continuation", () => { ], }); const resetReceipt = yield* Deferred.make(); - const harness = yield* makeCodexReplayHarness(transcript, (event) => - event.type === "turn_item.updated" && - event.turnItem.type === "error" && - event.turnItem.failure.resetAt === resetAt - ? Deferred.succeed(resetReceipt, undefined) - : Effect.void, + const readStarted = yield* Deferred.make(); + const readInterrupted = yield* Deferred.make(); + const harness = yield* makeCodexReplayHarness( + transcript, + (event) => + event.type === "turn_item.updated" && + event.turnItem.type === "error" && + event.turnItem.failure.resetAt === resetAt + ? Deferred.succeed(resetReceipt, undefined) + : Effect.void, + (method) => + method === "account/rateLimits/read" + ? Deferred.succeed(readStarted, undefined).pipe( + Effect.andThen( + scenario.name === "timed-out-read" + ? Effect.never.pipe( + Effect.onInterrupt(() => Deferred.succeed(readInterrupted, undefined)), + ) + : Effect.void, + ), + Effect.asVoid, + ) + : Effect.void, ); yield* harness.runtime.startTurn( makeCodexTestTurnInput({ @@ -5258,10 +5407,31 @@ 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.endsWith("-read")) yield* Deferred.await(readStarted); + if (scenario.name === "timed-out-read") { + yield* TestClock.adjust("3 seconds"); + yield* Deferred.await(readInterrupted); + assert.isNull(terminal.failure.resetAt); + assert.isFalse( + harness.events.some( + (event) => + event.type === "turn_item.updated" && + event.turnItem.type === "error" && + event.turnItem.failure.resetAt != null, + ), + ); + } + if (scenario.name === "known-reset") assert.isFalse(yield* Deferred.isDone(readStarted)); + if (["deferred-reset", "known-reset"].includes(scenario.name)) { + assert.equal(terminal.failure.resetAt, resetAt); + } + if (scenario.name === "failed-read" || scenario.name === "unexhausted-read") { + assert.isNull(terminal.failure.resetAt); + assert.equal(terminal.failure.message, message); + } if (scenario.name === "matching-details") assert.equal(terminal.failure.message, "Detailed provider allowance explanation."); - if (scenario.name === "late-reset") { + if (["late-reset", "stale-read", "empty-read"].includes(scenario.name)) { yield* Deferred.await(resetReceipt); const item = harness.events.find( (event) => diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts index 821cf051c3235..0049f9e609a98 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -4398,12 +4398,8 @@ 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), - ); - const resetAt = codexUsageLimitResetAt(yield* Ref.get(rateLimitSnapshot)); + const backfillUsageLimitResetAt = Effect.fn("CodexAdapterV2.backfillUsageLimitResetAt")( + function* (resetAt: string | null) { for (const item of (yield* Ref.get(limitedTurnItems)).values()) { if (resetAt === null || item.failure.resetAt != null) continue; const updated = { @@ -4420,6 +4416,17 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( turnItem: updated, }); } + }, + ); + + yield* client.handleServerNotification("account/rateLimits/updated", (payload) => + Effect.gen(function* () { + const previous = yield* Ref.getAndUpdate(rateLimitSnapshot, (previous) => + mergeCodexRateLimits(previous, payload.rateLimits), + ); + yield* backfillUsageLimitResetAt( + codexUsageLimitResetAt(yield* Ref.get(rateLimitSnapshot)), + ); const update = codexRateLimitsToUpdate(payload.rateLimits, previous); if (update && adapterOptions.onUsageLimits) { const checkedAt = DateTime.formatIso(yield* DateTime.now); @@ -5751,6 +5758,27 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( }, ); + const resolveUsageLimitResetAt = Effect.fn("CodexAdapterV2.resolveUsageLimitResetAt")( + function* () { + const cached = codexUsageLimitResetAt(yield* Ref.get(rateLimitSnapshot)); + if (cached !== null) return yield* backfillUsageLimitResetAt(cached); + // A rejected request may follow a stale percentage or an empty premium snapshot. + const response = yield* client + .request("account/rateLimits/read", null) + .pipe(Effect.timeout("3 seconds"), Effect.option); + if (Option.isSome(response)) { + const snapshot = + response.value.rateLimitsByLimitId?.codex ?? response.value.rateLimits; + yield* Ref.update(rateLimitSnapshot, (previous) => + mergeCodexRateLimits(previous, snapshot), + ); + } + yield* backfillUsageLimitResetAt( + codexUsageLimitResetAt(yield* Ref.get(rateLimitSnapshot)), + ); + }, + ); + const emitRootTerminal = Effect.fnUntraced(function* ( context: ActiveCodexTurnContext, event: CodexRootTerminalEvent, @@ -5800,6 +5828,10 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( yield* Ref.update(limitedTurnItems, (items) => new Map(items).set(context.providerThread.id, item), ); + // RPC responses use the notification reader, so refresh outside its handler. + if (item.failure.resetAt == null) { + yield* resolveUsageLimitResetAt().pipe(Effect.forkIn(scope)); + } } }); From 43c31ec2083561890a6e8d38092afda0790f88bb Mon Sep 17 00:00:00 2001 From: onerandomroamer Date: Sat, 31 Oct 2026 12:53:10 -0700 Subject: [PATCH 2/3] fix(server): publish refreshed Codex usage windows --- .../Adapters/CodexAdapterV2.test.ts | 19 +++++++++++++++++++ .../Adapters/CodexAdapterV2.ts | 7 ++++++- 2 files changed, 25 insertions(+), 1 deletion(-) diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts index 5f3c0ef95eb65..fb4fe114648b0 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts @@ -1150,6 +1150,9 @@ describe("CodexAdapterV2 post-settle continuation", () => { transcript.scenario, ).pipe(Effect.orDie); const changes = yield* makeChangeSignal; + const usageLimitsUpdates: Array< + Parameters>[0] + > = []; const continuationRequests: Array = []; const driver = yield* CodexReplay.makeReplayDriver( @@ -1196,6 +1199,10 @@ describe("CodexAdapterV2 post-settle continuation", () => { fileSystem, idAllocator, serverConfig, + onUsageLimits: (update) => + Effect.sync(() => { + usageLimitsUpdates.push(update); + }), continuationRequests: { offer: (request) => Effect.sync(() => { @@ -1259,6 +1266,7 @@ describe("CodexAdapterV2 post-settle continuation", () => { threadId, serverConfig, events, + usageLimitsUpdates, continuationRequests, terminalEvents, subagentUpdates, @@ -5440,6 +5448,17 @@ describe("CodexAdapterV2 post-settle continuation", () => { event.turnItem.failure.resetAt === resetAt, ); assert.isDefined(item); + if (scenario.name === "stale-read" || scenario.name === "empty-read") { + assert.deepEqual( + harness.usageLimitsUpdates + .at(-1) + ?.windows.map(({ id, usedPercent, resetsAt }) => ({ id, usedPercent, resetsAt })), + [ + { id: "primary", usedPercent: 100, resetsAt: "2033-05-18T03:33:20.000Z" }, + { id: "secondary", usedPercent: 100, resetsAt: resetAt }, + ], + ); + } } if (scenario.name === "retry") assert.equal(terminal.retry?.attempt, 1); if (scenario.name === "chatgpt-workspace-limit") { diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts index 0049f9e609a98..499757b7f39e3 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -5769,9 +5769,14 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( if (Option.isSome(response)) { const snapshot = response.value.rateLimitsByLimitId?.codex ?? response.value.rateLimits; - yield* Ref.update(rateLimitSnapshot, (previous) => + const previous = yield* Ref.getAndUpdate(rateLimitSnapshot, (previous) => mergeCodexRateLimits(previous, snapshot), ); + const update = codexRateLimitsToUpdate(snapshot, previous); + if (update && adapterOptions.onUsageLimits) { + const checkedAt = DateTime.formatIso(yield* DateTime.now); + yield* adapterOptions.onUsageLimits({ ...update, checkedAt }); + } } yield* backfillUsageLimitResetAt( codexUsageLimitResetAt(yield* Ref.get(rateLimitSnapshot)), From 02409c7a0af36c52f195b4d72b2fba0ff6083a48 Mon Sep 17 00:00:00 2001 From: onerandomroamer Date: Sat, 31 Oct 2026 13:01:17 -0700 Subject: [PATCH 3/3] fix(server): resolve Codex resets before terminal ingestion closes --- .../Adapters/CodexAdapterV2.test.ts | 131 +++++++++++++----- .../Adapters/CodexAdapterV2.ts | 64 ++++++--- 2 files changed, 144 insertions(+), 51 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts index fb4fe114648b0..cfbe0f0205f5c 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts @@ -5081,6 +5081,18 @@ describe("CodexAdapterV2 post-settle continuation", () => { notification: false, expectedClass: "usage_limit", }, + { + name: "newer-notification-read", + code: "usageLimitExceeded", + notification: false, + expectedClass: "usage_limit", + }, + { + name: "omitted-window-read", + code: "usageLimitExceeded", + notification: false, + expectedClass: "usage_limit", + }, { name: "known-reset", code: "usageLimitExceeded", @@ -5162,7 +5174,8 @@ describe("CodexAdapterV2 post-settle continuation", () => { ? [snapshot] : scenario.name === "stale-read" || scenario.name === "failed-read" || - scenario.name === "unexhausted-read" + scenario.name === "unexhausted-read" || + scenario.name === "newer-notification-read" ? [ { ...snapshot, @@ -5182,6 +5195,23 @@ describe("CodexAdapterV2 post-settle continuation", () => { }, ] : []), + ...(scenario.name === "omitted-window-read" + ? [ + { + ...snapshot, + frame: { + method: "account/rateLimits/updated", + params: { + rateLimits: { + limitId: "codex", + primary: { usedPercent: 100, resetsAt: 2000000000 }, + secondary: { usedPercent: 100 }, + }, + }, + }, + }, + ] + : []), ...(scenario.name === "stale-read" || scenario.name === "empty-read" ? [ { @@ -5297,6 +5327,26 @@ describe("CodexAdapterV2 post-settle continuation", () => { params: null, }, }, + ...(scenario.name === "newer-notification-read" + ? [ + { + ...snapshot, + frame: { + method: "account/rateLimits/updated", + params: { + rateLimits: { + limitId: "codex", + primary: { + usedPercent: 50, + resetsAt: 2000200000, + windowDurationMins: 300, + }, + }, + }, + }, + }, + ] + : []), { type: "emit_inbound" as const, label: "account/rateLimits/read", @@ -5309,7 +5359,9 @@ describe("CodexAdapterV2 post-settle continuation", () => { rateLimits: { limitId: "premium", primary: null, secondary: null }, rateLimitsByLimitId: { codex: - scenario.name === "stale-read" || scenario.name === "empty-read" + scenario.name === "stale-read" || + scenario.name === "empty-read" || + scenario.name === "newer-notification-read" ? { limitId: "codex", primary: { @@ -5323,14 +5375,19 @@ describe("CodexAdapterV2 post-settle continuation", () => { windowDurationMins: 10080, }, } - : { - limitId: "codex", - primary: { - usedPercent: 99, - resetsAt: 2000000000, - windowDurationMins: 300, + : scenario.name === "omitted-window-read" + ? { + limitId: "codex", + secondary: { usedPercent: 99, resetsAt: 2000100000 }, + } + : { + limitId: "codex", + primary: { + usedPercent: 99, + resetsAt: 2000000000, + windowDurationMins: 300, + }, }, - }, }, }, }), @@ -5409,6 +5466,12 @@ describe("CodexAdapterV2 post-settle continuation", () => { attemptId: RunAttemptId.make(`attempt-limit-${scenario.name}`), }), ); + if (scenario.name === "timed-out-read") { + yield* Deferred.await(readStarted); + assert.lengthOf(harness.terminalEvents(), 0); + yield* TestClock.adjust("3 seconds"); + yield* Deferred.await(readInterrupted); + } yield* harness.firstTerminal; const terminal = harness.terminalEvents()[0]; assert.equal(terminal?.status, "failed"); @@ -5417,8 +5480,6 @@ describe("CodexAdapterV2 post-settle continuation", () => { assert.equal(terminal.threadDisposition, "reusable"); if (scenario.name.endsWith("-read")) yield* Deferred.await(readStarted); if (scenario.name === "timed-out-read") { - yield* TestClock.adjust("3 seconds"); - yield* Deferred.await(readInterrupted); assert.isNull(terminal.failure.resetAt); assert.isFalse( harness.events.some( @@ -5430,35 +5491,43 @@ describe("CodexAdapterV2 post-settle continuation", () => { ); } if (scenario.name === "known-reset") assert.isFalse(yield* Deferred.isDone(readStarted)); - if (["deferred-reset", "known-reset"].includes(scenario.name)) { + if (["deferred-reset", "known-reset", "stale-read", "empty-read"].includes(scenario.name)) { assert.equal(terminal.failure.resetAt, resetAt); } - if (scenario.name === "failed-read" || scenario.name === "unexhausted-read") { + if ( + [ + "failed-read", + "unexhausted-read", + "omitted-window-read", + "newer-notification-read", + ].includes(scenario.name) + ) { assert.isNull(terminal.failure.resetAt); assert.equal(terminal.failure.message, message); } if (scenario.name === "matching-details") assert.equal(terminal.failure.message, "Detailed provider allowance explanation."); - if (["late-reset", "stale-read", "empty-read"].includes(scenario.name)) { + if (scenario.name === "newer-notification-read") { + assert.deepEqual( + harness.usageLimitsUpdates + .at(-1) + ?.windows.map(({ usedPercent, resetsAt }) => ({ usedPercent, resetsAt })), + [{ usedPercent: 50, resetsAt: "2033-05-20T11:06:40.000Z" }], + ); + } + if (scenario.name === "late-reset" && terminal.failure.resetAt !== resetAt) { yield* Deferred.await(resetReceipt); - const item = harness.events.find( - (event) => - event.type === "turn_item.updated" && - event.turnItem.type === "error" && - event.turnItem.failure.resetAt === resetAt, + } + if (scenario.name === "stale-read" || scenario.name === "empty-read") { + assert.deepEqual( + harness.usageLimitsUpdates + .at(-1) + ?.windows.map(({ id, usedPercent, resetsAt }) => ({ id, usedPercent, resetsAt })), + [ + { id: "primary", usedPercent: 100, resetsAt: "2033-05-18T03:33:20.000Z" }, + { id: "secondary", usedPercent: 100, resetsAt: resetAt }, + ], ); - assert.isDefined(item); - if (scenario.name === "stale-read" || scenario.name === "empty-read") { - assert.deepEqual( - harness.usageLimitsUpdates - .at(-1) - ?.windows.map(({ id, usedPercent, resetsAt }) => ({ id, usedPercent, resetsAt })), - [ - { id: "primary", usedPercent: 100, resetsAt: "2033-05-18T03:33:20.000Z" }, - { id: "secondary", usedPercent: 100, resetsAt: resetAt }, - ], - ); - } } if (scenario.name === "retry") assert.equal(terminal.retry?.attempt, 1); if (scenario.name === "chatgpt-workspace-limit") { diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts index 499757b7f39e3..25ea390283b57 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -5760,8 +5760,9 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( const resolveUsageLimitResetAt = Effect.fn("CodexAdapterV2.resolveUsageLimitResetAt")( function* () { - const cached = codexUsageLimitResetAt(yield* Ref.get(rateLimitSnapshot)); - if (cached !== null) return yield* backfillUsageLimitResetAt(cached); + const beforeRead = yield* Ref.get(rateLimitSnapshot); + const cached = codexUsageLimitResetAt(beforeRead); + if (cached !== null) return cached; // A rejected request may follow a stale percentage or an empty premium snapshot. const response = yield* client .request("account/rateLimits/read", null) @@ -5769,22 +5770,23 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( if (Option.isSome(response)) { const snapshot = response.value.rateLimitsByLimitId?.codex ?? response.value.rateLimits; - const previous = yield* Ref.getAndUpdate(rateLimitSnapshot, (previous) => - mergeCodexRateLimits(previous, snapshot), + // Main-allowance notifications replace this reference; keep any update that won the race. + const applied = yield* Ref.modify(rateLimitSnapshot, (current) => + current === beforeRead && (!snapshot.limitId || snapshot.limitId === "codex") + ? ([true, snapshot] as const) + : ([false, current] as const), ); - const update = codexRateLimitsToUpdate(snapshot, previous); + const update = applied ? codexRateLimitsToUpdate(snapshot, beforeRead) : undefined; if (update && adapterOptions.onUsageLimits) { const checkedAt = DateTime.formatIso(yield* DateTime.now); yield* adapterOptions.onUsageLimits({ ...update, checkedAt }); } } - yield* backfillUsageLimitResetAt( - codexUsageLimitResetAt(yield* Ref.get(rateLimitSnapshot)), - ); + return codexUsageLimitResetAt(yield* Ref.get(rateLimitSnapshot)); }, ); - const emitRootTerminal = Effect.fnUntraced(function* ( + const emitResolvedRootTerminal = Effect.fnUntraced(function* ( context: ActiveCodexTurnContext, event: CodexRootTerminalEvent, ) { @@ -5802,13 +5804,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); } @@ -5833,13 +5828,42 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( yield* Ref.update(limitedTurnItems, (items) => new Map(items).set(context.providerThread.id, item), ); - // RPC responses use the notification reader, so refresh outside its handler. - if (item.failure.resetAt == null) { - yield* resolveUsageLimitResetAt().pipe(Effect.forkIn(scope)); - } } }); + // Failure ingestion stops at the terminal; finish the RPC off the reader before emitting it. + const emitRootTerminal = Effect.fnUntraced(function* ( + context: ActiveCodexTurnContext, + event: CodexRootTerminalEvent, + ) { + if (event.status !== "completed") { + const continued = continuedRuns.get(context.providerTurnId) ?? [context]; + yield* Ref.update( + terminalizedNonCompletedNativeTurns, + (nativeTurns) => + new Set([...nativeTurns, ...continued.map((turn) => turn.nativeTurnId)]), + ); + } + if ( + event.status === "failed" && + event.failure.class === "usage_limit" && + event.failure.resetAt == null && + codexUsageLimitResetAt(yield* Ref.get(rateLimitSnapshot)) === null + ) { + yield* resolveUsageLimitResetAt().pipe( + Effect.flatMap((resetAt) => + emitResolvedRootTerminal(context, { + ...event, + failure: { ...event.failure, resetAt }, + }), + ), + Effect.forkIn(scope), + ); + return; + } + yield* emitResolvedRootTerminal(context, event); + }); + const emitOrDeferRootTerminal = Effect.fn("CodexAdapterV2.emitOrDeferRootTerminal")( function* (input: { readonly context: ActiveCodexTurnContext;