From 353a08721e5f9d399b522262ba851f455e9b8a3d Mon Sep 17 00:00:00 2001 From: 001005HS <99410048+001005HS@users.noreply.github.com> Date: Fri, 9 Oct 2026 21:49:53 +0900 Subject: [PATCH 1/3] feat(combos): metered targets and Anthropic quota recheck for subscription-first Claude chains One model id can now fail over across billing paths: an Anthropic OAuth subscription pool, Cursor's copy of the same model, then a pay-per-use Anthropic API key. - combo target `metered` + apiKeys[].allowMeteredComboTargets: issued keys never reach a pay-per-use target unless opted in; env token and loopback callers may. - anthropicAccountPool.quotaRecheckMs: caps a 429 account cooldown so a spent weekly window is re-offered instead of waiting days, and a recovered subscription serves again. - log a [combo] line naming the serving target index when a later target served. --- docs-site/src/content/docs/guides/combos.md | 1 + .../docs/reference/configuration/providers.md | 1 + .../docs/reference/configuration/routing.md | 2 +- scripts/test-layout/layout.json | 1 + src/cli/combo-input.ts | 5 +- src/combos/types.ts | 9 + src/config/schema/anthropic-account-pool.ts | 4 + src/config/schema/leaf-validators.ts | 2 + src/oauth/anthropic-routing.ts | 10 +- src/server/admission-model-scope.ts | 17 ++ src/server/management/combo-routes.ts | 13 +- src/server/responses/core-combo.ts | 14 +- src/types/anthropic-account-pool.ts | 7 + src/types/config.ts | 11 + .../anthropic-subscription-chain.test.ts | 283 ++++++++++++++++++ tests/codex-integration/combos.test.ts | 2 +- tests/fixtures/test-layout-expected.json | 2 +- 17 files changed, 374 insertions(+), 10 deletions(-) create mode 100644 tests/adapters/anthropic/anthropic-subscription-chain.test.ts diff --git a/docs-site/src/content/docs/guides/combos.md b/docs-site/src/content/docs/guides/combos.md index 60e3db08e6..bf1e99d03f 100644 --- a/docs-site/src/content/docs/guides/combos.md +++ b/docs-site/src/content/docs/guides/combos.md @@ -816,6 +816,7 @@ Combos are stored in the top-level `combos` object, keyed by combo id: | `targets` | Yes | — | Non-empty ordered array of configured `{ provider, model, weight?, lastResort? }` targets. Duplicate provider/model pairs are rejected. | | `targets[].weight` | No | `1` | Integer from 1 to 10,000. Used by round-robin and random; ignored by failover, least-used, reset-window, and JEV. | | `targets[].lastResort` | No | `false` | Marks an emergency-only target. Inert unless `cooldownWaitPolicy` is set. Never makes a target permanently ineligible: when no normal target can be reached it is dispatched as usual. | +| `targets[].metered` | No | `false` | Marks a pay-per-use target, such as an Anthropic API-key row placed after subscription targets. The environment token and loopback callers may reach it; an issued `apiKeys` entry may not unless it sets `allowMeteredComboTargets: true`, so its request stops at the last unmetered target and returns that target's error. | | `strategy` | No | `"failover"` | `"failover"`, `"round-robin"`, `"random"`, `"least-used"`, `"reset-window"`, or `"jev"`. JEV decides only the initial eligible target and effort; ordinary Combo fallback owns later attempts. | | `stickyLimit` | No | `1` | Integer from 1 to 100 successful requests per round-robin selection. Applies only to round-robin. | | `cooldownMs` | No | unset → upstream fallback (5 s for request-rate 429 codes `1302`/`1305`, 10 min for a spent usage window or a credential/billing failure, otherwise 60 s) | Integer from 1 to 600000. When set, applies as the per-target cooldown whenever no usable upstream `Retry-After` or Codex reset signal exists, including request-rate 429s; when unset, uses the upstream fallback. | diff --git a/docs-site/src/content/docs/reference/configuration/providers.md b/docs-site/src/content/docs/reference/configuration/providers.md index 3fa6e9d24d..8ff47c47b4 100644 --- a/docs-site/src/content/docs/reference/configuration/providers.md +++ b/docs-site/src/content/docs/reference/configuration/providers.md @@ -917,6 +917,7 @@ for the subscription and client conditions the pool is meant for. | --- | --- | --- | --- | | `anthropicAccountPool.enabled?` | `boolean` | `false` | Enable sticky session affinity and quota-ranked new-session selection. **429 failover is not gated here**: it activates whenever two or more usable accounts are stored, exactly like every other multi-credential provider, and cannot be switched off. | | `anthropicAccountPool.nativeMessages?` | `boolean` | `true` | Prefer native Claude Messages for eligible direct Anthropic routes when pooling is enabled. Explicit native rollout opt-outs still apply. `false` selects the legacy bridge for pooled requests; pool-off retains explicit rollout behavior. | +| `anthropicAccountPool.quotaRecheckMs?` | `number` | unset | Upper bound, in milliseconds (60000–86400000), on how long a 429-refused account stays cooled. A spent weekly window can state a reset days away; with this set the account is offered again after at most this long, answers 429 again if still spent, and serves at once if it recovered early. Applies to 429 refusals only; account 403 backoff is unchanged. | | `anthropicAccountPool.autoSwitchThreshold?` | `number` | `80` | For new sessions, when the active account reaches this threshold, choose the lowest known cached usage in the configured window; the account chosen does not itself have to be at or above the threshold. `0` disables **proactive** usage-based switching only — new-session selection and routing recovery after an eligible 429 still consult `quotaWindow`. | | `anthropicAccountPool.strategy?` | `"quota" \| "round-robin" \| "fill-first"` | `"quota"` | New-session strategy; `quota` ranks accounts by the window set by `quotaWindow`, and `fill-first` evaluates its drain threshold in that same window. | | `anthropicAccountPool.quotaWindow?` | `"five-hour" \| "weekly" \| "max-utilization"` | `"five-hour"` | The cached provider-reported utilization bar used for usage-aware account selection. `five-hour` keeps the original behavior. `weekly` scores the weekly bar and skips accounts whose 5-hour bar is exhausted while another eligible account remains, but falls back to exhausted candidates when none do. `max-utilization` scores the highest known bar, so it can use 5-hour usage before weekly usage is available; if neither is known, the account follows unknown-usage ordering. Known usage ranks before unknown usage under the opt-in `weekly` and `max-utilization` windows only; an omitted or explicit `five-hour` preserves the legacy ordering. If every eligible account is unknown, selection still returns one in eligible order. After the documented lower-5-hour tie-break, exact ties preserve eligible order. A healthy affinity-bound session is not proactively rebalanced. For new-session assignment and routing recovery after an eligible 429 replacement, `quota` ranks eligible candidates directly with this window; `fill-first` advances in stable order using this window's threshold and exhaustion rules; `round-robin` ignores it. Cooldown, failover limits, and reauthentication eligibility remain separate local state. Per-account weekly bars come from usage probes or observed response headers. | diff --git a/docs-site/src/content/docs/reference/configuration/routing.md b/docs-site/src/content/docs/reference/configuration/routing.md index e0d0535b85..761db21026 100644 --- a/docs-site/src/content/docs/reference/configuration/routing.md +++ b/docs-site/src/content/docs/reference/configuration/routing.md @@ -138,7 +138,7 @@ namespace, and cannot use reserved bare native families such as `gpt-*`, `o1-*`, | Key | Type | Default | Meaning | | --- | --- | --- | --- | -| `targets` | `{ provider: string; model: string; weight?: number; lastResort?: boolean }[]` | required | Ordered concrete routes. `weight` is 1–10000 and defaults to `1`. `lastResort` marks an emergency-only target; see `cooldownWaitPolicy`. | +| `targets` | `{ provider: string; model: string; weight?: number; lastResort?: boolean; metered?: boolean }[]` | required | Ordered concrete routes. `weight` is 1–10000 and defaults to `1`. `lastResort` marks an emergency-only target; see `cooldownWaitPolicy`. `metered` marks a pay-per-use target that issued API keys reach only with `allowMeteredComboTargets: true`; the environment token and loopback callers are not restricted. | | `strategy?` | `"failover" \| "round-robin" \| "random" \| "least-used" \| "reset-window" \| "jev"` | `"failover"` | Selection strategy. Target order is failover priority; weights shape round-robin and random draws; least-used follows recorded successes; reset-window follows the soonest quota reset; JEV makes one bounded decision for the initial eligible target and effort, then uses ordinary ordered fallback. | | `stickyLimit?` | `number` | `1` | Successful requests retained in one round-robin batch. Range 1–100. Applies only to round-robin. | | `cooldownMs?` | `number` | unset → upstream fallback (5 s for request-rate 429 codes `1302`/`1305`, otherwise 60 s) | Range 1–600000. When set, applies whenever no usable upstream `Retry-After` or Codex reset signal exists, including request-rate 429s; when unset, uses the upstream fallback. Upstream signals take precedence. An explicit upstream `Retry-After` is capped at 24 hours; reset-derived, configured, and fallback cooldowns are capped at 10 minutes. | diff --git a/scripts/test-layout/layout.json b/scripts/test-layout/layout.json index 0b7e20b7a1..5ff8307eed 100644 --- a/scripts/test-layout/layout.json +++ b/scripts/test-layout/layout.json @@ -159,6 +159,7 @@ "anthropic-account-pause.test.ts": "adapters/anthropic", "anthropic-account-threshold.test.ts": "adapters/anthropic", "anthropic-combo-account-cooldown.test.ts": "adapters/anthropic", + "anthropic-subscription-chain.test.ts": "adapters/anthropic", "cli-alias-json.test.ts": "cli", "cli-anthropic-account-threshold.test.ts": "cli", "anthropic-model-routes.test.ts": "adapters/anthropic", diff --git a/src/cli/combo-input.ts b/src/cli/combo-input.ts index e45f616a2a..2ef107645f 100644 --- a/src/cli/combo-input.ts +++ b/src/cli/combo-input.ts @@ -9,7 +9,7 @@ type ComboInput = { reasoningEffortMode?: "strict" | "adaptive"; nativeAlias?: boolean; }; -const TARGET_KEYS = new Set(["provider", "model", "weight", "reasoningEfforts", "modelProfile", "lastResort"]); +const TARGET_KEYS = new Set(["provider", "model", "weight", "reasoningEfforts", "modelProfile", "lastResort", "metered"]); export function parseComboTargets(value: unknown): Array> { if (!Array.isArray(value) || value.length === 0) throw new CliUsageError("Targets input must be a nonempty array"); @@ -41,6 +41,9 @@ export function parseComboTargets(value: unknown): Array if (Object.hasOwn(target, "lastResort") && typeof target.lastResort !== "boolean") { throw new CliUsageError("Target lastResort must be a boolean"); } + if (Object.hasOwn(target, "metered") && typeof target.metered !== "boolean") { + throw new CliUsageError("Target metered must be a boolean"); + } // Preserve raw slashes, target order, explicit false and omitted metadata. return { ...target }; }); diff --git a/src/combos/types.ts b/src/combos/types.ts index 6353db3b7a..d11a27baf9 100644 --- a/src/combos/types.ts +++ b/src/combos/types.ts @@ -39,6 +39,8 @@ export interface NormalizedComboTarget { weight: number; /** Emergency-only target, deferred under `cooldownWaitPolicy` (#5691). */ lastResort: boolean; + /** Pay-per-use target; issued keys reach it only with `allowMeteredComboTargets`. */ + metered: boolean; reasoningEfforts?: OcxComboDefaultEffort[]; /** Optional JEV decision description. */ modelProfile?: string; @@ -427,6 +429,12 @@ export function comboConfigIssues( message: `targets[${i}].lastResort must be a boolean`, }); } + if (target.metered !== undefined && typeof target.metered !== "boolean") { + issues.push({ + path: ["targets", i, "metered"], + message: `targets[${i}].metered must be a boolean`, + }); + } if (target.modelProfile !== undefined && (typeof target.modelProfile !== "string" || target.modelProfile.trim().length === 0 @@ -503,6 +511,7 @@ export function normalizeComboConfig(raw: OcxComboConfig): NormalizedComboConfig ? { modelProfile: target.modelProfile.trim() } : {}), lastResort: target.lastResort === true, + metered: target.metered === true, })), }; } diff --git a/src/config/schema/anthropic-account-pool.ts b/src/config/schema/anthropic-account-pool.ts index 6faac7c98c..2c9ca0669f 100644 --- a/src/config/schema/anthropic-account-pool.ts +++ b/src/config/schema/anthropic-account-pool.ts @@ -25,6 +25,10 @@ function poolError(value: unknown, path: string): string | undefined { const parsed = parseAnthropicModelRoutes(pool.routes); if (!parsed.ok) return `schema_invalid: ${path}.routes: ${parsed.error}`; } + if (pool.quotaRecheckMs !== undefined && (typeof pool.quotaRecheckMs !== "number" + || !Number.isInteger(pool.quotaRecheckMs) || pool.quotaRecheckMs < 60_000 || pool.quotaRecheckMs > 86_400_000)) { + return `schema_invalid: ${path}.quotaRecheckMs: must be an integer from 60000 to 86400000`; + } return undefined; } diff --git a/src/config/schema/leaf-validators.ts b/src/config/schema/leaf-validators.ts index d9e1d07226..70adf399ef 100644 --- a/src/config/schema/leaf-validators.ts +++ b/src/config/schema/leaf-validators.ts @@ -793,6 +793,8 @@ export const apiKeyEntrySchema = z.object({ // rather than promoting it. allowedProviders: z.array(z.string().trim().min(1).max(256)).optional(), allowedModels: z.array(z.string().trim().min(1).max(256)).optional(), + // Not `.catch`ed for the same reason: a damaged value must not grant metered spend. + allowMeteredComboTargets: z.boolean().optional(), }).passthrough(); /** diff --git a/src/oauth/anthropic-routing.ts b/src/oauth/anthropic-routing.ts index 5ee71f197d..3b029d644d 100644 --- a/src/oauth/anthropic-routing.ts +++ b/src/oauth/anthropic-routing.ts @@ -338,6 +338,12 @@ function createAnthropicRouting(instance: AnthropicInstanceId): AnthropicRouting return anthropicAccountPoolConfig(config).enabled === true; } + function anthropicQuotaRecheckMs(config: OcxConfig): number | undefined { + const value = anthropicAccountPoolConfig(config).quotaRecheckMs; + return typeof value === "number" && Number.isInteger(value) && value >= 60_000 && value <= 86_400_000 + ? value : undefined; + } + function anthropicAutoSwitchThreshold(config: OcxConfig): number { const value = anthropicAccountPoolConfig(config).autoSwitchThreshold; if (typeof value === "number" && Number.isInteger(value) && value >= 0 && value <= 100) return value; @@ -1084,7 +1090,9 @@ function createAnthropicRouting(instance: AnthropicInstanceId): AnthropicRouting // account is back in the rotation a minute later. const parsedRetry = parseRetryAfterMs(retryAfterHeader, now); const resetDerived = status === 429 && parsedRetry === undefined ? parseRateLimitReset(rateLimitHeaders, now) : undefined; - const cooldownMs = parsedRetry ?? resetDerived?.delayMs ?? (status === 403 ? ACCOUNT_REFUSAL_COOLDOWN_MS : DEFAULT_COOLDOWN_MS); + const statedCooldownMs = parsedRetry ?? resetDerived?.delayMs ?? (status === 403 ? ACCOUNT_REFUSAL_COOLDOWN_MS : DEFAULT_COOLDOWN_MS); + const recheckMs = anthropicQuotaRecheckMs(config); + const cooldownMs = status === 429 && recheckMs !== undefined ? Math.min(statedCooldownMs, recheckMs) : statedCooldownMs; const cooldownGeneration = noteAnthropicCooldownMutation(failedAccountId); upstreamHealth.set(failedAccountId, { cooldownUntil: now + cooldownMs, diff --git a/src/server/admission-model-scope.ts b/src/server/admission-model-scope.ts index 9390d2c3f3..98e2c2a06b 100644 --- a/src/server/admission-model-scope.ts +++ b/src/server/admission-model-scope.ts @@ -76,6 +76,23 @@ export function resolveAdmissionModelScope( return entry ? admissionModelScopeOf(entry) : undefined; } +/** + * May this request reach a combo target marked `metered` (pay-per-use billing)? + * + * The environment token and loopback admission are the operator's own clients, so they may. + * An issued key may not unless its entry opts in with `allowMeteredComboTargets`: a key with + * `allowedModels` naming a subscription model must not start spending pay-per-use credit the + * moment that subscription is exhausted, which is what a failover combo would otherwise do. + */ +export function admissionMayUseMeteredComboTargets( + config: Pick, + admission: DataPlaneAdmission | undefined, +): boolean { + if (!admission || admission.kind !== "configured") return true; + const entry = (config.apiKeys ?? []).find(key => key.id === admission.keyId); + return entry?.allowMeteredComboTargets === true; +} + /** * Does this scope admit this resolved destination? * diff --git a/src/server/management/combo-routes.ts b/src/server/management/combo-routes.ts index a1f64030cc..1699f43722 100644 --- a/src/server/management/combo-routes.ts +++ b/src/server/management/combo-routes.ts @@ -203,7 +203,7 @@ export async function handleComboRoutes(ctx: ManagementContext): Promise candidate.provider === provider && candidate.model === model, ); - return before?.lastResort ? { ...target, lastResort: true } : target; + // `metered` is carried the same way: dropping it on a GUI round-trip would open a + // pay-per-use target to every issued key. + return { + ...target, + ...(!Object.hasOwn(target, "lastResort") && before?.lastResort ? { lastResort: true } : {}), + ...(!Object.hasOwn(target, "metered") && before?.metered ? { metered: true } : {}), + }; }), } : {}), @@ -295,9 +301,10 @@ export async function handleComboRoutes(ctx: ManagementContext): Promise ({ + targets: normalizedBase.targets.map(({ lastResort, metered, modelProfile, ...target }) => ({ ...target, ...(lastResort ? { lastResort: true } : {}), + ...(metered ? { metered: true } : {}), ...(modelProfile ? { modelProfile } : {}), })), ...(normalizedAlias ? { alias: normalizedAlias } : {}), diff --git a/src/server/responses/core-combo.ts b/src/server/responses/core-combo.ts index 0d1f58163d..ab24192f75 100644 --- a/src/server/responses/core-combo.ts +++ b/src/server/responses/core-combo.ts @@ -4,7 +4,7 @@ import { isDeclaredReasoningEffort, resolveEffortAtOrBelow, } from "../../reasoning-effort"; -import { resolveAdmissionModelScope, routeAllowedByScope } from "../admission-model-scope"; +import { admissionMayUseMeteredComboTargets, resolveAdmissionModelScope, routeAllowedByScope } from "../admission-model-scope"; import { recordAttemptRequestedEffort } from "../request-log"; import { CODEX_TEXT_GUARDED_BUDGET_POLICY, @@ -446,8 +446,10 @@ export async function executeComboResponses( let comboPayloadReadable = false; const payloadEligible = (target: (typeof combo.targets)[number]): boolean => comboPayloadReadable || !unreadableEncryptedAgentTask || canDecryptUnreadableAgentTask(target); + const meteredAllowed = admissionMayUseMeteredComboTargets(config, options.admission); const targetEligible = (target: (typeof combo.targets)[number]): boolean => - (combo.strategy !== "jev" || target.provider !== JEV_PROVIDER_ID) + (meteredAllowed || !target.metered) + && (combo.strategy !== "jev" || target.provider !== JEV_PROVIDER_ID) && payloadEligible(target) && reasoningReplayEligible(target) && (protocolLanes?.pickable(target) ?? true); @@ -521,6 +523,9 @@ export async function executeComboResponses( now?: number; }) => pickComboTargetWithWait(config, comboId, { ...pickOptions, + // Every pick, including the recovery picks that skip `targetEligible`, keeps metered + // targets away from callers that may not spend on them. + eligible: target => (meteredAllowed || !target.metered) && (pickOptions.eligible?.(target) ?? true), waitForCooldownMs: combo.waitForCooldownMs, abortSignal: options.abortSignal, }); @@ -968,6 +973,11 @@ export async function executeComboResponses( ); attemptRetained = true; noteComboSuccess(comboId, combo, pick.target, pick.writerGeneration); + if (pick.targetIndex > 0) { + console.info( + `[combo] ${comboId}: served by target #${pick.targetIndex + 1} ${targetKey(pick.target)}${pick.target.metered ? " (metered)" : ""} after ${pick.attempted.length - 1} earlier target(s)`, + ); + } Object.assign(logCtx, childLog, { requestedModel, model: requestedModel, diff --git a/src/types/anthropic-account-pool.ts b/src/types/anthropic-account-pool.ts index a2e7b8be5c..c46092d267 100644 --- a/src/types/anthropic-account-pool.ts +++ b/src/types/anthropic-account-pool.ts @@ -30,4 +30,11 @@ export interface AnthropicAccountPoolConfig { quotaWindow?: OcxAccountPoolQuotaWindow; /** Ordered model allowlists; inactive while the pool is disabled. Stored account IDs only. */ routes?: AnthropicModelRoute[]; + /** + * Re-offer a quota-exhausted account after at most this many milliseconds instead of waiting + * for the stated reset (which can be days away for a weekly window). A still-spent account + * answers 429 again and is re-cooled; one that recovered early serves at once. Absent keeps + * the stated reset. Range 60000..86400000. + */ + quotaRecheckMs?: number; } diff --git a/src/types/config.ts b/src/types/config.ts index c5afa71c8b..de31616083 100644 --- a/src/types/config.ts +++ b/src/types/config.ts @@ -335,6 +335,12 @@ export interface OcxApiKeyEntry { * that is the only point at which the model about to be billed is known. */ allowedModels?: string[]; + /** + * Lets this key reach combo targets marked `metered`. Absent means no: an issued key stops + * at the last unmetered target, while the environment token and loopback callers are not + * restricted. + */ + allowMeteredComboTargets?: boolean; } export interface OcxPendingApiKeyRotation { @@ -1287,6 +1293,11 @@ export interface OcxComboTarget { * see `OcxComboConfig.cooldownWaitPolicy` (#5691). */ lastResort?: boolean; + /** + * Pay-per-use target (for example an API-key row behind subscription targets). Only the + * environment token, loopback callers and keys with `allowMeteredComboTargets` may reach it. + */ + metered?: boolean; } export interface OcxComboConfig { diff --git a/tests/adapters/anthropic/anthropic-subscription-chain.test.ts b/tests/adapters/anthropic/anthropic-subscription-chain.test.ts new file mode 100644 index 0000000000..1b06834b94 --- /dev/null +++ b/tests/adapters/anthropic/anthropic-subscription-chain.test.ts @@ -0,0 +1,283 @@ +/** + * One model id, three billing paths: an Anthropic OAuth subscription pool, Cursor's copy of the + * same model, then a pay-per-use Anthropic API key. A failover combo whose alias is the model id + * the clients already send gives that order; `metered` keeps issued keys off the API-key target, + * and the pool's `quotaRecheckMs` brings a spent subscription back before its weekly reset. + */ +import { afterEach, beforeEach, expect, mock, test } from "bun:test"; +import { mkdtempSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import type { ProviderAdapter } from "../../../src/adapters/base"; +import type { AdapterEvent, OcxConfig, OcxProviderConfig } from "../../../src/types"; +import type { DataPlaneAdmission } from "../../../src/server/auth-cors"; +import { acquireOwnedSpendHome } from "../../helpers/owned-spend-home"; +import { removeTreeWithRetry } from "../../helpers/remove-tree"; + +const MODEL = "claude-opus-5-5"; +const ISSUED_KEY_ID = "issued-key"; +let cursorLimited = false; +let cursorModels: string[] = []; +let cursorEfforts: unknown[] = []; + +const actualResolver = await import("../../../src/server/adapter-resolve"); +const actualResolveAdapter = actualResolver.resolveAdapter; +mock.module("../../../src/server/adapter-resolve", () => ({ + ...actualResolver, + resolveAdapter(provider: OcxProviderConfig, cacheRetention?: "none" | "short" | "long") { + if (provider.adapter !== "cursor") return actualResolveAdapter(provider, cacheRetention); + return { + name: "cursor", + buildRequest: () => ({ url: provider.baseUrl, method: "POST", headers: {}, body: "" }), + async *parseStream() { yield { type: "error", message: "fixture uses runTurn" } as AdapterEvent; }, + async runTurn(parsed: { modelId: string; options?: { reasoning?: unknown } }, _incoming: unknown, emit: (event: AdapterEvent) => void) { + cursorModels.push(parsed.modelId); + cursorEfforts.push(parsed.options?.reasoning); + if (cursorLimited) { + emit({ type: "error", status: 429, message: "You've hit your usage limit", code: "rate_limit_exceeded" }); + return; + } + emit({ type: "text_delta", text: "cursor answered" }); + emit({ type: "done" }); + }, + } as unknown as ProviderAdapter; + }, +})); + +const { handleChatCompletions } = await import("../../../src/server/chat-completions"); +const { saveCredential, getAccountSet, setActiveAccount } = await import("../../../src/oauth/store"); +const { + clearAnthropicAccountPoolState, forgetAnthropicFailoverQuorum, getAnthropicAccountHealthSnapshot, +} = await import("../../../src/oauth/anthropic-routing"); +const { clearComboSelectionState } = await import("../../../src/combos/resolve"); +const { clearComboTargetCooldowns } = await import("../../../src/combos/failover"); +const { clearGenericFailoverHealth } = await import("../../../src/oauth/generic-account-failover"); +const { clearAccountQuotaCache } = await import("../../../src/providers/quota"); + +type Lane = "subscription" | "api-key"; +type Sent = { lane: Lane; authorization: string | null; apiKey: string | null; body: Record }; +let sent: Sent[] = []; +let subscriptionSpent = false; +let apiKeyLimited = false; +let home = ""; +let releaseSpendHome: (() => void) | undefined; +const originalHome = process.env.OPENCODEX_HOME; + +function sse(): Response { + const event = (name: string, data: unknown) => `event: ${name}\ndata: ${JSON.stringify(data)}\n\n`; + return new Response([ + event("message_start", { type: "message_start", message: { id: "msg_chain", type: "message", role: "assistant", model: MODEL, content: [], stop_reason: null, usage: { input_tokens: 8, output_tokens: 0 } } }), + event("content_block_start", { type: "content_block_start", index: 0, content_block: { type: "text", text: "" } }), + event("content_block_delta", { type: "content_block_delta", index: 0, delta: { type: "text_delta", text: "anthropic answered" } }), + event("content_block_stop", { type: "content_block_stop", index: 0 }), + event("message_delta", { type: "message_delta", delta: { stop_reason: "end_turn" }, usage: { output_tokens: 3 } }), + event("message_stop", { type: "message_stop" }), + ].join(""), { headers: { "content-type": "text/event-stream" } }); +} + +/** The 2026-09-28 shape: a spent weekly window with an 88.5-hour Retry-After. */ +function weeklySpent(): Response { + return Response.json( + { type: "error", error: { type: "rate_limit_error", message: "usage limit reached" } }, + { status: 429, headers: { "anthropic-ratelimit-unified-7d-status": "rejected", "retry-after": "318747" } }, + ); +} + +function transport(lane: Lane): typeof fetch { + return (async (_input: RequestInfo | URL, init?: RequestInit) => { + const headers = new Headers(init?.headers); + sent.push({ + lane, + authorization: headers.get("authorization"), + apiKey: headers.get("x-api-key"), + body: JSON.parse(String(init?.body ?? "{}")) as Record, + }); + const limited = lane === "subscription" ? subscriptionSpent : apiKeyLimited; + return limited ? weeklySpent() : sse(); + }) as typeof fetch; +} + +function config(options: { combo?: boolean; quotaRecheckMs?: number; allowMetered?: boolean } = {}): OcxConfig { + const subscription = { adapter: "anthropic", baseUrl: "https://subscription.test", authMode: "oauth", models: [MODEL], fetch: transport("subscription") }; + const apiKey = { adapter: "anthropic", baseUrl: "https://api-key.test", authMode: "key", apiKey: "${CHAIN_TEST_ANTHROPIC_KEY}", models: [MODEL], fetch: transport("api-key") }; + return { + port: 0, + defaultProvider: "anthropic", + providers: { + anthropic: subscription as OcxProviderConfig, + cursor: { adapter: "cursor", baseUrl: "https://api2.cursor.sh", authMode: "oauth", models: [MODEL] }, + "anthropic-metered": apiKey as OcxProviderConfig, + }, + ...(options.quotaRecheckMs ? { anthropicAccountPool: { quotaRecheckMs: options.quotaRecheckMs } } : {}), + apiKeys: [{ + id: ISSUED_KEY_ID, name: "issued", key: "ocx_data_issued_key_for_chain_test", createdAt: "2026-10-09T00:00:00.000Z", + allowedModels: [MODEL], + ...(options.allowMetered ? { allowMeteredComboTargets: true } : {}), + }], + ...(options.combo === false ? {} : { + combos: { + "claude-chain": { + alias: `anthropic/${MODEL}`, + strategy: "failover", + targets: [ + { provider: "anthropic", model: MODEL }, + { provider: "cursor", model: MODEL }, + { provider: "anthropic-metered", model: MODEL, metered: true }, + ], + }, + }, + }), + } as OcxConfig; +} + +const LOOPBACK: DataPlaneAdmission = { kind: "loopback", source: "loopback" }; +const ISSUED: DataPlaneAdmission = { kind: "configured", keyId: ISSUED_KEY_ID, source: "bearer" }; + +async function chat(cfg: OcxConfig, admission: DataPlaneAdmission = LOOPBACK) { + const response = await handleChatCompletions(new Request("http://localhost/v1/chat/completions", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ + model: `anthropic/${MODEL}`, + stream: false, + reasoning_effort: "medium", + messages: [{ role: "system", content: "S".repeat(5000) }, { role: "user", content: "hi" }], + }), + }), cfg, { model: "", provider: "" }, { requestId: "chain-test", start: Date.now(), admission } as never); + return { status: response.status, text: await response.text() }; +} + +const lanes = () => sent.map(entry => entry.lane); + +beforeEach(async () => { + home = mkdtempSync(join(tmpdir(), "ocx-subscription-chain-")); + process.env.OPENCODEX_HOME = home; + process.env.CHAIN_TEST_ANTHROPIC_KEY = "sk-ant-chain-test"; + releaseSpendHome = acquireOwnedSpendHome(); + for (let index = 0; index < 2; index++) { + await saveCredential("anthropic", { + access: `subscription-access-${index}`, refresh: `subscription-refresh-${index}`, + expires: Date.now() + 3_600_000, accountId: `subscription-account-${index}`, + }); + } + await setActiveAccount("anthropic", getAccountSet("anthropic")!.accounts[0]!.id); + await saveCredential("cursor", { access: "cursor-access", refresh: "cursor-refresh", expires: Date.now() + 3_600_000, accountId: "cursor-account" }); + sent = []; cursorModels = []; cursorEfforts = []; + subscriptionSpent = false; cursorLimited = false; apiKeyLimited = false; + clearComboSelectionState(); clearComboTargetCooldowns(); clearAnthropicAccountPoolState(); + forgetAnthropicFailoverQuorum(); clearGenericFailoverHealth(); clearAccountQuotaCache(); +}); + +afterEach(() => { + releaseSpendHome?.(); + releaseSpendHome = undefined; + clearComboSelectionState(); clearComboTargetCooldowns(); clearAnthropicAccountPoolState(); + forgetAnthropicFailoverQuorum(); clearGenericFailoverHealth(); clearAccountQuotaCache(); + if (originalHome === undefined) delete process.env.OPENCODEX_HOME; + else process.env.OPENCODEX_HOME = originalHome; + delete process.env.CHAIN_TEST_ANTHROPIC_KEY; + removeTreeWithRetry(home); +}); + +test("a healthy subscription serves the aliased id with the same wire body as the plain route", async () => { + expect((await chat(config({ combo: false }))).status).toBe(200); + const plainBody = sent[0]!.body; + sent = []; + clearAnthropicAccountPoolState(); + + const response = await chat(config()); + expect(response.status).toBe(200); + expect(response.text).toContain("anthropic answered"); + expect(lanes()).toEqual(["subscription"]); + expect(sent[0]!.authorization).toStartWith("Bearer subscription-access-"); + expect(JSON.stringify({ ...sent[0]!.body, metadata: undefined })).toBe(JSON.stringify({ ...plainBody, metadata: undefined })); + expect(JSON.stringify(sent[0]!.body)).toContain("cache_control"); +}); + +test("every subscription account spent moves to Cursor's claude-opus-5-5 at the same effort", async () => { + subscriptionSpent = true; + const response = await chat(config()); + expect(response.status).toBe(200); + expect(response.text).toContain("cursor answered"); + expect(cursorModels).toEqual([MODEL]); + expect(cursorEfforts).toEqual(["medium"]); + expect(lanes().every(lane => lane === "subscription")).toBe(true); +}); + +test("subscription and Cursor limited moves a local caller to the API key with x-api-key and cache_control", async () => { + subscriptionSpent = true; + cursorLimited = true; + const response = await chat(config()); + expect(response.status).toBe(200); + const metered = sent.filter(entry => entry.lane === "api-key"); + expect(metered).toHaveLength(1); + expect(metered[0]!.apiKey).toBe("sk-ant-chain-test"); + expect(metered[0]!.authorization).toBeNull(); + expect(JSON.stringify(metered[0]!.body)).toContain("cache_control"); +}); + +test("all three limited returns the limit to the caller so its own fallback takes over", async () => { + subscriptionSpent = true; + cursorLimited = true; + apiKeyLimited = true; + const response = await chat(config()); + expect(response.status).toBe(429); + expect(sent.filter(entry => entry.lane === "api-key")).toHaveLength(1); +}); + +test("an issued key stops before the metered target and gets the error instead", async () => { + subscriptionSpent = true; + cursorLimited = true; + const response = await chat(config(), ISSUED); + expect(response.status).toBeGreaterThanOrEqual(400); + expect(response.text).not.toContain("answered"); + expect(cursorModels).toEqual([MODEL]); + expect(sent.filter(entry => entry.lane === "api-key")).toHaveLength(0); +}); + +test("an issued key that opts in may reach the metered target", async () => { + subscriptionSpent = true; + cursorLimited = true; + const response = await chat(config({ allowMetered: true }), ISSUED); + expect(response.status).toBe(200); + expect(sent.filter(entry => entry.lane === "api-key")).toHaveLength(1); +}); + +test("quotaRecheckMs caps a spent account's weekly cooldown so a recovered subscription is used again", async () => { + subscriptionSpent = true; + const recheckMs = 5 * 60_000; + const before = Date.now(); + expect((await chat(config({ quotaRecheckMs: recheckMs }))).text).toContain("cursor answered"); + const cooled = getAccountSet("anthropic")!.accounts + .map(account => getAnthropicAccountHealthSnapshot(account.id)) + .filter(snapshot => snapshot !== null); + expect(cooled.length).toBeGreaterThan(0); + for (const snapshot of cooled) { + expect(snapshot!.cooldownUntil!).toBeLessThanOrEqual(Date.now() + recheckMs); + expect(snapshot!.cooldownUntil!).toBeGreaterThanOrEqual(before + recheckMs); + } + + subscriptionSpent = false; + sent = []; cursorModels = []; + const later = Date.now() + recheckMs + 1; + const realNow = Date.now; + Date.now = () => later; + try { + const response = await chat(config({ quotaRecheckMs: recheckMs })); + expect(response.text).toContain("anthropic answered"); + } finally { + Date.now = realNow; + } + expect(lanes()).toEqual(["subscription"]); + expect(cursorModels).toEqual([]); +}); + +test("without quotaRecheckMs the stated 88-hour reset still holds the account out", async () => { + subscriptionSpent = true; + await chat(config()); + const snapshots = getAccountSet("anthropic")!.accounts + .map(account => getAnthropicAccountHealthSnapshot(account.id)) + .filter(snapshot => snapshot !== null); + expect(snapshots.length).toBeGreaterThan(0); + for (const snapshot of snapshots) expect(snapshot!.cooldownUntil! - Date.now()).toBeGreaterThan(80 * 3_600_000); +}); diff --git a/tests/codex-integration/combos.test.ts b/tests/codex-integration/combos.test.ts index dd76fa1ce5..d6c370bf14 100644 --- a/tests/codex-integration/combos.test.ts +++ b/tests/codex-integration/combos.test.ts @@ -1639,7 +1639,7 @@ describe("combo validation and normalization", () => { alias: null, nativeAlias: false, displayName: null, - targets: [{ provider: "a", model: "m1", weight: 2, lastResort: false }], + targets: [{ provider: "a", model: "m1", weight: 2, lastResort: false, metered: false }], }); expect(normalizeComboConfig({ targets: [{ provider: "a", model: "m1" }] }).defaultEffort).toBeNull(); const targetReasoningEfforts: OcxComboDefaultEffort[] = ["low", "high"]; diff --git a/tests/fixtures/test-layout-expected.json b/tests/fixtures/test-layout-expected.json index 1560995bd4..36a9afe700 100644 --- a/tests/fixtures/test-layout-expected.json +++ b/tests/fixtures/test-layout-expected.json @@ -137,7 +137,7 @@ "anthropic-account-metadata.test.ts": "adapters/anthropic", "anthropic-account-pool.test.ts": "adapters/anthropic", "anthropic-account-pause-outbound.test.ts": "adapters/anthropic", "anthropic-account-pause.test.ts": "adapters/anthropic", "anthropic-account-threshold.test.ts": "adapters/anthropic", - "anthropic-combo-account-cooldown.test.ts": "adapters/anthropic", "cli-alias-json.test.ts": "cli", + "anthropic-combo-account-cooldown.test.ts": "adapters/anthropic", "anthropic-subscription-chain.test.ts": "adapters/anthropic", "cli-alias-json.test.ts": "cli", "cli-anthropic-account-threshold.test.ts": "cli", "anthropic-model-routes.test.ts": "adapters/anthropic", "anthropic-agentrouter-language-framing.test.ts": "adapters/anthropic", "anthropic-baseurl-override.test.ts": "adapters/anthropic", From 4f8e7e0534aceffd8bcd6c1a53d1728f303c784c Mon Sep 17 00:00:00 2001 From: 001005HS <99410048+001005HS@users.noreply.github.com> Date: Sat, 10 Oct 2026 04:04:32 +0900 Subject: [PATCH 2/3] fix(combos): preserve subscription rechecks and target effort across failover --- docs-site/src/content/docs/guides/combos.md | 2 + src/cli/combo-input.ts | 9 +- src/combos/request.ts | 11 +- src/combos/types.ts | 11 ++ src/server/management/combo-routes.ts | 6 +- src/server/responses/core-combo-native.ts | 5 +- src/server/responses/core-combo.ts | 7 +- src/server/responses/run-turn-execution.ts | 16 ++- src/types/config.ts | 4 + structure/transports/responses-failover.md | 2 +- .../anthropic-subscription-chain.test.ts | 122 +++++++++++++++++- tests/codex-integration/combos.test.ts | 15 +++ 12 files changed, 192 insertions(+), 18 deletions(-) diff --git a/docs-site/src/content/docs/guides/combos.md b/docs-site/src/content/docs/guides/combos.md index bf1e99d03f..21659df793 100644 --- a/docs-site/src/content/docs/guides/combos.md +++ b/docs-site/src/content/docs/guides/combos.md @@ -815,6 +815,8 @@ Combos are stored in the top-level `combos` object, keyed by combo id: | --- | --- | --- | --- | | `targets` | Yes | — | Non-empty ordered array of configured `{ provider, model, weight?, lastResort? }` targets. Duplicate provider/model pairs are rejected. | | `targets[].weight` | No | `1` | Integer from 1 to 10,000. Used by round-robin and random; ignored by failover, least-used, reset-window, and JEV. | +| `targets[].reasoningEffort` | No | unset | Explicit per-target `low`, `medium`, `high`, `xhigh`, `max`, or `ultra` override. Other targets preserve the caller's effort. The operator owns target support; ordinary downstream validation still applies. | +| `targets[].serviceTier` | No | unset | Per-target `default` or `priority`. `priority` requests Fast through the destination's existing capability/wire policy; it does not enable an unsupported Fast lane. | | `targets[].lastResort` | No | `false` | Marks an emergency-only target. Inert unless `cooldownWaitPolicy` is set. Never makes a target permanently ineligible: when no normal target can be reached it is dispatched as usual. | | `targets[].metered` | No | `false` | Marks a pay-per-use target, such as an Anthropic API-key row placed after subscription targets. The environment token and loopback callers may reach it; an issued `apiKeys` entry may not unless it sets `allowMeteredComboTargets: true`, so its request stops at the last unmetered target and returns that target's error. | | `strategy` | No | `"failover"` | `"failover"`, `"round-robin"`, `"random"`, `"least-used"`, `"reset-window"`, or `"jev"`. JEV decides only the initial eligible target and effort; ordinary Combo fallback owns later attempts. | diff --git a/src/cli/combo-input.ts b/src/cli/combo-input.ts index 2ef107645f..4329429034 100644 --- a/src/cli/combo-input.ts +++ b/src/cli/combo-input.ts @@ -9,7 +9,7 @@ type ComboInput = { reasoningEffortMode?: "strict" | "adaptive"; nativeAlias?: boolean; }; -const TARGET_KEYS = new Set(["provider", "model", "weight", "reasoningEfforts", "modelProfile", "lastResort", "metered"]); +const TARGET_KEYS = new Set(["provider", "model", "weight", "reasoningEfforts", "modelProfile", "lastResort", "metered", "reasoningEffort", "serviceTier"]); export function parseComboTargets(value: unknown): Array> { if (!Array.isArray(value) || value.length === 0) throw new CliUsageError("Targets input must be a nonempty array"); @@ -27,6 +27,13 @@ export function parseComboTargets(value: unknown): Array || !Number.isInteger(target.weight) || target.weight < 1 || target.weight > 10_000)) { throw new CliUsageError("Target weight must be an integer from 1 to 10000"); } + if (target.reasoningEffort !== undefined + && (typeof target.reasoningEffort !== "string" || !isCodexReasoningEffort(target.reasoningEffort))) { + throw new CliUsageError("Target reasoningEffort must be low, medium, high, xhigh, max or ultra"); + } + if (target.serviceTier !== undefined && target.serviceTier !== "default" && target.serviceTier !== "priority") { + throw new CliUsageError("Target serviceTier must be default or priority"); + } if (Object.hasOwn(target, "reasoningEfforts") && (!Array.isArray(target.reasoningEfforts) || target.reasoningEfforts.length === 0 || target.reasoningEfforts.some((effort: unknown) => typeof effort !== "string" || !isCodexReasoningEffort(effort)) diff --git a/src/combos/request.ts b/src/combos/request.ts index d09cc14bd3..60c5f7781a 100644 --- a/src/combos/request.ts +++ b/src/combos/request.ts @@ -61,7 +61,7 @@ function responsesInputNodeHasImage(value: unknown): boolean { */ export function concreteComboRequestBody( body: unknown, - target: Pick, + target: Pick, defaultEffort: OcxComboDefaultEffort | null, targetReasoningEfforts: readonly string[] | undefined, reasoningEffortMode: OcxComboReasoningEffortMode = "strict", @@ -69,6 +69,15 @@ export function concreteComboRequestBody( ): Record { const clone = structuredClone(body) as Record; clone.model = `${target.provider}/${target.model}`; + if (target.serviceTier !== undefined) clone.service_tier = target.serviceTier; + if (target.reasoningEffort !== undefined) { + const reasoning = clone.reasoning; + const preserved = reasoning && typeof reasoning === "object" && !Array.isArray(reasoning) + ? reasoning as Record : {}; + stripAlternativeReasoningControls(clone); + clone.reasoning = { ...preserved, effort: target.reasoningEffort }; + return clone; + } if (defaultEffortMode === "force" && (!defaultEffort || !isCodexReasoningEffort(defaultEffort))) { throw new Error("force combo default effort requires a valid defaultEffort"); } diff --git a/src/combos/types.ts b/src/combos/types.ts index d11a27baf9..409f6be86d 100644 --- a/src/combos/types.ts +++ b/src/combos/types.ts @@ -36,6 +36,8 @@ export interface ComboValidationIssue { export interface NormalizedComboTarget { provider: string; model: string; + reasoningEffort?: OcxComboDefaultEffort; + serviceTier?: "default" | "priority"; weight: number; /** Emergency-only target, deferred under `cooldownWaitPolicy` (#5691). */ lastResort: boolean; @@ -397,6 +399,13 @@ export function comboConfigIssues( message: `targets[${i}].weight must be an integer from 1 to 10000`, }); } + if (target.reasoningEffort !== undefined + && (typeof target.reasoningEffort !== "string" || !isCodexReasoningEffort(target.reasoningEffort))) { + issues.push({ path: ["targets", i, "reasoningEffort"], message: "reasoningEffort must be low, medium, high, xhigh, max or ultra" }); + } + if (target.serviceTier !== undefined && target.serviceTier !== "default" && target.serviceTier !== "priority") { + issues.push({ path: ["targets", i, "serviceTier"], message: "serviceTier must be default or priority" }); + } if (target.reasoningEfforts !== undefined) { if (!Array.isArray(target.reasoningEfforts) || target.reasoningEfforts.length === 0) { issues.push({ @@ -504,6 +513,8 @@ export function normalizeComboConfig(raw: OcxComboConfig): NormalizedComboConfig provider: target.provider.trim(), model: target.model.trim(), weight: target.weight ?? 1, + ...(target.reasoningEffort !== undefined ? { reasoningEffort: target.reasoningEffort } : {}), + ...(target.serviceTier !== undefined ? { serviceTier: target.serviceTier } : {}), ...(target.reasoningEfforts !== undefined ? { reasoningEfforts: [...target.reasoningEfforts] } : {}), diff --git a/src/server/management/combo-routes.ts b/src/server/management/combo-routes.ts index 1699f43722..a409a6c12f 100644 --- a/src/server/management/combo-routes.ts +++ b/src/server/management/combo-routes.ts @@ -203,7 +203,7 @@ export async function handleComboRoutes(ctx: ManagementContext): Promise Object.hasOwn(target, key))) return target; const rawProvider = target.provider; const rawModel = target.model; if (typeof rawProvider !== "string" || typeof rawModel !== "string") return target; @@ -220,6 +220,10 @@ export async function handleComboRoutes(ctx: ManagementContext): Promise; -type ComboTarget = { provider: string; model: string }; +type ComboTarget = Pick; /** What the source needs to run one native child on the combo's open attempt. */ export interface NativeComboChildRun { @@ -263,6 +263,7 @@ function applyComboEffort( combo.defaultEffortMode, ); const reasoning = shaped.reasoning; + if (shaped.service_tier !== undefined) body.service_tier = shaped.service_tier; if (reasoning === undefined) { if (hadEffort) delete body.reasoning_effort; return; diff --git a/src/server/responses/core-combo.ts b/src/server/responses/core-combo.ts index ab24192f75..ab1f9faa7e 100644 --- a/src/server/responses/core-combo.ts +++ b/src/server/responses/core-combo.ts @@ -1123,12 +1123,17 @@ export async function executeComboResponses( const poolLocalRefusal = isAnthropicPoolLocalRefusal( config, pick.target.provider, failure.response.status, failedAccount, failureNow, ); + const poolInstance = configuredAnthropicInstance(config, pick.target.provider); + const rechecksPoolQuota = poolLocalRefusal && poolInstance !== undefined + && anthropicRoutingFor(poolInstance).anthropicAccountPoolConfig(config).quotaRecheckMs !== undefined; const nextPick = advanceComboAfterFailure(config, pick, { retryAfter: poolLocalRefusal ? undefined : failure.retryAfter, resetAt: failure.resetAt, cooldownMs: combo.cooldownMs, now: failureNow, - cooldownScope: failureCooldownScope, + // The pool already refuses locally until its account deadline. A second target + // cooldown would slide that deadline forward on each intervening request. + cooldownScope: rechecksPoolQuota ? "none" : failureCooldownScope, eligible: targetEligible, status: failure.response.status, code: failure.upstreamCode, diff --git a/src/server/responses/run-turn-execution.ts b/src/server/responses/run-turn-execution.ts index 54553d5508..855002473c 100644 --- a/src/server/responses/run-turn-execution.ts +++ b/src/server/responses/run-turn-execution.ts @@ -712,8 +712,11 @@ export async function executeResponsesRunTurn( if (preflight.error || preflight.empty) { runTurnAbort.abort(); queue.close(); - const message = preflight.error?.message ?? "Adapter ended before producing a response"; - const failure = formatErrorResponse(502, "upstream_error", redactSecretString(message)); + const classified = adapterFailureFromEvent(preflight.error ?? { + type: "error", status: 502, message: "Adapter ended before producing a response", + }); + const failure = formatErrorResponse(classified.httpStatus, classified.error.type, + redactSecretString(classified.error.message), { code: classified.error.code }); // A replay-unsafe heartbeat means the adapter already ran a local side effect, so the // combo must not send this turn to another target: the failure stays with this child. if (preflight.replayUnsafe) markResponseNonReplayable(failure); @@ -909,10 +912,11 @@ export async function executeResponsesRunTurn( ? classifyUndeclaredFirstTool(firstMeaningful) : undefined; if (!firstMeaningful || firstMeaningful.type === "error" || classifiedError) { - const message = classifiedError?.message ?? (firstMeaningful?.type === "error" - ? firstMeaningful.message - : "Adapter ended before producing a response"); - const failure = formatErrorResponse(502, "upstream_error", redactSecretString(message)); + const classified = adapterFailureFromEvent(classifiedError ?? (firstMeaningful?.type === "error" + ? firstMeaningful + : { type: "error", status: 502, message: "Adapter ended before producing a response" })); + const failure = formatErrorResponse(classified.httpStatus, classified.error.type, + redactSecretString(classified.error.message), { code: classified.error.code }); if (replayUnsafe) markResponseNonReplayable(failure); return failure; } diff --git a/src/types/config.ts b/src/types/config.ts index de31616083..5656438d89 100644 --- a/src/types/config.ts +++ b/src/types/config.ts @@ -1275,6 +1275,10 @@ export type OcxComboCooldownWaitPolicy = "before-last-resort"; export interface OcxComboTarget { provider: string; model: string; + /** Per-target effort override; other targets retain the caller's effort. */ + reasoningEffort?: OcxComboDefaultEffort; + /** Per-target speed tier, independent of the caller's tier. */ + serviceTier?: "default" | "priority"; /** Relative target weight for round-robin batches and random selection. Default 1; valid range 1..10000. */ weight?: number; /** diff --git a/structure/transports/responses-failover.md b/structure/transports/responses-failover.md index cc26a8ed31..a81bf11439 100644 --- a/structure/transports/responses-failover.md +++ b/structure/transports/responses-failover.md @@ -1,5 +1,5 @@ # Responses Failover And Replay - +`src/combos/request.ts` applies target `reasoningEffort`/`serviceTier` to a fresh child; `core-combo-native.ts` preserves these in native Chat, under the destination's Fast policy. With `quotaRecheckMs`, `core-combo.ts` leaves pool-local refusals to the account deadline, avoiding a second sliding target cooldown. `run-turn-execution.ts` preserves structured status/type/code at both combo preflight boundaries: context refusals can hop without a global 502 cooldown, while committed side effects remain non-replayable. Isolated HTTP/fake-clock coverage: `tests/adapters/anthropic/anthropic-subscription-chain.test.ts`. `src/server/responses/request-transport.ts` resolves the final Anthropic model ID once for each enabled-pool request and holds its model route through admission. The three account-refusal retry sites in `adapter-dispatch.ts`, `adapter-continuation.ts` and `sidecar-execution.ts` use the same route, keep the original refusal if no replacement exists inside it, and preserve existing send and output replay limits. A local cooldown returns 429 with the earliest known Retry-After among accounts the route can use when they recover; explicit fallback includes usable ordinary-pool accounts even when the saved route IDs have been removed, while a strict route stays route-scoped. Selection, refusal and 429-rotation logs use the rule’s 1-based `route:#` position, never its name. Classified 429 admission follows the [Anthropic account-pool contract](../providers/anthropic-account-pool.md#classified-429-admission). Classified account-entitlement 403 recovery follows the [Anthropic account-pool contract](../providers/anthropic-account-pool.md#account-entitlement-refusal-recovery). `src/server/responses/compaction-recovery-policy.ts` is a pure eligibility policy, not a dispatcher. diff --git a/tests/adapters/anthropic/anthropic-subscription-chain.test.ts b/tests/adapters/anthropic/anthropic-subscription-chain.test.ts index 1b06834b94..639ce4e3d0 100644 --- a/tests/adapters/anthropic/anthropic-subscription-chain.test.ts +++ b/tests/adapters/anthropic/anthropic-subscription-chain.test.ts @@ -17,6 +17,8 @@ import { removeTreeWithRetry } from "../../helpers/remove-tree"; const MODEL = "claude-opus-5-5"; const ISSUED_KEY_ID = "issued-key"; let cursorLimited = false; +let cursorOverflow = false; +let astraBodies: Record[] = []; let cursorModels: string[] = []; let cursorEfforts: unknown[] = []; @@ -33,6 +35,10 @@ mock.module("../../../src/server/adapter-resolve", () => ({ async runTurn(parsed: { modelId: string; options?: { reasoning?: unknown } }, _incoming: unknown, emit: (event: AdapterEvent) => void) { cursorModels.push(parsed.modelId); cursorEfforts.push(parsed.options?.reasoning); + if (cursorOverflow) { + emit({ type: "error", status: 400, message: "maximum context length exceeded", code: "context_length_exceeded" }); + return; + } if (cursorLimited) { emit({ type: "error", status: 429, message: "You've hit your usage limit", code: "rate_limit_exceeded" }); return; @@ -97,7 +103,7 @@ function transport(lane: Lane): typeof fetch { }) as typeof fetch; } -function config(options: { combo?: boolean; quotaRecheckMs?: number; allowMetered?: boolean } = {}): OcxConfig { +function config(options: { combo?: boolean; quotaRecheckMs?: number; allowMetered?: boolean; astra?: boolean } = {}): OcxConfig { const subscription = { adapter: "anthropic", baseUrl: "https://subscription.test", authMode: "oauth", models: [MODEL], fetch: transport("subscription") }; const apiKey = { adapter: "anthropic", baseUrl: "https://api-key.test", authMode: "key", apiKey: "${CHAIN_TEST_ANTHROPIC_KEY}", models: [MODEL], fetch: transport("api-key") }; return { @@ -106,7 +112,13 @@ function config(options: { combo?: boolean; quotaRecheckMs?: number; allowMetere providers: { anthropic: subscription as OcxProviderConfig, cursor: { adapter: "cursor", baseUrl: "https://api2.cursor.sh", authMode: "oauth", models: [MODEL] }, - "anthropic-metered": apiKey as OcxProviderConfig, + ...(options.astra ? {} : { "anthropic-metered": apiKey as OcxProviderConfig }), + ...(options.astra ? { astra: { + adapter: "openai-chat", baseUrl: "https://astra.test/v1", authMode: "key", apiKey: "fake-astra", + fastWire: { kind: "service-tier", canonicalToWire: { priority: "priority" }, foreignCallerTiers: "verbatim" }, + supportsServiceTier: true, + models: ["gpt-6-astra"], fetch: astraTransport, + } } : {}), }, ...(options.quotaRecheckMs ? { anthropicAccountPool: { quotaRecheckMs: options.quotaRecheckMs } } : {}), apiKeys: [{ @@ -122,7 +134,8 @@ function config(options: { combo?: boolean; quotaRecheckMs?: number; allowMetere targets: [ { provider: "anthropic", model: MODEL }, { provider: "cursor", model: MODEL }, - { provider: "anthropic-metered", model: MODEL, metered: true }, + ...(options.astra ? [] : [{ provider: "anthropic-metered", model: MODEL, metered: true }]), + ...(options.astra ? [{ provider: "astra", model: "gpt-6-astra", reasoningEffort: "high", serviceTier: "priority" }] : []), ], }, }, @@ -134,7 +147,12 @@ const LOOPBACK: DataPlaneAdmission = { kind: "loopback", source: "loopback" }; const ISSUED: DataPlaneAdmission = { kind: "configured", keyId: ISSUED_KEY_ID, source: "bearer" }; async function chat(cfg: OcxConfig, admission: DataPlaneAdmission = LOOPBACK) { - const response = await handleChatCompletions(new Request("http://localhost/v1/chat/completions", { + const server = Bun.serve({ hostname: "127.0.0.1", port: 0, + fetch: req => handleChatCompletions(req, cfg, { model: "", provider: "" }, + { requestId: "chain-test", start: Date.now(), admission } as never), + }); + try { + const response = await fetch(new URL("/v1/chat/completions", server.url), { method: "POST", headers: { "content-type": "application/json" }, body: JSON.stringify({ @@ -143,12 +161,24 @@ async function chat(cfg: OcxConfig, admission: DataPlaneAdmission = LOOPBACK) { reasoning_effort: "medium", messages: [{ role: "system", content: "S".repeat(5000) }, { role: "user", content: "hi" }], }), - }), cfg, { model: "", provider: "" }, { requestId: "chain-test", start: Date.now(), admission } as never); + }); return { status: response.status, text: await response.text() }; + } finally { + await server.stop(true); + } } const lanes = () => sent.map(entry => entry.lane); +const astraTransport: typeof fetch = Object.assign(async (_input: RequestInfo | URL, init?: RequestInit) => { + astraBodies.push(JSON.parse(String(init?.body ?? "{}"))); + return new Response([ + `data: ${JSON.stringify({ id: "astra", object: "chat.completion.chunk", model: "gpt-6-astra", choices: [{ index: 0, delta: { role: "assistant", content: "astra answered" }, finish_reason: null }] })}\n\n`, + `data: ${JSON.stringify({ id: "astra", object: "chat.completion.chunk", model: "gpt-6-astra", choices: [{ index: 0, delta: {}, finish_reason: "stop" }] })}\n\n`, + "data: [DONE]\n\n", + ].join(""), { headers: { "content-type": "text/event-stream" } }); +}, { preconnect: fetch.preconnect }); + beforeEach(async () => { home = mkdtempSync(join(tmpdir(), "ocx-subscription-chain-")); process.env.OPENCODEX_HOME = home; @@ -164,6 +194,7 @@ beforeEach(async () => { await saveCredential("cursor", { access: "cursor-access", refresh: "cursor-refresh", expires: Date.now() + 3_600_000, accountId: "cursor-account" }); sent = []; cursorModels = []; cursorEfforts = []; subscriptionSpent = false; cursorLimited = false; apiKeyLimited = false; + cursorOverflow = false; astraBodies = []; clearComboSelectionState(); clearComboTargetCooldowns(); clearAnthropicAccountPoolState(); forgetAnthropicFailoverQuorum(); clearGenericFailoverHealth(); clearAccountQuotaCache(); }); @@ -281,3 +312,84 @@ test("without quotaRecheckMs the stated 88-hour reset still holds the account ou expect(snapshots.length).toBeGreaterThan(0); for (const snapshot of snapshots) expect(snapshot!.cooldownUntil! - Date.now()).toBeGreaterThan(80 * 3_600_000); }); + +test("an intervening request during pool cooldown cannot postpone the five-minute subscription recheck", async () => { + subscriptionSpent = true; + const cfg = config({ quotaRecheckMs: 300_000 }); + const start = Date.now(); + const realNow = Date.now; + try { + Date.now = () => start; + expect((await chat(cfg)).text).toContain("cursor answered"); + Date.now = () => start + 240_000; + expect((await chat(cfg)).text).toContain("cursor answered"); + subscriptionSpent = false; + sent = []; cursorModels = []; + Date.now = () => start + 300_001; + expect((await chat(cfg)).text).toContain("anthropic answered"); + expect(lanes()).toEqual(["subscription"]); + expect(cursorModels).toEqual([]); + } finally { + Date.now = realNow; + } +}); + +for (const fallback of ["cursor", "astra"] as const) { + test(`subscription-only three-stage chain returns from ${fallback} after the original recheck deadline`, async () => { + const cfg = config({ quotaRecheckMs: 300_000, astra: true }); + subscriptionSpent = true; + cursorLimited = fallback !== "cursor"; + const start = Date.now(); + const realNow = Date.now; + Date.now = () => start; + try { + const first = await chat(cfg); + expect(first.status).toBe(200); + expect(first.text).toContain(`${fallback} answered`); + if (fallback === "astra") { + expect(lanes()).toEqual(["subscription", "subscription"]); + expect(cursorModels).toEqual([MODEL]); + expect(astraBodies).toHaveLength(1); + expect(astraBodies[0]?.reasoning_effort).toBe("high"); + expect(astraBodies[0]?.service_tier).toBe("priority"); + } + Date.now = () => start + 240_000; + expect((await chat(cfg)).status).toBe(200); + subscriptionSpent = false; + sent = []; cursorModels = []; astraBodies = []; + Date.now = () => start + 300_001; + const recovered = await chat(cfg); + expect(recovered.status).toBe(200); + expect(recovered.text).toContain("anthropic answered"); + expect(lanes()).toEqual(["subscription"]); + expect(cursorModels).toEqual([]); + expect(astraBodies).toEqual([]); + } finally { + Date.now = realNow; + } + }); +} + +test("Cursor context rejection hops to Astra without poisoning the next request", async () => { + subscriptionSpent = true; + cursorOverflow = true; + const cfg = config({ quotaRecheckMs: 300_000, astra: true }); + expect((await chat(cfg)).text).toContain("astra answered"); + expect(lanes()).not.toContain("api-key"); + cursorOverflow = false; + sent = []; cursorModels = []; + expect((await chat(cfg)).text).toContain("cursor answered"); + expect(cursorModels).toEqual([MODEL]); + expect(lanes()).not.toContain("api-key"); +}); + +test("buffered combo preflight preserves Cursor's structured rate-limit status", async () => { + cursorLimited = true; + const cfg = config({ astra: true }); + const combo = cfg.combos?.["claude-chain"]; + if (!combo) throw new Error("Missing fixture combo"); + combo.targets = [{ provider: "cursor", model: MODEL }]; + const response = await chat(cfg); + expect(response.status).toBe(429); + expect(JSON.parse(response.text).error.code).toBe("rate_limit_exceeded"); +}); diff --git a/tests/codex-integration/combos.test.ts b/tests/codex-integration/combos.test.ts index d6c370bf14..ace634eba4 100644 --- a/tests/codex-integration/combos.test.ts +++ b/tests/codex-integration/combos.test.ts @@ -297,6 +297,21 @@ describe("combo request cloning", () => { expect(concrete.input).not.toBe(raw.input); }); + test("target effort and tier override only that child while preserving the caller", () => { + const raw = { model: "combo/free", reasoning: { effort: "medium", summary: "auto" }, service_tier: "default", thinking: { budget_tokens: 1024 } }; + const concrete = concreteComboRequestBody(raw, { + provider: "openai", model: "gpt-6-astra", reasoningEffort: "high", serviceTier: "priority", + }, null, undefined); + expect(concrete.reasoning).toEqual({ effort: "high", summary: "auto" }); + expect(concrete.service_tier).toBe("priority"); + expect(concrete.thinking).toBeUndefined(); + expect(raw.reasoning.effort).toBe("medium"); + expect(raw.service_tier).toBe("default"); + const ordinary = concreteComboRequestBody(raw, target, null, undefined); + expect(ordinary.reasoning).toEqual(raw.reasoning); + expect(ordinary.service_tier).toBe("default"); + }); + test("combo target capability strips unsupported client reasoning controls", () => { expect(concreteComboRequestBody({ model: "combo/x", reasoning: null }, target, "high", []).reasoning).toBeNull(); expect(concreteComboRequestBody( From 5428b42440d68c73b8c12e14329e9e32cf9ff32e Mon Sep 17 00:00:00 2001 From: 001005HS <99410048+001005HS@users.noreply.github.com> Date: Sat, 10 Oct 2026 04:09:04 +0900 Subject: [PATCH 3/3] fix(combos): recognize Cursor context limits and record final reasoning wire --- src/adapters/openai-responses/passthrough.ts | 7 +++++++ src/combos/failover.ts | 1 + src/server/responses/core-combo.ts | 2 +- .../anthropic/anthropic-subscription-chain.test.ts | 2 +- .../adapters/openai/openai-chat-model-suffix.test.ts | 12 ++++++++++++ 5 files changed, 22 insertions(+), 2 deletions(-) diff --git a/src/adapters/openai-responses/passthrough.ts b/src/adapters/openai-responses/passthrough.ts index 4900ededa8..70f799d32a 100644 --- a/src/adapters/openai-responses/passthrough.ts +++ b/src/adapters/openai-responses/passthrough.ts @@ -574,6 +574,8 @@ export function createResponsesPassthroughAdapter(provider: OcxProviderConfig): validateFinalCustomToolCompatibility(finalBody, provider.supportsResponsesCustomTools); } const body = JSON.stringify(applyGithubCopilotContextTier(finalBody, provider, parsed.modelId, incoming.providerName)); + const wireReasoning = isPlainObject(finalBody) && isPlainObject(finalBody.reasoning) + ? finalBody.reasoning.effort : undefined; const releaseBodyObservation = translatorBudget.observeExternallyCapped( "passthrough_serialization", Buffer.byteLength(body, "utf8"), @@ -583,6 +585,11 @@ export function createResponsesPassthroughAdapter(provider: OcxProviderConfig): method: "POST", headers, body, + ...(typeof wireReasoning === "string" ? { reasoningLog: { + effectiveEffort: wireReasoning, + wireField: "reasoning.effort" as const, + wireValue: wireReasoning, + } } : {}), releaseBodyObservation, ...(convertedRoutedCustomToolNames ? { convertedRoutedCustomToolNames } : {}), ...(routedCustomToolRepairNames ? { routedCustomToolRepairNames } : {}), diff --git a/src/combos/failover.ts b/src/combos/failover.ts index d3d957f679..0d7b177cf4 100644 --- a/src/combos/failover.ts +++ b/src/combos/failover.ts @@ -634,6 +634,7 @@ const DEFINITE_CONTEXT_OVERFLOW_PHRASES = [ "exceed the context window", "context window exceeded", "context length exceeded", + "cursor context limit exceeded", "maximum context length", "maximum context window", "too many tokens", diff --git a/src/server/responses/core-combo.ts b/src/server/responses/core-combo.ts index ab1f9faa7e..c41343d1e6 100644 --- a/src/server/responses/core-combo.ts +++ b/src/server/responses/core-combo.ts @@ -918,7 +918,7 @@ export async function executeComboResponses( onResponseComplete: callbackGate.onResponseComplete, }); // Both lanes applied the initial choice before provider pins/caps; retain those transitions. - restoreOriginalRequestedEffort(childLog, initialJevDecision?.effort); + restoreOriginalRequestedEffort(childLog, initialJevDecision?.effort ?? pick.target.reasoningEffort); } catch (error) { callbackGate.discard(); if (options.abortSignal?.aborted) { diff --git a/tests/adapters/anthropic/anthropic-subscription-chain.test.ts b/tests/adapters/anthropic/anthropic-subscription-chain.test.ts index 639ce4e3d0..ab683caa93 100644 --- a/tests/adapters/anthropic/anthropic-subscription-chain.test.ts +++ b/tests/adapters/anthropic/anthropic-subscription-chain.test.ts @@ -36,7 +36,7 @@ mock.module("../../../src/server/adapter-resolve", () => ({ cursorModels.push(parsed.modelId); cursorEfforts.push(parsed.options?.reasoning); if (cursorOverflow) { - emit({ type: "error", status: 400, message: "maximum context length exceeded", code: "context_length_exceeded" }); + emit({ type: "error", status: 400, message: "Cursor context limit exceeded: Cursor Connect error resource limit exceeded: Error", code: "context_length_exceeded" }); return; } if (cursorLimited) { diff --git a/tests/adapters/openai/openai-chat-model-suffix.test.ts b/tests/adapters/openai/openai-chat-model-suffix.test.ts index 96ba37e1fc..b8dca2b3bb 100644 --- a/tests/adapters/openai/openai-chat-model-suffix.test.ts +++ b/tests/adapters/openai/openai-chat-model-suffix.test.ts @@ -16,6 +16,18 @@ function parsed(modelId: string): OcxParsedRequest { }; } +test("Responses reasoning metadata records the serialized effort rather than the caller label", async () => { + const adapter = withTestTranslatorBudget(createResponsesPassthroughAdapter({ + adapter: "openai-responses", baseUrl: "https://api.openai.com/v1", authMode: "key", apiKey: "test-key", + })); + const input = parsed("gpt-6-astra"); + input.options.reasoning = "medium"; + input._rawBody = { model: "gpt-6-astra", input: "OK", reasoning: { effort: "high" } }; + const request = await adapter.buildRequest(input, { headers: new Headers() }); + expect(JSON.parse(String(request.body)).reasoning.effort).toBe("high"); + expect(request.reasoningLog).toEqual({ effectiveEffort: "high", wireField: "reasoning.effort", wireValue: "high" }); +}); + function openaiChatProvider(): OcxProviderConfig { return { adapter: "openai-chat",