diff --git a/docs-site/src/content/docs/guides/combos.md b/docs-site/src/content/docs/guides/combos.md index 1e1e1fc7789..8445c2a5569 100644 --- a/docs-site/src/content/docs/guides/combos.md +++ b/docs-site/src/content/docs/guides/combos.md @@ -698,7 +698,11 @@ classify the original HTTP 403 as a hop. For streaming requests, the upstream HTTP status is not the final decision. OpenCodex buffers a bounded pre-output prefix of the selected child's Responses SSE. If the stream reports a retryable `response.failed` terminal before any text, reasoning, tool call, or other output event, the child -is recorded as failed and the combo may try its next eligible target. Once any output event begins, +is recorded as failed and the combo may try its next eligible target. A zero-output +`response.incomplete` follows the same path when its reason is missing or is `adapter_eof`, +`missing_terminal_event`, or `upstream_stall_timeout`. Explicit semantic reasons such as +`max_output_tokens` and `content_filter` remain terminal, as does an explicit null reason. +A reasonless final snapshot that already contains output items is also kept. Once any output event begins, the target is committed: a later stream failure is returned to the client and is never replayed on another provider, which prevents duplicate text and tool execution. If the pre-output buffer reaches its safety cap without a terminal or output boundary, OpenCodex also commits the current target diff --git a/src/server/responses/adapter-dispatch.ts b/src/server/responses/adapter-dispatch.ts index 38c48500f16..57fd020dfc1 100644 --- a/src/server/responses/adapter-dispatch.ts +++ b/src/server/responses/adapter-dispatch.ts @@ -2,6 +2,7 @@ import { recoverAutoPausedAccount } from "./account-auto-pause-recovery"; import { wasAccountAutoPaused, wasPermanentAccountRefusal } from "../../oauth/account-auto-pause"; import { bindAnthropicRejectionResponse } from "../../adapters/anthropic-rejection-memo"; import { deferPhysicalSendReporter } from "./deferred-send-reporter"; +import { usesApiKeyAccount } from "../../providers/label"; import { spendLedgerStorageErrorResponse } from "./spend-storage-error"; import { classifyAnthropic429 } from "../../oauth/anthropic-rate-limit-policy"; import { rotateAnthropicAccountOnResponseForInstance } from "../../oauth/anthropic-account-refusal"; @@ -400,7 +401,7 @@ export async function prepareAdapterExchange( const autoPauseEnabled = config.providers[route.providerName]?.accountAutoPause?.enabled === true; const billingPoolRequest = isAnthropicBillingKeyPool(route.provider); let billingRecoveryActive = false; - const genericRefetchCountsSends = () => sendBudgetState.adapterSendBudget?.spendEnforced === true + const genericRefetchCountsSends = () => options.comboAttempt === true || sendBudgetState.adapterSendBudget?.spendEnforced === true || transientRetryPolicyFor(route.provider) !== null || resetReplayPolicyFor(route.provider) !== null || billingPoolRequest || billingRecoveryActive || autoPauseEnabled; let handoffPermit: ResponsesSendBudget["pendingHopPermit"]; @@ -762,7 +763,9 @@ export async function prepareAdapterExchange( : undefined; let refetchPermitUsed = false; const refetchReporter = refetchAllowance ? transientSendReporter(refetchAllowance.permit) : undefined; - const deferredRefetch = billingPoolRequest && refetchReporter ? deferPhysicalSendReporter(refetchReporter) : undefined; + const deferredRefetch = (billingPoolRequest || options.comboAttempt) && refetchReporter + ? deferPhysicalSendReporter(refetchReporter) : undefined; + const pendingKeyMetadata = deferredRefetch && usesApiKeyAccount(route.provider); try { return await refetchWithPolicy( recoveryKind => { @@ -771,14 +774,14 @@ export async function prepareAdapterExchange( }); // Prepare key metadata before the guarded transport commits its attempt; // counters and permits still wait for the physical executor after pacing. - if (deferredRefetch) transportState.noteRoutedAttemptSend(retryEstimate, recoveryKind ?? recovery); + if (pendingKeyMetadata) transportState.noteRoutedAttemptSend(retryEstimate, recoveryKind ?? recovery); const confirmDispatch = () => { if (refetchAllowance?.permit && !refetchPermitUsed) { if (!refetchAllowance.permit.use()) throw new SendBudgetExhaustedError(safeHostLabel(retryRequest.url)); refetchPermitUsed = true; } - if (!deferredRefetch) transportState.noteRoutedAttemptSend(retryEstimate, recoveryKind ?? recovery); - // Eligible pools confirm after pacing at the physical executor; legacy + if (!pendingKeyMetadata) transportState.noteRoutedAttemptSend(retryEstimate, recoveryKind ?? recovery); + // Counted Combo/pool sends confirm at the physical executor; legacy // helper paths retain their existing thunk admission boundary. onDispatch?.(); if (preserveFailureResponse) replacementAdmitted = true; @@ -891,7 +894,7 @@ export async function prepareAdapterExchange( const adapterOwnsDispatch = transportState.activeAdapter.fetchResponse !== undefined; const hop = reserveCredentialHop("auth-recovery", `${route.providerName}|${route.modelId}|antigravity-auth`, - !adapterOwnsDispatch && (transientRetryPolicyFor(route.provider) !== null || resetReplayPolicyFor(route.provider) !== null)); + !adapterOwnsDispatch && genericRefetchCountsSends()); if (!hop.allowed) return null; try { const snapshot = await failoverAccountSnapshot(route.providerName, nextId); @@ -991,7 +994,7 @@ export async function prepareAdapterExchange( const adapterOwnsDispatch = transportState.activeAdapter.fetchResponse !== undefined; const hop = reserveCredentialHop("auth-recovery", `${route.providerName}|${route.modelId}|terminal-refresh-account`, - !adapterOwnsDispatch && (transientRetryPolicyFor(route.provider) !== null || resetReplayPolicyFor(route.provider) !== null)); + !adapterOwnsDispatch && genericRefetchCountsSends()); if (hop.allowed) { try { const admitted = await applyFailoverSnapshot(alternate); @@ -1361,12 +1364,31 @@ export async function prepareAdapterExchange( && anthropicInstance && transportState.anthropicPoolAccountId ) { - const nextAccountId = await rotateAnthropicAccountOnResponseForInstance(anthropicInstance, upstreamResponse, { - config, accountId: transportState.anthropicPoolAccountId, sessionKey: anthropicSessionKey, - model: route.modelId, requestKey: transportState, decision: transportState.anthropicRouteDecision, currentDecision: currentAnthropicRouteDecision, signal: upstream.signal, - canRetry: !sendBudgetExhausted() && transportState.anthropicPoolFailovers < ANTHROPIC_POOL_MAX_FAILOVERS_PER_REQUEST, - }); - if (!nextAccountId) break; + // Same admission as the other rotation ladders: the roster cap cannot see the shared + // send budget. A canRetry term, not a break -- the call also records the refusal. + const adapterOwnsDispatch = transportState.activeAdapter.fetchResponse !== undefined; + const hop = reserveCredentialHop( + "auth-recovery", + `${route.providerName}|${route.modelId}|anthropic-pool-rotation`, + !adapterOwnsDispatch && genericRefetchCountsSends(), + ); + let nextAccountId: string | null; + try { + nextAccountId = await rotateAnthropicAccountOnResponseForInstance(anthropicInstance, upstreamResponse, { + config, accountId: transportState.anthropicPoolAccountId, sessionKey: anthropicSessionKey, + model: route.modelId, requestKey: transportState, decision: transportState.anthropicRouteDecision, currentDecision: currentAnthropicRouteDecision, signal: upstream.signal, + // A denied reservation must not rotate on base headroom its bounds cannot see. + canRetry: hop.allowed && (hop.permit !== undefined || !sendBudgetExhausted()) + && transportState.anthropicPoolFailovers < ANTHROPIC_POOL_MAX_FAILOVERS_PER_REQUEST, + }); + } catch (rotationError) { + hop.permit?.release(); + throw rotationError; + } + if (!nextAccountId) { + hop.permit?.release(); + break; + } try { const admitted = await commitResolvedOAuthSelection(await anthropicRoutingFor(anthropicInstance).getAnthropicPoolAccessSnapshot(nextAccountId)); if (!admitted) throw new Error("OAuth selection changed during recovery"); @@ -1389,12 +1411,26 @@ export async function prepareAdapterExchange( }); sealRequestAttemptIdentity(logCtx.activeAttempt, logCtx.provider, transportState.activeAdapter.name, logCtx.accountLogLabel); recordAttemptCredentialSource(logCtx.activeAttempt, route.providerName, route.provider, transportState.activeAdapter.name); - const result = await rebuildAndRefetch(upstreamResponse.status === 401 ? "oauth-401" - : upstreamResponse.status === 403 ? "oauth-account-403" : "anthropic-oauth-429"); - if ("failed" in result) return result.failed; - upstreamResponse = result; + // The replay IS this hop's send; the dispatching layer settles the reservation. + sendBudgetState.pendingHopPermit = hop.permit; + let result: Response | { failed: Response }; + try { + result = await rebuildAndRefetch(upstreamResponse.status === 401 ? "oauth-401" + : upstreamResponse.status === 403 ? "oauth-account-403" : "anthropic-oauth-429", + () => { if (!adapterOwnsDispatch) hop.permit?.use(); }); + } finally { + sendBudgetState.pendingHopPermit = undefined; + } + if ("failed" in result) { + // A no-op once the boundary was reached; a refund when the rebuild died first. + hop.permit?.release(); + return result.failed; + } + upstreamResponse = result; if (isNonReplayableResponse(upstreamResponse)) continue recovery; - } catch { + } catch { + // release() refunds a pre-send hop and is a no-op once the send settled it. + hop.permit?.release(); break; } } @@ -1434,9 +1470,9 @@ export async function prepareAdapterExchange( const hop = reserveCredentialHop( "auth-recovery", `${route.providerName}|${route.modelId}|adapter-recovery-oauth-429`, - // Transient and opted-in reset helpers report physical sends; an adapter-owned + // Counted generic helpers report physical sends; an adapter-owned // ladder settles its own booking. - !adapterOwnsDispatch && (transientRetryPolicyFor(route.provider) !== null || resetReplayPolicyFor(route.provider) !== null), + !adapterOwnsDispatch && genericRefetchCountsSends(), ); if (!hop.allowed) break; const nextAccountId = rotateGenericOAuthAccountOnRefusal( @@ -1538,9 +1574,9 @@ export async function prepareAdapterExchange( const hop = reserveCredentialHop( "auth-recovery", `${route.providerName}|${route.modelId}|adapter-recovery-oauth-429`, - // Transient and opted-in reset helpers report physical sends; an adapter-owned + // Counted generic helpers report physical sends; an adapter-owned // ladder settles its own booking. - !adapterOwnsDispatch && (transientRetryPolicyFor(route.provider) !== null || resetReplayPolicyFor(route.provider) !== null), + !adapterOwnsDispatch && genericRefetchCountsSends(), ); if (!hop.allowed) break; const nextAccountId = rotateGenericOAuthAccountOn429( @@ -1659,7 +1695,7 @@ export async function prepareAdapterExchange( const hop = reserveCredentialHop( "auth-recovery", `${route.providerName}|${route.modelId}|adapter-recovery-oauth-verify`, - !adapterOwnsDispatch && (transientRetryPolicyFor(route.provider) !== null || resetReplayPolicyFor(route.provider) !== null), + !adapterOwnsDispatch && genericRefetchCountsSends(), ); if (!hop.allowed) break; // Auth-refusal rotation, not the rate-limit one: a verification refusal diff --git a/src/server/responses/combo-stream-preflight.ts b/src/server/responses/combo-stream-preflight.ts index 5cdc1b4ddb6..39554a2bd81 100644 --- a/src/server/responses/combo-stream-preflight.ts +++ b/src/server/responses/combo-stream-preflight.ts @@ -115,12 +115,14 @@ function retryableZeroOutputTerminal(payload: unknown): boolean { if (!payload || typeof payload !== "object" || Array.isArray(payload)) return false; const event = payload as { type?: unknown; - response?: { incomplete_details?: { reason?: unknown } }; + response?: { incomplete_details?: { reason?: unknown }; output?: unknown }; }; if (bareErrorIsRetryable(event)) return true; if (event.type === "response.failed") return true; if (event.type !== "response.incomplete") return false; const reason = event.response?.incomplete_details?.reason; + // A terminal-only upstream can carry output without preceding delta/item events. + if (reason === undefined) return !Array.isArray(event.response?.output) || event.response.output.length === 0; return typeof reason === "string" && RETRYABLE_ZERO_OUTPUT_INCOMPLETE_REASONS.has(reason); } diff --git a/src/server/responses/core-combo.ts b/src/server/responses/core-combo.ts index c96244d1718..ccc25879c7d 100644 --- a/src/server/responses/core-combo.ts +++ b/src/server/responses/core-combo.ts @@ -45,6 +45,7 @@ import { } from "../../combos"; import { formatErrorResponse } from "../../bridge"; import { SEND_BUDGET_EXHAUSTED_CODE } from "../../lib/errors"; +import { redactSecretString, sanitizeLogMetadataString } from "../../lib/redact"; import { expandPreviousResponseInput, previousResponseReplayFailure, @@ -1195,8 +1196,15 @@ export async function executeComboResponses( } return lastFailure; } + // A bare status made a revoked credential read as a provider outage. Log only the parsed + // code and message, never the raw envelope, as one redacted, bounded line. + const failureReason = sanitizeLogMetadataString( + redactSecretString([failure.upstreamCode, failure.upstreamMessage].filter(Boolean).join(": ")).replace(/\s+/g, " "), + 200, + ); console.warn( - `[combo] ${comboId}: ${targetKey(pick.target)} failed with ${failure.response.status} after ${Date.now() - started}ms`, + `[combo] ${comboId}: ${targetKey(pick.target)} failed with ${failure.response.status}` + + ` after ${Date.now() - started}ms${failureReason ? `: ${failureReason}` : ""}`, ); const failureNow = Date.now(); const attemptedTargets = pick.attempted; diff --git a/src/server/responses/core.ts b/src/server/responses/core.ts index efe171ec163..e3e77fd6cb1 100644 --- a/src/server/responses/core.ts +++ b/src/server/responses/core.ts @@ -12,6 +12,7 @@ import { createInferenceSendBudget } from "../inference/context"; import { finalizeOwnedTranslatorBudget, finalizeAccountLease } from "./core-lifetime"; import type { TranslatorBudget } from "../../lib/translator-budget"; import { executeComboResponses } from "./core-combo"; +import { registerTurn, unregisterTurn } from "../lifecycle"; import { runWithCompactionRecovery } from "./compaction-recovery"; /** * Route one `/v1/responses` request through the adapter pipeline: recovery loop, passthrough @@ -67,15 +68,17 @@ export async function handleComboResponses( logCtx: RequestLogContext, options: HandleResponsesOptions & { translatorBudget: TranslatorBudget }, ): Promise { - return executeComboResponses( - req, - rawBody, - comboId, - config, - logCtx, - options, - requestDispatchers, - ); + // Each child relay binds the turn's lease and unbinds it when its stream ends; the lease is + // released once nothing is bound. Hold it for the whole ladder so a refused child's relay + // cannot settle the turn the next target still runs in ("turn already settled"). An accepted + // child's stream is bound before this returns and keeps the lease until it ends. + const ladder = new AbortController(); + registerTurn(ladder, options.turnAdmissionLease); + try { + return await executeComboResponses(req, rawBody, comboId, config, logCtx, options, requestDispatchers); + } finally { + if (options.turnAdmissionLease && "bindAbortController" in options.turnAdmissionLease) unregisterTurn(ladder); + } } const requestDispatchers: ResponsesDispatchers = { handleResponses, handleComboResponses }; export { adapterNeedsForcedContinuation } from "./core-replay"; diff --git a/structure/INDEX.md b/structure/INDEX.md index 3677e16c754..e6c9e5e80bb 100644 --- a/structure/INDEX.md +++ b/structure/INDEX.md @@ -53,6 +53,7 @@ The wire surfaces a client actually talks to. | [`transports/native-reasoning-retention.md`](transports/native-reasoning-retention.md) | Optional native reasoning ciphertext forwarding on model/account switches, configuration and management controls. | | [`transports/responses-wire-shapes.md`](transports/responses-wire-shapes.md) | Mixed-wire model defaults, xAI agent-message continuation, declared-tool membership, and passthrough SSE stream shapes. | | [`transports/responses-failover.md`](transports/responses-failover.md) | Upstream reset retry, the ambiguous-resend gate, combo failover and commit boundaries, compaction routing, and output headroom. | +| [`transports/responses-combo-boundaries.md`](transports/responses-combo-boundaries.md) | Combo preflight output commitment, turn lease lifetime, and bounded hop warnings. | | [`transports/policy-fallback.md`](transports/policy-fallback.md) | Original policy authorization, concrete candidate retries and physical destination deduplication. | | [`transports/responses-spend.md`](transports/responses-spend.md) | Credential-hop and durable spend reservations, and what a spent budget tells the client. | | [`transports/streaming-health.md`](transports/streaming-health.md) | Heartbeat and stall deadlines, plus the opt-in WebSocket transport. | @@ -248,7 +249,7 @@ A source area can be described by more than one doc, because these docs are orga | `src/router.ts` | [`runtime.md`](runtime.md) | | `src/routing/` | [`catalog.md`](catalog.md) | | `src/routing/history/` | [`dashboard-and-usage.md`](dashboard-and-usage.md) | -| `src/server/` | [`providers/generic-account-pool.md`](providers/generic-account-pool.md)
[`runtime.md`](runtime.md)
[`codex-prompts.md`](codex-prompts.md)
[`catalog.md`](catalog.md)
[`subagents.md`](subagents.md)
[`transports/byte-accounting.md`](transports/byte-accounting.md)
[`transports/responses.md`](transports/responses.md)
[`transports/native-reasoning-retention.md`](transports/native-reasoning-retention.md)
[`transports/responses-wire-shapes.md`](transports/responses-wire-shapes.md)
[`transports/responses-failover.md`](transports/responses-failover.md)
[`transports/policy-fallback.md`](transports/policy-fallback.md)
[`transports/streaming-health.md`](transports/streaming-health.md)
[`transports/inventory.md`](transports/inventory.md)
[`data-planes/images.md`](data-planes/images.md)
[`data-planes/inbound-compat.md`](data-planes/inbound-compat.md)
[`providers/anthropic-progress.md`](providers/anthropic-progress.md)
[`providers-and-adapters.md`](providers-and-adapters.md)
[`providers/jev-decision.md`](providers/jev-decision.md)
[`providers/xai-grok.md`](providers/xai-grok.md)
[`adapters/registry.md`](adapters/registry.md)
[`gui-and-management-api.md`](gui-and-management-api.md)
[`provider-completion-diagnostic.md`](provider-completion-diagnostic.md)
[`dashboard-and-usage.md`](dashboard-and-usage.md)
[`clients/claude-desktop.md`](clients/claude-desktop.md)
[`ops/service-and-sidecars.md`](ops/service-and-sidecars.md) | +| `src/server/` | [`providers/generic-account-pool.md`](providers/generic-account-pool.md)
[`runtime.md`](runtime.md)
[`codex-prompts.md`](codex-prompts.md)
[`catalog.md`](catalog.md)
[`subagents.md`](subagents.md)
[`transports/byte-accounting.md`](transports/byte-accounting.md)
[`transports/responses.md`](transports/responses.md)
[`transports/native-reasoning-retention.md`](transports/native-reasoning-retention.md)
[`transports/responses-wire-shapes.md`](transports/responses-wire-shapes.md)
[`transports/responses-failover.md`](transports/responses-failover.md)
[`transports/responses-combo-boundaries.md`](transports/responses-combo-boundaries.md)
[`transports/policy-fallback.md`](transports/policy-fallback.md)
[`transports/streaming-health.md`](transports/streaming-health.md)
[`transports/inventory.md`](transports/inventory.md)
[`data-planes/images.md`](data-planes/images.md)
[`data-planes/inbound-compat.md`](data-planes/inbound-compat.md)
[`providers/anthropic-progress.md`](providers/anthropic-progress.md)
[`providers-and-adapters.md`](providers-and-adapters.md)
[`providers/jev-decision.md`](providers/jev-decision.md)
[`providers/xai-grok.md`](providers/xai-grok.md)
[`adapters/registry.md`](adapters/registry.md)
[`gui-and-management-api.md`](gui-and-management-api.md)
[`provider-completion-diagnostic.md`](provider-completion-diagnostic.md)
[`dashboard-and-usage.md`](dashboard-and-usage.md)
[`clients/claude-desktop.md`](clients/claude-desktop.md)
[`ops/service-and-sidecars.md`](ops/service-and-sidecars.md) | | `src/server/gui-pair-delivery.ts` | [`remote-link.md`](remote-link.md) | | `src/server/index.ts` | [`adapters/compatibility-lab.md`](adapters/compatibility-lab.md) | | `src/server/inference/advisor-diagnostics.ts` | [`advisor.md`](advisor.md) | diff --git a/structure/manifest.json b/structure/manifest.json index ec894791ee1..43e7ed5815b 100644 --- a/structure/manifest.json +++ b/structure/manifest.json @@ -326,6 +326,15 @@ "src/codex/" ] }, + { + "path": "transports/responses-combo-boundaries.md", + "tier": 3, + "title": "Combo Streaming Boundaries", + "scope": "Combo preflight output commitment, turn lease lifetime, and bounded hop warnings.", + "documents": [ + "src/server/" + ] + }, { "path": "transports/policy-fallback.md", "tier": 3, diff --git a/structure/transports/anthropic-thinking-recovery.md b/structure/transports/anthropic-thinking-recovery.md index 3739e683901..c82cbf29a73 100644 --- a/structure/transports/anthropic-thinking-recovery.md +++ b/structure/transports/anthropic-thinking-recovery.md @@ -22,6 +22,11 @@ signature recovery. Empty-completion retries in `src/server/responses/adapter-co rebuild through their existing pacing/send owner. Ordinary continuations after output cannot recover signatures. Anthropic has no `runTurn` recovery. +Combo signature repairs share the [physical-send allowance](responses-spend.md#prepaid-initial-sends) +with initial sends and later targets, including generic providers without an opt-in retry policy. +The HTTP and initial-stream repair paths settle each physical send once; a pacing refusal leaves +the unsent repair uncharged. Recovery preserves the original thinking classification and stripping. + Only a delivery-validated completed response with actual text/tool output and an observed successful terminal can memoize the rejection. Errors, incomplete/empty responses, missing terminals, cancellation and changed serving identity cannot memoize it. The existing thread and diff --git a/structure/transports/responses-combo-boundaries.md b/structure/transports/responses-combo-boundaries.md new file mode 100644 index 00000000000..7f58ddbc273 --- /dev/null +++ b/structure/transports/responses-combo-boundaries.md @@ -0,0 +1,51 @@ +# Combo Streaming Boundaries + +## Combo streaming commit boundary + +An HTTP 200 does not by itself commit a streaming combo child. The combo parent runs the child's +downstream Responses SSE through `src/server/responses/combo-stream-preflight.ts`, which owns one +reader and buffers only until one of these boundaries: + +Native post-header reset recovery shares that boundary, because it is asking the same question +about the same bytes. With `replayReadErrors`, the preflight reports the stage it observed — +`headers-only` before any parsed event, `protocol-prelude` after `response.created`, +`semantic-output` once anything else arrives, including a payload it could not parse — and the +resend gate decides. A `response.created` whose snapshot already carries output items is not a +prelude. + +- a non-control Responses event begins client-visible output or a tool/action item, after which the + target is committed and cross-target replay is forbidden; +- a `response.failed` terminal arrives first, in which case the terminal is converted back through + the ordinary bounded combo-failure classifier and may advance to the next declared target; +- a zero-output `response.incomplete` with no reason, `adapter_eof`, `missing_terminal_event`, or + `upstream_stall_timeout` follows the same failure path; explicit semantic or unknown reasons stay terminal; +- a top-level `error` arrives before output, in which case unknown, rate-limit, and server failures + may advance while errors explicitly classified as non-retryable 4xx remain committed; +- another completed/incomplete terminal or the aggregate preflight byte or retained-chunk cap is reached, + in which case the current target is committed conservatively. + +The buffered bytes are replayed unchanged before the reader continues. Native passthrough and eager +relay identity markers are restored on the wrapped response so Windows/Bun stream paths and deferred +logging retain their existing owners. A failed child keeps its physical attempt receipt and usage, +while the successful child remains the logical request result. + +`src/server/responses/core.ts` holds the admitted turn lease across the whole combo ladder. +Refused child relays can unbind without settling the parent; the accepted stream keeps its binding +after the ladder returns, until completion or cancellation. Buffered/exhausted exits release the +ladder binding in `finally`. `tests/server/server-combo-zero-output-failover.test.ts` covers failover, +accepted stream lifetime, cancellation, and exactly-once attached-lease release. +Plain admission leases without controller binding do not register or unregister a ladder controller. +Reasonless incomplete snapshots containing output items stay committed; an explicit `reason: null` +also stays terminal. `tests/routing/combo-stream-preflight.test.ts` pins these conservative boundaries. +Hop warnings in `src/server/responses/core-combo.ts` include only parsed upstream code/message, +redacted before whitespace flattening and bounded to 200 characters; unparsed bodies stay out. +`tests/responses/responses-combo-exhausted-error.test.ts` pins that log boundary. + +A `runTurn` adapter has the equivalent boundary in `preflightAdapterEvents` (`src/adapters/run-turn-queue.ts`), streaming and non-streaming alike: when its first meaningful event is a tool call the current request did not declare, and no earlier replay-unsafe heartbeat recorded a side effect, `src/server/responses/run-turn-execution.ts` projects the fail-closed undeclared-tool refusal as a pre-commit 502 so the combo can hop with the unchanged catalog. After any output or a replay-unsafe heartbeat the refusal stays with that child. Chat Completions and Anthropic Messages inbound requests do not use this classification. The same heartbeat also decides an ordinary pre-output adapter error or an empty end: after a replay-unsafe heartbeat the child's 502 is marked non-replayable, so the combo stops on it instead of sending the turn to the next target. + +HTTP 410 remains terminal by default. It advances and cools only the exact combo target when the +structured code or message explicitly identifies a model lifecycle event (end-of-life, retired, +deprecated, sunset, decommissioned, or no longer available). An unrelated application-level 410 is +not retried. + +> Decision record: [ADR-0071](../decisions/ADR-0071-combo-streaming-commit-boundary.md) diff --git a/structure/transports/responses-failover.md b/structure/transports/responses-failover.md index ebff7e8c852..d9231a74ca3 100644 --- a/structure/transports/responses-failover.md +++ b/structure/transports/responses-failover.md @@ -173,39 +173,7 @@ fabricated. ## Combo streaming commit boundary -An HTTP 200 does not by itself commit a streaming combo child. The combo parent runs the child's -downstream Responses SSE through `src/server/responses/combo-stream-preflight.ts`, which owns one -reader and buffers only until one of these boundaries: - -Native post-header reset recovery shares that boundary, because it is asking the same question -about the same bytes. With `replayReadErrors`, the preflight reports the stage it observed — -`headers-only` before any parsed event, `protocol-prelude` after `response.created`, -`semantic-output` once anything else arrives, including a payload it could not parse — and the -resend gate decides. A `response.created` whose snapshot already carries output items is not a -prelude. - -- a non-control Responses event begins client-visible output or a tool/action item, after which the - target is committed and cross-target replay is forbidden; -- a `response.failed` terminal arrives first, in which case the terminal is converted back through - the ordinary bounded combo-failure classifier and may advance to the next declared target; -- a top-level `error` arrives before output, in which case unknown, rate-limit, and server failures - may advance while errors explicitly classified as non-retryable 4xx remain committed; -- a completed/incomplete terminal or the aggregate preflight byte or retained-chunk cap is reached, - in which case the current target is committed conservatively. - -The buffered bytes are replayed unchanged before the reader continues. Native passthrough and eager -relay identity markers are restored on the wrapped response so Windows/Bun stream paths and deferred -logging retain their existing owners. A failed child keeps its physical attempt receipt and usage, -while the successful child remains the logical request result. - -A `runTurn` adapter has the equivalent boundary in `preflightAdapterEvents` (`src/adapters/run-turn-queue.ts`), streaming and non-streaming alike: when its first meaningful event is a tool call the current request did not declare, and no earlier replay-unsafe heartbeat recorded a side effect, `src/server/responses/run-turn-execution.ts` projects the fail-closed undeclared-tool refusal as a pre-commit 502 so the combo can hop with the unchanged catalog. After any output or a replay-unsafe heartbeat the refusal stays with that child. Chat Completions and Anthropic Messages inbound requests do not use this classification. The same heartbeat also decides an ordinary pre-output adapter error or an empty end: after a replay-unsafe heartbeat the child's 502 is marked non-replayable, so the combo stops on it instead of sending the turn to the next target. - -HTTP 410 remains terminal by default. It advances and cools only the exact combo target when the -structured code or message explicitly identifies a model lifecycle event (end-of-life, retired, -deprecated, sunset, decommissioned, or no longer available). An unrelated application-level 410 is -not retried. - -> Decision record: [ADR-0071](../decisions/ADR-0071-combo-streaming-commit-boundary.md) +See [Combo streaming boundaries](responses-combo-boundaries.md#combo-streaming-commit-boundary) for preflight, turn-lease and hop-warning ownership. Usage consumers preserve positive incomplete-history metadata and connected CLI usage follows the client-scoped hub contract, both specified in [usage accounting](../dashboard-and-usage.md#usage-accounting): readable totals are not a complete ledger, and local management and account data stay separate. Codex pool settings and their consumers follow the [reset-first ordering contract](../providers/openai-accounts.md#reset-first-account-ordering), including independent-quota fallback and preserved affinity. The shared atomic replacement publisher also identifies explicit Remote Workspace file writes as `remote-workspace`. Remote Workspace uses a separate, explicitly enabled server surface with structural WebSocket callbacks and awaited per-server cleanup; [its contract](../remote-workspace.md) owns that integration and records its isolated owner and support limits. diff --git a/structure/transports/responses-spend.md b/structure/transports/responses-spend.md index 579d5c35722..83b6912bca8 100644 --- a/structure/transports/responses-spend.md +++ b/structure/transports/responses-spend.md @@ -11,6 +11,11 @@ That reservation occupies the shared ceiling immediately; it is not a completed `comboInitialSend` hands only that child's single-use permit to `request-send-budget.ts`. `initialSendAllowance` includes that one prepaid send while `remainingBaseSends` keeps every booking spent. Other pending external bookings never become this child's free headroom. +Generic Combo rebuilds use the same shared allowance even without an opt-in retry or spend +policy. Their reporter settles at physical executor admission after pacing, using the exact +prepaid recovery permit when present; unsent repairs and credential hops remain refundable. +Anthropic pool rotations, including direct requests, reserve an `auth-recovery` hop against the +shared allowance and may use its existing final-recovery reserve; account-count limits still apply. Configured transient totals subtract an owner-local counter updated by physical receipts, adapter observers and retry-helper reports, not shared-ledger deltas. Earlier Combo targets, later bookings by other owners and the child's unsent booking do not consume that total; the shared ceiling and diff --git a/tests/adapters/anthropic/anthropic-combo-account-cooldown.test.ts b/tests/adapters/anthropic/anthropic-combo-account-cooldown.test.ts index 559ade3e2e1..3e9521ee77e 100644 --- a/tests/adapters/anthropic/anthropic-combo-account-cooldown.test.ts +++ b/tests/adapters/anthropic/anthropic-combo-account-cooldown.test.ts @@ -23,6 +23,8 @@ import { getAccountSet, saveCredential, setActiveAccount } from "../../../src/oa import { clearAccountQuotaCache, resetProviderQuotaReconcileStateForTests } from "../../../src/providers/quota"; import { clearResponseStateForTests, flushResponseState } from "../../../src/responses/state"; import { handleResponses } from "../../../src/server/responses"; +import { createRequestExecutionBudget } from "../../../src/lib/request-execution-budget"; +import type { RequestExecutionBudget } from "../../../src/lib/request-execution-budget"; import type { OcxConfig, OcxProviderConfig } from "../../../src/types"; import { acquireOwnedSpendHome } from "../../helpers/owned-spend-home"; import { removeTreeWithRetry } from "../../helpers/remove-tree"; @@ -85,11 +87,11 @@ function configFor(reply: () => Response, poolEnabled = true): OcxConfig { }; } -async function post(config: OcxConfig): Promise { +async function post(config: OcxConfig, sendBudget?: RequestExecutionBudget): Promise { const response = await handleResponses(new Request("http://localhost/v1/responses", { method: "POST", headers: { "content-type": "application/json" }, body: JSON.stringify({ model: "combo/pooled", input: "Answer briefly", stream: false }), - }), config, { model: "", provider: "" }); + }), config, { model: "", provider: "" }, sendBudget ? { sendBudget } : undefined); await response.text(); return response; } @@ -131,7 +133,8 @@ test("a spent pool account does not cool the combo target for the accounts that let refuse = true; const config = configFor(() => refuse ? spent() : answered()); - const refused = await post(config); + const budget = createRequestExecutionBudget(); + const refused = await post(config, budget); expect(refused.status).toBe(429); // Only the accounts this request actually reached are cooled, and the pool -- not the combo // layer -- is what cooled them. @@ -140,6 +143,11 @@ test("a spent pool account does not cool the combo target for the accounts that expect(cooled.length).toBeLessThan(ids.length); // The last refused account must be recorded even though the request spent its retry budget. expect(cooled).toHaveLength(4); + // The reservation IS the charge, so the shared counter must equal the sends the transport + // actually made: four physical sends, four booked -- an unsettled booking or a double charge + // would each break this equality even when the account list still says four. + expect(sentTokens).toHaveLength(4); + expect(budget.used).toBe(sentTokens.length); expect(isComboTargetInCooldown("pooled", TARGET)).toBe(false); refuse = false; @@ -160,6 +168,58 @@ test("a spent pool account does not cool the combo target for the accounts that expect(cooled).not.toContain(ids[servedIndex]!); }, 20_000); +test("a denied hop reservation with base headroom sends nothing more but still records the refusal", async () => { + const ids = await seed(5); + const config = configFor(spent); + // Refuse every deferred (pre-dispatch) booking once two physical sends have left: the second + // hop is denied while baseSendAllowance still has room, the way a shared-total or spend- + // observer refusal denies it. Post-dispatch reports stay recordable (alreadySent). + const budget = createRequestExecutionBudget(undefined, undefined, { + charge: options => options?.deferDispatch === true ? sentTokens.length < 2 : true, + refund: () => {}, + }); + const response = await post(config, budget); + expect(response.status).toBe(429); + // The denied hop produced neither a physical send nor an account switch beyond the recorded + // refusal, and it left no unsettled reservation: the counter equals the sends on the wire. + expect(sentTokens).toHaveLength(2); + expect(budget.used).toBe(sentTokens.length); + const cooled = ids.filter(id => getAnthropicAccountHealthSnapshot(id) !== null); + expect(cooled).toHaveLength(2); + expect(isComboTargetInCooldown("pooled", TARGET)).toBe(false); +}, 20_000); + +test("a rotation-helper throw after hop reservation releases the permit and still propagates", async () => { + await seed(2); + const config = configFor(spent); + // First send goes out with the real pool config; once the refusal lands, the helper's own + // config reads start throwing -- a stand-in for the verdict/store reads that can reject + // inside the rotation helper after the hop was already reserved. + const pool = config.anthropicAccountPool; + Object.defineProperty(config, "anthropicAccountPool", { + configurable: true, + get() { if (sentTokens.length > 0) throw new Error("synthetic pool-config read failure"); return pool; }, + }); + let hopCharges = 0; + let refunds = 0; + const budget = createRequestExecutionBudget(undefined, undefined, { + charge: options => { if (options?.deferDispatch === true && sentTokens.length > 0) hopCharges += 1; return true; }, + refund: () => { refunds += 1; }, + }); + let thrown: unknown; + try { await post(config, budget); } catch (error) { thrown = error; } + // The helper's throw is the request's rejection itself: the policy-fallback seam converts + // only pacing overloads, so nothing synthesizes a 5xx here and post() must not resolve. + expect(thrown).toBeInstanceOf(Error); + expect((thrown as Error).message).toBe("synthetic pool-config read failure"); + // The hop really was reserved before the throw, and its release refunded the charge: + // the counter equals the one physical send that actually left. + expect(sentTokens).toHaveLength(1); + expect(hopCharges).toBe(1); + expect(refunds).toBe(1); + expect(budget.used).toBe(sentTokens.length); +}, 20_000); + test("every pool account spent still cools the combo target", async () => { await seed(2); const config = configFor(spent); diff --git a/tests/responses/responses-anthropic-thinking-signature-recovery.test.ts b/tests/responses/responses-anthropic-thinking-signature-recovery.test.ts index 64c31ff6a95..df11c72c34b 100644 --- a/tests/responses/responses-anthropic-thinking-signature-recovery.test.ts +++ b/tests/responses/responses-anthropic-thinking-signature-recovery.test.ts @@ -10,7 +10,15 @@ import { encodeReasoningEnvelope } from "../../src/responses/reasoning-envelope" import { clearReasoningReplayCacheForTests } from "../../src/responses/reasoning-replay-cache"; import { resetThoughtSignatureReplayForTests } from "../../src/responses/thought-signature-replay"; import * as thoughtSignatureReplay from "../../src/responses/thought-signature-replay"; +import * as pacing from "../../src/providers/request-pacing"; +import { saveCredential, getAccountSet, setActiveAccount } from "../../src/oauth/store"; +import { clearGenericFailoverHealth } from "../../src/oauth/generic-account-failover"; +import { createSpendReservationLedger, DEFAULT_SPEND_RESERVATION_POLICY } from "../../src/lib/spend-reservation-ledger"; +import { createRequestSpendTracker } from "../../src/server/responses/request-spend"; import { handleResponses } from "../../src/server/responses/core"; +import { clearComboSelectionState, clearComboTargetCooldowns } from "../../src/combos"; +import { clearKeyCooldowns } from "../../src/providers/key-failover"; +import { comboExecutionBudgetPolicy } from "../../src/server/responses/core-combo"; import type { RequestLogContext } from "../../src/server/request-log"; import type { OcxConfig } from "../../src/types"; import { acquireOwnedSpendHome } from "../helpers/owned-spend-home"; @@ -31,6 +39,10 @@ beforeEach(() => { clearReasoningReplayCacheForTests(); resetThoughtSignatureReplayForTests(); resetWorkflowBudgetsForTest(); + clearComboSelectionState(); + clearComboTargetCooldowns(); + clearKeyCooldowns(); + clearGenericFailoverHealth(); }); afterEach(() => { @@ -40,6 +52,10 @@ afterEach(() => { clearReasoningReplayCacheForTests(); resetThoughtSignatureReplayForTests(); resetWorkflowBudgetsForTest(); + clearComboSelectionState(); + clearComboTargetCooldowns(); + clearKeyCooldowns(); + clearGenericFailoverHealth(); if (originalHome === undefined) delete process.env.OPENCODEX_HOME; else process.env.OPENCODEX_HOME = originalHome; removeTreeWithRetry(testDir); @@ -143,6 +159,160 @@ function expectStrippedRetry(bodies: string[]): void { expect(bodies[1]).toContain("next question"); } +async function comboRequest(count: number, stream = false) { + const cfg = config(); + const template = cfg.providers["anthropic-test"]!; + const targets = Array.from({ length: count }, (_, index) => ({ provider: `target-${index}`, model: "claude-sonnet-4-5" })); + cfg.providers = Object.fromEntries(targets.map(({ provider }) => [provider, { + ...template, baseUrl: `https://${provider}.example.test`, apiKey: `synthetic-${provider}`, + }])); + cfg.combos = { replay: { strategy: "failover", targets } }; + const body = await request(stream).json() as Record; + body.model = "combo/replay"; + return { cfg, body, req: () => new Request("http://localhost/v1/responses", { + method: "POST", headers: { "content-type": "application/json" }, body: JSON.stringify(body), + }) }; +} + +for (const streamError of [false, true]) test(`Combo signature repairs count every physical send (streamError=${streamError})`, async () => { + const { cfg, req } = await comboRequest(4, streamError); + const budget = createRequestExecutionBudget(); + const sends: Array<{ host: string; body: string }> = []; + globalThis.fetch = (async (input: RequestInfo | URL, init?: RequestInit) => { + const host = new URL(String(input)).host; + const serialized = String(init?.body); + sends.push({ host, body: serialized }); + return serialized.includes(SIGNATURE) ? streamError ? streamRejection() : rejection() + : rejection(500, "upstream unavailable after repair"); + }) as typeof fetch; + const log: RequestLogContext = { model: "", provider: "" }; + const response = await handleResponses(req(), cfg, log, { sendBudget: budget }); + await response.text(); + const limit = comboExecutionBudgetPolicy(4).maxTotalModelSends; + // A streamed repair failure is terminal in the committed child; buffered failures hop. + expect(sends.length).toBe(streamError ? 2 : limit); + expect(sends.length).toBeLessThanOrEqual(limit); + expect(budget.used).toBe(sends.length); + expect((log.attempts ?? []).reduce((count, attempt) => count + attempt.sendCount, 0)).toBe(sends.length); + expect(sends.filter(send => send.host === "target-3.example.test")).toHaveLength(streamError ? 0 : 1); + expect(sends.every(send => send.body.includes("earlier answer") && send.body.includes("next question"))).toBe(true); +}); + +test("Combo prepaid fast repair settles once before signature recovery", async () => { + const { cfg, req } = await comboRequest(1); + cfg.fastMode = true; + cfg.providers["target-0"]!.fastEnabled = true; + cfg.providers["target-0"]!.supportsServiceTier = true; + cfg.providers["target-0"]!.fastWire = { kind: "anthropic-speed", canonicalToWire: { priority: "fast" } }; + cfg.providers["target-0"]!.models = ["claude-opus-5-5"]; + cfg.combos!.replay!.targets[0]!.model = "claude-opus-5-5"; + const bodies = fakeUpstream(index => index === 1 + ? Response.json({ type: "error", error: { type: "rate_limit_error", message: "Usage credits are required for fast mode." } }, { status: 429 }) + : index === 2 ? rejection() : success(false)); + const budget = createRequestExecutionBudget(); + const response = await handleResponses(req(), cfg, { model: "", provider: "" }, { sendBudget: budget }); + expect(response.status, await response.clone().text()).toBe(200); + expect(await response.text()).toContain("Recovered answer"); + expect(bodies).toHaveLength(3); + expect(budget.used).toBe(3); + expect(JSON.parse(bodies[0]!).speed).toBe("fast"); + expect(JSON.parse(bodies[1]!).speed).toBeUndefined(); + expect(bodies[1]).toContain(SIGNATURE); + expect(bodies[2]).not.toContain(SIGNATURE); +}); + +for (const prepaid of [false, true]) test(`Combo repair refused before dispatch releases its send (prepaid=${prepaid})`, async () => { + const { cfg, req } = await comboRequest(1); + if (prepaid) { + cfg.fastMode = true; + Object.assign(cfg.providers["target-0"]!, { fastEnabled: true, supportsServiceTier: true, + fastWire: { kind: "anthropic-speed", canonicalToWire: { priority: "fast" } } }); + } + const bodies = fakeUpstream(() => prepaid + ? rejection(400, "Fast mode is not enabled for your organization.") : rejection()); + const wait = pacing.waitForProviderRequestSlot; + let admissions = 0; + const spy = spyOn(pacing, "waitForProviderRequestSlot").mockImplementation(async (...args) => { + if (++admissions > 1) throw new Error("synthetic local pacing refusal"); + return wait(...args); + }); + const budget = createRequestExecutionBudget(); + const log: RequestLogContext = { model: "", provider: "" }; + try { + const response = await handleResponses(req(), cfg, log, { sendBudget: budget }); + await response.text(); + expect(admissions).toBe(2); + expect(bodies).toHaveLength(1); + expect(budget.used).toBe(1); + expect(log.attempts?.[0]?.sendCount).toBe(1); + if (prepaid) expect(JSON.parse(bodies[0]!).speed).toBe("fast"); + } finally { spy.mockRestore(); } +}); + +for (const refused of [false, true]) test(`Combo OAuth hop settles its exact prepaid send (refused=${refused})`, async () => { + for (const id of ["first", "second"]) await saveCredential("xai", { + access: `synthetic-${id}`, refresh: `synthetic-refresh-${id}`, + expires: Date.now() + 3_600_000, accountId: id, + }, { addAccount: true }); + await setActiveAccount("xai", getAccountSet("xai")!.accounts[0]!.id); + const cfg = { defaultProvider: "xai", providers: { xai: { + adapter: "openai-chat", baseUrl: "https://api.x.ai/v1", authMode: "oauth", models: ["test"], proxy: "direct", + } }, combos: { replay: { strategy: "failover", targets: [{ provider: "xai", model: "test" }] } } } as OcxConfig; + const bodies = fakeUpstream(index => index === 1 + ? Response.json({ error: { message: "quota exhausted" } }, { status: 429 }) + : Response.json({ id: "chat-test", choices: [{ index: 0, message: { role: "assistant", content: "ok" }, finish_reason: "stop" }] })); + const budget = createRequestExecutionBudget(); + const log: RequestLogContext = { model: "", provider: "" }; + const wait = pacing.waitForProviderRequestSlot; + let admissions = 0; + const spy = spyOn(pacing, "waitForProviderRequestSlot").mockImplementation(async (...args) => { + if (++admissions > 1 && refused) throw new Error("synthetic local pacing refusal"); + return wait(...args); + }); + try { + const response = await handleResponses(new Request("http://localhost/v1/responses", { + method: "POST", headers: { "content-type": "application/json" }, + body: JSON.stringify({ model: "combo/replay", input: "hello", stream: false, store: false }), + }), cfg, log, { sendBudget: budget }); + await response.text(); + expect(response.status).toBe(refused ? 502 : 200); + expect(bodies).toHaveLength(refused ? 1 : 2); + expect(budget.used).toBe(bodies.length); + expect((log.attempts ?? []).reduce((count, attempt) => count + attempt.sendCount, 0)).toBe(bodies.length); + } finally { spy.mockRestore(); } +}); + +for (const mode of ["spend", "transient", "reset"] as const) test(`Combo signature repair preserves ${mode} accounting`, async () => { + const { cfg, req } = await comboRequest(1); + if (mode === "transient") cfg.providers["target-0"]!.transientRetryOn5xx = { enabled: true, attempts: 3 }; + if (mode === "reset") cfg.providers["target-0"]!.retryOnReset = {}; + const log: RequestLogContext = { model: "", provider: "", spendPoolId: "target-0", usageLogInputTokens: 10, spendOutputCeilingTokens: 100 }; + const lines: string[] = []; + const ledger = createSpendReservationLedger({ salt: "combo-repair-spend", policy: { + ...DEFAULT_SPEND_RESERVATION_POLICY, pool: { maxTokens: 1_000_000 }, + }, journal: { read: () => [], append: line => { lines.push(line); } } }); + const tracker = mode === "spend" ? createRequestSpendTracker(log, undefined, ledger) : undefined; + const budget = createRequestExecutionBudget(undefined, undefined, tracker); + const bodies = fakeUpstream(index => { + if (index === 1) return rejection(); + if (mode === "reset" && index === 2) throw Object.assign(new Error("synthetic pre-header reset"), { code: "ECONNRESET" }); + if (mode === "transient" && index === 2) return rejection(503, "upstream unavailable"); + return success(false); + }); + try { + const response = await handleResponses(req(), cfg, log, { sendBudget: budget }); + // Anthropic does not opt into the key-auth Chat transient ladder. + expect(response.status, await response.clone().text()).toBe(mode === "transient" ? 503 : 200); + await response.text(); + expect(bodies).toHaveLength(mode === "reset" ? 3 : 2); + expect(budget.used).toBe(bodies.length); + expect((log.attempts ?? []).reduce((count, attempt) => count + attempt.sendCount, 0)).toBe(bodies.length); + expect(bodies[0]).toContain(SIGNATURE); + expect(bodies.slice(1).every(body => !body.includes(SIGNATURE))).toBe(true); + if (mode === "spend") expect(budget.physicalStarted).toBe(bodies.length); + } finally { tracker?.settle(undefined); } +}); + function streamedFailure(text: string): { error: { type: string; code: string; message: string }; retryable?: boolean; diff --git a/tests/responses/responses-combo-exhausted-error.test.ts b/tests/responses/responses-combo-exhausted-error.test.ts index c0c1410004c..23915b9f602 100644 --- a/tests/responses/responses-combo-exhausted-error.test.ts +++ b/tests/responses/responses-combo-exhausted-error.test.ts @@ -1,4 +1,4 @@ -import { afterEach, beforeEach, describe, expect, test } from "bun:test"; +import { afterEach, beforeEach, describe, expect, spyOn, test } from "bun:test"; import { clearComboSelectionState, clearComboTargetCooldowns, coolComboTarget, snapshotComboQuotaCooldowns } from "../../src/combos"; import { clearKeyCooldowns } from "../../src/providers/key-failover"; import { clearCachedProviderQuotas, setCachedProviderQuotaForTests } from "../../src/providers/quota-routing-cache"; @@ -284,6 +284,52 @@ describe("combo exhaustion reports the primary's quota refusal", () => { expect(logCtx.attempts).toHaveLength(2); }); + test("the hop warning names the upstream refusal on one redacted line", async () => { + const key = "sk-proj-" + "a1B2c3D4e5F6g7H8i9J0k1L2m3N4o5P6"; + globalThis.fetch = (async () => Response.json( + { error: { code: "invalid_api_key", message: `Incorrect API key provided: ${key}.\nRotate it.` } }, + { status: 401 }, + )) as unknown as typeof fetch; + const warn = spyOn(console, "warn").mockImplementation(() => {}); + try { + await loggedCombo(config, { model: "", provider: "" }); + const line = warn.mock.calls.map(call => String(call[0])).find(text => text.includes("primary/model-primary failed with")); + expect(line).toMatch(/failed with 401 after \d+ms: invalid_api_key: Incorrect API key provided: .* Rotate it\.$/); + expect(line).not.toContain(key); + } finally { + warn.mockRestore(); + } + }); + + test("the hop warning redacts credential framing before flattening parsed codes", async () => { + const credential = "synthetic-multipart-credential"; + globalThis.fetch = (async () => Response.json({ error: { + code: `name="authorization"\r\n\r\n${credential}`, + message: "upstream refused", + } }, { status: 502 })) as unknown as typeof fetch; + const warn = spyOn(console, "warn").mockImplementation(() => {}); + try { + await loggedCombo(config, { model: "", provider: "" }); + const line = warn.mock.calls.map(call => String(call[0])).find(text => text.includes("primary/model-primary failed with")); + expect(line).toBeDefined(); + expect(line!.includes(credential)).toBe(false); + expect(line).toContain("[REDACTED]"); + expect(line).not.toMatch(/[\r\n]/); + } finally { warn.mockRestore(); } + }); + + test("the hop warning never logs an unparsed upstream body", async () => { + globalThis.fetch = (async () => new Response("upstream said: session=abc123 private detail", { status: 502 })) as unknown as typeof fetch; + const warn = spyOn(console, "warn").mockImplementation(() => {}); + try { + await loggedCombo(config, { model: "", provider: "" }); + const line = warn.mock.calls.map(call => String(call[0])).find(text => text.includes("primary/model-primary failed with")); + expect(line).toMatch(/failed with 502 after \d+ms$/); + } finally { + warn.mockRestore(); + } + }); + test("a fallback 5xx is still the answer", async () => { upstream(502, "bad gateway"); const response = await handleResponses(request(), config, { model: "", provider: "" } as RequestLogContext); diff --git a/tests/routing/combo-stream-preflight.test.ts b/tests/routing/combo-stream-preflight.test.ts index 29e6d34c3e7..07e6db0d39c 100644 --- a/tests/routing/combo-stream-preflight.test.ts +++ b/tests/routing/combo-stream-preflight.test.ts @@ -116,7 +116,35 @@ describe("combo stream preflight", () => { } }); - test("does not replay semantic incompletes that another provider cannot safely repair", async () => { + test("converts a zero-output incomplete without a stated reason into a retryable failure", async () => { + for (const response of [ + { id: "r1", status: "incomplete" }, + { id: "r1", status: "incomplete", incomplete_details: null }, + ]) { + const result = await preflightComboStreamResponse(sse( + { type: "response.created", response: { id: "r1", status: "in_progress" } }, + { type: "response.incomplete", response }, + ), { model: "m1", provider: "a" }); + expect(result.kind).toBe("failed"); + expect(result.response.status).toBe(502); + } + }); + + for (const item of [ + { type: "message", role: "assistant", content: [{ type: "output_text", text: "already answered" }] }, + { type: "function_call", name: "lookup", call_id: "call-snapshot", arguments: "{}" }, + ]) test(`does not replay a reasonless incomplete carrying terminal snapshot output (${item.type})`, async () => { + const source = sse({ type: "response.incomplete", response: { + id: "r1", status: "incomplete", output: [item], + } }); + const expected = await source.clone().text(); + const result = await preflightComboStreamResponse(source, { model: "m1", provider: "a" }); + expect(result.kind).toBe("accepted"); + expect(await result.response.text()).toBe(expected); + }); + + for (const reason of ["max_output_tokens", "content_filter", "future_semantic_reason", "", null]) + test(`does not replay semantic or explicit unknown incomplete reasons (${JSON.stringify(reason)})`, async () => { const source = sse( { type: "response.created", response: { id: "r1", status: "in_progress" } }, { @@ -124,7 +152,7 @@ describe("combo stream preflight", () => { response: { id: "r1", status: "incomplete", - incomplete_details: { reason: "max_output_tokens" }, + incomplete_details: { reason }, }, }, ); @@ -134,7 +162,8 @@ describe("combo stream preflight", () => { expect(await result.response.text()).toBe(expected); }); - test("does not replay transport incompletes after output commits the target", async () => { + for (const reason of ["adapter_eof", undefined]) + test(`does not replay incompletes after output commits the target (${reason ?? "missing reason"})`, async () => { const source = sse( { type: "response.created", response: { id: "r1", status: "in_progress" } }, { type: "response.output_text.delta", delta: "visible" }, @@ -143,7 +172,7 @@ describe("combo stream preflight", () => { response: { id: "r1", status: "incomplete", - incomplete_details: { reason: "adapter_eof" }, + incomplete_details: { reason }, }, }, ); diff --git a/tests/server/server-combo-zero-output-failover.test.ts b/tests/server/server-combo-zero-output-failover.test.ts index 72afd27a812..29be15efb12 100644 --- a/tests/server/server-combo-zero-output-failover.test.ts +++ b/tests/server/server-combo-zero-output-failover.test.ts @@ -12,6 +12,7 @@ import { clearComboRecallForTests } from "../../src/server/responses/combo-sessi import { clearKeyCooldowns } from "../../src/providers/key-failover"; import { clearCodexUpstreamHealth } from "../../src/codex/routing"; import { clearRequestLogsForTests, type RequestLogContext } from "../../src/server/request-log"; +import { activeRegistryMetrics, getActiveTurnCount, tryAdmitTurn } from "../../src/server/lifecycle"; import { clearResponseStateForTests, flushResponseState, @@ -220,6 +221,93 @@ describe("combo zero-output bare Responses error failover", () => { }); }); + test("a refused child relay does not settle the parent turn's admission lease", async () => { + // Each passthrough child binds the parent's lease to its own relay controller. Cancelling the + // refused first stream unbinds that controller; it must not settle the turn the next target + // still runs in, or the second child is aborted on arrival ("turn already settled"). + const sse = (...events: unknown[]) => new Response(events.map(event => `data: ${JSON.stringify(event)}\n\n`).join(""), + { headers: { "content-type": "text/event-stream" } }); + const a = serve(() => sse({ type: "response.created", response: { id: "ra", status: "in_progress" } }, { type: "response.failed", + response: { id: "ra", status: "failed", error: { type: "server_error", code: "upstream_server_error", message: "busy" } } })); + const b = serve(() => sse({ type: "response.completed", response: responsesSuccess("lease backup", "m2") })); + const config = comboConfig({ + a: provider("openai-responses", baseUrl(a), "key-a"), + b: provider("openai-responses", baseUrl(b), "key-b"), + }); + const turnAdmissionLease = tryAdmitTurn(); + if (!turnAdmissionLease) throw new Error("test turn admission unavailable"); + try { + const response = await handleResponses(new Request("http://localhost/v1/responses", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ model: "combo/free", input: "hello", stream: true }), + }), config, { model: "", provider: "" }, { turnAdmissionLease }); + expect(response.status).toBe(200); + expect(await response.text()).toContain("lease backup"); + expect(getActiveTurnCount()).toBe(0); + } finally { turnAdmissionLease.release(); } + }); + + test("a plain admission lease does not record an unmatched ladder unbind", async () => { + const a = serve(() => Response.json(responsesSuccess("plain lease", "m1"))); + const config = comboConfig({ a: provider("openai-responses", baseUrl(a), "key-a"), + b: provider("openai-responses", baseUrl(a), "key-b") }); + const misses = activeRegistryMetrics().activeTurns!.releaseMisses; + const response = await handleResponses(new Request("http://localhost/v1/responses", { + method: "POST", headers: { "content-type": "application/json" }, + body: JSON.stringify({ model: "combo/free", input: "hello", stream: false }), + }), config, { model: "", provider: "" }, { turnAdmissionLease: { release() {} } }); + expect(response.status).toBe(200); + await response.text(); + expect(activeRegistryMetrics().activeTurns!.releaseMisses).toBe(misses); + }); + + for (const exit of ["complete", "cancel", "abort"] as const) + test(`accepted fallback keeps its turn lease until stream ${exit}`, async () => { + const encoder = new TextEncoder(); + const frame = (event: unknown) => encoder.encode(`data: ${JSON.stringify(event)}\n\n`); + const a = serve(() => new Response(frame({ type: "response.failed", response: { + id: "ra", status: "failed", error: { type: "server_error", message: "busy" }, + } }), { headers: { "content-type": "text/event-stream" } })); + let finish!: () => void; + const b = serve(() => new Response(new ReadableStream({ start(controller) { + controller.enqueue(frame({ type: "response.output_text.delta", delta: "held fallback" })); + finish = () => { + controller.enqueue(frame({ type: "response.completed", response: responsesSuccess("held fallback", "m2") })); + controller.close(); + }; + } }), { headers: { "content-type": "text/event-stream" } })); + const config = comboConfig({ a: provider("openai-responses", baseUrl(a), "key-a"), + b: provider("openai-responses", baseUrl(b), "key-b") }); + const caller = new AbortController(); + const lease = tryAdmitTurn(undefined, caller.signal); + if (!lease) throw new Error("test turn admission unavailable"); + let releases = 0; + lease.attach({ release() { releases++; } }); + const misses = activeRegistryMetrics().activeTurns!.releaseMisses; + try { + const response = await handleResponses(new Request("http://localhost/v1/responses", { + method: "POST", headers: { "content-type": "application/json" }, signal: caller.signal, + body: JSON.stringify({ model: "combo/free", input: "hello", stream: true }), + }), config, { model: "", provider: "" }, { turnAdmissionLease: lease }); + expect(response.status).toBe(200); + expect(getActiveTurnCount()).toBe(1); + expect(releases).toBe(0); + if (exit === "cancel") await response.body!.cancel(); + else if (exit === "abort") { + caller.abort(); + await response.text().catch(() => ""); + } else { + finish(); + expect(await response.text()).toContain("held fallback"); + } + expect(getActiveTurnCount()).toBe(0); + expect(releases).toBe(1); + expect(activeRegistryMetrics().activeTurns!.releaseMisses).toBe(misses); + } finally { lease.release(); } + expect(releases).toBe(1); + }); + test("undeclared first adapter tool call hops without changing the request catalog", async () => { const requests: Record[] = []; const upstream = (content: string) => serve(async request => {