diff --git a/.env.example b/.env.example index c811d012..862a5e25 100644 --- a/.env.example +++ b/.env.example @@ -182,6 +182,8 @@ JWT_SECRET=your-jwt-secret-key # Provider API Keys (Optional - if you want to use different keys for different providers) OPENAI_API_KEY=your-openai-api-key ANTHROPIC_API_KEY=your-anthropic-api-key +# Optional spares, used in order after a key is refused (docs/anthropic.md). +# ANTHROPIC_API_KEY_1= # Required only for Anthropic keys that can act on multiple workspaces. ANTHROPIC_WORKSPACE_ID= CEREBRAS_API_KEY=your-cerebras-api-key diff --git a/docs/anthropic.md b/docs/anthropic.md index d9642405..b0524083 100644 --- a/docs/anthropic.md +++ b/docs/anthropic.md @@ -18,6 +18,47 @@ Set `ANTHROPIC_API_KEY` in the server environment or Cloudflare Worker secrets. The Worker forwards it to the Container. Restart the server, or deploy the updated Worker and Container, after changing configuration. +## Spare keys and failover + +Add spare keys as `ANTHROPIC_API_KEY_1`, `ANTHROPIC_API_KEY_2`, and so on: + +```bash +npx wrangler secret put ANTHROPIC_API_KEY_1 +``` + +Requests use the first key that is not resting, in that order, so the spare +carries traffic only after the primary is refused. When the Claude API refuses a +key, the gateway rests it and sends the same request once on the next key. This +applies to unified routes, native `/anthropic/v1/*` routes, and the +`auto:intelligence` chain. Refusals that rest a key: + +| Response | Rest | +| --- | --- | +| 400 `invalid_request_error`, "Your credit balance is too low…" | 1 hour | +| 400 `invalid_request_error`, "You have reached your specified … API usage limits" | 1 hour | +| 429 with `error.details.error_code` `enforced_spend_limit_reached` | 1 hour | +| 402 `billing_error` | 1 hour | +| 401 `authentication_error` | 5 minutes | +| 429 rate limit | `retry-after`, or 1 minute, for that model | + +These rests apply with `MODEL_COOLDOWN_ENABLED=true`, the production setting, +and are capped by `MODEL_COOLDOWN_MAX_SECONDS` (default 3600). A 403 can be +specific to one model, so it and other 400 errors return to the caller without +trying another key. A rested key returns to service when its rest ends. If every key +is resting, the request fails with 429 `model_cooldown`. Responses from unified +routes carry `X-MultiLLM-Credential-Attempts` with the number of keys tried. +`GET /admin/providers/usage` lists each key's slot name, whether it is the +active key or resting, and its dispatch and refusal counts; see +[provider usage](provider-usage.md#credential-slots). + +Keys from different Anthropic organizations do not share thinking signatures: +a signed thinking block replays only in the organization that created it. +Prompt-cache affinity keeps a conversation on the key that served it, so a +conversation moves to the spare only when its key is refused. Its next request +may then fail with an invalid `signature` error; drop the earlier thinking +blocks, or start a new conversation. `ANTHROPIC_WORKSPACE_ID` is sent with every +key, so leave it empty when the keys belong to different organizations. + For a key that can act on multiple Anthropic workspaces, also set `ANTHROPIC_WORKSPACE_ID` to the intended workspace. The server sends it as `anthropic-workspace-id` for generation and model discovery. Caller headers diff --git a/docs/deployment-cloudflare.md b/docs/deployment-cloudflare.md index 752eee4d..a08be9d6 100644 --- a/docs/deployment-cloudflare.md +++ b/docs/deployment-cloudflare.md @@ -255,6 +255,8 @@ Provider secrets are optional and should be set only when used: ```bash npx wrangler secret put OPENAI_API_KEY +npx wrangler secret put ANTHROPIC_API_KEY +npx wrangler secret put ANTHROPIC_API_KEY_1 npx wrangler secret put OPENROUTER_API_KEY npx wrangler secret put OPENCODE_GO_API_KEY npx wrangler secret put CODEX_EASY_API_KEY diff --git a/docs/provider-usage.md b/docs/provider-usage.md index 586fb206..5a178438 100644 --- a/docs/provider-usage.md +++ b/docs/provider-usage.md @@ -94,6 +94,46 @@ an account-quota API. Its console remains the authoritative source. NanoGPT usage lookup follows configured key preference order and advances past definitive credential rejections. It does not expose which key succeeded. +## Credential slots + +Providers with a key pool (the Claude API and Codex Everywhere) add +`credential_slots`: one entry per configured key, in selection order, named by +its environment variable and never by its value. + +```json +"credential_slots": [ + { + "slot": "ANTHROPIC_API_KEY", + "active": false, + "resting_seconds": 3412, + "requests": 41, + "errors": 1, + "refusals": 1, + "last_used_at": "2026-10-26T13:05:12Z", + "last_refusal": "anthropic_credit_exhausted", + "last_refusal_at": "2026-10-26T13:05:12Z" + }, + { + "slot": "ANTHROPIC_API_KEY_1", + "active": true, + "resting_seconds": 0, + "requests": 6, + "errors": 0, + "refusals": 0, + "last_used_at": "2026-10-26T13:09:40Z", + "last_refusal": null, + "last_refusal_at": null + } +] +``` + +`active` marks the first key that is not resting, which new conversations use. +`resting_seconds` covers rests that apply to every model; a rate-limit rest for a +single model is not shown. The counters come from the Container process and +start again at zero when it restarts, for example after a deploy. They count +dispatches, not spend: the Claude Console's Usage and Cost pages, filtered by +API key, are the authoritative record of each key's credit. + ## Caching and timeouts Authoritative results are cached independently per provider. Local request and diff --git a/providers/anthropic.py b/providers/anthropic.py index df865d16..5245fea4 100644 --- a/providers/anthropic.py +++ b/providers/anthropic.py @@ -1,5 +1,6 @@ """Direct Claude API authentication and model metadata.""" +import json from collections.abc import Mapping from typing import Any @@ -9,6 +10,70 @@ "anthropic-beta": "Anthropic-Beta", "x-client-request-id": "X-Client-Request-ID", } +# An account that ran out of credit or reached a spend limit stays refused until its owner +# acts, so its key rests for the longest cooldown instead of being retried every minute. +ANTHROPIC_REFUSAL_REST_SECONDS = 3600 +_REFUSAL_BODY_LIMIT = 65_536 +_SPEND_LIMIT_MESSAGES = ( + "You have reached your specified API usage limits", + "You have reached your specified workspace API usage limits", +) + + +def _header(headers: Any, name: str) -> str: + getter = getattr(headers, "get", None) + if getter is None: + return "" + value = getter(name) + if value is None: + value = next( + (item for key, item in headers.items() if str(key).lower() == name.lower()), + None, + ) + return value if isinstance(value, str) else "" + + +def anthropic_credential_refusal(response: Any) -> str | None: + """Name the account-wide refusal in a Claude API error response, or return None. + + Anthropic reports an empty credit balance and a spend limit set by the account owner + as 400 invalid_request_error, the same shape as a malformed request, so only the + message tells them apart. A tier spend cap is a 429 with a distinct error code. Reading + the small JSON error body caches it on the response, so relays still forward it. + """ + status = getattr(response, "status_code", None) + if status == 402: + return "billing_error" + if status not in (400, 429): + return None + headers = getattr(response, "headers", None) or {} + if "json" not in _header(headers, "Content-Type").lower(): + return None + length = _header(headers, "Content-Length") + if length and (not length.isdigit() or int(length) > _REFUSAL_BODY_LIMIT): + return None + try: + body = response.content + except Exception: # A body that cannot be read is not evidence of a refusal. + return None + try: + error = json.loads(body).get("error") if len(body) <= _REFUSAL_BODY_LIMIT else None + except (AttributeError, TypeError, ValueError, RecursionError): + return None + if not isinstance(error, Mapping): + return None + if status == 429: + details = error.get("details") + code = details.get("error_code") if isinstance(details, Mapping) else None + return "spend_cap" if code == "enforced_spend_limit_reached" else None + message = error.get("message") + if error.get("type") != "invalid_request_error" or not isinstance(message, str): + return None + if message.startswith("Your credit balance is too low"): + return "credit_exhausted" + if message.startswith(_SPEND_LIMIT_MESSAGES): + return "spend_limit" + return None def with_anthropic_headers( diff --git a/routes/provider_credentials.py b/routes/provider_credentials.py index 9c7e3c3a..ae900640 100644 --- a/routes/provider_credentials.py +++ b/routes/provider_credentials.py @@ -5,6 +5,7 @@ from error_handlers import APIError from services.credential_context import observation_context, selection_context +from services.credential_pool import is_key_rejection from services.model_cooldown import ModelCooldownCapacity, ModelCooldownExhausted from providers.nanogpt import is_nanogpt_paygo_rejection from routes.auto_routes import AutoRouteCandidateUnavailable @@ -60,6 +61,19 @@ def _is_nanogpt_routing_refusal(response) -> bool: ) +# Providers whose unified requests move to the next configured key after a key refusal. +ROTATING_PROVIDERS = frozenset({"nanogpt", "anthropic"}) + + +def is_anthropic_key_refusal(response) -> bool: + """The Claude API refused the key (auth, billing, credit, spend limit or rate limit), + so the request generated nothing and may run on the next key. Dispatch already rested + the refused key, and the next selection skips it.""" + return bool(getattr(response, "multillm_credential_refusal", None)) or is_key_rejection( + getattr(response, "status_code", None) + ) + + def _record_nanogpt_token_result( app, provider: str, @@ -96,9 +110,9 @@ def _request_with_provider_token_rotation( billing_refusal_is_routing: bool = False, *, model=None, quota_bucket=None, ): - """Send once per usable NanoGPT key after definite pre-generation failures.""" + """Send once per usable NanoGPT or Claude API key after definite pre-generation failures.""" attempt_limit = 1 - if provider == "nanogpt": + if provider in ROTATING_PROVIDERS: attempt_limit = max(1, len(auth_service_cls.get_api_keys(provider))) attempted_tokens: set[str] = set() @@ -147,7 +161,10 @@ def _request_with_provider_token_rotation( # Every key would refuse this suffix identically, so rotating on # would only surface an unrelated key's error to the caller. break - if provider != "nanogpt" or not is_nanogpt_credential_rejection( + if provider == "anthropic": + if not is_anthropic_key_refusal(response): + break + elif provider != "nanogpt" or not is_nanogpt_credential_rejection( response.status_code ): break @@ -162,6 +179,6 @@ def _add_credential_attempt_headers( provider: str, attempts: int, ) -> Response: - if provider == "nanogpt": + if provider in ROTATING_PROVIDERS: response.headers["X-MultiLLM-Credential-Attempts"] = str(attempts) return response diff --git a/routes/proxy.py b/routes/proxy.py index 8ff05fd6..446b6901 100644 --- a/routes/proxy.py +++ b/routes/proxy.py @@ -51,8 +51,10 @@ stream_upstream_response, ) from routes.dashboard_chat import register_dashboard_chat_routes +from routes.provider_credentials import is_anthropic_key_refusal from services.nanogpt_key_pool import NanoGPTKeyPool, NanoGPTKeyPoolExhausted from services.credential_context import dispatch_context, observation_context, proxy_selection_context +from services.model_cooldown import ModelCooldownCapacity, ModelCooldownExhausted from services.model_cooldown import settings as cooldown_settings from services.nanogpt_speed_breaker import NanoGPTSpeedBreaker from services.provider_access_policy import provider_route_scope @@ -445,6 +447,24 @@ def send_to_url(target_url: str): else: response = send_to_url(url) + if api_provider == "anthropic": + # A refused Claude API key generated nothing: send once on each other key. + # Dispatch rested the refused key, so the next selection skips it. + attempted_tokens = {auth_token} + while is_anthropic_key_refusal(response): + try: + next_token = auth_service_cls.get_api_key(api_provider, **credential_scope) + except (ModelCooldownExhausted, ModelCooldownCapacity): + break + if not next_token or next_token in attempted_tokens: + break + attempted_tokens.add(next_token) + response.close() + headers = proxy_service_cls.prepare_headers( + request.headers, api_provider, next_token, upstream_path=path, + ) + response = send_to_url(url) + if speed_routed_body is not None and is_nanogpt_paygo_rejection( getattr(response, "status_code", None), getattr(response, "content", None), diff --git a/services/auth_primitives.py b/services/auth_primitives.py index c3ab426d..2746e3ea 100644 --- a/services/auth_primitives.py +++ b/services/auth_primitives.py @@ -72,6 +72,8 @@ def provider_credential_env_names(provider: str) -> tuple[str, ...]: ) if provider == "groq": return ("GROQ_API_KEY", "GROQ_API_KEY_1", "GROQ_API_KEY_2") + if provider == "anthropic": + return ("ANTHROPIC_API_KEY", "ANTHROPIC_API_KEY_1") if provider == "nanogpt": return ( "NANOGPT_API_KEY", diff --git a/services/credential_pool.py b/services/credential_pool.py index 446b3e01..aa3df81a 100644 --- a/services/credential_pool.py +++ b/services/credential_pool.py @@ -1,19 +1,22 @@ """Several keys for one provider: use the first that is not resting, and rest refused ones. -Codex Everywhere pools accept a base key and numbered spares (``, `_1`, -`_2`, ...). Each request uses the first key in that order that is not resting. A key +Codex Everywhere and Claude API pools accept a base key and numbered spares (``, +`_1`, `_2`, ...). Each request uses the first key in that order that is not resting. A key the provider refuses (401, 402 or 403, which includes an empty balance) rests for five minutes and a rate-limited key (429) for one minute, so traffic moves to the next key and -returns once the rest ends. State is process-local. +returns once the rest ends. State is process-local, including the per-key counters that +`slots` reports to operators by environment-variable name. """ from __future__ import annotations +import math import os import re import threading import time from collections.abc import Mapping +from datetime import datetime, timezone from typing import ClassVar from providers.codex_everywhere import CODEX_EVERYWHERE_POOLS @@ -30,6 +33,7 @@ POOLED_KEY_ENVS = { **{pool.provider: pool.credential_env for pool in CODEX_EVERYWHERE_POOLS}, "ce-image": "CODEX_EVERYWHERE_API_KEY_GPT_IMAGE", + "anthropic": "ANTHROPIC_API_KEY", } @@ -41,14 +45,15 @@ def is_key_rejection(status: int | None) -> bool: class CredentialPool: _lock: ClassVar[threading.Lock] = threading.Lock() _resting: ClassVar[dict[tuple[str, str], float]] = {} + _observed: ClassVar[dict[tuple[str, str], dict]] = {} @staticmethod def pooled(provider: str | None) -> bool: return provider in POOLED_KEY_ENVS @staticmethod - def keys(provider: str, environ: Mapping[str, str] | None = None) -> list[str]: - """The provider's configured keys: the base name first, then _1, _2, ... ascending.""" + def named_keys(provider: str, environ: Mapping[str, str] | None = None) -> list[tuple[str, str]]: + """(environment variable, key) pairs: the base name first, then _1, _2, ... ascending.""" base = POOLED_KEY_ENVS.get(provider) if base is None: return [] @@ -59,12 +64,57 @@ def keys(provider: str, environ: Mapping[str, str] | None = None) -> list[str]: for name in source if (match := numbered.fullmatch(name)) ) - keys: list[str] = [] + named: list[tuple[str, str]] = [] for name in (base, *(name for _, name in spares)): key = usable_credential(source.get(name), name) - if key and key.strip() not in keys: - keys.append(key.strip()) - return keys + if key and key.strip() not in {known for _, known in named}: + named.append((name, key.strip())) + return named + + @staticmethod + def keys(provider: str, environ: Mapping[str, str] | None = None) -> list[str]: + """The provider's configured keys: the base name first, then _1, _2, ... ascending.""" + return [key for _, key in CredentialPool.named_keys(provider, environ)] + + @classmethod + def slots(cls, provider: str, *, now: float | None = None) -> list[dict]: + """Each configured key by environment-variable name, never its value: whether it is + resting, and what this process sent with it since it started.""" + current = time.monotonic() if now is None else now + rows = [] + with cls._lock: + for name, key in cls.named_keys(provider): + observed = cls._observed.get((provider, key), {}) + rows.append({ + "slot": name, + "active": False, + "resting_seconds": math.ceil(max(0.0, cls._resting.get((provider, key), 0) - current)), + "requests": observed.get("requests", 0), + "errors": observed.get("errors", 0), + "refusals": observed.get("refusals", 0), + "last_used_at": observed.get("last_used_at"), + "last_refusal": observed.get("last_refusal"), + "last_refusal_at": observed.get("last_refusal_at"), + }) + active = next((row for row in rows if not row["resting_seconds"]), None) + if active is not None: + active["active"] = True + return rows + + @classmethod + def _observe(cls, provider: str, key: str, status: int, outcome: UpstreamOutcome | None) -> None: + classified = outcome or classify_upstream_outcome(status) + refused = classified.credential_health in {"rejected", "throttled"} or is_key_rejection(status) + at = datetime.now(timezone.utc).isoformat(timespec="seconds").replace("+00:00", "Z") + with cls._lock: + entry = cls._observed.setdefault((provider, key), {"requests": 0, "errors": 0, "refusals": 0}) + entry["requests"] += 1 + entry["errors"] += int(status >= 400) + entry["last_used_at"] = at + if refused: + entry["refusals"] += 1 + entry["last_refusal"] = classified.reason if outcome is not None else f"http_{status}" + entry["last_refusal_at"] = at @classmethod def available( @@ -197,6 +247,7 @@ def record( ) -> None: if not key or provider not in POOLED_KEY_ENVS: return + cls._observe(provider, key, status, outcome) reset_schedule.record_result_observation( provider, key, status, model=model, quota_bucket=quota_bucket, outcome=outcome, retry_after_seconds=retry_after_seconds, @@ -264,21 +315,20 @@ def record_headers( usage_windows: list[Mapping] | None = None, now: float | None = None, ) -> None: - """Record the result for the bearer key a request carried.""" + """Record the result for the bearer or X-Api-Key credential a request carried.""" if provider not in POOLED_KEY_ENVS: return - authorization = next( - ( - value - for name, value in headers.items() - if name.lower() == "authorization" - ), - "", - ) + sent = {name.lower(): value for name, value in headers.items()} + authorization = sent.get("authorization", "") if isinstance(authorization, str) and authorization.startswith("Bearer "): + key = authorization[len("Bearer ") :].strip() + else: + api_key = sent.get("x-api-key") + key = api_key.strip() if isinstance(api_key, str) else "" + if key: cls.record( provider, - authorization[len("Bearer ") :].strip(), + key, status, model=model, quota_bucket=quota_bucket, @@ -293,3 +343,4 @@ def record_headers( def reset(cls) -> None: with cls._lock: cls._resting = {} + cls._observed = {} diff --git a/services/intelligence_gateway.py b/services/intelligence_gateway.py index e55dc359..4a4983cb 100644 --- a/services/intelligence_gateway.py +++ b/services/intelligence_gateway.py @@ -227,7 +227,9 @@ def events(self): self.unresolved = False last_error = rejection(head) # A refused key generated nothing, so the same model may run on the next key. - self._key_refused = is_key_rejection(head.status_code) + self._key_refused = is_key_rejection(head.status_code) or getattr( + head, "credential_refused", False + ) if self.request.explicit and not (self._key_refused and spare_keys): break if not self.escalations: diff --git a/services/intelligence_transport.py b/services/intelligence_transport.py index 84083d39..4e8995f8 100644 --- a/services/intelligence_transport.py +++ b/services/intelligence_transport.py @@ -53,6 +53,9 @@ def remaining(deadline, cancelled): class ResponseHead: status_code: int headers: dict + # Dispatch classified the error as an account-wide key refusal (for example an empty + # Claude API credit balance reported as 400), so the next key may run the same model. + credential_refused: bool = False class Exchange: @@ -119,7 +122,8 @@ def _run(self): for name in ("Content-Type", "Retry-After") if name in response.headers } - self._put(("head", ResponseHead(response.status_code, headers))) + refused = bool(getattr(response, "multillm_credential_refusal", None)) + self._put(("head", ResponseHead(response.status_code, headers, refused))) size = 0 # Definite rejections need no error-body read, which could otherwise # disclose secrets or postpone a safe fallback. diff --git a/services/managed_turn.py b/services/managed_turn.py index cd0a3e98..6cebfea5 100644 --- a/services/managed_turn.py +++ b/services/managed_turn.py @@ -278,6 +278,7 @@ def __init__(self, response, upstream): self.headers = CaseInsensitiveDict(response.headers) self.response, self.upstream = response, upstream self.raw = None + self.multillm_credential_refusal = getattr(upstream, "multillm_credential_refusal", None) def iter_content(self, **kwargs): yield from self.response.response diff --git a/services/provider_usage_service.py b/services/provider_usage_service.py index 41c2e790..4f7e7e79 100644 --- a/services/provider_usage_service.py +++ b/services/provider_usage_service.py @@ -155,7 +155,7 @@ def _base_entry( status = "pending" stats = self.metrics_service.get_provider_stats(provider) cost = _mapping(provider_cost) - return { + entry = { "provider": provider, "configured": configured, "supports_authoritative_usage": supported, @@ -180,6 +180,12 @@ def _base_entry( }, "error": None, } + from services.credential_pool import CredentialPool + + if CredentialPool.pooled(provider): + # Process-local per-key state, named by environment variable; never key material. + entry["credential_slots"] = CredentialPool.slots(provider) + return entry def _authoritative_usage(self, provider: str) -> dict[str, Any]: cached = self._cached(provider) diff --git a/services/proxy_service.py b/services/proxy_service.py index add230b4..15c5631c 100644 --- a/services/proxy_service.py +++ b/services/proxy_service.py @@ -23,7 +23,11 @@ from services.transport_policy import RAW_PASSTHROUGH_PROVIDERS from services.secret_firewall import protect_body from providers.cline_pass import cline_completion_payload -from providers.anthropic import with_anthropic_headers +from providers.anthropic import ( + ANTHROPIC_REFUSAL_REST_SECONDS, + anthropic_credential_refusal, + with_anthropic_headers, +) from services.provider_request_headers import filter_upstream_headers from providers.codex_everywhere import ( CODEX_EVERYWHERE_ANTHROPIC_PROVIDERS, @@ -3325,9 +3329,20 @@ def make_request( ) from services.credential_context import dispatch_context, observation_context from services.pool_reset_schedule import response_usage_observation - CredentialPool.record_headers(api_provider, headers, response.status_code, - **(cooldown_context or dispatch_context(api_provider, data, url)), - **observation_context(response), **response_usage_observation(response)) + observation = {**observation_context(response), **response_usage_observation(response)} + status = response.status_code + if api_provider == "anthropic" and (refusal := anthropic_credential_refusal(response)): + # The account, not this request, was refused: rest its key so the + # rotation loops and later requests move to the next configured key. + response.multillm_credential_refusal = refusal + status = 402 + observation.update( + outcome=UpstreamOutcome(False, "neutral", "rejected", f"anthropic_{refusal}"), + credential_wide_auth=True, + retry_after_seconds=ANTHROPIC_REFUSAL_REST_SECONDS, + ) + CredentialPool.record_headers(api_provider, headers, status, + **(cooldown_context or dispatch_context(api_provider, data, url)), **observation) return response # Check if this is a streaming request diff --git a/services/upstream_transport.py b/services/upstream_transport.py index 0cc9fb74..b3a030b1 100644 --- a/services/upstream_transport.py +++ b/services/upstream_transport.py @@ -25,6 +25,13 @@ def iter_stream_content(response: requests.Response, *, pii_context=None) -> Ite def _stream_content(response: requests.Response) -> Iterator[bytes]: + if getattr(response, "_content_consumed", False) and isinstance( + getattr(response, "_content", None), bytes + ): + # The body was already read (an error body classified at dispatch); the socket + # is drained, so replay the cached bytes. + yield from response.iter_content(chunk_size=64 * 1024) + return raw_response = getattr(response, "raw", None) raw_read1 = getattr(raw_response, "read1", None) if callable(raw_read1): diff --git a/tests/outbound_inventory.json b/tests/outbound_inventory.json index 979cfae5..72d3ac81 100644 --- a/tests/outbound_inventory.json +++ b/tests/outbound_inventory.json @@ -47,7 +47,7 @@ "providers/anthropic.py": { "kind": "non-provider", "reason": "Runtime module pinned independently of HTTP client spelling; network attempts require runtime admission.", - "sha256": "28ec570d523c8326c073ad8fc0aa051719e0aaaf2e8a468efe9e1061cdd9c528" + "sha256": "775f4b5241684e69c80a62b23cd61a9b41de7e39e5931e2a4ea83c027e95a8fe" }, "providers/base.py": { "kind": "non-provider", @@ -342,12 +342,12 @@ "routes/provider_credentials.py": { "kind": "guarded", "reason": "Provider adapter, protocol helper or orchestration; delegates network I/O to the guarded provider clients.", - "sha256": "7ed444a62db36f68a4bffa9f4e74c762212114989c8a00efc9fbb0c2f059d1a4" + "sha256": "13d312e94c56ab52be80ddd480d83077faa3b1ae89b06c8046af1c72452c43ce" }, "routes/proxy.py": { "kind": "guarded", "reason": "Provider adapter, protocol helper or orchestration; delegates network I/O to the guarded provider clients.", - "sha256": "265e183e46865675ea3963f49546f650b3d0a04f98742f2dec63c7da862868a7" + "sha256": "60f84a4c5029694f6a5b8b9be3d54e03b08262b22c49c33203b12d8d04f8396f" }, "routes/public_pages.py": { "kind": "non-provider", @@ -487,7 +487,7 @@ "services/auth_primitives.py": { "kind": "non-provider", "reason": "Runtime module pinned independently of HTTP client spelling; network attempts require runtime admission.", - "sha256": "0236bdad078e24d37db938cabe1db820094f05605f7ebd87c89597b0232143b8" + "sha256": "62d1d194fc94dd9bc313569975f0eeea10ca95b42af9341d198e5db8195ede06" }, "services/auth_service.py": { "kind": "non-provider", @@ -652,7 +652,7 @@ "services/credential_pool.py": { "kind": "non-provider", "reason": "Runtime module pinned independently of HTTP client spelling; network attempts require runtime admission.", - "sha256": "aa6ca6ddfb0137452cc53ecff300d13f383c39836a3005a2a56e7481d6e90771" + "sha256": "4620e65913f7ef29aa514bc92251bc88f74f09082de9b73f1759409efec43f6d" }, "services/credits_admission.py": { "kind": "non-provider", @@ -822,7 +822,7 @@ "services/intelligence_gateway.py": { "kind": "non-provider", "reason": "Runtime module pinned independently of HTTP client spelling; network attempts require runtime admission.", - "sha256": "efac340e944c0444222a44a484c1abf2caf219458b4eaa5ed0f093ac143d0c91" + "sha256": "acaa597b2f94a9fc0cd351564bdca3405aff00fd742572a0e5a167daa0a4ae3d" }, "services/intelligence_output.py": { "kind": "non-provider", @@ -857,7 +857,7 @@ "services/intelligence_transport.py": { "kind": "non-provider", "reason": "Runtime module pinned independently of HTTP client spelling; network attempts require runtime admission.", - "sha256": "2c3e234e66159f49d9cb55fcf586f5034d6adab9a169d564ac10d66372defd66" + "sha256": "52453244c54b25bcba978bc345b15e9d5355f9bac703572e0692825e037f036f" }, "services/judge_routing.py": { "kind": "non-provider", @@ -917,7 +917,7 @@ "services/managed_turn.py": { "kind": "guarded", "reason": "Provider adapter, protocol helper or orchestration; delegates network I/O to the guarded provider clients.", - "sha256": "c619b5fd489667332193b797375f43eae84700cef42dd779b88bacac8f6c7f55" + "sha256": "caa27eb5c748c419ba758abc3eb44c03a6ccef7bcee0cbab4b06da30a7b1ed93" }, "services/mcp_contract_drift.py": { "kind": "non-provider", @@ -1197,7 +1197,7 @@ "services/provider_usage_service.py": { "kind": "non-provider", "reason": "Runtime module pinned independently of HTTP client spelling; network attempts require runtime admission.", - "sha256": "86e844a9e00cd2fc3609131f527b60f853f0d868f9bcd12bfb3f9956e053fbb6" + "sha256": "1e514b5c0d08b4cbb444a0c73c2591505f7ca8da934c78f700edba5a0d23b1d7" }, "services/proxy_documentation_service.py": { "kind": "guarded", @@ -1207,7 +1207,7 @@ "services/proxy_service.py": { "kind": "guarded", "reason": "Provider adapter, protocol helper or orchestration; delegates network I/O to the guarded provider clients.", - "sha256": "e4241056d6a511a101367ec27d46e6147ffd62f86a4feb91f6a133954f5ca505" + "sha256": "3afa672d3710422437638b2cb68e17a31d7cba9213ad63b64d976978d9bce164" }, "services/rate_limit_d1.py": { "kind": "non-provider", @@ -1437,7 +1437,7 @@ "services/upstream_transport.py": { "kind": "guarded", "reason": "Provider adapter, protocol helper or orchestration; delegates network I/O to the guarded provider clients.", - "sha256": "c27e7589a1efbc44b8bdd546199f087c4ec08c852e811156c6be149f0fa0690c" + "sha256": "2cbca089d5fdbe7305cc0f049e3c4a747c2d5601e63c8268da6e1ad994b6020f" }, "services/usage_ledger.py": { "kind": "non-provider", @@ -1572,7 +1572,7 @@ "worker/container-env.mjs": { "kind": "guarded", "reason": "Provider adapter, protocol helper or orchestration; delegates network I/O to the guarded provider clients.", - "sha256": "56f07b1a9d2f6451be990e16038e30d55afbee1dbf5a5e5ea82b30235558d790" + "sha256": "f5dc040ed498889a81747ddde2bf7ea7750af3fffa3396e75e82c92438326a9a" }, "worker/context-canary.mjs": { "kind": "non-provider", diff --git a/tests/test_anthropic_key_failover.py b/tests/test_anthropic_key_failover.py new file mode 100644 index 00000000..38133d1e --- /dev/null +++ b/tests/test_anthropic_key_failover.py @@ -0,0 +1,336 @@ +"""Claude API key pool: refusal classification, key rests and failover to the spare key.""" + +import io +import json +import os +from unittest.mock import patch + +import pytest +import requests + +from providers.anthropic import anthropic_credential_refusal +from tests.intelligence_fixtures import IntelligenceApiTestCase +from tests.intelligence_fixtures import upstream as cached_upstream +from tests.unified_api_test_case import UnifiedApiTestCase + +MODEL = "claude-haiku-5-5" +SELECTOR = f"anthropic:{MODEL}" +ADMIN = {"Authorization": "Bearer admin-test-key"} +KEYS = {"ANTHROPIC_API_KEY": "key-a", "ANTHROPIC_API_KEY_1": "key-b"} +CREDIT = { + "type": "error", + "error": { + "type": "invalid_request_error", + "message": "Your credit balance is too low to access the Anthropic API. " + "Please go to Plans & Billing to upgrade or purchase credits.", + }, +} +MESSAGE = { + "id": "msg_haiku", + "type": "message", + "role": "assistant", + "model": MODEL, + "content": [{"type": "text", "text": "Hello"}], + "stop_reason": "end_turn", + "stop_sequence": None, + "usage": {"input_tokens": 8, "output_tokens": 4}, +} + + +def error(error_type, message, **details): + body = {"type": "error", "error": {"type": error_type, "message": message}} + if details: + body["error"]["details"] = details + return body + + +def upstream(body, status=200, content_type="application/json"): + """A response whose body still sits on the socket, like a streamed upstream reply.""" + response = requests.Response() + response.status_code = status + response.raw = io.BytesIO(body if isinstance(body, bytes) else json.dumps(body).encode()) + response.headers["Content-Type"] = content_type + return response + + +@pytest.fixture +def pool(monkeypatch): + from services import pool_reset_schedule + from services.credential_pool import CredentialPool + from services.model_cooldown import model_cooldown + + for name, value in KEYS.items(): + monkeypatch.setenv(name, value) + monkeypatch.setattr(pool_reset_schedule, "pool_reset_schedule", pool_reset_schedule.PoolResetSchedule()) + CredentialPool.reset() + model_cooldown.reset() + yield CredentialPool + CredentialPool.reset() + model_cooldown.reset() + + +@pytest.mark.parametrize( + ("status", "body", "expected"), + [ + (400, CREDIT, "credit_exhausted"), + (400, error("invalid_request_error", "You have reached your specified API usage limits. " + "You will regain access on 2026-11-01 at 00:00 UTC."), "spend_limit"), + (400, error("invalid_request_error", "You have reached your specified workspace API usage limits."), + "spend_limit"), + (429, error("rate_limit_error", "You have reached your API usage limits.", + error_code="enforced_spend_limit_reached"), "spend_cap"), + (402, error("billing_error", "Payment details are invalid."), "billing_error"), + (400, error("invalid_request_error", "messages.1.content.0: Invalid `signature` in `thinking` block."), None), + (400, error("invalid_request_error", "max_tokens: Field required"), None), + (429, error("rate_limit_error", "Number of request tokens has exceeded your rate limit."), None), + (401, error("authentication_error", "invalid x-api-key"), None), + (200, MESSAGE, None), + ], +) +def test_only_account_wide_refusals_are_classified(status, body, expected): + assert anthropic_credential_refusal(upstream(body, status)) == expected + + +def test_unreadable_or_oversized_error_bodies_are_not_refusals(): + assert anthropic_credential_refusal(upstream(b"credit balance is too low", 400, "text/plain")) is None + assert anthropic_credential_refusal(upstream(b"{not json", 400)) is None + oversized = upstream(CREDIT, 400) + oversized.headers["Content-Length"] = "65537" + assert anthropic_credential_refusal(oversized) is None + + +def test_a_classified_body_is_still_relayed_from_the_cache(): + from services.upstream_transport import _stream_content + + response = upstream(CREDIT, 400) + assert anthropic_credential_refusal(response) == "credit_exhausted" + assert response.raw.read() == b"", "classification drained the socket" + assert json.loads(b"".join(_stream_content(response))) == CREDIT + + +def test_the_base_key_comes_first_and_x_api_key_results_rest_it(pool): + from services.auth_service import AuthService + + assert pool.pooled("anthropic") + assert AuthService.get_api_keys("anthropic") == ["key-a", "key-b"] + assert AuthService.get_api_key("anthropic") == "key-a" + pool.record_headers("anthropic", {"X-Api-Key": "key-a"}, 401) + assert AuthService.get_api_key("anthropic") == "key-b" + + +@pytest.mark.parametrize("production_flags", [False, True]) +def test_dispatch_rests_a_key_whose_account_ran_out_of_credit(pool, monkeypatch, production_flags): + from services.auth_service import AuthService + from services.credential_context import selection_context + from services.proxy_service import ProxyService + + flag = "true" if production_flags else "false" + monkeypatch.setenv("MODEL_COOLDOWN_ENABLED", flag) + monkeypatch.setenv("POOL_RESET_SCHEDULING_ENABLED", flag) + + def send(body, status): + with patch.object(ProxyService, "_make_base_request", return_value=upstream(body, status)): + return ProxyService.make_request( + method="POST", + url="https://api.anthropic.com/v1/messages", + headers={"X-Api-Key": "key-a", "Anthropic-Version": "2023-06-01"}, + params={}, + data=json.dumps({"model": MODEL, "max_tokens": 16, "messages": []}).encode(), + api_provider="anthropic", + use_cache=False, + force_raw_passthrough=True, + ) + + scope = selection_context("anthropic", MODEL) + response = send(error("invalid_request_error", "max_tokens: Field required"), 400) + assert getattr(response, "multillm_credential_refusal", None) is None + assert AuthService.get_api_key("anthropic", **scope) == "key-a", "a bad request is not the key's fault" + + response = send(CREDIT, 400) + assert response.status_code == 400 + assert response.multillm_credential_refusal == "credit_exhausted" + assert json.loads(response.content) == CREDIT + assert AuthService.get_api_key("anthropic", **scope) == "key-b" + assert AuthService.get_api_key("anthropic", **selection_context("anthropic", "claude-other")) == "key-b", ( + "an empty balance refuses every model" + ) + + +def test_provider_usage_reports_each_key_by_slot_name_only(pool): + from services.provider_usage_service import ProviderUsageService + from services.upstream_outcome import UpstreamOutcome + from tests.test_provider_usage_service import FakeAuthService, FakeMetricsService + + pool.record("anthropic", "key-a", 200) + pool.record("anthropic", "key-a", 402, retry_after_seconds=3600, credential_wide_auth=True, + outcome=UpstreamOutcome(False, "neutral", "rejected", "anthropic_credit_exhausted")) + pool.record("anthropic", "key-b", 200) + service = ProviderUsageService(config={}, auth_service=FakeAuthService(keys={"anthropic": "key-b"}), + metrics_service=FakeMetricsService(), proxy_service=None) + + entry = service.snapshot(["anthropic"])["providers"][0] + first, spare = entry["credential_slots"] + assert (first["slot"], first["active"], first["requests"], first["errors"], first["refusals"]) == ( + "ANTHROPIC_API_KEY", False, 2, 1, 1) + assert first["last_refusal"] == "anthropic_credit_exhausted" and first["resting_seconds"] > 0 + assert (spare["slot"], spare["active"], spare["requests"], spare["refusals"]) == ("ANTHROPIC_API_KEY_1", True, 1, 0) + assert spare["last_refusal"] is None and spare["last_used_at"] + serialized = json.dumps(entry) + assert "key-a" not in serialized and "key-b" not in serialized + assert "credential_slots" not in service.snapshot(["openrouter"])["providers"][0] + + +class AnthropicKeyFailoverRouteTest(UnifiedApiTestCase): + def setUp(self): + super().setUp() + from services.credential_pool import CredentialPool + from services.model_cooldown import model_cooldown + + os.environ.update(KEYS) + CredentialPool.reset() + model_cooldown.reset() + self.addCleanup(CredentialPool.reset) + self.addCleanup(model_cooldown.reset) + + def upstreams(self, *replies): + """Serve replies in order through the real dispatch path and record each key used.""" + from services.proxy_service import ProxyService + + self.sent_keys = [] + pending = list(replies) + + def reply(**kwargs): + self.sent_keys.append(kwargs["headers"].get("X-Api-Key")) + item = pending.pop(0) + return item if isinstance(item, requests.Response) else upstream(*item) + + return patch.object(ProxyService, "_make_base_request", side_effect=reply) + + def chat(self): + return self.client.post( + "/v1/chat/completions", + json={"model": SELECTOR, "messages": [{"role": "user", "content": "Hello"}]}, + headers=ADMIN, + ) + + def test_unified_chat_moves_to_the_spare_when_credit_runs_out(self): + with self.upstreams((CREDIT, 400), (MESSAGE, 200)): + result = self.chat() + self.assertEqual(result.status_code, 200) + self.assertEqual(result.get_json()["choices"][0]["message"]["content"], "Hello") + self.assertEqual(self.sent_keys, ["key-a", "key-b"]) + self.assertEqual(result.headers["X-MultiLLM-Credential-Attempts"], "2") + with self.upstreams((MESSAGE, 200)): + self.assertEqual(self.chat().status_code, 200) + self.assertEqual(self.sent_keys, ["key-b"], "the refused key rests") + from services.credential_pool import CredentialPool + + first, spare = CredentialPool.slots("anthropic") + self.assertEqual((first["refusals"], first["last_refusal"], first["active"]), (1, "anthropic_credit_exhausted", False)) + self.assertEqual((spare["requests"], spare["active"]), (2, True)) + + def test_unified_stream_moves_to_the_spare_when_credit_runs_out(self): + from tests.test_protocol_routes import ANTHROPIC_STREAM, sse_upstream + + with self.upstreams((CREDIT, 400), sse_upstream(ANTHROPIC_STREAM)): + result = self.client.post( + "/v1/chat/completions", + json={"model": SELECTOR, "messages": [{"role": "user", "content": "Hello"}], "stream": True}, + headers=ADMIN, + ) + body = result.get_data(as_text=True) + self.assertEqual(result.status_code, 200) + self.assertIn("Salut", body) + self.assertTrue(body.rstrip().endswith("data: [DONE]")) + self.assertEqual(self.sent_keys, ["key-a", "key-b"]) + + def test_unified_request_errors_are_not_repeated_on_the_spare(self): + invalid = error("invalid_request_error", "messages: Field required") + with self.upstreams((invalid, 400)): + result = self.chat() + self.assertEqual(result.status_code, 400) + self.assertEqual(self.sent_keys, ["key-a"]) + + def test_native_messages_move_to_the_spare_after_a_key_refusal(self): + with self.upstreams((error("authentication_error", "invalid x-api-key"), 401), (MESSAGE, 200)): + result = self.client.post( + "/anthropic/v1/messages", + json={"model": MODEL, "max_tokens": 16, "messages": [{"role": "user", "content": "Hi"}]}, + headers={"X-Api-Key": "admin-test-key", "Anthropic-Version": "2023-06-01"}, + ) + self.assertEqual(result.get_json(), MESSAGE) + self.assertEqual(result.status_code, 200) + self.assertEqual(self.sent_keys, ["key-a", "key-b"]) + + def test_native_messages_relay_the_last_refusal_when_every_key_is_refused(self): + with self.upstreams((CREDIT, 400), (CREDIT, 400)): + result = self.client.post( + "/anthropic/v1/messages", + json={"model": MODEL, "max_tokens": 16, "messages": [{"role": "user", "content": "Hi"}]}, + headers={"X-Api-Key": "admin-test-key", "Anthropic-Version": "2023-06-01"}, + ) + self.assertEqual(result.get_json(), CREDIT) + self.assertEqual(result.status_code, 400) + self.assertEqual(self.sent_keys, ["key-a", "key-b"]) + + +class AnthropicKeyFailoverProductionFlagsTest(AnthropicKeyFailoverRouteTest): + """The same routes with production's scoped cooldowns, reset scheduling and cache affinity.""" + + def setUp(self): + super().setUp() + from services import pool_reset_schedule, prompt_cache_affinity + + os.environ.update( + { + "MODEL_COOLDOWN_ENABLED": "true", + "POOL_RESET_SCHEDULING_ENABLED": "true", + "PROMPT_CACHE_AFFINITY_ENABLED": "true", + } + ) + for module, name, value in ( + (pool_reset_schedule, "pool_reset_schedule", pool_reset_schedule.PoolResetSchedule()), + (prompt_cache_affinity, "store", prompt_cache_affinity.PromptCacheAffinity()), + ): + replaced = patch.object(module, name, value) + replaced.start() + self.addCleanup(replaced.stop) + + +class AnthropicKeyFailoverIntelligenceTest(IntelligenceApiTestCase): + def test_haiku_runs_on_the_spare_key_before_falling_back_to_another_model(self): + from services.credential_pool import CredentialPool + from services.intelligence_anthropic import HAIKU_INTELLIGENCE_MODEL + from services.intelligence_store import IntelligenceStore + from services.proxy_service import ProxyService + from tests.test_intelligence_policy import candidate, policy + + CredentialPool.reset() + self.addCleanup(CredentialPool.reset) + IntelligenceStore.seed( + policy( + allow_paid_overage=True, + principal_daily_tokens=1_048_576, + global_daily_tokens=1_048_576, + candidates=[ + candidate(HAIKU_INTELLIGENCE_MODEL, billing="payg", context_window=1_000_000), + candidate("openai:fallback", quality_tier=5), + ], + ) + ) + sent = [] + replies = [upstream(CREDIT, 400), cached_upstream(MESSAGE)] + + def reply(**kwargs): + sent.append(kwargs["headers"].get("X-Api-Key")) + return replies.pop(0) + + with ( + patch.object(self.app_module.AuthService, "get_api_keys", return_value=["key-a", "key-b"]), + patch.object(ProxyService, "_make_base_request", side_effect=reply), + ): + response = self.post(routing={"allow_paid_overage": True}) + assert response.status_code == 200 + assert response.json["model"] == HAIKU_INTELLIGENCE_MODEL + assert response.json["multillm"]["attempts"] == 2 + assert sent == ["key-a", "key-b"] diff --git a/tests/test_anthropic_worker.mjs b/tests/test_anthropic_worker.mjs index 6ae088ea..55714dee 100644 --- a/tests/test_anthropic_worker.mjs +++ b/tests/test_anthropic_worker.mjs @@ -5,10 +5,12 @@ import { collectContainerEnv } from "../worker/container-env.mjs"; import { isApiRequestPath } from "../worker/api-paths.mjs"; import { loadWorkerModule } from "./helpers/load_cloudflare_worker.mjs"; -test("container receives the Anthropic secret and provider limits", async () => { +test("container receives the Anthropic secrets, spare keys and provider limits", async () => { const { MultiLLMProxyContainer } = await loadWorkerModule(); const source = { ANTHROPIC_API_KEY: "anthropic-test-key", + ANTHROPIC_API_KEY_1: "anthropic-spare-test-key", + ANTHROPIC_API_KEY_12: "anthropic-later-spare-test-key", ANTHROPIC_WORKSPACE_ID: "workspace-test", ANTHROPIC_RATE_LIMIT_RPM: "50", ANTHROPIC_MAX_PROMPT_TOKENS: "1000000", @@ -22,6 +24,7 @@ test("container receives the Anthropic secret and provider limits", async () => assert.equal(container.envVars[name], source[name]); } assert.equal(forwarded.UNRELATED_SECRET, undefined); + assert.equal(collectContainerEnv({ ANTHROPIC_API_KEY_SPARE: "not-a-numbered-spare" }).ANTHROPIC_API_KEY_SPARE, undefined); }); test("Anthropic native routes use edge API handling", () => { diff --git a/tests/test_haiku_intelligence.py b/tests/test_haiku_intelligence.py index a6e57682..687b71e4 100644 --- a/tests/test_haiku_intelligence.py +++ b/tests/test_haiku_intelligence.py @@ -205,6 +205,16 @@ def test_unconfigured_primary_uses_the_existing_fallback(self): ) ), ), + # Claude API keys form a pool, which the transport reads as a list. + patch.object( + self.app_module.AuthService, + "get_api_keys", + side_effect=( + lambda provider: ( + [] if provider == "anthropic" else ["synthetic-provider-key"] + ) + ), + ), self.requests(return_value=upstream(completion())) as send, ): response = self.haiku_post() diff --git a/worker/container-env.mjs b/worker/container-env.mjs index d18abb2a..e946a3aa 100644 --- a/worker/container-env.mjs +++ b/worker/container-env.mjs @@ -282,8 +282,9 @@ const DYNAMIC_ENV_PATTERNS = [ /^GROQ_API_KEY_\d+$/, /^NANOGPT_API_KEY_\d+$/, /^NANO_GPT_KEY_\d+$/, - // Codex Everywhere spares: _1, _2, ... (services/credential_pool.py). + // Codex Everywhere and Claude API spares: _1, _2, ... (services/credential_pool.py). /^CODEX_EVERYWHERE_API_KEY_[A-Z_]+_\d+$/, + /^ANTHROPIC_API_KEY_\d+$/, /^(?:ANTHROPIC|AZURE|CEREBRAS|CHUTES|CLINE_PASS|CODEX_EASY|GEMINI|GOOGLEAI|GROQ|HYPERBOLIC|KIMI_CODE|LINKAPI|MIMO|NANOGPT|NAVYAI|NINETEEN|OPENAI|OPENCODE|OPENCODE_GO|OPENROUTER|PALM|SAMBANOVA|SCALEWAY|TOGETHER|XAI)_(?:RATE_LIMIT_RPM|RATE_LIMIT_TPM|DAILY_REQUEST_LIMIT|MAX_REQUEST_BYTES|MAX_PROMPT_TOKENS|MAX_OUTPUT_TOKENS)$/, ];