fix(server): codex usage-limit failures keep their reset time

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

@@ -6018,6 +6018,148 @@ describe("CodexAdapterV2 post-settle continuation", () => {
       ),
   );
 
+  // regression: #27773 a 429 on a run's first request still reports the reset time
+  it.effect(
+    "recovers resetAt from a fresh rate-limits read when the 429 never primed the snapshot",
+    () =>
+      Effect.scoped(
+        Effect.gen(function* () {
+          const nativeThreadId = "native-limit-cold-429";
+          const nativeTurnId = "turn-limit-cold-429";
+          const message = "Provider stopped this request.";
+          const resetAt = "2033-05-19T07:20:00.000Z";
+          const transcript = makeCodexReplayTranscript({
+            scenario: "codex-limit-cold-429",
+            entries: [
+              ...codexReplayPreamble({ nativeThreadId, nativeTurnId, prompt: "Continue." }),
+              {
+                type: "emit_inbound",
+                label: "error",
+                frame: {
+                  method: "error",
+                  params: {
+                    threadId: nativeThreadId,
+                    turnId: nativeTurnId,
+                    willRetry: false,
+                    error: { message, codexErrorInfo: "usageLimitExceeded" },
+                  },
+                },
+              },
+              // The 429's own notification is an empty non-codex snapshot the adapter must ignore.
+              {
+                type: "emit_inbound",
+                label: "account/rateLimits/updated",
+                frame: {
+                  method: "account/rateLimits/updated",
+                  params: {
+                    rateLimits: { limitId: "premium", primary: null, secondary: null },
+                  },
+                },
+              },
+              {
+                type: "emit_inbound",
+                label: "turn/completed",
+                frame: {
+                  method: "turn/completed",
+                  params: {
+                    threadId: nativeThreadId,
+                    turn: {
+                      ...makeCodexReplayTurn({ id: nativeTurnId, status: "failed" }),
+                      error: { message, codexErrorInfo: "usageLimitExceeded" },
+                    },
+                  },
+                },
+              },
+              {
+                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,
+                  result: {
+                    rateLimits: { limitId: "premium", primary: null, secondary: null },
+                    rateLimitsByLimitId: {
+                      codex: {
+                        limitId: "codex",
+                        primary: {
+                          usedPercent: 100,
+                          resetsAt: 2000100000,
+                          windowDurationMins: 300,
+                        },
+                      },
+                    },
+                  },
+                },
+              },
+            ],
+          });
+          const resetReceipt = yield* Deferred.make<void>();
+          const readRequests: Array<string> = [];
+          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) =>
+              Effect.sync(() => {
+                readRequests.push(method);
+              }),
+          );
+          yield* harness.runtime.startTurn(
+            makeCodexTestTurnInput({
+              threadId: harness.threadId,
+              providerThread: harness.providerThread,
+              now: yield* DateTime.now,
+              text: "Continue.",
+              attemptId: RunAttemptId.make("attempt-limit-cold-429"),
+            }),
+          );
+          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");
+          yield* Deferred.await(resetReceipt);
+          assert.include(readRequests, "account/rateLimits/read");
+          const item = harness.events.find(
+            (event) =>
+              event.type === "turn_item.updated" &&
+              event.turnItem.type === "error" &&
+              event.turnItem.failure.resetAt === resetAt,
+          );
+          assert.isDefined(item);
+          assert.equal(terminal.failure.resetAt, null);
+          assert.isBelow(harness.events.indexOf(terminal), harness.events.indexOf(item!));
+          assert.lengthOf(
+            harness.events.filter(
+              (event) =>
+                event.type === "turn_item.updated" && event.usageLimitResetAtEnrichment === true,
+            ),
+            1,
+          );
+          assert.isTrue(
+            item?.type === "turn_item.updated" && item.usageLimitResetAtEnrichment === true,
+          );
+        }).pipe(
+          Effect.provide(
+            Layer.mergeAll(
+              IdAllocator.layer,
+              McpProviderSessions.layer,
+              NodeServices.layer,
+              NodeCrypto.layer,
+            ),
+          ),
+        ),
+      ),
+  );
+
   // regression: #28032 an unlisted codexErrorInfo dropped the failed turn/completed
   it.effect("fails the turn when Codex reports an error info the schema does not list", () =>
     Effect.scoped(

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

@@ -21,6 +21,7 @@ import {
   buildCodexInitializeParams,
   codexAppServerEnvironment,
   codexDefaultReasoningEffortOverride,
+  RATE_LIMITS_PROBE_TIMEOUT_MS,
 } from "../../provider/CodexProvider.ts";
 import {
   codexRateLimitsToUpdate,
@@ -86,6 +87,7 @@ import * as Context from "effect/Context";
 import * as Digest from "effect/crypto/Digest";
 import * as DateTime from "effect/DateTime";
 import * as Deferred from "effect/Deferred";
+import * as Duration from "effect/Duration";
 import * as Effect from "effect/Effect";
 import * as Fiber from "effect/Fiber";
 import * as FileSystem from "effect/FileSystem";
@@ -4491,29 +4493,58 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* (
           }).pipe(Effect.orDie),
         );
 
+        const applyRateLimitSnapshot = Effect.fnUntraced(function* (
+          update: CodexRateLimitSnapshot,
+        ) {
+          yield* Ref.update(rateLimitSnapshot, (previous) =>
+            mergeCodexRateLimits(previous, update),
+          );
+          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,
+              usageLimitResetAtEnrichment: true,
+            });
+          }
+        });
+
+        // A 429 that kills a run's first request never delivers a codex-family
+        // notification, so a usage-limit failure without a reset time reads
+        // fresh limits straight from the app-server.
+        const backfillRateLimitSnapshot = Effect.fn("CodexAdapterV2.backfillRateLimitSnapshot")(
+          function* (): Effect.fn.Return<void> {
+            const response = yield* client.request("account/rateLimits/read", null).pipe(
+              Effect.timeoutOption(Duration.millis(RATE_LIMITS_PROBE_TIMEOUT_MS)),
+              Effect.map(Option.getOrUndefined),
+              Effect.catch((error) =>
+                Effect.logWarning("Codex rate-limit read failed.", { cause: error }).pipe(
+                  Effect.as(undefined),
+                ),
+              ),
+            );
+            if (response === undefined) return;
+            yield* applyRateLimitSnapshot(
+              response.rateLimitsByLimitId?.codex ?? response.rateLimits,
+            );
+          },
+        );
+
         yield* client.handleServerNotification("account/rateLimits/updated", (payload) =>
           Effect.gen(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);
-              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 previous = yield* Ref.get(rateLimitSnapshot);
+            yield* applyRateLimitSnapshot(payload.rateLimits);
             const update = codexRateLimitsToUpdate(payload.rateLimits, previous);
             if (update && adapterOptions.onUsageLimits) {
               const checkedAt = DateTime.formatIso(yield* DateTime.now);
@@ -5920,6 +5951,11 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* (
             yield* Ref.update(limitedTurnItems, (items) =>
               new Map(items).set(context.providerThread.id, item),
             );
+            // Forked because the read's own response arrives on the protocol
+            // reader fiber this terminal handler runs on.
+            if (current.failure.resetAt == null) {
+              yield* backfillRateLimitSnapshot().pipe(Effect.forkIn(scope));
+            }
           }
         });
 

