fix(codex): bound the stored Pool 401 recovery budget by account

devlog/_plan/260829_bugpr_lane_h_residual_issues/170_pr2895_pool_401_recovery_budget.md

@@ -0,0 +1,111 @@
+# Lane P — #2895 / #2892 gap 5: one recovery budget for a stored Pool 401
+
+Carries contributor PR #2895 (`luvs01`, `a838b071c`) onto current `dev` and corrects the one
+blocker in it. The contributor's commit is preserved with its authorship; this unit is the
+follow-up commit on top.
+
+## What the contributor got right
+
+#2889 gave an ordinary stored Codex Pool account one generation-fenced forced refresh plus one
+same-account replay after a pre-stream 401. Gap 5 of #2892 is that the replay's *result* was not
+treated as final: a replay 429/402 could still be composed with another Pool account, a remembered
+compact model, a combo target, or a policy-fallback candidate — so a single logical request could
+spend several accounts' quota after the budget was already used.
+
+The contributor's structure is sound and is kept as-is: the boolean `codexMain401ReplayAttempted`
+becomes a tri-state `codex401ReplayKind` (`"main" | "stored" | null`), an
+`onStoredPool401ReplayDispatched` signal is threaded through combo and policy fallback, and compact
+guards both its pool-rotation and remembered-model paths. `main-pool` keeps its full recovery
+breadth, which is correct — a native main 401 is not a stored-account budget.
+
+## The blocker: the budget bounds accounts, not rescue
+
+The original patch enforced the bound with one line in `src/server/responses/core.ts`:
+
+```ts
+if (codex401ReplayKind === "stored" && upstreamResponse.status >= 400) break;
+```
+
+That break sits *above* two recovery ladders that send to the account already paying:
+
+- `shouldRetryCodexPoolAccountModel400` (`:4200`) — an allow-listed gated-model 400, retried on
+  the **same** account when the refreshed roster still grants the model
+  (`retryCodexPoolOnAlternateAccount` sets `retryAuthCtx = firstAuthCtx` for exactly that case).
+- `attemptOpaqueBlobRecovery` (`:4249`) — a rejected opaque reasoning/compaction blob, where the
+  one-shot rebuild strips the blob and resends to the same refreshed account.
+
+Neither charges a different account, so neither is inside the budget #2892 asked to bound. With the
+broad break, `401 → refresh → invalid_encrypted_content` became a user-visible 400 where the
+rebuild would have succeeded. A regression proves it: restoring that one line turns
+*a stored-account replay may still rebuild a rejected opaque blob on the same account* red.
+
+The corrected boundary is stated in terms of what is actually scarce — **another account's quota**,
+not further sends:
+
+- a quota failure (429/402) after a stored replay has no same-account move left, so `sameAccountOnly`
+  makes it terminal by refusing the alternate;
+- a gated-model 400 keeps its ladder, because it can retry the account the refreshed roster still
+  grants, and `sameAccountOnly` refuses only the alternate resolution;
+- opaque-blob recovery is untouched.
+
+That is **one** mechanism, not two. An earlier revision of this fix also broke at the pool-retry
+site on a non-400 outcome, and review showed no test could tell the difference: `sameAccountOnly`
+already produced the identical result by returning `no-alternate`. The redundant break is gone
+rather than kept as unjustifiable control flow.
+
+`sameAccountOnly` is a new field on the retry args rather than a check at the call site, because
+the decision belongs where the alternate is resolved — the existing `fixedAccount` guard already
+lives on that line and means the same thing for a different reason.
+
+## The timing defect in the dispatch signal
+
+The signal fired immediately before `fetchWithHeaderTimeout`, but that helper awaits
+`pacing.waitForPacing()` (`src/server/responses/fetch-helpers.ts:121`) and only then invokes the
+executor. A rejected pacing admission therefore marked the budget spent for a send that never
+reached the network, and the request lost its fallback for nothing.
+
+`storedPoolReplayDispatchNotifier` wraps the executor so the signal fires at the last moment before
+the send. It deliberately re-exposes `waitForPacing` and `unpacedFetch`: `fetchWithHeaderTimeout`
+reads both off the executor, so a plain function wrapper would drop provider pacing — and a wrapper
+that kept `waitForPacing` but dropped `unpacedFetch` would pace twice. Both are covered by named
+mutations.
+
+## Verification
+
+196 pass / 0 fail across the pool-401, native-main, policy-fallback, fetch-helper, opaque-blob,
+pool-rotation, compaction-routing, combo-recovery, stream-preflight, and request-pacing suites.
+`bun x tsc --noEmit` clean; `privacy:scan` green.
+
+Named mutations, each turning its own test red:
+
+| Mutation | Test that fails |
+| --- | --- |
+| restore the broad `status >= 400` break | opaque blob rebuilt on the same account |
+| `sameAccountOnly: false` | gated-model 400 after a stored replay reaches an alternate |
+| disable the combo dispatch gate | all four combo cases reach the backup target |
+| notify eagerly at the core call site | replay stuck in the pacing queue signals a dispatch |
+| drop the `notified` guard | one notifier signals twice across two sends |
+| drop `unpacedFetch` from the wrapper | pacing applied twice |
+
+Three process notes, all from tests that looked fine and were not:
+
+1. The gated-model test was **vacuous on the first attempt**. The injected entitlement resolver
+   reported only the other account as entitled, so initial selection picked that account and the
+   stored 401 never happened — it passed with one send and no refresh. It now returns both accounts
+   on the first resolution and only the alternate from the retry resolution onward. Its name was
+   also wrong: it asserts the *refusal* of an alternate, not a same-account retry, and now says so.
+2. The pacing test began as a **helper unit test only**, which review showed could not catch the
+   defect it was written for: restoring eager notification at the core call site left it green. It
+   is now an integration test through `handleResponses`, and two details had to be right for it to
+   bite at all — `route.provider` is a snapshot taken at routing time, so enabling pacing
+   mid-flight does nothing (the module-level queue depth is what changes under a live request), and
+   the request must not be a combo, because `handleComboResponses` installs its own dispatch
+   callback for the child and would swallow the caller's.
+3. "Signals exactly once" was **not mutation-protected** while the test invoked the notifier once.
+   It now sends twice through one notifier.
+
+## Not in this unit
+
+Gaps 1–4 of #2892 (refresh-flight abort ownership, superseding-generation freshness, rotated-grant
+fan-out to inactive aliases, atomic generation validation) remain open and are the other PR that
+issue asks for.

