fix(server): Codex usage-limit failures on a fresh session carry their reset time

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

@@ -727,6 +727,23 @@ function makeCodexReplayTranscript(input: {
   };
 }
 
+function codexRateLimitsReadEntries(
+  response: { readonly result: unknown } | { readonly error: unknown },
+): Array<CodexReplay.CodexAppServerReplayEntry> {
+  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,
@@ -5597,6 +5614,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",
@@ -5791,6 +5814,33 @@ describe("CodexAdapterV2 post-settle continuation", () => {
                 },
               },
             },
+            ...(scenario.expectedClass === "usage_limit" &&
+            scenario.name !== "deferred-reset" &&
+            !("expectedCode" in scenario)
+              ? 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" || "resetAfter" in scenario ? [snapshot] : []),
             ...(scenario.name === "deferred-reset"
               ? [
@@ -5858,7 +5908,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 ("resetAfter" in scenario) {
@@ -5980,6 +6032,9 @@ describe("CodexAdapterV2 post-settle continuation", () => {
                   },
                 },
               },
+              ...(expectedClass === "usage_limit"
+                ? codexRateLimitsReadEntries(CODEX_RATE_LIMITS_READ_FAILURE)
+                : []),
             ],
           });
           const harness = yield* makeCodexReplayHarness(transcript);

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

@@ -967,6 +967,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<OrchestrationV2ProviderTurn>,
@@ -4491,35 +4492,38 @@ 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 snapshot = yield* Ref.get(rateLimitSnapshot);
+          for (const item of (yield* Ref.get(limitedTurnItems)).values()) {
+            const resetAt = failureResetAt(item.failure, snapshot);
+            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 snapshot = yield* Ref.get(rateLimitSnapshot);
-            for (const item of (yield* Ref.get(limitedTurnItems)).values()) {
-              const resetAt = failureResetAt(item.failure, snapshot);
-              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) =>
@@ -5871,7 +5875,7 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* (
           },
         );
 
-        const emitRootTerminal = Effect.fnUntraced(function* (
+        const publishRootTerminal = Effect.fnUntraced(function* (
           context: ActiveCodexTurnContext,
           event: CodexRootTerminalEvent,
         ) {
@@ -5888,23 +5892,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,
@@ -5923,6 +5910,53 @@ 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 &&
+            !classifyCodexManagedError(event.failure.code)?.usageLimit &&
+            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;