fix(responses): count Combo recovery sends in the shared send budget

src/server/responses/adapter-dispatch.ts

@@ -2,6 +2,7 @@ import { recoverAutoPausedAccount } from "./account-auto-pause-recovery";
 import { wasAccountAutoPaused, wasPermanentAccountRefusal } from "../../oauth/account-auto-pause";
 import { bindAnthropicRejectionResponse } from "../../adapters/anthropic-rejection-memo";
 import { deferPhysicalSendReporter } from "./deferred-send-reporter";
+import { usesApiKeyAccount } from "../../providers/label";
 import { spendLedgerStorageErrorResponse } from "./spend-storage-error";
 import { classifyAnthropic429 } from "../../oauth/anthropic-rate-limit-policy";
 import { rotateAnthropicAccountOnResponseForInstance } from "../../oauth/anthropic-account-refusal";
@@ -400,7 +401,7 @@ export async function prepareAdapterExchange(
   const autoPauseEnabled = config.providers[route.providerName]?.accountAutoPause?.enabled === true;
   const billingPoolRequest = isAnthropicBillingKeyPool(route.provider);
   let billingRecoveryActive = false;
-  const genericRefetchCountsSends = () => sendBudgetState.adapterSendBudget?.spendEnforced === true
+  const genericRefetchCountsSends = () => options.comboAttempt === true || sendBudgetState.adapterSendBudget?.spendEnforced === true
     || transientRetryPolicyFor(route.provider) !== null || resetReplayPolicyFor(route.provider) !== null
     || billingPoolRequest || billingRecoveryActive || autoPauseEnabled;
   let handoffPermit: ResponsesSendBudget["pendingHopPermit"];
@@ -762,7 +763,9 @@ export async function prepareAdapterExchange(
             : undefined;
           let refetchPermitUsed = false;
           const refetchReporter = refetchAllowance ? transientSendReporter(refetchAllowance.permit) : undefined;
-          const deferredRefetch = billingPoolRequest && refetchReporter ? deferPhysicalSendReporter(refetchReporter) : undefined;
+          const deferredRefetch = (billingPoolRequest || options.comboAttempt) && refetchReporter
+            ? deferPhysicalSendReporter(refetchReporter) : undefined;
+          const pendingKeyMetadata = deferredRefetch && usesApiKeyAccount(route.provider);
           try {
             return await refetchWithPolicy(
               recoveryKind => {
@@ -771,14 +774,14 @@ export async function prepareAdapterExchange(
                 });
                 // Prepare key metadata before the guarded transport commits its attempt;
                 // counters and permits still wait for the physical executor after pacing.
-                if (deferredRefetch) transportState.noteRoutedAttemptSend(retryEstimate, recoveryKind ?? recovery);
+                if (pendingKeyMetadata) transportState.noteRoutedAttemptSend(retryEstimate, recoveryKind ?? recovery);
                 const confirmDispatch = () => {
                   if (refetchAllowance?.permit && !refetchPermitUsed) {
                     if (!refetchAllowance.permit.use()) throw new SendBudgetExhaustedError(safeHostLabel(retryRequest.url));
                     refetchPermitUsed = true;
                   }
-                  if (!deferredRefetch) transportState.noteRoutedAttemptSend(retryEstimate, recoveryKind ?? recovery);
-                  // Eligible pools confirm after pacing at the physical executor; legacy
+                  if (!pendingKeyMetadata) transportState.noteRoutedAttemptSend(retryEstimate, recoveryKind ?? recovery);
+                  // Counted Combo/pool sends confirm at the physical executor; legacy
                   // helper paths retain their existing thunk admission boundary.
                   onDispatch?.();
                   if (preserveFailureResponse) replacementAdmitted = true;
@@ -891,7 +894,7 @@ export async function prepareAdapterExchange(
       const adapterOwnsDispatch = transportState.activeAdapter.fetchResponse !== undefined;
       const hop = reserveCredentialHop("auth-recovery",
         `${route.providerName}|${route.modelId}|antigravity-auth`,
-        !adapterOwnsDispatch && (transientRetryPolicyFor(route.provider) !== null || resetReplayPolicyFor(route.provider) !== null));
+        !adapterOwnsDispatch && genericRefetchCountsSends());
       if (!hop.allowed) return null;
       try {
         const snapshot = await failoverAccountSnapshot(route.providerName, nextId);
@@ -991,7 +994,7 @@ export async function prepareAdapterExchange(
               const adapterOwnsDispatch = transportState.activeAdapter.fetchResponse !== undefined;
               const hop = reserveCredentialHop("auth-recovery",
                 `${route.providerName}|${route.modelId}|terminal-refresh-account`,
-                !adapterOwnsDispatch && (transientRetryPolicyFor(route.provider) !== null || resetReplayPolicyFor(route.provider) !== null));
+                !adapterOwnsDispatch && genericRefetchCountsSends());
               if (hop.allowed) {
                 try {
                   const admitted = await applyFailoverSnapshot(alternate);
@@ -1361,12 +1364,31 @@ export async function prepareAdapterExchange(
         && anthropicInstance
         && transportState.anthropicPoolAccountId
       ) {
-        const nextAccountId = await rotateAnthropicAccountOnResponseForInstance(anthropicInstance, upstreamResponse, {
-          config, accountId: transportState.anthropicPoolAccountId, sessionKey: anthropicSessionKey,
-          model: route.modelId, requestKey: transportState, decision: transportState.anthropicRouteDecision, currentDecision: currentAnthropicRouteDecision, signal: upstream.signal,
-          canRetry: !sendBudgetExhausted() && transportState.anthropicPoolFailovers < ANTHROPIC_POOL_MAX_FAILOVERS_PER_REQUEST,
-        });
-        if (!nextAccountId) break;
+        // Same admission as the other rotation ladders: the roster cap cannot see the shared
+        // send budget. A canRetry term, not a break -- the call also records the refusal.
+        const adapterOwnsDispatch = transportState.activeAdapter.fetchResponse !== undefined;
+        const hop = reserveCredentialHop(
+          "auth-recovery",
+          `${route.providerName}|${route.modelId}|anthropic-pool-rotation`,
+          !adapterOwnsDispatch && genericRefetchCountsSends(),
+        );
+        let nextAccountId: string | null;
+        try {
+          nextAccountId = await rotateAnthropicAccountOnResponseForInstance(anthropicInstance, upstreamResponse, {
+            config, accountId: transportState.anthropicPoolAccountId, sessionKey: anthropicSessionKey,
+            model: route.modelId, requestKey: transportState, decision: transportState.anthropicRouteDecision, currentDecision: currentAnthropicRouteDecision, signal: upstream.signal,
+            // A denied reservation must not rotate on base headroom its bounds cannot see.
+            canRetry: hop.allowed && (hop.permit !== undefined || !sendBudgetExhausted())
+              && transportState.anthropicPoolFailovers < ANTHROPIC_POOL_MAX_FAILOVERS_PER_REQUEST,
+          });
+        } catch (rotationError) {
+          hop.permit?.release();
+          throw rotationError;
+        }
+        if (!nextAccountId) {
+          hop.permit?.release();
+          break;
+        }
         try {
           const admitted = await commitResolvedOAuthSelection(await anthropicRoutingFor(anthropicInstance).getAnthropicPoolAccessSnapshot(nextAccountId));
           if (!admitted) throw new Error("OAuth selection changed during recovery");
@@ -1389,12 +1411,26 @@ export async function prepareAdapterExchange(
           });
           sealRequestAttemptIdentity(logCtx.activeAttempt, logCtx.provider, transportState.activeAdapter.name, logCtx.accountLogLabel);
           recordAttemptCredentialSource(logCtx.activeAttempt, route.providerName, route.provider, transportState.activeAdapter.name);
-         const result = await rebuildAndRefetch(upstreamResponse.status === 401 ? "oauth-401"
-           : upstreamResponse.status === 403 ? "oauth-account-403" : "anthropic-oauth-429");
-         if ("failed" in result) return result.failed;
-         upstreamResponse = result;
+          // The replay IS this hop's send; the dispatching layer settles the reservation.
+          sendBudgetState.pendingHopPermit = hop.permit;
+          let result: Response | { failed: Response };
+          try {
+            result = await rebuildAndRefetch(upstreamResponse.status === 401 ? "oauth-401"
+              : upstreamResponse.status === 403 ? "oauth-account-403" : "anthropic-oauth-429",
+              () => { if (!adapterOwnsDispatch) hop.permit?.use(); });
+          } finally {
+            sendBudgetState.pendingHopPermit = undefined;
+          }
+          if ("failed" in result) {
+            // A no-op once the boundary was reached; a refund when the rebuild died first.
+            hop.permit?.release();
+            return result.failed;
+          }
+          upstreamResponse = result;
           if (isNonReplayableResponse(upstreamResponse)) continue recovery;
-       } catch {
+        } catch {
+          // release() refunds a pre-send hop and is a no-op once the send settled it.
+          hop.permit?.release();
           break;
         }
       }
@@ -1434,9 +1470,9 @@ export async function prepareAdapterExchange(
         const hop = reserveCredentialHop(
           "auth-recovery",
           `${route.providerName}|${route.modelId}|adapter-recovery-oauth-429`,
-          // Transient and opted-in reset helpers report physical sends; an adapter-owned
+          // Counted generic helpers report physical sends; an adapter-owned
           // ladder settles its own booking.
-          !adapterOwnsDispatch && (transientRetryPolicyFor(route.provider) !== null || resetReplayPolicyFor(route.provider) !== null),
+          !adapterOwnsDispatch && genericRefetchCountsSends(),
         );
         if (!hop.allowed) break;
         const nextAccountId = rotateGenericOAuthAccountOnRefusal(
@@ -1538,9 +1574,9 @@ export async function prepareAdapterExchange(
         const hop = reserveCredentialHop(
           "auth-recovery",
           `${route.providerName}|${route.modelId}|adapter-recovery-oauth-429`,
-          // Transient and opted-in reset helpers report physical sends; an adapter-owned
+          // Counted generic helpers report physical sends; an adapter-owned
           // ladder settles its own booking.
-          !adapterOwnsDispatch && (transientRetryPolicyFor(route.provider) !== null || resetReplayPolicyFor(route.provider) !== null),
+          !adapterOwnsDispatch && genericRefetchCountsSends(),
         );
         if (!hop.allowed) break;
         const nextAccountId = rotateGenericOAuthAccountOn429(
@@ -1659,7 +1695,7 @@ export async function prepareAdapterExchange(
         const hop = reserveCredentialHop(
           "auth-recovery",
           `${route.providerName}|${route.modelId}|adapter-recovery-oauth-verify`,
-          !adapterOwnsDispatch && (transientRetryPolicyFor(route.provider) !== null || resetReplayPolicyFor(route.provider) !== null),
+          !adapterOwnsDispatch && genericRefetchCountsSends(),
         );
         if (!hop.allowed) break;
         // Auth-refusal rotation, not the rate-limit one: a verification refusal

structure/transports/anthropic-thinking-recovery.md

@@ -22,6 +22,11 @@ signature recovery. Empty-completion retries in `src/server/responses/adapter-co
 rebuild through their existing pacing/send owner. Ordinary continuations after output cannot
 recover signatures. Anthropic has no `runTurn` recovery.
 
+Combo signature repairs share the [physical-send allowance](responses-spend.md#prepaid-initial-sends)
+with initial sends and later targets, including generic providers without an opt-in retry policy.
+The HTTP and initial-stream repair paths settle each physical send once; a pacing refusal leaves
+the unsent repair uncharged. Recovery preserves the original thinking classification and stripping.
+
 Only a delivery-validated completed response with actual text/tool output and an observed
 successful terminal can memoize the rejection. Errors, incomplete/empty responses, missing
 terminals, cancellation and changed serving identity cannot memoize it. The existing thread and

structure/transports/responses-spend.md

@@ -11,6 +11,9 @@ That reservation occupies the shared ceiling immediately; it is not a completed
 `comboInitialSend` hands only that child's single-use permit to `request-send-budget.ts`.
 `initialSendAllowance` includes that one prepaid send while `remainingBaseSends` keeps every
 booking spent. Other pending external bookings never become this child's free headroom.
+Generic Combo rebuilds use the same shared allowance even without an opt-in retry or spend
+policy. Their reporter settles at physical executor admission after pacing, using the exact
+prepaid recovery permit when present; unsent repairs and credential hops remain refundable.
 Configured transient totals subtract an owner-local counter updated by physical receipts, adapter
 observers and retry-helper reports, not shared-ledger deltas. Earlier Combo targets, later bookings
 by other owners and the child's unsent booking do not consume that total; the shared ceiling and

tests/adapters/anthropic/anthropic-combo-account-cooldown.test.ts

@@ -23,6 +23,8 @@ import { getAccountSet, saveCredential, setActiveAccount } from "../../../src/oa
 import { clearAccountQuotaCache, resetProviderQuotaReconcileStateForTests } from "../../../src/providers/quota";
 import { clearResponseStateForTests, flushResponseState } from "../../../src/responses/state";
 import { handleResponses } from "../../../src/server/responses";
+import { createRequestExecutionBudget } from "../../../src/lib/request-execution-budget";
+import type { RequestExecutionBudget } from "../../../src/lib/request-execution-budget";
 import type { OcxConfig, OcxProviderConfig } from "../../../src/types";
 import { acquireOwnedSpendHome } from "../../helpers/owned-spend-home";
 import { removeTreeWithRetry } from "../../helpers/remove-tree";
@@ -85,11 +87,11 @@ function configFor(reply: () => Response, poolEnabled = true): OcxConfig {
   };
 }
 
-async function post(config: OcxConfig): Promise<Response> {
+async function post(config: OcxConfig, sendBudget?: RequestExecutionBudget): Promise<Response> {
   const response = await handleResponses(new Request("http://localhost/v1/responses", {
     method: "POST", headers: { "content-type": "application/json" },
     body: JSON.stringify({ model: "combo/pooled", input: "Answer briefly", stream: false }),
-  }), config, { model: "", provider: "" });
+  }), config, { model: "", provider: "" }, sendBudget ? { sendBudget } : undefined);
   await response.text();
   return response;
 }
@@ -131,7 +133,8 @@ test("a spent pool account does not cool the combo target for the accounts that
   let refuse = true;
   const config = configFor(() => refuse ? spent() : answered());
 
-  const refused = await post(config);
+  const budget = createRequestExecutionBudget();
+  const refused = await post(config, budget);
   expect(refused.status).toBe(429);
   // Only the accounts this request actually reached are cooled, and the pool -- not the combo
   // layer -- is what cooled them.
@@ -140,6 +143,11 @@ test("a spent pool account does not cool the combo target for the accounts that
   expect(cooled.length).toBeLessThan(ids.length);
   // The last refused account must be recorded even though the request spent its retry budget.
   expect(cooled).toHaveLength(4);
+  // The reservation IS the charge, so the shared counter must equal the sends the transport
+  // actually made: four physical sends, four booked -- an unsettled booking or a double charge
+  // would each break this equality even when the account list still says four.
+  expect(sentTokens).toHaveLength(4);
+  expect(budget.used).toBe(sentTokens.length);
   expect(isComboTargetInCooldown("pooled", TARGET)).toBe(false);
 
   refuse = false;
@@ -160,6 +168,58 @@ test("a spent pool account does not cool the combo target for the accounts that
   expect(cooled).not.toContain(ids[servedIndex]!);
 }, 20_000);
 
+test("a denied hop reservation with base headroom sends nothing more but still records the refusal", async () => {
+  const ids = await seed(5);
+  const config = configFor(spent);
+  // Refuse every deferred (pre-dispatch) booking once two physical sends have left: the second
+  // hop is denied while baseSendAllowance still has room, the way a shared-total or spend-
+  // observer refusal denies it. Post-dispatch reports stay recordable (alreadySent).
+  const budget = createRequestExecutionBudget(undefined, undefined, {
+    charge: options => options?.deferDispatch === true ? sentTokens.length < 2 : true,
+    refund: () => {},
+  });
+  const response = await post(config, budget);
+  expect(response.status).toBe(429);
+  // The denied hop produced neither a physical send nor an account switch beyond the recorded
+  // refusal, and it left no unsettled reservation: the counter equals the sends on the wire.
+  expect(sentTokens).toHaveLength(2);
+  expect(budget.used).toBe(sentTokens.length);
+  const cooled = ids.filter(id => getAnthropicAccountHealthSnapshot(id) !== null);
+  expect(cooled).toHaveLength(2);
+  expect(isComboTargetInCooldown("pooled", TARGET)).toBe(false);
+}, 20_000);
+
+test("a rotation-helper throw after hop reservation releases the permit and still propagates", async () => {
+  await seed(2);
+  const config = configFor(spent);
+  // First send goes out with the real pool config; once the refusal lands, the helper's own
+  // config reads start throwing -- a stand-in for the verdict/store reads that can reject
+  // inside the rotation helper after the hop was already reserved.
+  const pool = config.anthropicAccountPool;
+  Object.defineProperty(config, "anthropicAccountPool", {
+    configurable: true,
+    get() { if (sentTokens.length > 0) throw new Error("synthetic pool-config read failure"); return pool; },
+  });
+  let hopCharges = 0;
+  let refunds = 0;
+  const budget = createRequestExecutionBudget(undefined, undefined, {
+    charge: options => { if (options?.deferDispatch === true && sentTokens.length > 0) hopCharges += 1; return true; },
+    refund: () => { refunds += 1; },
+  });
+  let thrown: unknown;
+  try { await post(config, budget); } catch (error) { thrown = error; }
+  // The helper's throw is the request's rejection itself: the policy-fallback seam converts
+  // only pacing overloads, so nothing synthesizes a 5xx here and post() must not resolve.
+  expect(thrown).toBeInstanceOf(Error);
+  expect((thrown as Error).message).toBe("synthetic pool-config read failure");
+  // The hop really was reserved before the throw, and its release refunded the charge:
+  // the counter equals the one physical send that actually left.
+  expect(sentTokens).toHaveLength(1);
+  expect(hopCharges).toBe(1);
+  expect(refunds).toBe(1);
+  expect(budget.used).toBe(sentTokens.length);
+}, 20_000);
+
 test("every pool account spent still cools the combo target", async () => {
   await seed(2);
   const config = configFor(spent);