src/server/responses/compact.ts

@@ -163,6 +163,10 @@ interface CompactHandoffRoute {
  */
 const compactHandoffRoutes = new Map<string, CompactHandoffRoute>();
 
+export function clearCompactHandoffRoutesForTests(): void {
+  compactHandoffRoutes.clear();
+}
+
 function pruneCompactHandoffRoutes(now: number): void {
   for (const [key, entry] of compactHandoffRoutes) {
     if (now - entry.lastUsedAt > COMPACT_HANDOFF_ROUTE_TTL_MS) compactHandoffRoutes.delete(key);
@@ -740,6 +744,7 @@ export async function handleResponsesCompact(
     // actually happens, so every recorder call names the context that produced it.
     let outcomeCtx = authCtx;
     let upstream: Response;
+    let storedPool401ReplayAttempted = false;
     try {
       // Same connect timeout + keep-alive reset + transient-5xx recovery as /v1/responses —
       // compact hits the same ChatGPT host and must soft-avoid / clear affinity (#186).
@@ -777,6 +782,7 @@ export async function handleResponsesCompact(
     ) {
       await upstream.body?.cancel().catch(() => undefined);
       const poolAuthCtx = authCtx.kind === "pool" ? authCtx : undefined;
+      storedPool401ReplayAttempted = poolAuthCtx !== undefined;
       const poolReplay = poolAuthCtx
         ? await refreshPoolCompactContext({
           req,
@@ -837,6 +843,7 @@ export async function handleResponsesCompact(
     // — reporting exhausted retries while another pool account sat idle (#913).
     if (
       (upstream.status === 429 || upstream.status === 402)
+      && !storedPool401ReplayAttempted
       && usesCodexForwardPoolAuth(authCtx, route.provider)
       && !authCtx.fixedAccount
       && route.codexAccountMode
@@ -947,7 +954,7 @@ export async function handleResponsesCompact(
     if (buffered.ok) {
       inspectResponseLogJson(logCtx, await buffered.clone().text());
       forgetCompactHandoffRoute(req);
-    } else if (quotaFailure) {
+    } else if (quotaFailure && !storedPool401ReplayAttempted) {
       const fallbackModel = compactHandoffRoute(req, raw.model);
       if (fallbackModel && !req.signal.aborted) {
         const fallbackReq = new Request(req.url, {

src/server/responses/core.ts

@@ -321,7 +321,7 @@ import type { EffectiveSubagentRoster, SpawnAgentSurface } from "../../codex/cat
 import { buildToolBridgeMaps, collabSurface, injectDeveloperMessage, multiAgentGuidanceText } from "./collaboration";
 import { mapCodexAuthContextErrorToResponse, nativeMainRefreshFailureResponse } from "./codex-auth-error";
 import { hasUnreadableEncryptedAgentTask, looksLikeBackendCiphertext, sanitizeEncryptedContentInPlace } from "./encrypted-payload";
-import { fetchWithHeaderTimeout, providerFetch, safeHostLabel, safeOriginLabel } from "./fetch-helpers";
+import { fetchWithHeaderTimeout, providerFetch, safeHostLabel, safeOriginLabel, storedPoolReplayDispatchNotifier } from "./fetch-helpers";
 import { classifyTransportFailureKind, transportErrorCode } from "../../lib/upstream-reachability";
 import {
   acquireUpstreamHostAdmission,
@@ -922,6 +922,15 @@ interface CodexPoolAccountRetryArgs {
   firstAuthCtx: Extract<CodexAuthContext, { kind: "pool" | "main-pool" }>;
   firstResponse: Response;
   outcomeStatus: number;
+  /**
+   * Forbid resolving a DIFFERENT account for this retry.
+   *
+   * Set when a stored Pool 401 already spent this logical request's account budget on its own
+   * refresh and replay. The same-account gated-model retry above stays available, because it
+   * sends to the account that was already paying; only the alternate-account resolution below is
+   * out of budget.
+   */
+  sameAccountOnly?: boolean;
   upstream: AbortController;
   connectMs: number;
   passthroughEstimate?: number;
@@ -1084,7 +1093,9 @@ async function retryCodexPoolOnAlternateAccount(
   }
   // Exact account selectors may retry the same confirmed account above, but must never resolve
   // an alternate. Quota failures and a refreshed entitlement miss remain terminal.
-  if (!retryAuthCtx && firstAuthCtx.fixedAccount) return { kind: "no-alternate" };
+  if (!retryAuthCtx && (firstAuthCtx.fixedAccount || args.sameAccountOnly === true)) {
+    return { kind: "no-alternate" };
+  }
   try {
     retryAuthCtx ??= await resolveCodexAuthContext(
         req.headers,
@@ -1433,6 +1444,8 @@ export interface HandleResponsesOptions {
   deferCodexResetDerivedCooldown?: boolean;
   /** 030-owned handoff when a child consumed the original failure under bounds. */
   onConsumedComboFailure?: (failure: ConsumedComboFailure) => void;
+  /** A stored Pool credential was refreshed and its one allowed same-account replay was sent. */
+  onStoredPool401ReplayDispatched?: () => void;
   /** Caller-owned for Chat/Claude replay; omitted only at genuine Responses ingress. */
   translatorBudget?: TranslatorBudget;
   /**
@@ -2289,6 +2302,7 @@ export async function handleComboResponses(
       attemptRetained = true;
     };
     let consumedChildFailure: ConsumedComboFailure | undefined;
+    let storedPool401ReplayDispatched = false;
     const callbackGate = createChildPassthroughCallbackGate(options);
     let response: Response;
     try {
@@ -2315,6 +2329,7 @@ export async function handleComboResponses(
         onCodexAuthContextResolved: value => { resolvedAuth = value; },
         setTerminalOutcomeRecorder: value => { terminalRecorder = value; },
         onConsumedComboFailure: value => { consumedChildFailure = value; },
+        onStoredPool401ReplayDispatched: () => { storedPool401ReplayDispatched = true; },
         onNativePassthroughTerminal: callbackGate.onTerminal,
         onNativePassthroughCancel: callbackGate.onCancel,
       });
@@ -2420,6 +2435,10 @@ export async function handleComboResponses(
     (logCtx.attempts ??= []).push(attempt);
     attemptRetained = true;
     lastFailure = failure.response;
+    if (storedPool401ReplayDispatched) {
+      adoptFailedChildLog(childLog);
+      return lastFailure;
+    }
     if (comboFailureDecision(failure.response.status, failure.classificationText, {
       code: failure.upstreamCode,
     }) === "stop") {
@@ -3839,7 +3858,7 @@ async function handleResponsesInner(
 
     const opaqueBlobRecoveryGuard: OpaqueBlobRecoveryGuard = { attempted: false };
     let oauth401ReplayAttempted = false;
-    let codexMain401ReplayAttempted = false;
+    let codex401ReplayKind: "main" | "stored" | null = null;
     const rateLimitPolicy = rateLimitRetryPolicyFor(route.provider);
     let rateLimitRetries = 0;
     const rebuildAndRefetch = async (
@@ -3914,9 +3933,9 @@ async function handleResponsesInner(
       upstreamResponse.status === 401
       && (authCtx.kind === "main-pool" || authCtx.kind === "pool")
       && usesCodexForwardPoolAuth(authCtx, route.provider)
-      && !codexMain401ReplayAttempted
+      && codex401ReplayKind === null
     ) {
-      codexMain401ReplayAttempted = true;
+      codex401ReplayKind = authCtx.kind === "pool" ? "stored" : "main";
       try { void upstreamResponse.body?.cancel().catch(() => {}); } catch { /* already consumed */ }
       const poolAuthCtx = authCtx.kind === "pool" ? authCtx : undefined;
       const poolReplay = poolAuthCtx
@@ -3978,10 +3997,19 @@ async function handleResponsesInner(
           upstream.signal,
           connectMs,
           parsed.stream,
-          providerFetch(route.provider, options.codexWsRuntimeIdentity, {
-            providerName: route.providerName,
-            modelId: route.modelId,
-          }),
+          // The replay-dispatched signal is what bounds the rest of this logical request, so it
+          // has to describe a send that actually happened. fetchWithHeaderTimeout awaits pacing
+          // admission BEFORE calling the executor, so signalling at the call site would spend the
+          // budget even when a rejected pacing wait means nothing reaches the network. Wrapping
+          // the executor moves the signal to the last moment before the send, where a throw from
+          // here on is a genuine transport attempt.
+          storedPoolReplayDispatchNotifier(
+            providerFetch(route.provider, options.codexWsRuntimeIdentity, {
+              providerName: route.providerName,
+              modelId: route.modelId,
+            }),
+            codex401ReplayKind === "stored" ? options.onStoredPool401ReplayDispatched : undefined,
+          ),
           route.provider.authMode === "forward",
         ).then(response => {
           settleObservedHostResponse();
@@ -3995,7 +4023,7 @@ async function handleResponsesInner(
       continue passthroughRecovery;
     }
 
-    if (codexMain401ReplayAttempted && upstreamResponse.status === 401) break;
+    if (codex401ReplayKind !== null && upstreamResponse.status === 401) break;
 
     // Native Responses providers return before the generic adapter recovery loop below. Keep
     // their OAuth contract identical: one pre-stream 401 forces a credential refresh and one
@@ -4196,6 +4224,14 @@ async function handleResponsesInner(
       }
 
       if (poolRetryOutcome !== undefined) {
+        // A stored Pool 401 spent this request's account budget on its own refresh and replay, so
+        // nothing afterwards may be paid for out of a DIFFERENT account. One flag carries that,
+        // rather than a status check here as well: a quota failure has no same-account move, so
+        // `sameAccountOnly` makes it terminal by refusing the alternate; the gated-model 400
+        // ladder does have one — retrying the account the refreshed roster still grants — and
+        // keeps it. An earlier revision also broke here on a non-400 outcome, which no test could
+        // justify because this flag already produced the identical result.
+        const storedReplaySpent = codex401ReplayKind === "stored";
         const retry = await retryCodexPoolOnAlternateAccount({
           req,
           config,
@@ -4206,6 +4242,7 @@ async function handleResponsesInner(
           firstAuthCtx: authCtx,
           firstResponse: upstreamResponse,
           outcomeStatus: poolRetryOutcome,
+          sameAccountOnly: storedReplaySpent,
           upstream,
           connectMs,
           passthroughEstimate,

src/server/responses/fetch-helpers.ts

@@ -108,6 +108,48 @@ export function providerFetch(
 
 
 
+/**
+ * Wrap a provider fetch so `onDispatch` fires immediately before the send, not before pacing.
+ *
+ * `fetchWithHeaderTimeout` awaits `waitForPacing` and only then calls the executor, so a caller
+ * that signals at the call site records a dispatch even when a rejected pacing wait means nothing
+ * reached the network. That matters when the signal bounds later recovery: the request would lose
+ * its fallback on the strength of a send that never happened.
+ *
+ * The pacing surface is preserved deliberately. `waitForPacing` and `unpacedFetch` are read off
+ * the executor by `fetchWithHeaderTimeout`, so a plain function wrapper would silently drop
+ * provider pacing and double-send the slot.
+ */
+export function storedPoolReplayDispatchNotifier(
+  executor: ProviderFetch,
+  onDispatch: (() => void) | undefined,
+): ProviderFetch {
+  if (!onDispatch) return executor;
+  let notified = false;
+  const notifyOnce = (): void => {
+    if (notified) return;
+    notified = true;
+    onDispatch();
+  };
+  const unpacedSource = executor.unpacedFetch ?? executor;
+  const unpaced = Object.assign(
+    (input: Parameters<typeof globalThis.fetch>[0], init?: RequestInit) => {
+      notifyOnce();
+      return unpacedSource(input, init);
+    },
+    { preconnect: unpacedSource.preconnect },
+  ) as ProviderFetch["unpacedFetch"];
+  const wrapped = async (input: Parameters<typeof globalThis.fetch>[0], init?: RequestInit) => {
+    await executor.waitForPacing?.(init?.signal ?? undefined);
+    return unpaced!(input, init);
+  };
+  return Object.assign(wrapped, {
+    preconnect: executor.preconnect,
+    waitForPacing: executor.waitForPacing,
+    unpacedFetch: unpaced,
+  }) as ProviderFetch;
+}
+
 export async function fetchWithHeaderTimeout(
   url: string,
   init: Omit<RequestInit, "signal">,

src/server/responses/policy-fallback.ts

@@ -116,16 +116,21 @@ export async function handleResponsesWithPolicyFallback(
 ): Promise<Response> {
   const runCore = deps.runCore ?? handleResponsesCore;
   let requestBodyReadNotified = false;
-  const coreOptions: CoreOptions = options.onRequestBodyRead
-    ? {
-      ...options,
+  let storedPool401ReplayDispatched = false;
+  const coreOptions: CoreOptions = {
+    ...options,
+    ...(options.onRequestBodyRead ? {
       onRequestBodyRead: () => {
         if (requestBodyReadNotified) return;
         requestBodyReadNotified = true;
         options.onRequestBodyRead?.();
       },
-    }
-    : options;
+    } : {}),
+    onStoredPool401ReplayDispatched: () => {
+      storedPool401ReplayDispatched = true;
+      options.onStoredPool401ReplayDispatched?.();
+    },
+  };
   let rawBody: Record<string, unknown> | null = null;
   try {
     const parsed = await readJsonRequestBody(req.clone());
@@ -150,7 +155,7 @@ export async function handleResponsesWithPolicyFallback(
     candidateKey({ provider: initialTrace.selected.provider, model: initialTrace.selected.model }),
   ]);
 
-  while (await shouldHopPolicyCandidate(response, req.signal)) {
+  while (!storedPool401ReplayDispatched && await shouldHopPolicyCandidate(response, req.signal)) {
     if (req.signal.aborted) return response;
     const next = rankPolicyFallbackCandidates(initialTrace, tried)[0];
     if (!next) return response;