Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
72 changes: 41 additions & 31 deletions src/server/responses/passthrough-dispatch.ts
Original file line number Diff line number Diff line change
Expand Up @@ -694,10 +694,13 @@ export async function preparePassthroughExchange(
* provider can narrow this request's sends exactly and cannot widen the bound that exists to
* stop per-request amplification (#4546).
*/
const transientSendPolicy = () => transientRetryPolicyFor(route.provider);
const transientSendAttempts = (): number => transientSendCapFor(
transientRetryPolicyFor(route.provider)?.attempts,
transientSendPolicy()?.attempts,
sendBudgetState.sendsUsed,
);
const configuredTransientSendBudgetExhausted = (): boolean =>
transientSendPolicy() !== null && transientSendAttempts() === 0;
/**
* Refuse a built body that exceeds the operator's configured ceiling, before it is sent.
*
Expand Down Expand Up @@ -897,14 +900,15 @@ export async function preparePassthroughExchange(
recordAttemptCredentialSource(logCtx.activeAttempt, route.providerName, route.provider, retryAdapter.name);
const rebuiltBodyRefusal = refuseOversizedOutboundBody(request);
if (rebuiltBodyRefusal) return { failed: rebuiltBodyRefusal };
// The base allowance is spent first; once it is gone this leg may still draw the one
// shared final-recovery reserve, which is what keeps a validated sanitized rebuild
// after a 5xx streak alive at four total sends instead of dying at three. Reserved
// outside the try so the finally can hand it back if the leg never reached its send.
// The base allowance is spent first. An unconfigured provider may then draw the one shared
// final-recovery reserve, which keeps a validated sanitized rebuild after a 5xx streak alive
// at four total sends instead of dying at three. An explicit provider total cannot widen.
// Reserve outside the try so the finally can hand it back if the leg never reaches its send.
const allowance = recoverySendAllowance(
transientSendAttempts(),
recoveryClassFor(recovery),
`${route.providerName}|${route.modelId}|${recovery}`,
{ allowFinalRecoveryReserve: transientSendPolicy() === null },
);
try {
return await fetchWithTransientRetry(
Expand Down Expand Up @@ -1414,25 +1418,28 @@ export async function preparePassthroughExchange(
// eviction, or an older transcript). Inspect only a bounded clone of a 4xx whose exact outbound
// Responses body still carries opaque state, then rebuild once through the ordinary adapter
// sanitation path. A second rejection falls through unchanged because the guard stays armed.
const opaqueBlobRecovery = await attemptOpaqueBlobRecovery({
response: upstreamResponse,
outboundBody: request.body,
adapterName: transportState.adapter.name,
parsed,
guard: opaqueBlobRecoveryGuard,
signal: upstream.signal,
}, rebuildAndRefetch);
if (opaqueBlobRecovery.kind === "failed") return opaqueBlobRecovery.response;
if (opaqueBlobRecovery.kind === "recovered") {
upstreamResponse = opaqueBlobRecovery.response;
continue passthroughRecovery;
if (!configuredTransientSendBudgetExhausted()) {

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Preserve the upstream response when recovery has no allowance

When transientRetryOn5xx.attempts is 4–10 and the initial transient ladder consumes all three base sends, this predicate remains true because the configured total still has headroom. A recovery-eligible final response (for example, a 502 encrypted-function-output rejection) is therefore consumed and cancelled by attemptOpaqueBlobRecovery, but rebuildAndRefetch then receives zero attempts because line 911 disables the final reserve for every configured policy; the client gets a synthetic request_send_budget_exhausted 429 instead of the original upstream response. Gate the recovery on its effective allowance before mutating the response, or permit the reserve while the configured total still has headroom.

AGENTS.md reference: src/AGENTS.md:L19-L19

Useful? React with 👍 / 👎.

const opaqueBlobRecovery = await attemptOpaqueBlobRecovery({
response: upstreamResponse,
outboundBody: request.body,
adapterName: transportState.adapter.name,
parsed,
guard: opaqueBlobRecoveryGuard,
signal: upstream.signal,
}, rebuildAndRefetch);
if (opaqueBlobRecovery.kind === "failed") return opaqueBlobRecovery.response;
if (opaqueBlobRecovery.kind === "recovered") {
upstreamResponse = opaqueBlobRecovery.response;
continue passthroughRecovery;
}
}

const recoveryContentType = upstreamResponse.headers.get("content-type")?.toLowerCase() ?? "";
const streamedFunctionOutputCandidate = upstreamResponse.ok
&& !!upstreamResponse.body
&& (recoveryContentType.includes("text/event-stream") || (!recoveryContentType && parsed.stream))
&& !opaqueBlobRecoveryGuard.attempted
&& !configuredTransientSendBudgetExhausted()
&& outboundResponsesBodyCarriesEncryptedFunctionOutput(request.body);
if (streamedFunctionOutputCandidate) {
const preflightLog: RequestLogContext = { model: logCtx.model, provider: logCtx.provider };
Expand All @@ -1449,19 +1456,21 @@ export async function preparePassthroughExchange(
if (options.abortSignal?.aborted) return transportFailureResponse(options.abortSignal.reason);
upstreamResponse = preflight.response;
if (preflight.kind === "failed") {
const streamedOpaqueRecovery = await attemptOpaqueBlobRecovery({
response: upstreamResponse,
outboundBody: request.body,
adapterName: transportState.adapter.name,
parsed,
guard: opaqueBlobRecoveryGuard,
signal: upstream.signal,
}, rebuildAndRefetch);
if (streamedOpaqueRecovery.kind === "failed") return streamedOpaqueRecovery.response;
if (streamedOpaqueRecovery.kind === "recovered") {
resetStreamedOpaqueBlobLogContext(logCtx);
upstreamResponse = streamedOpaqueRecovery.response;
continue passthroughRecovery;
if (!configuredTransientSendBudgetExhausted()) {
const streamedOpaqueRecovery = await attemptOpaqueBlobRecovery({
response: upstreamResponse,
outboundBody: request.body,
adapterName: transportState.adapter.name,
parsed,
guard: opaqueBlobRecoveryGuard,
signal: upstream.signal,
}, rebuildAndRefetch);
if (streamedOpaqueRecovery.kind === "failed") return streamedOpaqueRecovery.response;
if (streamedOpaqueRecovery.kind === "recovered") {
resetStreamedOpaqueBlobLogContext(logCtx);
upstreamResponse = streamedOpaqueRecovery.response;
continue passthroughRecovery;
}
}
logCtx.upstreamError = preflightLog.upstreamError;
logCtx.terminalHttpStatus = preflightLog.terminalHttpStatus;
Expand All @@ -1480,6 +1489,7 @@ export async function preparePassthroughExchange(
upstream.signal,
);
if (uploadRejectionBody !== undefined
&& !configuredTransientSendBudgetExhausted()
&& isTransientConsoleGoUploadRejection({
status: upstreamResponse.status,
errorBody: uploadRejectionBody,
Expand Down Expand Up @@ -1518,7 +1528,7 @@ export async function preparePassthroughExchange(
requested: parsed.options.reasoning,
rejectionText,
});
if (downgrade) {
if (downgrade && !configuredTransientSendBudgetExhausted()) {
reasoningEffortDowngradeGuard.attempted = true;
parsed.options.reasoning = downgrade.effort;
try { void upstreamResponse.body?.cancel().catch(() => {}); } catch { /* already consumed/closed */ }
Expand Down
10 changes: 8 additions & 2 deletions src/server/responses/request-send-budget.ts
Original file line number Diff line number Diff line change
Expand Up @@ -145,16 +145,22 @@ export function createResponsesSendBudget(
* The base allowance is spent first. Once it is gone a recovery class may still draw the
* single shared final-recovery reserve -- which is what keeps the validated sanitized rebuild
* after a 5xx streak alive at four total sends -- but an account move and a rebuild cannot
* each take one. `countedExternally` is set because these legs run through the retry helper,
* which reports the same send again through `onSendsConsumed`.
* each take one. A caller with an exact provider total can suppress that reserve. The
* `countedExternally` flag exists because these legs run through the retry helper, which reports
* the same send again through `onSendsConsumed`.
*/
const recoverySendAllowance = (
cap: number,
sendClass: SendClass,
targetKey: string,
options: { allowFinalRecoveryReserve?: boolean } = {},
): { attempts: number; permit?: SingleUseDispatchPermit } => {
const base = remainingTransientSendBudget(cap);
if (base > 0) return { attempts: base };
// A provider-configured transient total is an exact physical-send ceiling. Once it is
// exhausted, the request-wide recovery reserve must not silently widen it. The default stays
// permissive so unconfigured providers retain the guarded profile's fourth recovery send.
if (options.allowFinalRecoveryReserve === false) return { attempts: 0 };
if (pendingHopPermit) {
const hopPermit = pendingHopPermit;
pendingHopPermit = undefined;
Expand Down
5 changes: 5 additions & 0 deletions structure/transports/responses.md
Original file line number Diff line number Diff line change
Expand Up @@ -888,6 +888,11 @@ because its replay is the next loop iteration. `run-turn-execution.ts` always ha
down, because a runTurn adapter is by definition the layer that sends. The passthrough ladder keeps
the shape it already had: reserve with `countedExternally: true` and pass the permit to the rebuild.

An explicit provider `transientRetryOn5xx.attempts` value is the exact physical-send total for that
request. Once spent, a passthrough rebuild receives no final-recovery reserve and returns the
original upstream response. The guarded profile's shared reserve remains available only when the
provider leaves that transient policy unconfigured; its existing hop-permit settlement is unchanged.

What must not happen is a ladder that charges and then returns through a path that neither confirms
nor releases. That is not a lost send; it is a send the request never made, spending an allowance a
later recovery in the same request then cannot have. `tests/lib/execution-budget-permits.test.ts`
Expand Down
18 changes: 18 additions & 0 deletions tests/responses/responses-console-go-upload-retry.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -109,6 +109,24 @@ describe("Console Go transient upload refusal recovery", () => {
expect(logCtx.activeAttempt?.recoveryKinds).toEqual(["console-go-upload-retry"]);
});

test("a configured one-send total returns the original refusal without using the reserve", async () => {
const cfg = config();
cfg.providers.go!.transientRetryOn5xx = { attempts: 1 };
let sends = 0;
globalThis.fetch = (async () => {
sends += 1;
return refusal();
}) as typeof fetch;
const logCtx: RequestLogContext = { model: "", provider: "" };

const response = await handleResponses(request(), cfg, logCtx);

expect(response.status).toBe(400);
expect(await response.text()).toBe(UPLOAD_REFUSAL);
expect(sends).toBe(1);
expect(logCtx.activeAttempt?.recoveryKinds).toEqual([]);
});

test("does not replay a different 400 from the same wire", async () => {
let sends = 0;
globalThis.fetch = (async () => {
Expand Down
46 changes: 44 additions & 2 deletions tests/responses/responses-passthrough-transient-policy.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -34,8 +34,9 @@ describe("the Responses passthrough lane reads the provider transient policy", (

test("the ladder is resolved from the provider row, not a constant", () => {
expect(packed).toContain(dense(
"const transientSendAttempts = (): number => transientSendCapFor(\n"
+ " transientRetryPolicyFor(route.provider)?.attempts,\n"
"const transientSendPolicy = () => transientRetryPolicyFor(route.provider);\n"
+ "const transientSendAttempts = (): number => transientSendCapFor(\n"
+ " transientSendPolicy()?.attempts,\n"
+ " sendBudgetState.sendsUsed,\n"
+ ");",
));
Expand Down Expand Up @@ -132,4 +133,45 @@ describe("a configured ladder is bounded by the request budget", () => {
// And a narrower configured value still wins over the remaining allowance.
expect(state.remainingTransientSendBudget(transientSendCapFor(2, state.sendsUsed))).toBe(0);
});

test("a configured one-send total cannot draw the final recovery reserve", () => {
const budget = createRequestExecutionBudget();
const state = sendBudgetFor(budget);

state.noteTransientSends(1);
const cap = transientSendCapFor(1, state.sendsUsed);
expect(cap).toBe(0);

// Compose the real rebuild allowance rather than checking only the exhaustion predicate.
const allowance = state.recoverySendAllowance(
cap,
"repair",
"provider|model|reasoning-effort-downgrade",
{ allowFinalRecoveryReserve: false },
);
expect(allowance).toEqual({ attempts: 0 });
expect(budget.used).toBe(1);
expect(budget.reserveSpent).toBe(false);
});

test("an unconfigured provider retains the shared final recovery reserve", () => {
const budget = createRequestExecutionBudget();
const state = sendBudgetFor(budget);

state.noteTransientSends(TRANSIENT_RETRY_MAX_ATTEMPTS);
const cap = transientSendCapFor(undefined, state.sendsUsed);
const allowance = state.recoverySendAllowance(
cap,
"repair",
"provider|model|reasoning-effort-downgrade",
);

expect(allowance.attempts).toBe(1);
expect(allowance.permit).toBeDefined();
expect(budget.used).toBe(4);
expect(budget.reserveSpent).toBe(true);
allowance.permit?.release();
expect(budget.used).toBe(3);
expect(budget.reserveSpent).toBe(false);
});
});
Loading