apps/server/src/orchestration-v2/EventSink.ts

@@ -346,7 +346,38 @@ const layerBase: Layer.Layer<
                   event.payload,
                   event.payload.runId === null ? undefined : runOrdinals.get(event.payload.runId),
                 )
-                .pipe(Effect.map((payload) => ({ ...event, payload })))
+                .pipe(
+                  Effect.flatMap((payload) =>
+                    Effect.gen(function* () {
+                      if (
+                        event.driver === "codex" &&
+                        payload.type === "error" &&
+                        payload.failure.class === "usage_limit" &&
+                        payload.failure.resetAt == null
+                      ) {
+                        // The direct enrichment lane can commit before the queued terminal.
+                        const existing = yield* projectionStore.getTurnItem({
+                          threadId: payload.threadId,
+                          itemId: payload.id,
+                        });
+                        if (
+                          existing?.type === "error" &&
+                          existing.failure.class === "usage_limit" &&
+                          existing.failure.resetAt != null
+                        ) {
+                          return {
+                            ...event,
+                            payload: {
+                              ...payload,
+                              failure: { ...payload.failure, resetAt: existing.failure.resetAt },
+                            },
+                          };
+                        }
+                      }
+                      return { ...event, payload };
+                    }),
+                  ),
+                )
             : Effect.succeed(event),
         { concurrency: 1 },
       );