From cfb380ec8ad2fb49d99b7bcb40ef89d502861a49 Mon Sep 17 00:00:00 2001 From: Abhyudaya Baiju Tharakan Date: Mon, 26 Oct 2026 18:29:54 +0530 Subject: [PATCH 1/4] feat(anthropic): fail over to spare Claude API keys ANTHROPIC_API_KEY_1, _2, ... join ANTHROPIC_API_KEY in the credential pool. Requests use the first key that is not resting, so the spare only carries traffic after the primary is refused. The Claude API reports an empty credit balance and an owner-set spend limit as 400 invalid_request_error, and a tier spend cap as a 429 with enforced_spend_limit_reached. Raw dispatch now reads that small JSON error body, rests the key for an hour and marks the response. Unified routes, native /anthropic/v1/* routes and the auto:intelligence chain then send the request once on the next key. Other 400 errors still go straight back to the caller. The Worker forwards ANTHROPIC_API_KEY_ to the Container. A body read for classification is replayed from the cache, so relays still forward Anthropic's error. --- .env.example | 2 + docs/anthropic.md | 38 ++++ docs/deployment-cloudflare.md | 2 + providers/anthropic.py | 65 +++++++ routes/provider_credentials.py | 25 ++- routes/proxy.py | 20 ++ services/auth_primitives.py | 2 + services/credential_pool.py | 24 +-- services/intelligence_gateway.py | 4 +- services/intelligence_transport.py | 6 +- services/managed_turn.py | 1 + services/proxy_service.py | 23 ++- services/upstream_transport.py | 7 + tests/outbound_inventory.json | 22 +-- tests/test_anthropic_key_failover.py | 268 +++++++++++++++++++++++++++ tests/test_anthropic_worker.mjs | 5 +- worker/container-env.mjs | 3 +- 17 files changed, 482 insertions(+), 35 deletions(-) create mode 100644 tests/test_anthropic_key_failover.py 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..0b812621 100644 --- a/docs/anthropic.md +++ b/docs/anthropic.md @@ -18,6 +18,44 @@ 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. + +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/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..c8ed9ed4 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..a749d02f 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..8a734ea7 100644 --- a/services/credential_pool.py +++ b/services/credential_pool.py @@ -1,7 +1,7 @@ """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. @@ -30,6 +30,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", } @@ -264,21 +265,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, 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/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..2d150437 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": "7366df3b5e05343f762542a1e8d6e3157904298acc68399740c3b04847960d06" }, "routes/proxy.py": { "kind": "guarded", "reason": "Provider adapter, protocol helper or orchestration; delegates network I/O to the guarded provider clients.", - "sha256": "265e183e46865675ea3963f49546f650b3d0a04f98742f2dec63c7da862868a7" + "sha256": "7df1a655aa4704127f0c0064b949689bfecbb917d7d650148cf08b60cbc9a09e" }, "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": "d8c3ce5401d837d0341d9a5555942e36730cde29ce350354f2f008b4f47dd375" }, "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", @@ -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..47243fbc --- /dev/null +++ b/tests/test_anthropic_key_failover.py @@ -0,0 +1,268 @@ +"""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" + ) + + +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")) + return upstream(*pending.pop(0)) + + 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") + + 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 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/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)$/, ]; From 9789d6a2e1ef491547588fd3e562d8a62ecc3707 Mon Sep 17 00:00:00 2001 From: Abhyudaya Baiju Tharakan Date: Mon, 26 Oct 2026 18:37:29 +0530 Subject: [PATCH 2/4] test(anthropic): cover key failover for streams and production cooldown flags Rerun the route failover tests with scoped cooldowns, reset scheduling and prompt-cache affinity enabled, as production runs them, and add a streamed chat whose first key ran out of credit. The shared refusal check is now public, since the native route imports it. --- routes/provider_credentials.py | 4 +-- routes/proxy.py | 4 +-- tests/outbound_inventory.json | 4 +-- tests/test_anthropic_key_failover.py | 41 +++++++++++++++++++++++++++- 4 files changed, 46 insertions(+), 7 deletions(-) diff --git a/routes/provider_credentials.py b/routes/provider_credentials.py index c8ed9ed4..ae900640 100644 --- a/routes/provider_credentials.py +++ b/routes/provider_credentials.py @@ -65,7 +65,7 @@ def _is_nanogpt_routing_refusal(response) -> bool: ROTATING_PROVIDERS = frozenset({"nanogpt", "anthropic"}) -def _is_anthropic_key_refusal(response) -> bool: +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.""" @@ -162,7 +162,7 @@ def _request_with_provider_token_rotation( # would only surface an unrelated key's error to the caller. break if provider == "anthropic": - if not _is_anthropic_key_refusal(response): + if not is_anthropic_key_refusal(response): break elif provider != "nanogpt" or not is_nanogpt_credential_rejection( response.status_code diff --git a/routes/proxy.py b/routes/proxy.py index a749d02f..446b6901 100644 --- a/routes/proxy.py +++ b/routes/proxy.py @@ -51,7 +51,7 @@ stream_upstream_response, ) from routes.dashboard_chat import register_dashboard_chat_routes -from routes.provider_credentials import _is_anthropic_key_refusal +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 @@ -451,7 +451,7 @@ def send_to_url(target_url: str): # 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): + while is_anthropic_key_refusal(response): try: next_token = auth_service_cls.get_api_key(api_provider, **credential_scope) except (ModelCooldownExhausted, ModelCooldownCapacity): diff --git a/tests/outbound_inventory.json b/tests/outbound_inventory.json index 2d150437..56f62f53 100644 --- a/tests/outbound_inventory.json +++ b/tests/outbound_inventory.json @@ -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": "7366df3b5e05343f762542a1e8d6e3157904298acc68399740c3b04847960d06" + "sha256": "13d312e94c56ab52be80ddd480d83077faa3b1ae89b06c8046af1c72452c43ce" }, "routes/proxy.py": { "kind": "guarded", "reason": "Provider adapter, protocol helper or orchestration; delegates network I/O to the guarded provider clients.", - "sha256": "7df1a655aa4704127f0c0064b949689bfecbb917d7d650148cf08b60cbc9a09e" + "sha256": "60f84a4c5029694f6a5b8b9be3d54e03b08262b22c49c33203b12d8d04f8396f" }, "routes/public_pages.py": { "kind": "non-provider", diff --git a/tests/test_anthropic_key_failover.py b/tests/test_anthropic_key_failover.py index 47243fbc..c6c77dda 100644 --- a/tests/test_anthropic_key_failover.py +++ b/tests/test_anthropic_key_failover.py @@ -177,7 +177,8 @@ def upstreams(self, *replies): def reply(**kwargs): self.sent_keys.append(kwargs["headers"].get("X-Api-Key")) - return upstream(*pending.pop(0)) + item = pending.pop(0) + return item if isinstance(item, requests.Response) else upstream(*item) return patch.object(ProxyService, "_make_base_request", side_effect=reply) @@ -199,6 +200,21 @@ def test_unified_chat_moves_to_the_spare_when_credit_runs_out(self): self.assertEqual(self.chat().status_code, 200) self.assertEqual(self.sent_keys, ["key-b"], "the refused key rests") + 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)): @@ -229,6 +245,29 @@ def test_native_messages_relay_the_last_refusal_when_every_key_is_refused(self): 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 8ac554d3a56d753014a2b33b62240e20d33419b3 Mon Sep 17 00:00:00 2001 From: Abhyudaya Baiju Tharakan Date: Mon, 26 Oct 2026 18:53:08 +0530 Subject: [PATCH 3/4] test(haiku): leave the unconfigured primary without pooled keys Claude API keys now form a credential pool, so the intelligence transport reads them with get_api_keys. The fixture that marks the primary unconfigured patches that list too. --- tests/test_haiku_intelligence.py | 10 ++++++++++ 1 file changed, 10 insertions(+) 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() From a6a0192edec8f1a97d4c7dce0e85cf217e8c8ca4 Mon Sep 17 00:00:00 2001 From: Abhyudaya Baiju Tharakan Date: Mon, 26 Oct 2026 18:39:54 +0530 Subject: [PATCH 4/4] feat(usage): show each pooled key's state by slot name GET /admin/providers/usage now adds credential_slots for providers with a key pool, such as the Claude API and Codex Everywhere. Each configured key appears under its environment-variable name, never its value. Entries show whether the key is active or resting, plus this process's dispatch, error and refusal counts and its last refusal reason. Operators can see which Anthropic key is serving traffic and why a key moved aside. The counters are process-local and restart with the Container. The Claude Console stays the record of each key's spend. --- docs/anthropic.md | 3 ++ docs/provider-usage.md | 40 +++++++++++++++++ services/credential_pool.py | 65 +++++++++++++++++++++++++--- services/provider_usage_service.py | 8 +++- tests/outbound_inventory.json | 4 +- tests/test_anthropic_key_failover.py | 29 +++++++++++++ 6 files changed, 139 insertions(+), 10 deletions(-) diff --git a/docs/anthropic.md b/docs/anthropic.md index 0b812621..b0524083 100644 --- a/docs/anthropic.md +++ b/docs/anthropic.md @@ -47,6 +47,9 @@ 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. 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/services/credential_pool.py b/services/credential_pool.py index 8a734ea7..aa3df81a 100644 --- a/services/credential_pool.py +++ b/services/credential_pool.py @@ -4,16 +4,19 @@ `_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 @@ -42,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 [] @@ -60,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( @@ -198,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, @@ -293,3 +343,4 @@ def record_headers( def reset(cls) -> None: with cls._lock: cls._resting = {} + cls._observed = {} 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/tests/outbound_inventory.json b/tests/outbound_inventory.json index 56f62f53..72d3ac81 100644 --- a/tests/outbound_inventory.json +++ b/tests/outbound_inventory.json @@ -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": "d8c3ce5401d837d0341d9a5555942e36730cde29ce350354f2f008b4f47dd375" + "sha256": "4620e65913f7ef29aa514bc92251bc88f74f09082de9b73f1759409efec43f6d" }, "services/credits_admission.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", diff --git a/tests/test_anthropic_key_failover.py b/tests/test_anthropic_key_failover.py index c6c77dda..38133d1e 100644 --- a/tests/test_anthropic_key_failover.py +++ b/tests/test_anthropic_key_failover.py @@ -156,6 +156,30 @@ def send(body, status): ) +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() @@ -199,6 +223,11 @@ def test_unified_chat_moves_to_the_spare_when_credit_runs_out(self): 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