feat(responses): put combo hops and adapter inner retries on the shared send budget (#4546)

scripts/test-layout/layout.json

@@ -181,6 +181,8 @@
     "adapter-buffered-tool-conformance.test.ts": "adapters",
     "adapter-error-inline.test.ts": "adapters",
     "adapter-event-oauth-failover.test.ts": "oauth",
+    "adapter-inner-send-budget-wiring.test.ts": "adapters",
+    "adapter-inner-send-budget.test.ts": "adapters",
     "adapter-registry-authority.test.ts": "adapters",
     "adapter-resolve.test.ts": "server",
     "adapter-tool-conformance.test.ts": "adapters",
@@ -1179,6 +1181,7 @@
     "responses-reasoning-summary-passthrough.test.ts": "responses",
     "responses-routed-web-search-fields.test.ts": "responses",
     "responses-self-named-namespace-scrub.test.ts": "responses",
+    "responses-send-budget-counts.test.ts": "responses",
     "responses-shadow-intercept.test.ts": "responses",
     "responses-show-thinking-summary.test.ts": "responses",
     "responses-snapshot-repair-server.test.ts": "responses",

src/adapters/base.ts

@@ -1,6 +1,7 @@
 import type { AdapterEvent, OcxParsedRequest } from "../types";
 import type { TranslatorBudget } from "../lib/translator-budget";
 import type { RequestExecutionBudget } from "../lib/request-execution-budget";
+import type { AttemptRecoveryKind } from "../usage/log";
 import type { AdapterTierMetadata } from "../providers/fastwire";
 
 /** Metadata about the caller's incoming request, for auth-forwarding adapters. */
@@ -20,6 +21,16 @@ export interface IncomingMeta {
    * the anthropic and openai-chat adapters; others ignore it.
    */
   imageTierBias?: number;
+  /**
+   * The enclosing request's send budget, for adapters that own their upstream transport.
+   *
+   * A `runTurn` adapter never receives an `AdapterFetchContext`, so the budget that bounds every
+   * other leg could not reach it: Cursor re-sends a whole turn up to three times inside one
+   * adapter call, and the request cap counted that as one send. Optional, and absent means
+   * unlimited, because adapter unit tests build a meta with neither a budget nor a request
+   * behind it (#4546).
+   */
+  sendBudget?: RequestExecutionBudget;
 }
 
 export interface ProviderAdapter {
@@ -147,6 +158,16 @@ export interface AdapterFetchContext {
    * adapter entry as one send is how a nested 3x3 ladder stayed invisible to a request cap.
    */
   sendBudget?: RequestExecutionBudget;
+  /**
+   * Observes every physical upstream send this adapter makes, including its own inner retries.
+   *
+   * `ordinal` counts from 1 within this fetch call, so a caller that already recorded the entry
+   * send records only ordinals above 1 and an adapter that never retries internally logs exactly
+   * what it logs today. Kiro and Cursor were unpinnable without this: they report one send per
+   * adapter call however many requests they actually made, so their inner ladders were invisible
+   * to `sendCount` and no regression could assert a count for them (#4546).
+   */
+  onPhysicalSend?: (send: { ordinal: number; recovery?: AttemptRecoveryKind }) => void;
 }
 
 /**

src/adapters/cursor.ts

@@ -403,6 +403,10 @@ export function createCursorAdapter(provider: OcxProviderConfig, deps: CursorAda
                 }
               }
             },
+            // Cursor's retry ladder re-sends the WHOLE turn, so each attempt is a physical send
+            // the enclosing request pays for. A meta without a budget -- every adapter unit test,
+            // and any caller predating this -- keeps the adapter's own three attempts (#4546).
+            incoming.sendBudget ? { sendBudget: incoming.sendBudget } : {},
           );
         };
 

src/adapters/cursor/transport-retry.ts

@@ -1,6 +1,8 @@
 import type { CursorRunRequest, CursorServerMessage } from "./types";
 import type { CursorTransport, CursorTransportFactory, CursorTransportFactoryInput } from "./transport";
-import { abortError, retryBackoffDelayMs, sleepWithAbort } from "../../lib/upstream-retry";
+import type { RequestExecutionBudget } from "../../lib/request-execution-budget";
+import type { AttemptRecoveryKind } from "../../usage/log";
+import { SendBudgetExhaustedError, abortError, retryBackoffDelayMs, sleepWithAbort } from "../../lib/upstream-retry";
 import { debugProviderDiagnostic } from "../../lib/debug";
 import { isCursorRootEnvelopeError, safeCursorErrorMessage } from "./cursor-errors";
 
@@ -11,6 +13,27 @@ export const CURSOR_RETRY_ATTEMPTS = 3;
 export const CURSOR_RETRY_BASE_MS = 250;
 export const CURSOR_RETRY_MAX_MS = 2_000;
 
+/**
+ * Fixed identity for the Cursor upstream in the request budget's target ledger. A literal, not
+ * anything derived from the turn: the ledger is read back in diagnostics, so it must not become
+ * a place where a session or credential identity leaks.
+ */
+export const CURSOR_BUDGET_TARGET_KEY = "cursor";
+
+/**
+ * How one Cursor turn participates in the enclosing logical request (#4546).
+ *
+ * Both fields are optional and the whole object defaults to empty, which is what keeps a
+ * context-free unit call unlimited: this transport is exercised directly by tests that build no
+ * request at all, and a mandatory budget would have made every one of them a budget test.
+ */
+export interface CursorTurnExecutionOptions {
+  /** Absent means unlimited; present means every retry is a physical send the request pays for. */
+  sendBudget?: RequestExecutionBudget;
+  /** Observes each physical run request; `ordinal` counts from 1 within this turn. */
+  onPhysicalSend?: (send: { ordinal: number; recovery?: AttemptRecoveryKind }) => void;
+}
+
 /**
  * True only for clearly transient failures that occur BEFORE the run request is committed to the
  * wire (connection refused/reset/timeout, immediate HTTP/2 GOAWAY, gRPC/Connect "unavailable").
@@ -66,16 +89,38 @@ function requestUncommitted(transport: CursorTransport): boolean {
  *  - the failing transport reports the run request was not committed to the wire,
  *  - the error is a transient pre-commit failure.
  * Otherwise the error propagates (the adapter maps it to a user-facing message).
+ *
+ * `execution` carries the enclosing request's send budget. Each attempt here is a real re-send
+ * of the whole turn, so an outer cap that counted one adapter entry counted at most a third of
+ * what went upstream; when a budget is present every attempt is admitted against it and an
+ * exhausted request stops before opening another transport (#4546).
  */
 export async function runCursorTurnWithRetry(
   makeTransport: (input: CursorTransportFactoryInput) => CursorTransport,
   input: CursorTransportFactoryInput,
   request: CursorRunRequest,
   signal: AbortSignal | undefined,
   onEvent: (message: CursorServerMessage, transport: CursorTransport) => void,
+  execution: CursorTurnExecutionOptions = {},
 ): Promise<void> {
   for (let attempt = 0; ; attempt++) {
     if (signal?.aborted) throw abortError(signal);
+    // Admitted before the transport is built: a refused send must not open a connection, and
+    // the refusal must reach the adapter as the typed exhaustion rather than as a run failure
+    // that the retry predicate below could read as transient.
+    const decision = execution.sendBudget?.reserveDispatch({
+      sendClass: "transient",
+      targetKey: CURSOR_BUDGET_TARGET_KEY,
+    });
+    if (decision && (!decision.allowed || !decision.permit.use())) {
+      throw new SendBudgetExhaustedError(CURSOR_BUDGET_TARGET_KEY);
+    }
+    execution.onPhysicalSend?.({
+      ordinal: attempt + 1,
+      // Cursor retries only pre-commit transport failures, so every retry send is the
+      // connection-reset class; there is no re-send of a turn the server may have accepted.
+      ...(attempt > 0 ? { recovery: "connection-reset" as const } : {}),
+    });
     const transport = makeTransport(input);
     let emittedAny = false;
     let closed = false;

src/adapters/kiro-retry.ts

@@ -1,4 +1,5 @@
 import type { AdapterFetchContext, AdapterRequest } from "./base";
+import type { AttemptRecoveryKind } from "../usage/log";
 import { classifyKiroHttpError, safeKiroHttpErrorMessage } from "./kiro-errors";
 import { normalizeUpstreamHttpErrorResponse } from "./upstream-http-error";
 import { readBoundedResponseBody } from "../lib/bounded-body";
@@ -159,6 +160,7 @@ async function fetchWithResetRecovery(
   url: string,
   ctx: AdapterFetchContext,
   timeoutMs: number,
+  notePhysicalSend: (reset: boolean) => void,
 ): Promise<Response> {
   let lastError: unknown;
   for (let attempt = 0; attempt < RESET_ATTEMPTS; attempt++) {
@@ -170,6 +172,9 @@ async function fetchWithResetRecovery(
     if (decision && (!decision.allowed || !decision.permit.use())) {
       throw new SendBudgetExhaustedError(url);
     }
+    // Reported after admission and before dispatch, so a refused send is never counted and an
+    // admitted one is counted exactly once whichever way the fetch below settles.
+    notePhysicalSend(attempt > 0);
     try {
       const headers = new Headers(request.headers);
       const recovered = attempt > 0;
@@ -252,22 +257,23 @@ async function fetchKiroAttempt(
   request: AdapterRequest,
   ctx: AdapterFetchContext,
   timeoutMs: number,
+  notePhysicalSend: (reset: boolean) => void,
 ): Promise<Response> {
   const legacy = legacyUrl(request.url);
   let response: Response;
   try {
-    response = await fetchWithResetRecovery(request, request.url, ctx, timeoutMs);
+    response = await fetchWithResetRecovery(request, request.url, ctx, timeoutMs, notePhysicalSend);
   } catch (error) {
     if (!legacy || !endpointConnectFailure(error)) throw error;
-    return fetchWithResetRecovery(request, legacy, ctx, timeoutMs);
+    return fetchWithResetRecovery(request, legacy, ctx, timeoutMs, notePhysicalSend);
   }
 
   if (legacy && !response.ok) {
     const inspected = await inspectEndpointHttpFailure(response, ctx.abortSignal);
     response = inspected.response;
     if (inspected.fallback) {
       cancelResponseBodyBestEffort(response);
-      response = await fetchWithResetRecovery(request, legacy, ctx, timeoutMs);
+      response = await fetchWithResetRecovery(request, legacy, ctx, timeoutMs, notePhysicalSend);
     }
   }
   return response;
@@ -281,12 +287,25 @@ async function fetchKiroAttempt(
 export async function fetchKiroWithRetry(request: AdapterRequest, ctx: AdapterFetchContext = {}): Promise<Response> {
   const timeoutMs = ctx.timeoutMs ?? 200_000;
   let probeToken: symbol | undefined;
+  // One ordinal sequence for the whole call, across the throttle loop, the endpoint fallback
+  // and the reset ladder nested inside it. The caller records ordinal 1 itself, so this is what
+  // turns "one adapter call" back into the physical count the request actually made.
+  let physicalSends = 0;
+  let throttleRound = 0;
+  const notePhysicalSend = (reset: boolean): void => {
+    physicalSends += 1;
+    const recovery: AttemptRecoveryKind | undefined = reset
+      ? "connection-reset"
+      : throttleRound > 0 ? "rate-limit-429" : undefined;
+    ctx.onPhysicalSend?.({ ordinal: physicalSends, ...(recovery ? { recovery } : {}) });
+  };
   try {
     for (let attempt = 0; attempt < THROTTLE_ATTEMPTS; attempt++) {
+      throttleRound = attempt;
       if (!probeToken) probeToken = await enterKiroThrottleGate(ctx.abortSignal);
       else await waitForKiroCooldown(ctx.abortSignal);
 
-      const response = await fetchKiroAttempt(request, ctx, timeoutMs);
+      const response = await fetchKiroAttempt(request, ctx, timeoutMs, notePhysicalSend);
       const throttle = await inspectKiroThrottle(response, ctx.abortSignal);
       if (!throttle || !throttle.transient) {
         releaseKiroThrottleProbe(probeToken);

src/adapters/kiro/adapter.ts

@@ -45,6 +45,10 @@ import {
   type KiroWireClient,
 } from "./wire";
 
+/** The physical-send observer an `AdapterFetchContext` may carry, and the record it receives. */
+type KiroPhysicalSendObserver = NonNullable<AdapterFetchContext["onPhysicalSend"]>;
+type KiroPhysicalSend = Parameters<KiroPhysicalSendObserver>[0];
+
 // Adapter
 export function createKiroAdapter(provider: OcxProviderConfig): ProviderAdapter {
   // Per-request closure (resolveAdapter builds a fresh adapter per request — server.ts:440 — so this
@@ -62,6 +66,25 @@ export function createKiroAdapter(provider: OcxProviderConfig): ProviderAdapter
   // Captured the same way as the abort signal, because the text-fallback rebuild below runs
   // outside the fetchResponse frame and used to construct a context without either (#4546).
   let requestSendBudget: RequestExecutionBudget | undefined;
+  // Captured for the same reason, and needed for the same leg to be COUNTABLE rather than merely
+  // bounded: the rebuild's sends were paid for out of the request budget but reported by nobody,
+  // so no regression could pin how many requests one Kiro turn actually makes.
+  let requestOnPhysicalSend: KiroPhysicalSendObserver | undefined;
+  // One ordinal sequence across the whole turn. `fetchKiroWithRetry` numbers from 1 inside each
+  // call, and the caller reads ordinal 1 as the send it already recorded itself; forwarding the
+  // rebuild's raw ordinals would therefore drop its first send — the very send that makes the
+  // fallback a second request rather than a continuation of the first.
+  let physicalSendsObserved = 0;
+  const forwardPhysicalSend = (
+    send: KiroPhysicalSend,
+    ordinalBase: number,
+    defaultRecovery?: KiroPhysicalSend["recovery"],
+  ): void => {
+    const ordinal = ordinalBase + send.ordinal;
+    if (ordinal > physicalSendsObserved) physicalSendsObserved = ordinal;
+    const recovery = send.recovery ?? defaultRecovery;
+    requestOnPhysicalSend?.({ ordinal, ...(recovery ? { recovery } : {}) });
+  };
 
   const build = async (
     parsed: OcxParsedRequest,
@@ -208,13 +231,22 @@ export function createKiroAdapter(provider: OcxProviderConfig): ProviderAdapter
       retryBodyReservation.commitRetained();
       retryBodyRetained = true;
       budget.releaseRetained(retryBodyUpperBound - retryBodyBytes, { kind: "request_copies" });
+      // Fixed before the rebuild dispatches, so the leg's ordinals continue the first attempt's
+      // sequence even though this call's own counter restarts at 1.
+      const fallbackOrdinalBase = physicalSendsObserved;
       const response = await fetchKiroWithRetry(retry.request, {
         abortSignal: requestAbortSignal,
         returnRawErrors: true,
         stream: true,
         // The text-fallback rebuild used to construct a fresh context and drop the budget,
         // so everything after the first send escaped the per-request cap.
         ...(requestSendBudget ? { sendBudget: requestSendBudget } : {}),
+        // And reported nothing, so the sends it paid for were invisible. Its own first send is
+        // the completion retry itself: the first attempt produced progress without a final
+        // answer, which is the same recovery class the generic empty-completion guard records.
+        ...(requestOnPhysicalSend
+          ? { onPhysicalSend: (send: KiroPhysicalSend) => forwardPhysicalSend(send, fallbackOrdinalBase, "empty-completion") }
+          : {}),
       });
       return {
         response,
@@ -286,7 +318,16 @@ export function createKiroAdapter(provider: OcxProviderConfig): ProviderAdapter
       // both the first Kiro request and its one allowed completion retry.
       if (ctx?.abortSignal) requestAbortSignal = ctx.abortSignal;
       if (ctx?.sendBudget) requestSendBudget = ctx.sendBudget;
-      return fetchKiroWithRetry(request, ctx);
+      if (ctx?.onPhysicalSend) requestOnPhysicalSend = ctx.onPhysicalSend;
+      // Reset per fetch call, because `ordinal` is defined within one call and the caller records
+      // ordinal 1 of each new attempt itself. The text fallback that follows this attempt then
+      // continues THIS attempt's sequence rather than an earlier one's.
+      physicalSendsObserved = 0;
+      // Routed through the same forwarder as the fallback so both legs share one ordinal
+      // sequence; a context without an observer is passed through untouched.
+      return fetchKiroWithRetry(request, requestOnPhysicalSend
+        ? { ...ctx, onPhysicalSend: (send: KiroPhysicalSend) => forwardPhysicalSend(send, 0) }
+        : ctx);
     },
 
     formatErrorBody(status: number, headers: Headers, payloadText: string): string {