fix(lib): make the dispatch permit the charge, and close the uncounted send paths (#4546)

scripts/test-layout/layout.json

@@ -1444,7 +1444,8 @@
     "main-device-reauth-api.test.ts": "codex-integration",
     "main-device-reauth-ui.test.ts": "gui",
     "adapter-input-media-guard.test.ts": "adapters",
-    "chat-media-translation.test.ts": "responses"
+    "chat-media-translation.test.ts": "responses",
+    "execution-budget-permits.test.ts": "lib"
   },
   "migrated": [
     "adapters",

src/lib/request-execution-budget.ts

@@ -72,17 +72,28 @@ export interface DispatchIntent {
   readonly replaySafe?: boolean;
   /**
    * True when the physical send is already reported through another counter -- the retry
-   * helpers' `onSendsConsumed` hook. The permit then books the reserve, alternate-target and
-   * transition ledgers but leaves `used` to that reporter, because charging both is how a
-   * four-send cap silently becomes a two-send cap.
+   * helpers' `onSendsConsumed` hook. The send is still booked at reservation time, because an
+   * advisory reservation cannot stop a concurrent leg; what changes is that the booking is
+   * PENDING, and the first send the external reporter names settles it instead of adding a
+   * second charge. Charging both is how a four-send cap silently becomes a two-send cap.
    */
   readonly countedExternally?: boolean;
 }
 
 export interface SingleUseDispatchPermit {
   readonly sendClass: SendClass;
-  /** Consume exactly once. A second call returns false and charges nothing. */
+  /**
+   * Confirm the dispatch this permit already paid for. The reservation is the charge, so this
+   * charges nothing; it is how a leg proves it is the one that sent. A second call returns
+   * false, which is what keeps a retry thunk from sending twice on one permit.
+   */
   use(): boolean;
+  /**
+   * Hand back a reservation that never dispatched -- a credential move that found no alternate,
+   * a rebuild abandoned before the send. Idempotent, and a no-op once the permit was used or
+   * once an external send reporter already settled it.
+   */
+  release(): void;
 }
 
 export type DispatchDecision =
@@ -103,6 +114,9 @@ export interface RequestExecutionBudget extends TransientSendBudget {
    * Sends still available from the base allowance, capped by a layer's own maximum.
    * Returns 0 when the allowance is gone -- it never floors to 1, because a floor of 1 is
    * what let every recovery leg send one more time forever.
+   *
+   * A reserved-but-unconfirmed send is spent for this purpose. The alternative -- counting only
+   * confirmed sends -- is what let two legs read the same remainder and both dispatch.
    */
   remainingBaseSends(cap: number): number;
   readonly reserveSpent: boolean;
@@ -124,13 +138,30 @@ export function createRequestExecutionBudget(
   policy: RequestExecutionBudgetPolicy = CODEX_TEXT_GUARDED_BUDGET_POLICY,
   logicalRequestId?: string,
 ): RequestExecutionBudget {
+  let spent = 0;
+  // Reservations whose physical send is reported by a retry helper rather than by the permit.
+  // They are already charged; the reporter's first send settles one instead of charging again.
+  let pendingExternalSends = 0;
   let reserveSpent = false;
   let alternateTargetSends = 0;
   let targetTransitions = 0;
   let lastTargetKey: string | undefined;
 
   const budget: RequestExecutionBudget = {
-    used: 0,
+    get used(): number { return spent; },
+    set used(next: number) {
+      // The retry helpers report their real send count by assigning through this field. A
+      // reservation taken with `countedExternally` has already booked one of those sends, so
+      // the report settles the pending booking first and only the surplus is charged.
+      const delta = next - spent;
+      if (delta <= 0) {
+        spent = Math.max(0, next);
+        return;
+      }
+      const settled = Math.min(delta, pendingExternalSends);
+      pendingExternalSends -= settled;
+      spent += delta - settled;
+    },
     logicalRequestId: logicalRequestId ?? `lr-${Date.now().toString(36)}-${(logicalRequestSeq += 1).toString(36)}`,
     policyVersion: REQUEST_BUDGET_POLICY_VERSION,
     policy,
@@ -140,11 +171,11 @@ export function createRequestExecutionBudget(
     get lastTargetKey() { return lastTargetKey; },
     remainingBaseSends(cap: number): number {
       const capped = Number.isFinite(cap) ? Math.trunc(cap) : 0;
-      return Math.max(0, Math.min(capped, policy.baseSendAllowance - budget.used));
+      return Math.max(0, Math.min(capped, policy.baseSendAllowance - spent));
     },
     reserveDispatch(intent: DispatchIntent): DispatchDecision {
       if (intent.replaySafe === false) return { allowed: false, reason: "not-replay-safe" };
-      if (budget.used >= policy.maxTotalModelSends) return { allowed: false, reason: "total-exhausted" };
+      if (spent >= policy.maxTotalModelSends) return { allowed: false, reason: "total-exhausted" };
 
       const changesTarget = lastTargetKey !== undefined && lastTargetKey !== intent.targetKey;
       const isAlternateTarget = changesTarget || intent.sendClass === "account-failover"
@@ -159,7 +190,7 @@ export function createRequestExecutionBudget(
       // The base allowance is spent first. Only once it is gone does a recovery class reach
       // for the single shared reserve -- an account move and a validated rebuild cannot each
       // take one.
-      const drawsReserve = budget.remainingBaseSends(policy.baseSendAllowance) === 0;
+      const drawsReserve = policy.baseSendAllowance - spent <= 0;
       if (drawsReserve) {
         if (!RESERVE_FUNDED_CLASSES.has(intent.sendClass)) {
           return { allowed: false, reason: "base-allowance-exhausted" };
@@ -169,29 +200,47 @@ export function createRequestExecutionBudget(
         }
       }
 
-      let consumed = false;
+      // THE RESERVATION IS THE CHARGE. Deciding here and charging in `use()` left a window in
+      // which two legs read the same remainder, both received a permit, and both dispatched:
+      // one remaining send admitted two physical sends, which is the per-request multiplication
+      // this budget exists to stop. Everything is booked now; `release()` is the way back.
+      const previousTargetKey = lastTargetKey;
+      spent += 1;
+      if (intent.countedExternally === true) pendingExternalSends += 1;
+      if (drawsReserve) reserveSpent = true;
+      if (isAlternateTarget) alternateTargetSends += 1;
+      if (changesTarget) targetTransitions += 1;
+      lastTargetKey = intent.targetKey;
+
+      let settled: "open" | "used" | "released" = "open";
       return {
         allowed: true,
         permit: {
           sendClass: intent.sendClass,
           use(): boolean {
-            if (consumed) return false;
-            consumed = true;
-            // Charged here, immediately before the physical send, rather than reported after
-            // the helper returns: a counter that is only reconciled afterwards cannot stop two
-            // concurrent legs that both read the same remainder.
-            if (intent.countedExternally !== true) budget.used += 1;
-            if (drawsReserve) reserveSpent = true;
-            if (isAlternateTarget) alternateTargetSends += 1;
-            if (changesTarget) targetTransitions += 1;
-            lastTargetKey = intent.targetKey;
+            if (settled !== "open") return false;
+            settled = "used";
             return true;
           },
+          release(): void {
+            if (settled !== "open") return;
+            settled = "released";
+            // An externally counted reservation the reporter already settled paid for a send
+            // that physically happened. Refunding it would hand the request a free send back.
+            if (intent.countedExternally === true) {
+              if (pendingExternalSends === 0) return;
+              pendingExternalSends -= 1;
+            }
+            spent -= 1;
+            if (drawsReserve) reserveSpent = false;
+            if (isAlternateTarget) alternateTargetSends -= 1;
+            if (changesTarget) targetTransitions -= 1;
+            lastTargetKey = previousTargetKey;
+          },
         },
       };
     },
   };
-  if (lastTargetKey === undefined) lastTargetKey = undefined;
   return budget;
 }
 

src/lib/upstream-retry.ts

@@ -357,17 +357,22 @@ export interface ResetRetryOptions {
   label?: string;
   /** Total upstream sends allowed, including the first one. Not a per-layer retry count. */
   attempts?: number;
-}
-
-export interface TransientRetryOptions extends ResetRetryOptions {
-  /** Test seam: per-attempt slow budget override (defaults to TRANSIENT_RETRY_SLOW_ATTEMPT_MS). */
-  slowAttemptMs?: number;
   /**
    * Reports how many upstream sends this call actually consumed, so a caller that spans
    * several legs of one request (initial send, then a 429/account-recovery refetch) can
    * keep them on ONE budget instead of handing each leg a fresh one.
+   *
+   * It lives on the RESET options, not on the transient ones, because every leg that falls
+   * back to reset-only retry -- the non-policy adapter initial send, and every
+   * `rebuildAndRefetch` recovery kind whose provider has no transient policy -- was not merely
+   * uncounted but UNCOUNTABLE: the callback existed on a type those call sites never reach.
    */
   onSendsConsumed?: (sends: number) => void;
+}
+
+export interface TransientRetryOptions extends ResetRetryOptions {
+  /** Test seam: per-attempt slow budget override (defaults to TRANSIENT_RETRY_SLOW_ATTEMPT_MS). */
+  slowAttemptMs?: number;
   /**
    * How long this caller can wait on an honoured `Retry-After`, defaulting to
    * {@link RETRY_AFTER_CEILING_MS}. It is a deadline, never a clamp: an instruction inside it
@@ -455,6 +460,10 @@ export async function fetchWithResetRetry(
   let sawReset = false;
   for (let attempt = 0; attempt < attempts; attempt++) {
     if (opts.abortSignal?.aborted) throw abortError(opts.abortSignal);
+    // Reported before the await, one physical send at a time: a send that rejects has still
+    // been made, and this helper leaves through four exits (return, reset give-up, non-reset
+    // rethrow, abort), so a per-send report is the only shape that is correct on all of them.
+    opts.onSendsConsumed?.(1);
     try {
       return await doFetch(attempt === 0 ? firstRecovery : "connection-reset");
     } catch (err) {
@@ -517,13 +526,22 @@ export async function fetchWithTransientRetry(
   // more send -- the loop condition alone was never enough, because every later recovery leg
   // called this helper again and the floor funded each of them.
   const remaining = () => Math.max(0, budget - sent);
+  // The inner reset layer now has its own `onSendsConsumed`, and these are the same physical
+  // sends `countedFetch` already counts. Forwarding the reporter down the `remaining()` path
+  // would report each of them twice, which is how a four-send cap becomes a two-send cap. One
+  // send is counted once, by the outermost layer that owns the budget.
+  const innerResetOptions = (): ResetRetryOptions => ({
+    ...opts,
+    attempts: remaining(),
+    onSendsConsumed: undefined,
+  });
   // Reported in `finally` rather than at each exit: this function returns from five places
   // and throws from one, and a caller sharing the budget across request legs must be told the
   // real count on every one of them.
   try {
   if (budget === 0) throw new SendBudgetExhaustedError(opts.label);
   let attemptStart = Date.now();
-  let res = await fetchWithResetRetry(countedFetch, { ...opts, attempts: remaining() });
+  let res = await fetchWithResetRetry(countedFetch, innerResetOptions());
   for (let attempt = 0; sent < budget; attempt++) {
     // A non-replayable gateway status was settled after the request body had already left
     // for the origin; retrying it here is the automatic resend the marker exists to forbid.
@@ -561,7 +579,7 @@ export async function fetchWithTransientRetry(
     attemptStart = Date.now();
     transientStatuses.push(res.status);
     try {
-      res = await fetchWithResetRetry(countedFetch, { ...opts, attempts: remaining() }, "transient-5xx");
+      res = await fetchWithResetRetry(countedFetch, innerResetOptions(), "transient-5xx");
     } catch (err) {
       // Keep the prior 5xx evidence attached: the origin already responded, so
       // this rejection is not pre-connection and must not classify as neutral.

src/server/responses/compact.ts

@@ -682,6 +682,13 @@ export async function handleResponsesCompact(
   // Combo-resolved targets skip native compact so failover can advance through the
   // combo target list when the picked model returns 429/5xx — the routed path below
   // dispatches through handleResponses → handleComboResponses with full failover.
+  //
+  // One holder for the WHOLE logical compact, declared above the native branch because the
+  // routed fallback below is not a different request: a native attempt that 404s, or a quota
+  // failure that hands off, continues here. The routed turn used to call handleResponses with
+  // no budget at all, so `handleResponsesInner` minted a fresh four after the native attempt
+  // had already spent some of the first one.
+  const sendBudget: RequestExecutionBudget = options.sendBudget ?? createRequestExecutionBudget();
   if (supportsNativeResponsesCompactEndpoint(route.providerName, route.provider) && !accountGatedCompactWireModel && !route.combo) {
     if (req.signal.aborted) {
       return formatErrorResponse(499, "client_cancelled", "Client cancelled compact request");
@@ -774,9 +781,6 @@ export async function handleResponsesCompact(
     // so routed-model reasoning items (reasoning_text content) don't 400 the ChatGPT backend.
     const compactBody = sanitizeReasoningInputContent(compactBodyRaw) as typeof compactBodyRaw;
     const compactUrl = `${base}/responses/compact`;
-    // One holder for this logical compact, inherited by the handoff child so a second model
-    // does not start over with a fresh four.
-    const sendBudget: RequestExecutionBudget = options.sendBudget ?? createRequestExecutionBudget();
     const compactTargetKey = `${route.providerName}|${route.modelId}|compact`;
     const actualCompactHostKey = upstreamHostHealthKey(
       route.providerName,
@@ -1223,7 +1227,10 @@ export async function handleResponsesCompact(
     body: JSON.stringify(internalBody),
   });
   linkRequestSessionLane(req, internalReq);
-  const response = await handleResponses(internalReq, config, logCtx, { abortSignal: req.signal, turnAdmissionLease, ...(admission ? { admission } : {}) });
+  // The routed compaction turn is a handoff inside the same logical request, so it draws the
+  // REMAINDER. Minting here is what let a native attempt spend three sends and the routed
+  // fallback spend four more.
+  const response = await handleResponses(internalReq, config, logCtx, { abortSignal: req.signal, turnAdmissionLease, sendBudget, ...(admission ? { admission } : {}) });
   if (!response.ok) return response;
   let json: { output?: unknown[]; status?: unknown; error?: unknown };
   if (response.headers.get("content-type")?.includes("text/event-stream")) {