From df16fcf326439710e2d2d6ffa6c532107853f2c5 Mon Sep 17 00:00:00 2001 From: Epinephrine Date: Sun, 18 Oct 2026 17:24:30 -0700 Subject: [PATCH 1/3] fix(responses): count Combo recovery sends in the shared send budget --- src/server/responses/adapter-dispatch.ts | 27 +-- .../transports/anthropic-thinking-recovery.md | 5 + structure/transports/responses-spend.md | 3 + ...hropic-thinking-signature-recovery.test.ts | 170 ++++++++++++++++++ 4 files changed, 193 insertions(+), 12 deletions(-) diff --git a/src/server/responses/adapter-dispatch.ts b/src/server/responses/adapter-dispatch.ts index 38c48500f16..c34bd45e2c3 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); @@ -1434,9 +1437,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 +1541,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 +1662,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/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-spend.md b/structure/transports/responses-spend.md index 579d5c35722..bf68d1eacce 100644 --- a/structure/transports/responses-spend.md +++ b/structure/transports/responses-spend.md @@ -11,6 +11,9 @@ 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. 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/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; From 89eb1450ee0881db3727290af8ffccdddfdacc12 Mon Sep 17 00:00:00 2001 From: Epinephrine Date: Sun, 18 Oct 2026 18:00:07 -0700 Subject: [PATCH 2/3] fix(responses): preserve reserved Anthropic pool recovery sends --- src/server/responses/adapter-dispatch.ts | 55 ++++++++++++---- .../anthropic-combo-account-cooldown.test.ts | 63 ++++++++++++++++++- 2 files changed, 104 insertions(+), 14 deletions(-) diff --git a/src/server/responses/adapter-dispatch.ts b/src/server/responses/adapter-dispatch.ts index c34bd45e2c3..57fd020dfc1 100644 --- a/src/server/responses/adapter-dispatch.ts +++ b/src/server/responses/adapter-dispatch.ts @@ -1364,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"); @@ -1392,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; } } diff --git a/tests/adapters/anthropic/anthropic-combo-account-cooldown.test.ts b/tests/adapters/anthropic/anthropic-combo-account-cooldown.test.ts index 559ade3e2e1..126fce2d7c7 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,55 @@ 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 outcome: Response | undefined; + try { outcome = await post(config, budget); } catch { /* a helper throw may surface as a rejection */ } + // 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).toBeGreaterThanOrEqual(1); + expect(budget.used).toBe(sentTokens.length); + if (outcome) expect(outcome.status).toBeGreaterThanOrEqual(500); +}, 20_000); + test("every pool account spent still cools the combo target", async () => { await seed(2); const config = configFor(spent); From 1c8003a35a453de053115c09c4fde3e279b80975 Mon Sep 17 00:00:00 2001 From: Epinephrine Date: Sun, 18 Oct 2026 18:25:02 -0700 Subject: [PATCH 3/3] test(adapters): assert exact combo cooldown rejection and single refund --- .../anthropic-combo-account-cooldown.test.ts | 11 +++++++---- 1 file changed, 7 insertions(+), 4 deletions(-) diff --git a/tests/adapters/anthropic/anthropic-combo-account-cooldown.test.ts b/tests/adapters/anthropic/anthropic-combo-account-cooldown.test.ts index 126fce2d7c7..3e9521ee77e 100644 --- a/tests/adapters/anthropic/anthropic-combo-account-cooldown.test.ts +++ b/tests/adapters/anthropic/anthropic-combo-account-cooldown.test.ts @@ -206,15 +206,18 @@ test("a rotation-helper throw after hop reservation releases the permit and stil charge: options => { if (options?.deferDispatch === true && sentTokens.length > 0) hopCharges += 1; return true; }, refund: () => { refunds += 1; }, }); - let outcome: Response | undefined; - try { outcome = await post(config, budget); } catch { /* a helper throw may surface as a rejection */ } + 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).toBeGreaterThanOrEqual(1); + expect(refunds).toBe(1); expect(budget.used).toBe(sentTokens.length); - if (outcome) expect(outcome.status).toBeGreaterThanOrEqual(500); }, 20_000); test("every pool account spent still cools the combo target", async () => {