fix(adapters): charge adapter-owned retry sends to the request send budget

scripts/test-layout/layout.json

@@ -188,6 +188,7 @@
     "adapter-event-oauth-failover.test.ts": "oauth",
     "adapter-inner-send-budget-wiring.test.ts": "adapters",
     "adapter-inner-send-budget.test.ts": "adapters",
+    "physical-send.test.ts": "adapters",
     "adapter-registry-authority.test.ts": "adapters",
     "adapter-resolve.test.ts": "server",
     "adapter-tool-conformance.test.ts": "adapters",

src/adapters/command-code.ts

@@ -13,6 +13,8 @@ import { commandCodeReasoningEfforts, refreshCommandCodeReasoningEfforts } from
 import { identifyRoutedModel } from "./identity";
 import { buildNonOpenAIToolCatalogNudgeForTools } from "./tool-catalog-nudge";
 import { parseDataUrl } from "./image";
+import { createAdapterPhysicalSend } from "./physical-send";
+import { SendBudgetExhaustedError } from "../lib/upstream-retry";
 
 // Retain the short ids emitted by the first local integration. New requests use the live catalog's
 // provider-native IDs directly; this map is compatibility-only and is not a model fallback list.
@@ -469,7 +471,7 @@ async function fetchCommandCode(request: AdapterRequest, ctx: AdapterFetchContex
   const timer = setTimeout(() => timeout.abort(new DOMException("Timeout elapsed", "TimeoutError")), ctx?.timeoutMs ?? 200_000);
   const callerSignal = ctx?.abortSignal ?? new AbortController().signal;
   try {
-    return await (ctx?.executor ?? executor)(request.url, {
+    return await executor(request.url, {
       method: request.method,
       headers: request.headers,
       body: request.body,
@@ -556,7 +558,8 @@ export function createCommandCodeAdapter(provider: OcxProviderConfig): ProviderA
       };
     },
     async fetchResponse(request: AdapterRequest, ctx?: AdapterFetchContext): Promise<Response> {
-      const response = await fetchCommandCode(request, ctx, executor);
+      const send = createAdapterPhysicalSend(ctx, executor);
+      const response = await send({ url: request.url, dispatch: physical => fetchCommandCode(request, ctx, physical) });
       if (response.ok) return response;
       const currentEffort = (() => {
         try { return (JSON.parse(request.body) as { params?: { reasoning_effort?: unknown } }).params?.reasoning_effort; } catch { return undefined; }
@@ -577,8 +580,14 @@ export function createCommandCodeAdapter(provider: OcxProviderConfig): ProviderA
       if (!refreshed || refreshed.includes(currentEffort)) return response;
       const retry = requestWithoutReasoningEffort(request);
       if (!retry) return response;
-      try { void response.body?.cancel(); } catch { /* already closed */ }
-      return fetchCommandCode(retry, ctx, executor);
+      try {
+        return await send({ url: retry.url, sendClass: "repair", recovery: "reasoning-effort-downgrade",
+          beforeDispatch: () => { try { void response.body?.cancel().catch(() => {}); } catch { /* already closed */ } },
+          dispatch: physical => fetchCommandCode(retry, ctx, physical) });
+      } catch (error) {
+        if (error instanceof SendBudgetExhaustedError) return response;
+        throw error;
+      }
     },
     async *parseStream(response: Response, budget: TranslatorBudget): AsyncGenerator<AdapterEvent> {
       let sawFinish = false;

src/adapters/google-http.ts

@@ -1,4 +1,7 @@
 import type { AdapterFetchContext, AdapterRequest } from "./base";
+import { createAdapterPhysicalSend } from "./physical-send";
+import type { SendClass } from "../lib/request-execution-budget";
+import type { AttemptRecoveryKind } from "../usage/log";
 import { isQuotaExhaustedBody, retryableGoogleStatus, safeGoogleHttpErrorMessage } from "./google-errors";
 import { repairGoogleInvalidRequestBody } from "./google-wire-compiler";
 import { normalizeUpstreamHttpErrorResponse, readDisplaySafeErrorPayloadText } from "./upstream-http-error";
@@ -8,6 +11,8 @@ import {
   fetchWithAttemptDeadline,
   retryBackoffDelayMs,
   sleepWithAbort,
+  SendBudgetExhaustedError,
+  isConnectionResetError,
 } from "../lib/upstream-retry";
 
 const GOOGLE_RETRY_ATTEMPTS = 3;
@@ -41,18 +46,30 @@ export async function fetchGoogleWithRetry(
 ): Promise<Response> {
   const repairInvalid400 = opts.repairInvalid400 ?? true;
   const timeoutMs = ctx.timeoutMs ?? 200_000;
-  const executor = ctx.executor ?? globalThis.fetch;
+  const send = createAdapterPhysicalSend(ctx);
   let lastError: unknown;
   let activeRequest = request;
   let compatibilityReplayUsed = false;
+  let pendingResponse: Response | undefined;
+  let retryDelayMs = 0;
+  let sendClass: SendClass = "transient";
+  let recovery: AttemptRecoveryKind | undefined;
   for (let attempt = 0; attempt < GOOGLE_RETRY_ATTEMPTS; attempt++) {
     if (ctx.abortSignal?.aborted) throw abortError(ctx.abortSignal);
     try {
-      const res = await fetchWithAttemptDeadline(activeRequest.url, {
-        method: activeRequest.method,
-        headers: activeRequest.headers,
-        body: activeRequest.body,
-      }, timeoutMs, ctx.abortSignal, ctx.stream, executor);
+      const res = await send({ url: activeRequest.url, sendClass, recovery,
+        beforeDispatch: async () => {
+          if (retryDelayMs > 0) await sleepWithAbort(retryDelayMs, ctx.abortSignal);
+          if (pendingResponse) cancelResponseBodyBestEffort(pendingResponse);
+          pendingResponse = undefined;
+        },
+        dispatch: executor => fetchWithAttemptDeadline(activeRequest.url, {
+          method: activeRequest.method, headers: activeRequest.headers, body: activeRequest.body,
+        }, timeoutMs, ctx.abortSignal, ctx.stream, executor),
+      });
+      retryDelayMs = 0;
+      sendClass = "transient";
+      recovery = undefined;
       if (res.status === 400 && repairInvalid400 && !compatibilityReplayUsed) {
         let payloadText = "";
         try {
@@ -64,7 +81,8 @@ export async function fetchGoogleWithRetry(
         if (repairedBody !== undefined) {
           compatibilityReplayUsed = true;
           activeRequest = { ...activeRequest, body: repairedBody };
-          cancelResponseBodyBestEffort(res);
+          pendingResponse = res;
+          sendClass = "repair";
           attempt--; // The changed-request replay is separate from transient retry accounting.
           continue;
         }
@@ -75,7 +93,7 @@ export async function fetchGoogleWithRetry(
       // A 429 may be a transient rate limit (retry) or hard quota exhaustion (do NOT retry —
       // it won't recover for hours and burns retries). Peek the body to tell them apart.
       if (res.status === 429) {
-        const peekTarget = ctx.returnRawErrors ? res.clone() : res;
+        const peekTarget = res.clone();
         const peek = await readDisplaySafeErrorPayloadText(peekTarget, ctx.abortSignal);
         if (isQuotaExhaustedBody(peek)) {
           return ctx.returnRawErrors ? res : normalizeUpstreamHttpErrorResponse(res, {
@@ -84,20 +102,27 @@ export async function fetchGoogleWithRetry(
           });
         }
       }
-      cancelResponseBodyBestEffort(res);
-      await sleepWithAbort(retryBackoffDelayMs(attempt, {
+      pendingResponse = res;
+      recovery = res.status === 429 ? "rate-limit-429" : "transient-5xx";
+      retryDelayMs = retryBackoffDelayMs(attempt, {
         baseDelayMs: GOOGLE_RETRY_BASE_MS,
         maxDelayMs: GOOGLE_RETRY_MAX_MS,
         headers: res.headers,
-      }), ctx.abortSignal);
+      });
     } catch (err) {
       if (ctx.abortSignal?.aborted) throw err;
+      if (err instanceof SendBudgetExhaustedError) {
+        if (pendingResponse) return ctx.returnRawErrors ? pendingResponse : normalizeFinalGoogleError(label, pendingResponse, ctx.abortSignal);
+        throw err;
+      }
       lastError = err;
       if (attempt === GOOGLE_RETRY_ATTEMPTS - 1) throw err;
-      await sleepWithAbort(retryBackoffDelayMs(attempt, {
+      sendClass = "transient";
+      recovery = isConnectionResetError(err) ? "connection-reset" : undefined;
+      retryDelayMs = retryBackoffDelayMs(attempt, {
         baseDelayMs: GOOGLE_RETRY_BASE_MS,
         maxDelayMs: GOOGLE_RETRY_MAX_MS,
-      }), ctx.abortSignal);
+      });
     }
   }
   throw lastError ?? new Error(`${label} fetch failed`);

src/adapters/mimo-free.ts

@@ -6,6 +6,8 @@ import { recordOwnedConfigPath } from "../lib/config-ownership";
 import type { OcxProviderConfig, OcxParsedRequest } from "../types";
 import { createOpenAIChatAdapter } from "./openai-chat";
 import type { ProviderAdapter, AdapterRequest, IncomingMeta } from "./base";
+import { createAdapterPhysicalSend } from "./physical-send";
+import { SendBudgetExhaustedError } from "../lib/upstream-retry";
 
 const BOOTSTRAP_URL = "https://api.xiaomimimo.com/api/free-ai/bootstrap";
 export const MIMO_CHAT_URL = "https://api.xiaomimimo.com/api/free-ai/openai/chat";
@@ -248,33 +250,46 @@ export function createMimoFreeAdapter(provider: OcxProviderConfig): ProviderAdap
     },
 
     async fetchResponse(request: AdapterRequest, ctx): Promise<Response> {
-      const response = await fetch(request.url, {
+      const send = createAdapterPhysicalSend(ctx);
+      const response = await send({ url: request.url, dispatch: executor => executor(request.url, {
         method: request.method,
         redirect: "manual",
         headers: request.headers as Record<string, string>,
         body: request.body,
         signal: ctx?.abortSignal,
-      });
+      }) });
 
       // Retry predicate: 401 (expired/invalid JWT) retries ONCE with a fresh token.
       // 403 is NOT retried — Xiaomi uses it for anti-abuse "Illegal access" and there is
       // no documented token-expiry signature that would mark a 403 as retryable.
       if (response.status === 401) {
-        // Drain the first response body before issuing the retry.
-        try { await response.body?.cancel(); } catch { /* already consumed */ }
-        resetMimoJwtCache();
-        const freshJwt = await getMimoJwt(ctx?.abortSignal);
-        const retryHeaders = {
-          ...(request.headers as Record<string, string>),
-          "Authorization": `Bearer ${freshJwt}`,
-        };
-        return fetch(request.url, {
-          method: request.method,
-          redirect: "manual",
-          headers: retryHeaders,
-          body: request.body,
-          signal: ctx?.abortSignal,
-        });
+        let retryHeaders = request.headers;
+        try {
+          return await send({ url: request.url, sendClass: "auth-recovery", recovery: "oauth-401",
+            beforeDispatch: async () => {
+              // Drain the first response body and refresh the JWT only after admission: a
+              // refused replay still returns THIS response to the caller, body intact.
+              // Draining comes first within the block because getMimoJwt issues its own
+              // network call and may throw, and the 401 body would then never be released.
+              try { void response.body?.cancel().catch(() => {}); } catch { /* already consumed */ }
+              resetMimoJwtCache();
+              const freshJwt = await getMimoJwt(ctx?.abortSignal);
+              retryHeaders = {
+                ...(request.headers as Record<string, string>),
+                "Authorization": `Bearer ${freshJwt}`,
+              };
+            },
+            dispatch: executor => executor(request.url, {
+              method: request.method,
+              redirect: "manual",
+              headers: retryHeaders,
+              body: request.body,
+              signal: ctx?.abortSignal,
+            }) });
+        } catch (error) {
+          if (error instanceof SendBudgetExhaustedError) return response;
+          throw error;
+        }
       }
 
       return response;

src/adapters/physical-send.ts

@@ -0,0 +1,50 @@
+import type { AdapterFetchContext } from "./base";
+import type { SendClass } from "../lib/request-execution-budget";
+import type { AttemptRecoveryKind } from "../usage/log";
+import { abortError, SendBudgetExhaustedError } from "../lib/upstream-retry";
+
+type PacedFetch = typeof globalThis.fetch & {
+  waitForPacing?: (signal?: AbortSignal) => Promise<void>;
+  unpacedFetch?: typeof globalThis.fetch;
+};
+
+/** One ordinal sequence per adapter fetchResponse call, across all of its inference retries.
+ * Consumption starts at underlying executor invocation; its own later preflight may still fail. */
+export function createAdapterPhysicalSend(ctx: AdapterFetchContext = {}, fallback = globalThis.fetch) {
+  const executor = (ctx.executor ?? fallback) as PacedFetch;
+  let ordinal = 0;
+  return async (options: {
+    url: string;
+    sendClass?: SendClass;
+    recovery?: AttemptRecoveryKind;
+    /** Runs only after admission, e.g. backoff and cancellation of a superseded response. */
+    beforeDispatch?: () => void | Promise<void>;
+    dispatch: (executor: typeof globalThis.fetch) => Promise<Response>;
+  }): Promise<Response> => {
+    if (ctx.abortSignal?.aborted) throw abortError(ctx.abortSignal);
+    const decision = ctx.sendBudget?.reserveDispatch({
+      sendClass: options.sendClass ?? "transient", targetKey: options.url,
+    });
+    if (decision && !decision.allowed) throw new SendBudgetExhaustedError(options.url);
+    const permit = decision?.allowed ? decision.permit : undefined;
+    let dispatched = false;
+    const physicalExecutor = (async (input, init) => {
+      if (ctx.abortSignal?.aborted) throw abortError(ctx.abortSignal);
+      if (init?.signal?.aborted) throw abortError(init.signal);
+      if (dispatched || (permit && !permit.use())) throw new SendBudgetExhaustedError(options.url);
+      dispatched = true;
+      ordinal += 1;
+      ctx.onPhysicalSend?.({ ordinal, ...(options.recovery ? { recovery: options.recovery } : {}) });
+      return (executor.unpacedFetch ?? executor)(input, init);
+    }) as typeof globalThis.fetch;
+    try {
+      await executor.waitForPacing?.(ctx.abortSignal);
+      if (ctx.abortSignal?.aborted) throw abortError(ctx.abortSignal);
+      await options.beforeDispatch?.();
+      if (ctx.abortSignal?.aborted) throw abortError(ctx.abortSignal);
+      return await options.dispatch(physicalExecutor);
+    } finally {
+      permit?.release();
+    }
+  };
+}

tests/adapters/google/google-vertex-http.test.ts

@@ -1,4 +1,7 @@
-import { afterEach, describe, expect, test } from "bun:test";
+import { afterEach, describe, expect, spyOn, test } from "bun:test";
+import * as retry from "../../../src/lib/upstream-retry";
+import { createRequestExecutionBudget } from "../../../src/lib/request-execution-budget";
+import { budgetOwner } from "../../helpers/send-budget-owner";
 import type { AdapterRequest } from "../../../src/adapters/base";
 import { fetchAntigravityWithRetry, fetchDirectGeminiWithRetry, fetchVertexWithRetry } from "../../../src/adapters/google-http";
 import { safeVertexHttpErrorMessage, retryableGoogleStatus } from "../../../src/adapters/google-errors";
@@ -30,6 +33,43 @@ function vertexError(code: number, status: string, message: string): string {
 }
 
 describe("vertex retry fetch", () => {
+  for (const [name, fetchResponse] of [["Vertex", fetchVertexWithRetry], ["Antigravity", fetchAntigravityWithRetry]] as const) {
+    test.each([400, 429, 503, "reset"] as const)(`${name} prepaid final send prevents another inference or backoff (%s)`, async status => {
+      const parent = createRequestExecutionBudget();
+      parent.used = 3;
+      const { owner, dispose } = budgetOwner(parent);
+      const raw = status === 400 ? vertexError(400, "INVALID_ARGUMENT", "tools.0.custom.input_schema: JSON schema is invalid") : `fixture ${status}`;
+      const first = status === "reset" ? Object.assign(new Error("fixture reset"), { code: "ECONNRESET" })
+        : new Response(raw, { status, headers: { "Retry-After": "60" } });
+      const fixture = mockFetch([first, new Response("unexpected replay")]);
+      const waits = spyOn(retry, "sleepWithAbort").mockImplementation(async () => {});
+      const ordinals: number[] = [];
+      try {
+        const hop = owner.reserveCredentialHop("auth-recovery", request.url, true);
+        if (!hop.allowed || !hop.permit) throw new Error("Expected final prepaid send");
+        owner.pendingHopPermit = hop.permit;
+        const scope = owner.adapterDispatchBudget;
+        if (!scope) throw new Error("Expected an adapter dispatch budget");
+        const result = fetchResponse({ ...request, body: JSON.stringify({ request: {
+          contents: [{ role: "user", parts: [{ text: "hi" }] }],
+          tools: [{ functionDeclarations: [{ name: "replace_in_files", parameters: {
+            type: "object", properties: { occurrence_ids: { type: "array", items: { type: "string" } } },
+          } }] }],
+        } }) }, { sendBudget: scope, returnRawErrors: true, onPhysicalSend: send => ordinals.push(send.ordinal) });
+        if (status === "reset") await expect(result).rejects.toBeInstanceOf(retry.SendBudgetExhaustedError);
+        else {
+          const response = await result;
+          expect(response).toBe(first);
+          expect(await response.text()).toBe(raw);
+        }
+        expect(fixture.calls).toHaveLength(1);
+        expect(waits).not.toHaveBeenCalled();
+        expect(parent.used).toBe(4);
+        expect(ordinals).toEqual([1]);
+      } finally { waits.mockRestore(); dispose(); }
+    });
+  }
+
   test("successful response bodies survive beyond the response-header timeout", async () => {
     globalThis.fetch = (async () => new Response(new ReadableStream<Uint8Array>({
       async start(controller) {