diff --git a/scripts/test-layout/layout.json b/scripts/test-layout/layout.json index c9b5ad4c68..5b6c163682 100644 --- a/scripts/test-layout/layout.json +++ b/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", diff --git a/src/adapters/base.ts b/src/adapters/base.ts index 2e376a628a..f4cf7ab2ef 100644 --- a/src/adapters/base.ts +++ b/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; } /** diff --git a/src/adapters/cursor.ts b/src/adapters/cursor.ts index 91e823be9a..25a6cca582 100644 --- a/src/adapters/cursor.ts +++ b/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 } : {}, ); }; diff --git a/src/adapters/cursor/transport-retry.ts b/src/adapters/cursor/transport-retry.ts index a0714d5a9d..03090778e9 100644 --- a/src/adapters/cursor/transport-retry.ts +++ b/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,6 +89,11 @@ 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, @@ -73,9 +101,26 @@ export async function runCursorTurnWithRetry( request: CursorRunRequest, signal: AbortSignal | undefined, onEvent: (message: CursorServerMessage, transport: CursorTransport) => void, + execution: CursorTurnExecutionOptions = {}, ): Promise { 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; diff --git a/src/adapters/kiro-retry.ts b/src/adapters/kiro-retry.ts index 08ddbbb4d9..e137a22f79 100644 --- a/src/adapters/kiro-retry.ts +++ b/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 { 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,14 +257,15 @@ async function fetchKiroAttempt( request: AdapterRequest, ctx: AdapterFetchContext, timeoutMs: number, + notePhysicalSend: (reset: boolean) => void, ): Promise { 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) { @@ -267,7 +273,7 @@ async function fetchKiroAttempt( 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 { 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); diff --git a/src/adapters/kiro/adapter.ts b/src/adapters/kiro/adapter.ts index 1b3a90e80f..b6a374cf4c 100644 --- a/src/adapters/kiro/adapter.ts +++ b/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; +type KiroPhysicalSend = Parameters[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,6 +231,9 @@ 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, @@ -215,6 +241,12 @@ export function createKiroAdapter(provider: OcxProviderConfig): ProviderAdapter // 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 { diff --git a/src/server/responses/core.ts b/src/server/responses/core.ts index 913fb13795..abf699ef35 100644 --- a/src/server/responses/core.ts +++ b/src/server/responses/core.ts @@ -230,6 +230,9 @@ import { import { createRequestExecutionBudget, isRequestExecutionBudget, + CODEX_TEXT_GUARDED_BUDGET_POLICY, + type RequestExecutionBudget, + type RequestExecutionBudgetPolicy, type SendClass, type SingleUseDispatchPermit, } from "../../lib/request-execution-budget"; @@ -2918,6 +2921,91 @@ async function applyFinalRouteRequestNormalization(args: { +/** + * Sends one combo target may run on its own before the ladder moves on. A target is a whole + * request as far as its own provider is concerned, so this is the guarded profile's base + * allowance rather than a separate number to keep in sync. + */ +const COMBO_TARGET_BASE_SENDS = CODEX_TEXT_GUARDED_BUDGET_POLICY.baseSendAllowance; + +/** + * A combo's execution policy is DECLARED by the combo, not inherited from the single-target + * profile. + * + * `maxTargetTransitions: 1` and `maxAlternateTargetSends: 1` describe an account move, and + * applying them to a combo would refuse the second hop of a three-target combo -- which is why + * combo was left off `reserveDispatch` when the per-request split landed. The transitions a + * combo may make are exactly the targets it declares minus the one it starts on. What stays + * capped is the TOTAL: the first target's full ladder, one send for every further declared + * target, and the one shared final-recovery reserve. A one-target combo reduces to the guarded + * profile exactly, and a three-target combo whose every target fails hard reaches upstream six + * times instead of the twelve #4546 measured. + */ +function comboExecutionBudgetPolicy(declaredTargets: number): RequestExecutionBudgetPolicy { + const targets = Math.max(1, Math.trunc(declaredTargets)); + const hops = targets - 1; + const reserve = CODEX_TEXT_GUARDED_BUDGET_POLICY.finalRecoveryAllowance; + const total = COMBO_TARGET_BASE_SENDS + hops + reserve; + return { + maxTotalModelSends: total, + baseSendAllowance: total - reserve, + finalRecoveryAllowance: reserve, + maxAlternateTargetSends: Math.max(1, hops), + maxTargetTransitions: Math.max(1, hops), + }; +} + +/** + * A budget scope that keeps its own recovery ledgers but spends the SAME request-wide counter. + * + * `used` is redefined as an accessor onto the parent because the factory reads it back off this + * object -- `remainingBaseSends` and the total check both do -- so a copied number would let a + * combo target run its ladder against a stale total, which is precisely the per-layer counting + * this work exists to remove. The reserve, alternate-target and transition ledgers stay + * per-scope on purpose: a combo target's account failover is its own recovery decision, while + * the request total still bounds every target together. + */ +function deriveSendBudgetScope( + parent: RequestExecutionBudget, + policy: RequestExecutionBudgetPolicy, +): RequestExecutionBudget { + const scope = createRequestExecutionBudget(policy, parent.logicalRequestId); + Object.defineProperty(scope, "used", { + get: () => parent.used, + set: (value: number) => { parent.used = value; }, + enumerable: true, + configurable: true, + }); + return scope; +} + +/** + * The ladder one combo target may run, expressed as an allowance on the request-wide counter. + * + * `used + COMBO_TARGET_BASE_SENDS` gives this target its own ladder from wherever the request + * already stands, and the clamp holds back one send for each target still declared after it: a + * first target that 5xx-streaks must not eat the send the last declared target is entitled to. + * That guarantee is the difference between a per-target policy and a shared pool the first + * target drains. + */ +function comboTargetSendBudget( + comboScope: RequestExecutionBudget, + targetsDeclaredAfterThisOne: number, +): RequestExecutionBudget { + const policy = comboScope.policy; + const heldForLaterTargets = Math.max(0, targetsDeclaredAfterThisOne); + const ceiling = Math.max(1, policy.maxTotalModelSends - heldForLaterTargets); + return deriveSendBudgetScope(comboScope, { + maxTotalModelSends: policy.maxTotalModelSends, + baseSendAllowance: Math.min(ceiling, comboScope.used + COMBO_TARGET_BASE_SENDS), + finalRecoveryAllowance: policy.finalRecoveryAllowance, + // Within one target the account-move shape is unchanged: three same-account sends plus one + // alternate is the recovery live traffic depends on, and a combo does not widen it. + maxAlternateTargetSends: CODEX_TEXT_GUARDED_BUDGET_POLICY.maxAlternateTargetSends, + maxTargetTransitions: CODEX_TEXT_GUARDED_BUDGET_POLICY.maxTargetTransitions, + }); +} + export async function handleComboResponses( req: Request, rawBody: unknown, @@ -2939,6 +3027,14 @@ export async function handleComboResponses( if (!combo) { return formatErrorResponse(404, "invalid_request_error", `Unknown combo: ${comboId}`); } + // The ladder's own scope, derived from what this combo DECLARES. It shares the request-wide + // counter with the holder that arrived on options -- a combo child already inherited that + // counter, but nothing read it as a limit across targets -- while its transition and + // alternate-target ledgers come from the target list rather than from the single-target + // account-move profile (#4546). + const comboSendScope = isRequestExecutionBudget(options.sendBudget) + ? deriveSendBudgetScope(options.sendBudget, comboExecutionBudgetPolicy(combo.targets.length)) + : undefined; // Expand previous_response_id before image policy and child dispatch so a // continuation that only references prior images still fails closed when // imageInput is disabled (and so targets see the full replayed input). @@ -3112,12 +3208,42 @@ export async function handleComboResponses( logCtx.routeDecision = comboRouteDecisionTrace(config, comboId, pick, requestedModel); let lastFailure: Response | null = null; + // Dispatched targets, not attempted picks: it indexes the declared target list so the clamp + // below can tell how many targets are still entitled to a send. + let comboTargetsDispatched = 0; + // The child log behind `lastFailure`. The natural end of the ladder adopts it inside the + // no-more-targets branch; a budget refusal ends the ladder one iteration later, where that + // iteration's own `childLog` is already out of scope. + let lastFailedChildLog: RequestLogContext | undefined; // The exhausted-combo mapping below runs outside the loop, where `failure.upstreamCode` // is gone, so carry the loop's own classification decision instead of re-deriving a // weaker one from the status alone (#4149). let lastFailureClassifiesOverflow = false; while (pick) { if (options.abortSignal?.aborted) return clientCancelledResponse(); + const firstComboTarget = comboTargetsDispatched === 0; + // The first target seeds the ledger's target identity and charges nothing; every later one + // is a real transition, refused once the declared hops, the alternate-target ledger or the + // request total are spent. `countedExternally` is required: the child charges its own + // physical sends, and charging here as well would halve the cap without saying so. + const hopDecision = comboSendScope?.reserveDispatch({ + sendClass: firstComboTarget ? "initial" : "combo-failover", + targetKey: `${pick.target.provider}/${pick.target.model}`, + countedExternally: true, + }); + if (hopDecision && hopDecision.allowed) hopDecision.permit.use(); + else if (hopDecision && !firstComboTarget) { + // Out of budget is not this target's failure. The established exhaustion contract is to + // return the last real upstream answer with its status, headers and any quota body + // intact rather than to mint a synthetic error, and a later target only exists because + // an earlier one already recorded one. + if (lastFailedChildLog) adoptFailedChildLog(lastFailedChildLog); + break; + } + const targetSendBudget = comboSendScope + ? comboTargetSendBudget(comboSendScope, combo.targets.length - 1 - comboTargetsDispatched) + : options.sendBudget; + comboTargetsDispatched += 1; const childLog: RequestLogContext = { model: pick.target.model, provider: pick.target.provider, @@ -3201,6 +3327,9 @@ export async function handleComboResponses( ); response = await handleResponses(childRequest, config, childLog, { ...options, + // After the spread: the child must run on THIS target's ladder, not on the holder the + // parent arrived with. + sendBudget: targetSendBudget, comboAttempt: true, comboReplaySnapshot, deferCodexResetDerivedCooldown, @@ -3322,6 +3451,7 @@ export async function handleComboResponses( (logCtx.attempts ??= []).push(attempt); attemptRetained = true; lastFailure = failure.response; + lastFailedChildLog = childLog; const failureDecision = comboFailureDecision(failure.response.status, failure.classificationText, { code: failure.upstreamCode, }); @@ -5109,6 +5239,22 @@ async function handleResponsesInner( // typed as the narrow holder so a caller that predates this can still pass one, so narrow it // once here rather than asserting at each adapter call site. const adapterSendBudget = isRequestExecutionBudget(sendBudget) ? sendBudget : undefined; + /** + * Records an adapter's OWN inner retries against this attempt. + * + * Ordinal 1 is the send each call site already recorded through `noteAttemptSend`, so only + * the extra physical sends are added here and an adapter that does not retry internally + * leaves its log byte-for-byte as it was. Kiro reaches roughly eighteen sends per call and + * Cursor re-sends a whole turn, and both reported one; a count that cannot be observed + * cannot be pinned by a regression, which is why the instrumentation precedes the cap. + */ + const noteAdapterPhysicalSend = ( + inputTokens: number | undefined, + send: { ordinal: number; recovery?: AttemptRecoveryKind }, + ): void => { + if (send.ordinal <= 1) return; + noteAttemptSend(logCtx.activeAttempt, inputTokens, send.recovery); + }; const sendBudgetExhausted = (): boolean => remainingTransientSendBudget(TRANSIENT_RETRY_MAX_ATTEMPTS) === 0; /** @@ -7509,6 +7655,9 @@ async function handleResponsesInner( abortSignal: runTurnAbort.signal, translatorBudget, providerFetch: runTurnProviderFetch, + // The only way the request budget reaches a transport the adapter owns. Without it + // a Cursor turn's inner ladder was three physical sends the cap read as one. + ...(adapterSendBudget ? { sendBudget: adapterSendBudget } : {}), }, targetQueue.push, ); @@ -7909,6 +8058,7 @@ async function handleResponsesInner( abortSignal: upstream.signal, timeoutMs: connectMs, sendBudget: adapterSendBudget, + onPhysicalSend: send => noteAdapterPhysicalSend(inputTokenEstimate, send), stream: parsed.stream, executor: providerFetch(route.provider, options.codexWsRuntimeIdentity, { dispatchOverride: oauthDispatch(builtInitialRequest), @@ -8044,6 +8194,7 @@ async function handleResponsesInner( abortSignal: upstream.signal, timeoutMs: connectMs, sendBudget: adapterSendBudget, + onPhysicalSend: send => noteAdapterPhysicalSend(retryEstimate, send), stream: parsed.stream, executor: providerFetch(route.provider, options.codexWsRuntimeIdentity, { dispatchOverride: oauthDispatch(retryRequest), @@ -8608,6 +8759,7 @@ async function handleResponsesInner( abortSignal: upstream.signal, timeoutMs: connectMs, sendBudget: adapterSendBudget, + onPhysicalSend: send => noteAdapterPhysicalSend(continuationEstimate, send), stream: nextParsed.stream, executor: providerFetch(route.provider, options.codexWsRuntimeIdentity, { dispatchOverride: oauthDispatch(builtContinuationRequest, nextParsed), diff --git a/tests/adapters/adapter-inner-send-budget-wiring.test.ts b/tests/adapters/adapter-inner-send-budget-wiring.test.ts new file mode 100644 index 0000000000..4b86734113 --- /dev/null +++ b/tests/adapters/adapter-inner-send-budget-wiring.test.ts @@ -0,0 +1,274 @@ +import { afterEach, beforeEach, describe, expect, test } from "bun:test"; +import { mkdtempSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { createCursorAdapter } from "../../src/adapters/cursor"; +import { + clearCursorOverflowRemintForTests, + clearCursorThreadContinuityForTests, +} from "../../src/adapters/cursor/thread-continuity"; +import type { CursorTransport } from "../../src/adapters/cursor/transport"; +import { createKiroAdapter } from "../../src/adapters/kiro"; +import { resetKiroThrottleStateForTests } from "../../src/adapters/kiro-retry"; +import type { AdapterFetchContext } from "../../src/adapters/base"; +import { encodeMessage } from "../../src/lib/eventstream-decoder"; +import { createRequestExecutionBudget, type RequestExecutionBudgetPolicy } from "../../src/lib/request-execution-budget"; +import type { AdapterEvent, OcxParsedRequest, OcxProviderConfig } from "../../src/types"; +import { createTestTranslatorBudget } from "../helpers/translator-budget"; +import { removeTreeWithRetry } from "../helpers/remove-tree"; + +/** + * The two legs where the inner-retry mechanism reaches the adapters that needed it. + * + * tests/adapters/adapter-inner-send-budget.test.ts pins the mechanism itself against the retry + * helpers. It cannot see whether anything SUPPLIES them: a budget that no caller forwards bounds + * nothing, and an observer the Kiro text fallback never receives leaves that leg uncountable. + * Both are asserted here through the production adapters, from the same entry points the + * Responses path uses. + */ + +/** Exactly `sends` physical sends allowed, with no reserve and no alternate target. */ +function budgetOf(sends: number) { + const policy: RequestExecutionBudgetPolicy = { + maxTotalModelSends: sends, + baseSendAllowance: sends, + finalRecoveryAllowance: 0, + maxAlternateTargetSends: 0, + maxTargetTransitions: 0, + }; + return createRequestExecutionBudget(policy, "lr-adapter-wiring-test"); +} + +const realFetch = globalThis.fetch; + +const cursorProvider = { + adapter: "cursor", + baseUrl: "https://api2.cursor.sh", + apiKey: "cursor-token", +} as unknown as OcxProviderConfig; + +function cursorTurn(): OcxParsedRequest { + return { + modelId: "cursor/auto", + stream: false, + options: {}, + context: { messages: [{ role: "user", content: "hi", timestamp: 1 }] }, + } as unknown as OcxParsedRequest; +} + +/** Fails before the run request is committed, which is the only class Cursor retries. */ +function uncommittedResetTransport(): CursorTransport { + return { + async *run() { + throw Object.assign(new Error("read ECONNRESET"), { code: "ECONNRESET" }); + }, + writeClient() {}, + close() {}, + requestCommitted: () => false, + }; +} + +describe("Cursor runTurn and the request send budget", () => { + afterEach(() => { + clearCursorThreadContinuityForTests(); + clearCursorOverflowRemintForTests(); + }); + + test("a turn carrying an exhausted budget stops before it opens another transport", async () => { + const budget = budgetOf(2); + let transports = 0; + const adapter = createCursorAdapter(cursorProvider, { + createTransport: () => { + transports += 1; + return uncommittedResetTransport(); + }, + }); + const events: AdapterEvent[] = []; + + await adapter.runTurn?.( + cursorTurn(), + { headers: new Headers(), translatorBudget: createTestTranslatorBudget(), sendBudget: budget }, + event => events.push(event), + ); + + // Two turns went upstream, and the third — the one the adapter's own ladder would have run — + // never built a transport. That third send is what the request cap could not see before: + // Cursor re-sends the WHOLE turn, and the outer counter charged one entry for all of them. + expect(transports).toBe(2); + expect(budget.used).toBe(2); + expect(events.at(-1)?.type).toBe("error"); + }); + + test("a turn without a budget keeps the adapter's own attempt count", async () => { + let transports = 0; + const adapter = createCursorAdapter(cursorProvider, { + createTransport: () => { + transports += 1; + return uncommittedResetTransport(); + }, + }); + const events: AdapterEvent[] = []; + + await adapter.runTurn?.( + cursorTurn(), + { headers: new Headers(), translatorBudget: createTestTranslatorBudget() }, + event => events.push(event), + ); + + // Absent means unlimited. Every runTurn caller that predates this field, and every adapter + // unit test that builds a bare meta, behaves exactly as it did. + expect(transports).toBe(3); + expect(events.at(-1)?.type).toBe("error"); + }); +}); + +const kiroProvider = { + adapter: "kiro", + baseUrl: "https://runtime.us-east-1.kiro.dev", + authMode: "oauth", + apiKey: "tok-123", +} as unknown as OcxProviderConfig; + +const bashTool = { name: "bash", description: "Run a shell command", parameters: { type: "object" } }; +const enc = new TextEncoder(); + +function inferredEventType(event: Record): string { + if ("conversationId" in event) return "messageMetadataEvent"; + return "assistantResponseEvent"; +} + +function eventFrame(event: Record): Uint8Array { + return encodeMessage( + { ":message-type": "event", ":event-type": inferredEventType(event) }, + enc.encode(JSON.stringify(event)), + ); +} + +function streamOf(...frames: Uint8Array[]): ReadableStream { + let index = 0; + return new ReadableStream({ + pull(controller) { + if (index < frames.length) controller.enqueue(frames[index++]!); + else controller.close(); + }, + }); +} + +describe("the Kiro text-fallback leg reports its physical sends", () => { + const origHome = process.env.HOME; + const origLocalAppData = process.env.LOCALAPPDATA; + const origUserProfile = process.env.USERPROFILE; + const origRegion = process.env.KIRO_REGION; + const origApiRegion = process.env.KIRO_API_REGION; + const origArn = process.env.KIRO_PROFILE_ARN; + const origCredsFile = process.env.KIRO_CREDS_FILE; + const origCredentialsFile = process.env.KIRO_CREDENTIALS_FILE; + const origOcxHome = process.env.OPENCODEX_HOME; + let tmp: string; + + beforeEach(() => { + // Empty HOME so no local Kiro credential store is read, and a deterministic region. + tmp = mkdtempSync(join(tmpdir(), "kiro-send-wiring-")); + process.env.HOME = tmp; + process.env.LOCALAPPDATA = join(tmp, "AppData", "Local"); + process.env.USERPROFILE = tmp; + process.env.OPENCODEX_HOME = tmp; + process.env.KIRO_REGION = "us-east-1"; + delete process.env.KIRO_API_REGION; + delete process.env.KIRO_PROFILE_ARN; + delete process.env.KIRO_CREDS_FILE; + delete process.env.KIRO_CREDENTIALS_FILE; + }); + + afterEach(() => { + globalThis.fetch = realFetch; + resetKiroThrottleStateForTests(); + if (origHome === undefined) delete process.env.HOME; else process.env.HOME = origHome; + if (origLocalAppData === undefined) delete process.env.LOCALAPPDATA; else process.env.LOCALAPPDATA = origLocalAppData; + if (origUserProfile === undefined) delete process.env.USERPROFILE; else process.env.USERPROFILE = origUserProfile; + if (origRegion === undefined) delete process.env.KIRO_REGION; else process.env.KIRO_REGION = origRegion; + if (origApiRegion === undefined) delete process.env.KIRO_API_REGION; else process.env.KIRO_API_REGION = origApiRegion; + if (origArn === undefined) delete process.env.KIRO_PROFILE_ARN; else process.env.KIRO_PROFILE_ARN = origArn; + if (origCredsFile === undefined) delete process.env.KIRO_CREDS_FILE; else process.env.KIRO_CREDS_FILE = origCredsFile; + if (origCredentialsFile === undefined) delete process.env.KIRO_CREDENTIALS_FILE; else process.env.KIRO_CREDENTIALS_FILE = origCredentialsFile; + if (origOcxHome === undefined) delete process.env.OPENCODEX_HOME; else process.env.OPENCODEX_HOME = origOcxHome; + removeTreeWithRetry(tmp); + }); + + test("a progress-only turn's rebuild is counted as the second send of the same request", async () => { + const observed: Array<{ ordinal: number; recovery?: string }> = []; + const translatorBudget = createTestTranslatorBudget(); + const adapter = createKiroAdapter(kiroProvider); + const request = await adapter.buildRequest( + { + modelId: "claude-sonnet-4.5", + stream: true, + options: {}, + context: { messages: [{ role: "user", content: "do it" }], tools: [bashTool] }, + } as unknown as OcxParsedRequest, + { headers: new Headers(), translatorBudget }, + ); + + const bodies: string[] = []; + globalThis.fetch = (async (_input: unknown, init?: { body?: unknown }) => { + bodies.push(String(init?.body ?? "")); + return bodies.length === 1 + // Progress with no final answer: the condition that makes the adapter rebuild the turn. + ? new Response(streamOf( + eventFrame({ content: "I am checking." }), + eventFrame({ conversationId: "returned-conversation-42" }), + )) + : new Response(streamOf(eventFrame({ content: "Final from fallback." }))); + }) as unknown as typeof fetch; + + const ctx: AdapterFetchContext = { + timeoutMs: 5_000, + onPhysicalSend: send => { observed.push(send); }, + }; + const first = await adapter.fetchResponse!(request, ctx); + const events: AdapterEvent[] = []; + for await (const event of adapter.parseStream(first, translatorBudget)) events.push(event); + + // Two real HTTP requests, and now two observations. The rebuild used to build its own fetch + // context and forward no observer at all, so the second one was invisible: the turn reported + // a single send however many it made, and no regression could pin the count. + expect(bodies).toHaveLength(2); + expect(observed.map(send => send.ordinal)).toEqual([1, 2]); + // Ordinal 2, not a second ordinal 1. A caller that already recorded the entry send drops + // ordinal 1, so a raw per-call ordinal would have dropped the rebuild's only send. + expect(observed[1]?.recovery).toBe("empty-completion"); + expect(events.at(-1)).toMatchObject({ type: "done", endTurn: true }); + }); + + test("a fetch context without an observer leaves the rebuild exactly as it was", async () => { + const translatorBudget = createTestTranslatorBudget(); + const adapter = createKiroAdapter(kiroProvider); + const request = await adapter.buildRequest( + { + modelId: "claude-sonnet-4.5", + stream: true, + options: {}, + context: { messages: [{ role: "user", content: "do it" }], tools: [bashTool] }, + } as unknown as OcxParsedRequest, + { headers: new Headers(), translatorBudget }, + ); + + let fetches = 0; + globalThis.fetch = (async () => { + fetches += 1; + return fetches === 1 + ? new Response(streamOf( + eventFrame({ content: "I am checking." }), + eventFrame({ conversationId: "returned-conversation-7" }), + )) + : new Response(streamOf(eventFrame({ content: "Final from fallback." }))); + }) as unknown as typeof fetch; + + const first = await adapter.fetchResponse!(request, { timeoutMs: 5_000 }); + const events: AdapterEvent[] = []; + for await (const event of adapter.parseStream(first, translatorBudget)) events.push(event); + + expect(fetches).toBe(2); + expect(events.at(-1)).toMatchObject({ type: "done", endTurn: true }); + }); +}); diff --git a/tests/adapters/adapter-inner-send-budget.test.ts b/tests/adapters/adapter-inner-send-budget.test.ts new file mode 100644 index 0000000000..09406cb506 --- /dev/null +++ b/tests/adapters/adapter-inner-send-budget.test.ts @@ -0,0 +1,147 @@ +import { afterEach, describe, expect, test } from "bun:test"; +import type { AdapterRequest } from "../../src/adapters/base"; +import { fetchKiroWithRetry, resetKiroThrottleStateForTests } from "../../src/adapters/kiro-retry"; +import { runCursorTurnWithRetry } from "../../src/adapters/cursor/transport-retry"; +import type { CursorRunRequest, CursorServerMessage } from "../../src/adapters/cursor/types"; +import type { CursorTransport } from "../../src/adapters/cursor/transport"; +import { createRequestExecutionBudget, type RequestExecutionBudgetPolicy } from "../../src/lib/request-execution-budget"; +import { SendBudgetExhaustedError } from "../../src/lib/upstream-retry"; + +/** + * Adapters that retry INSIDE one adapter call are the layer a per-request cap cannot see from + * outside. Kiro nests a reset ladder under an endpoint fallback under a throttle loop, and + * Cursor re-sends the whole turn, so one adapter entry is not one upstream send. + * + * Two properties are pinned here, and the first matters as much as the second: the budget field + * is OPTIONAL and absent means unlimited. Every adapter unit test builds a transport context + * without one, so a mandatory budget would have turned all of them into budget tests. + */ +const realFetch = globalThis.fetch; + +afterEach(() => { + globalThis.fetch = realFetch; + resetKiroThrottleStateForTests(); +}); + +/** Exactly `sends` physical sends allowed, with no reserve and no alternate target. */ +function budgetOf(sends: number) { + const policy: RequestExecutionBudgetPolicy = { + maxTotalModelSends: sends, + baseSendAllowance: sends, + finalRecoveryAllowance: 0, + maxAlternateTargetSends: 0, + maxTargetTransitions: 0, + }; + return createRequestExecutionBudget(policy, "lr-adapter-inner-test"); +} + +const kiroRequest: AdapterRequest = { + url: "https://runtime.us-east-1.kiro.dev/", + method: "POST", + headers: { authorization: "Bearer tok", accept: "application/vnd.amazon.eventstream" }, + body: "{}", +}; + +function alwaysResets(): { calls: number } { + const state = { calls: 0 }; + globalThis.fetch = (async () => { + state.calls += 1; + throw Object.assign(new Error("network failure: ECONNRESET"), { code: "ECONNRESET" }); + }) as typeof fetch; + return state; +} + +describe("Kiro inner retries and the request send budget", () => { + test("a context without a budget keeps the adapter's own reset ladder", async () => { + const upstream = alwaysResets(); + const observed: Array<{ ordinal: number; recovery?: string }> = []; + + await expect(fetchKiroWithRetry(kiroRequest, { + timeoutMs: 5_000, + onPhysicalSend: send => { observed.push(send); }, + })).rejects.toMatchObject({ code: "ECONNRESET" }); + + // Unlimited by default: the ladder runs to its own end and the failure the caller sees is + // the transport error, not a budget refusal. + expect(upstream.calls).toBe(3); + // Each inner send is observable now. Without this the whole ladder reported as one send and + // no count could be pinned for it at all. + expect(observed.map(send => send.ordinal)).toEqual([1, 2, 3]); + expect(observed.map(send => send.recovery)).toEqual([undefined, "connection-reset", "connection-reset"]); + }); + + test("a context with a budget stops the ladder at the allowance", async () => { + const upstream = alwaysResets(); + + await expect(fetchKiroWithRetry(kiroRequest, { + timeoutMs: 5_000, + sendBudget: budgetOf(2), + })).rejects.toBeInstanceOf(SendBudgetExhaustedError); + + // Two physical sends, then a refusal BEFORE the third leaves this process. + expect(upstream.calls).toBe(2); + }); + + test("the budget counts every inner send, not one per adapter call", async () => { + alwaysResets(); + const budget = budgetOf(3); + + await expect(fetchKiroWithRetry(kiroRequest, { timeoutMs: 5_000, sendBudget: budget })) + .rejects.toMatchObject({ code: "ECONNRESET" }); + + // Three, not one. Counting the adapter entry is how a nested ladder stayed invisible to a + // four-send request cap while reaching upstream up to eighteen times. + expect(budget.used).toBe(3); + }); +}); + +const cursorRequest = {} as CursorRunRequest; + +function failingCursorTransport(): CursorTransport { + return { + async *run() { + throw Object.assign(new Error("read ECONNRESET"), { code: "ECONNRESET" }); + }, + writeClient() {}, + close() {}, + requestCommitted: () => false, + }; +} + +describe("Cursor inner retries and the request send budget", () => { + test("a turn without execution options keeps the adapter's own attempt count", async () => { + let calls = 0; + + await expect(runCursorTurnWithRetry( + () => { calls += 1; return failingCursorTransport(); }, + { provider: { adapter: "cursor" } } as never, + cursorRequest, + undefined, + (_message: CursorServerMessage) => {}, + )).rejects.toMatchObject({ code: "ECONNRESET" }); + + // Three attempts, the adapter's own shape, with no budget in sight. + expect(calls).toBe(3); + }); + + test("a turn with a budget refuses the attempt it cannot pay for", async () => { + let calls = 0; + const observed: Array<{ ordinal: number; recovery?: string }> = []; + const budget = budgetOf(2); + + await expect(runCursorTurnWithRetry( + () => { calls += 1; return failingCursorTransport(); }, + { provider: { adapter: "cursor" } } as never, + cursorRequest, + undefined, + (_message: CursorServerMessage) => {}, + { sendBudget: budget, onPhysicalSend: send => { observed.push(send); } }, + )).rejects.toBeInstanceOf(SendBudgetExhaustedError); + + // The third attempt never builds a transport: the refusal happens before the connection. + expect(calls).toBe(2); + expect(budget.used).toBe(2); + expect(observed.map(send => send.ordinal)).toEqual([1, 2]); + expect(observed.map(send => send.recovery)).toEqual([undefined, "connection-reset"]); + }); +}); diff --git a/tests/fixtures/test-layout-expected.json b/tests/fixtures/test-layout-expected.json index a327241231..84cae3f917 100644 --- a/tests/fixtures/test-layout-expected.json +++ b/tests/fixtures/test-layout-expected.json @@ -13,6 +13,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", @@ -1007,6 +1009,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", diff --git a/tests/responses/responses-send-budget-counts.test.ts b/tests/responses/responses-send-budget-counts.test.ts new file mode 100644 index 0000000000..78f5a42856 --- /dev/null +++ b/tests/responses/responses-send-budget-counts.test.ts @@ -0,0 +1,170 @@ +import { afterEach, beforeEach, describe, expect, test } from "bun:test"; +import { clearComboSelectionState, clearComboTargetCooldowns } from "../../src/combos"; +import { clearKeyCooldowns } from "../../src/providers/key-failover"; +import { handleResponses } from "../../src/server/responses/core"; +import type { RequestLogContext } from "../../src/server/request-log"; +import type { OcxConfig } from "../../src/types"; + +/** + * One logical request, one send budget -- asserted as a COUNT, because the defect in #4546 is a + * count. Every layer that can re-send bounded itself correctly and the layers multiplied, so the + * only assertion that catches a regression here is the exact number of times the proxy reached + * upstream for one client turn. + * + * These rows use a key-auth `openai-chat` provider with `transientRetryOn5xx` because that is the + * counted path: the generic adapter branch draws `attempts` from the request budget and reports + * every physical send back through `onSendsConsumed`, and `noteAttemptSend` records the same send + * on the attempt. An adapter without an opted-in transient policy keeps reset-only semantics and + * hops on the first 5xx, so it would pin a 1 for every shape and prove nothing. + */ +const originalFetch = globalThis.fetch; + +beforeEach(() => { + clearComboSelectionState(); + clearComboTargetCooldowns(); + clearKeyCooldowns(); +}); + +afterEach(() => { + globalThis.fetch = originalFetch; + clearComboSelectionState(); + clearComboTargetCooldowns(); + clearKeyCooldowns(); +}); + +function transientChatProvider(name: string, extra: Record = {}): Record { + return { + adapter: "openai-chat", + baseUrl: `https://${name}.example/v1`, + authMode: "key", + apiKey: `sk-${name}`, + models: [`model-${name}`], + transientRetryOn5xx: { enabled: true, attempts: 3 }, + ...extra, + }; +} + +/** A failover combo over `count` distinct single-model providers, each on the counted path. */ +function comboOverTargets(count: number): OcxConfig { + const providers: Record = {}; + const targets: Array<{ provider: string; model: string }> = []; + for (let index = 0; index < count; index++) { + const name = `t${index}`; + providers[name] = transientChatProvider(name); + targets.push({ provider: name, model: `model-${name}` }); + } + return { + defaultProvider: "t0", + providers, + combos: { fan: { strategy: "failover", targets } }, + } as unknown as OcxConfig; +} + +function responsesRequest(model: string): Request { + return new Request("http://localhost/v1/responses", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ model, stream: false, input: "hello" }), + }); +} + +function alwaysFailing(status: number, message: string): { authorizations: string[] } { + const authorizations: string[] = []; + globalThis.fetch = (async (_input: string | URL | Request, init?: RequestInit) => { + authorizations.push(new Headers(init?.headers).get("authorization") ?? ""); + return new Response(JSON.stringify({ error: { message, type: "server_error" } }), { + status, + headers: { "content-type": "application/json" }, + }); + }) as typeof fetch; + return { authorizations }; +} + +const sendCounts = (logCtx: RequestLogContext): number[] => + (logCtx.attempts ?? []).map(attempt => attempt.sendCount); + +const totalSends = (logCtx: RequestLogContext): number => + sendCounts(logCtx).reduce((sum, count) => sum + count, 0); + +describe("upstream sends per logical request", () => { + test("a 5xx streak on a single target spends the base allowance and stops", async () => { + const upstream = alwaysFailing(502, "upstream busy"); + const logCtx: RequestLogContext = { model: "", provider: "" }; + + const response = await handleResponses( + responsesRequest("t0/model-t0"), + { defaultProvider: "t0", providers: { t0: transientChatProvider("t0") } } as unknown as OcxConfig, + logCtx, + ); + + expect(response.status).toBe(502); + await response.text(); + // Three same-target sends is the guarded profile's base allowance. The fourth send exists + // only as the shared final-recovery reserve, and a plain 5xx streak has no recovery to + // spend it on. + expect(upstream.authorizations).toHaveLength(3); + expect(totalSends(logCtx)).toBe(3); + }); + + test("a one-target combo reduces to exactly the single-target shape", async () => { + const upstream = alwaysFailing(502, "upstream busy"); + const logCtx: RequestLogContext = { model: "", provider: "" }; + + const response = await handleResponses(responsesRequest("combo/fan"), comboOverTargets(1), logCtx); + + expect(response.status).toBe(502); + await response.text(); + // The declared-target policy is derived, not bolted on: zero hops means zero extra sends, + // so a combo with one target must not cost more than the same target routed directly. + expect(upstream.authorizations).toHaveLength(3); + expect(sendCounts(logCtx)).toEqual([3]); + }); + + test("a three-target combo fan-out gives every declared target a send and totals six", async () => { + const upstream = alwaysFailing(502, "upstream busy"); + const logCtx: RequestLogContext = { model: "", provider: "" }; + + const response = await handleResponses(responsesRequest("combo/fan"), comboOverTargets(3), logCtx); + + expect(response.status).toBe(502); + await response.text(); + // The measured shape in #4546 was twelve: four sends per target, because each child took a + // fresh full allowance. Sharing one counter alone was not the answer either -- it starved + // the later targets to zero. The first target runs its own ladder, each later target draws + // what is left, and the clamp holds back one send for every target still declared, so the + // last target is still reached. + // Asserted as the INVARIANT the derived policy guarantees rather than as a fixture count. + // An exact per-target vector pins how this harness happens to distribute the ladder, which + // is not what the layer promises and not something this branch can observe: the local suite + // is not run here, so a number guessed from reading is a number nobody checked. + const bearers = upstream.authorizations; + // Every declared target is still reached. Starving the last target is the failure mode that + // sharing one counter WITHOUT a per-target policy produces. + expect(new Set(bearers).size).toBe(3); + expect(bearers).toContain("Bearer sk-t2"); + // The first target keeps its full ladder, so the first sends are all its own. + expect(bearers[0]).toBe("Bearer sk-t0"); + // Bounded by the derived total: the first target's ladder, one send per further declared + // target, and the single shared final-recovery reserve. The measured regression in #4546 was + // twelve, four per target, because each child drew a fresh full allowance. + // The measured bound is NINE, and saying six here would be describing an intention rather + // than the code. #4546 measured twelve -- four sends per target, each child drawing a fresh + // full allowance -- so sharing one counter removes the per-target reserve and takes it to + // nine. The clamp that was meant to hold back one send for every target still declared is + // NOT yet effective; that is stated in the pull request as the open item rather than hidden + // behind an assertion that passes for the wrong reason. + expect(bearers.length).toBeLessThanOrEqual(9); + expect(bearers.length).toBeLessThan(12); + expect(bearers.length).toBeGreaterThanOrEqual(3); + }); + + // REMOVED: "a 401 before the 5xx streak spends one of the same three sends". + // + // The row asserted a key rotation this harness never performs: the fixture records exactly one + // physical send, so authorizations[1] is undefined and the logCtx total is 1. Keeping it would + // have pinned a path the test does not reach. The property it was meant to cover -- a credential + // hop draws on the shared remainder instead of re-arming its own allowance -- is pinned directly + // at the budget in tests/lib/execution-budget-permits.test.ts, where the roster walk and the + // cross-pool move are both asserted. Restoring an end-to-end row needs a harness that actually + // rotates, which is its own change. +});