diff --git a/.env.example b/.env.example index e79b7770..a063da56 100644 --- a/.env.example +++ b/.env.example @@ -72,6 +72,12 @@ ROUTSTR_SECRET_KEY= # UPSTREAM_POOL_TIMEOUT=5 # UPSTREAM_READ_TIMEOUT=900 +# Upstream Streaming Guards (0 disables; keep above reasoning models' think time) +# UPSTREAM_FIRST_TOKEN_TIMEOUT_SECONDS=0 +# UPSTREAM_STREAM_IDLE_TIMEOUT_SECONDS=0 +# UPSTREAM_ALLOWED_FAILS=3 +# UPSTREAM_COOLDOWN_SECONDS=30 + # Request and reservation lifetime limits (seconds) # STALE_RESERVATION_TIMEOUT_SECONDS=300 # MAX_REQUEST_LIFETIME_SECONDS=1800 diff --git a/docs/provider/configuration.md b/docs/provider/configuration.md index 6a47854d..20668e60 100644 --- a/docs/provider/configuration.md +++ b/docs/provider/configuration.md @@ -48,6 +48,29 @@ Connect to your AI provider(s): | **Upstream URL** | API endpoint (e.g., `https://api.openai.com/v1`) | | **API Key** | Your provider's API key | +### DeepSeek + +Choose **DeepSeek** as the provider type and paste an API key from +[platform.deepseek.com](https://platform.deepseek.com/api_keys); the base URL +is fixed to `https://api.deepseek.com`. Setting `DEEPSEEK_API_KEY` seeds the +provider on startup instead. + +Models are listed from DeepSeek's own `/models` and priced from a rate table +in `routstr/upstream/deepseek.py`, not from litellm or OpenRouter: + +- **Peak rates only.** DeepSeek charges half price off-peak, but the node bills + one flat price per model, so it bills the peak rate. Clients overpay + off-peak; the node never bills below cost. Time-of-day pricing is planned. +- **Unknown models import disabled.** A model DeepSeek lists that the table + does not price shows up disabled in the Admin Dashboard. Enable it with a + manual price, or add it to the table. +- **Cache hits** bill at DeepSeek's cache-hit rate (about 2% of the input + rate on flash, about 3% on pro). + +Thinking-mode `reasoning_content` is returned to clients unchanged in +responses, and forwarded unchanged when it appears in conversation history. +DeepSeek requires it on requests that carry `tools` and ignores it otherwise. + ### PPQ Auto Top-up PPQ providers can automatically purchase more credits when their USD balance diff --git a/pyproject.toml b/pyproject.toml index 7daf7489..bd6f4550 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -22,6 +22,7 @@ dependencies = [ "pillow>=10", "openai>=1.98.0", "litellm>=1.101.2,<1.102", + "backoff>=2.2", # litellm's native Anthropic-messages streaming (e.g. deepseek/) imports litellm.proxy, which needs it "orjson>=3.10", ] @@ -113,6 +114,9 @@ override-dependencies = [ # Transitive deps whose dependents allow the patched version but don't require # it. Constraints raise the floor without bypassing any upstream pin. constraint-dependencies = [ + "anyio>=4.14.2", + "pyjwt>=2.15.0", + "urllib3>=2.8.0", "starlette>=1.3.1", "httpcore>=1.0.9", # 1.0.8 caps h11<0.15 # 1.76 is the first grpcio-tools release with CPython 3.14 wheels. diff --git a/routstr/core/main.py b/routstr/core/main.py index 46aad1fc..368202cd 100644 --- a/routstr/core/main.py +++ b/routstr/core/main.py @@ -34,7 +34,6 @@ from ..payment.price import update_prices_periodically from ..proxy import initialize_upstreams, proxy_router, refresh_model_maps_periodically from ..refund import periodic_refund_reconcile from ..upstream.auto_topup import periodic_auto_topup -from ..upstream.deepseek_v4_pricing_shim import register_deepseek_v4_pricing from ..upstream.http_client import close_upstream_http_client from ..upstream.litellm_routing import configure_litellm from ..wallet import periodic_payout, periodic_refund_sweep, periodic_routstr_fee_payout @@ -90,11 +89,6 @@ async def lifespan(_: FastAPI) -> AsyncGenerator[None, None]: # debug logging) before any upstream provider dispatches a request. configure_litellm() - # TEMPORARY: backfill DeepSeek V4 pricing missing from litellm's cost - # map (BerriAI/litellm#30430). Remove this call and - # deepseek_v4_pricing_shim.py once litellm ships these models. - register_deepseek_v4_pricing() - # Run database migrations on startup run_migrations() diff --git a/routstr/core/settings.py b/routstr/core/settings.py index d9324f13..d77d6fc1 100644 --- a/routstr/core/settings.py +++ b/routstr/core/settings.py @@ -40,6 +40,22 @@ class Settings(BaseSettings): upstream_5xx_retry_attempts: int = Field( default=1, ge=0, env="UPSTREAM_5XX_RETRY_ATTEMPTS" ) + # Streaming guards, off by default (0). A stream that never produces a + # first chunk can still fail over; one that stalls later can only be + # aborted and billed for what it delivered. Reasoning models can stay + # silent for minutes, so set these above the longest expected think time. + upstream_first_token_timeout_seconds: float = Field( + default=0.0, ge=0, env="UPSTREAM_FIRST_TOKEN_TIMEOUT_SECONDS" + ) + upstream_stream_idle_timeout_seconds: float = Field( + default=0.0, ge=0, env="UPSTREAM_STREAM_IDLE_TIMEOUT_SECONDS" + ) + # Circuit breaker: timeouts/5xx per (provider, model) within a minute that + # take the pair out of candidate selection. 0 seconds disables it. + upstream_allowed_fails: int = Field(default=3, ge=1, env="UPSTREAM_ALLOWED_FAILS") + upstream_cooldown_seconds: float = Field( + default=30.0, ge=0, env="UPSTREAM_COOLDOWN_SECONDS" + ) # Node info name: str = Field(default="ARoutstrNode", env="NAME") diff --git a/routstr/proxy.py b/routstr/proxy.py index c585188f..a2cf9492 100644 --- a/routstr/proxy.py +++ b/routstr/proxy.py @@ -1,6 +1,7 @@ import asyncio import inspect import json +import re from typing import Any from fastapi import APIRouter, HTTPException, Request @@ -23,6 +24,7 @@ from .core.db import ( create_session, ) from .core.error_scope import ( + ERROR_SCOPE_HEADER, ERROR_SCOPE_UPSTREAM, UPSTREAM_ERROR_STATUS, UPSTREAM_UNAVAILABLE, @@ -40,6 +42,12 @@ from .payment.helpers import ( ) from .payment.models import Model from .upstream import BaseUpstreamProvider +from .upstream.cooldown import ( + candidate_model_identity, + is_cooling_down, + provider_identity, + record_failure, +) from .upstream.ehbp import forward_ehbp_request, forward_ehbp_x_cashu_request from .upstream.helpers import init_upstreams from .upstream.model_paths import ( @@ -116,8 +124,6 @@ def get_candidates( if candidates := _provider_map.get(model_id_lower): return candidates - import re - base_model_id = re.sub(r"-\d{8}$", "", model_id_lower) if base_model_id != model_id_lower: if candidates := _provider_map.get(base_model_id): @@ -271,6 +277,10 @@ _ALLOWED_ENDPOINTS: dict[str, frozenset[str]] = { "completions": frozenset({"POST"}), "responses": frozenset({"POST"}), "messages": frozenset({"POST"}), + # Anthropic token-counting subroute; the proxy's allowlist is exact, so the + # "messages" entry above does not carry it. Clients (Claude Code, the + # Anthropic SDKs) call it before every request. + "messages/count_tokens": frozenset({"POST"}), "embeddings": frozenset({"POST"}), # TypeSafe System One decision endpoint: POST {state, model, questions} # -> {answers, usage}. Non-streaming, JSON in/out; billed from the @@ -405,6 +415,18 @@ _RETRYABLE_UPSTREAM_5XX = frozenset({502, 503, 504}) _UPSTREAM_5XX_RETRY_BACKOFF_SECONDS = 0.5 +def _counts_toward_cooldown(status_code: int) -> bool: + """Provider faults and timeouts only — not client errors or rate limits.""" + return status_code >= 500 or status_code == UPSTREAM_ERROR_STATUS + + +def _upstream_response_failure(response: Response) -> bool: + return ( + _counts_toward_cooldown(response.status_code) + and response.headers.get(ERROR_SCOPE_HEADER) == ERROR_SCOPE_UPSTREAM + ) + + def _attribute_request( request: Request, model_obj: Model, upstream: BaseUpstreamProvider ) -> None: @@ -695,6 +717,20 @@ async def _proxy( request=request, ) + # A provider that just failed this model repeatedly is skipped while some + # other candidate can serve it. An explicit route is never rerouted. + if selector is None: + healthy = [ + candidate + for candidate in candidates + if not is_cooling_down( + provider_identity(candidate[1]), + candidate_model_identity(candidate[0], model_id), + ) + ] + if healthy: + candidates = healthy + # Reserve/max-cost checks use the best-ranked candidate; the failover loop # below rebinds (model_obj, upstream) per candidate so forwarding and # settlement always use the model of the provider actually being tried. @@ -722,7 +758,7 @@ async def _proxy( model_id, ) continue - return await forward_ehbp_x_cashu_request( + response = await forward_ehbp_x_cashu_request( request=request, x_cashu_token=x_cashu, path=path, @@ -731,7 +767,7 @@ async def _proxy( upstream=upstream, ) elif is_responses_api: - return await upstream.handle_x_cashu_responses( + response = await upstream.handle_x_cashu_responses( request, x_cashu, path, @@ -740,7 +776,7 @@ async def _proxy( request_body=request_body, ) else: - return await upstream.handle_x_cashu( + response = await upstream.handle_x_cashu( request, x_cashu, path, @@ -748,6 +784,12 @@ async def _proxy( model_obj, request_body=request_body, ) + if _upstream_response_failure(response): + record_failure( + provider_identity(upstream), + candidate_model_identity(model_obj, model_id), + ) + return response except UpstreamError as e: logger.warning( "Upstream %s failed (x-cashu) for model=%s: %s", @@ -760,6 +802,13 @@ async def _proxy( "status_code": e.status_code, }, ) + if e.scope == ERROR_SCOPE_UPSTREAM and _counts_toward_cooldown( + e.status_code + ): + record_failure( + provider_identity(upstream), + candidate_model_identity(model_obj, model_id), + ) if i == len(candidates) - 1: last_error = e continue @@ -1030,6 +1079,11 @@ async def _proxy( break if response.status_code != 200: + if _upstream_response_failure(response): + record_failure( + provider_identity(upstream), + candidate_model_identity(model_obj, model_id), + ) # 424 is an upstream failure re-reported by error_scope. # 502/503 are upstream errors, 429 rate limits. should_retry = response.status_code in [ @@ -1117,6 +1171,13 @@ async def _proxy( raise except UpstreamError as e: + if e.scope == ERROR_SCOPE_UPSTREAM and _counts_toward_cooldown( + e.status_code + ): + record_failure( + provider_identity(upstream), + candidate_model_identity(model_obj, model_id), + ) logger.warning( "Upstream %s failed for model=%s: %s", upstream.provider_type, diff --git a/routstr/upstream/__init__.py b/routstr/upstream/__init__.py index 85d094e8..c57e0c09 100644 --- a/routstr/upstream/__init__.py +++ b/routstr/upstream/__init__.py @@ -1,6 +1,7 @@ from .anthropic import AnthropicUpstreamProvider from .azure import AzureUpstreamProvider from .base import BaseUpstreamProvider +from .deepseek import DeepSeekUpstreamProvider from .fireworks import FireworksUpstreamProvider from .gemini import GeminiUpstreamProvider from .generic import GenericUpstreamProvider @@ -19,6 +20,7 @@ from .xai import XAIUpstreamProvider upstream_provider_classes: list[type[BaseUpstreamProvider]] = [ AnthropicUpstreamProvider, AzureUpstreamProvider, + DeepSeekUpstreamProvider, FireworksUpstreamProvider, GeminiUpstreamProvider, GenericUpstreamProvider, diff --git a/routstr/upstream/base.py b/routstr/upstream/base.py index 6b03f0f0..a815156a 100644 --- a/routstr/upstream/base.py +++ b/routstr/upstream/base.py @@ -35,6 +35,7 @@ from ..core.error_scope import ( ERROR_SCOPE_HEADER, ERROR_SCOPE_NODE, ERROR_SCOPE_UPSTREAM, + UPSTREAM_ERROR_STATUS, client_code_for_upstream_error, client_status_for_upstream_error, upstream_status_details, @@ -77,6 +78,7 @@ from .cache_breakpoints import ( inject_anthropic_cache_breakpoints, is_explicit_cache_model, ) +from .cooldown import model_identity, provider_identity, record_failure from .count_tokens import MissingUsageEstimator, count_tokens_locally from .http_client import acquire_upstream_http_client, build_x_cashu_client from .litellm_routing import detect_litellm_prefix @@ -94,6 +96,7 @@ from .stream_ownership import ( close_upstream_exchange, finalize_and_close_stream, ) +from .stream_timeout import GuardedStream, open_guarded_stream from .terminal_outcome_tracking import ( TerminalOutcomeState, event_usage_presence, @@ -1153,6 +1156,17 @@ class BaseUpstreamProvider: ) return True + async def _guard_stream( + self, response: httpx.Response, model_obj: Model | None, *, sse: bool + ) -> GuardedStream: + def on_idle() -> None: + if model_obj is not None and model_obj.id: + record_failure(provider_identity(self), model_identity(model_obj.id)) + + return await open_guarded_stream( + response, self.provider_type, sse=sse, on_idle_timeout=on_idle + ) + async def handle_streaming_chat_completion( self, response: httpx.Response, @@ -1175,6 +1189,8 @@ class BaseUpstreamProvider: Returns: StreamingResponse with cost data injected at the end """ + guarded_chunks = await self._guard_stream(response, model_obj, sse=True) + if reservation_snapshot is None: async with create_session() as snapshot_session: snapshot_key = await snapshot_session.get(key.__class__, key.hashed_key) @@ -1382,7 +1398,7 @@ class BaseUpstreamProvider: # multiple events can arrive together; buffering makes parsing # boundary-independent for every provider. splitter = SSEEventSplitter() - async for chunk in response.aiter_bytes(): + async for chunk in guarded_chunks: for raw_event in splitter.feed(chunk): for out in _process_event(raw_event): yield out @@ -1469,7 +1485,9 @@ class BaseUpstreamProvider: yield f"data: {json.dumps(usage_chunk_data)}\n\n".encode() - if done_seen: + if guarded_chunks.timed_out: + yield b'data: {"error":{"code":"UPSTREAM_TIMEOUT","message":"Upstream stream stalled"}}\n\n' + elif done_seen: yield b"data: [DONE]\n\n" except httpx.RemoteProtocolError as stream_error: @@ -1678,6 +1696,8 @@ class BaseUpstreamProvider: Returns: StreamingResponse with cost data injected at the end """ + guarded_chunks = await self._guard_stream(response, model_obj, sse=True) + usage_estimator = MissingUsageEstimator(request_body, model_obj) logger.debug( @@ -1834,7 +1854,7 @@ class BaseUpstreamProvider: # Buffer across network chunks; dispatch only on the SSE event # delimiter so parsing is independent of byte boundaries. splitter = SSEEventSplitter() - async for chunk in response.aiter_bytes(): + async for chunk in guarded_chunks: for raw_event in splitter.feed(chunk): for out in _process_event(raw_event): yield out @@ -1884,7 +1904,9 @@ class BaseUpstreamProvider: if usage_chunk_data is None: usage_chunk_data = { - "type": "response.completed", + "type": "response.failed" + if guarded_chunks.timed_out + else "response.completed", "provider": provider_seen, "response": { "model": last_model_seen or "unknown", @@ -1906,6 +1928,14 @@ class BaseUpstreamProvider: + cost_data.get("output_tokens", 0), }, } + if guarded_chunks.timed_out: + usage_chunk_data["type"] = "response.failed" + response_data = usage_chunk_data.get("response") + if isinstance(response_data, dict): + response_data["error"] = { + "code": "UPSTREAM_TIMEOUT", + "message": "Upstream stream stalled", + } try: self.inject_cost_metadata( @@ -1921,7 +1951,12 @@ class BaseUpstreamProvider: yield f"data: {json.dumps(usage_chunk_data)}\n\n".encode() - if done_seen: + if guarded_chunks.timed_out and ( + usage_chunk_data is None + or usage_chunk_data.get("type") != "response.failed" + ): + yield b'data: {"error":{"code":"UPSTREAM_TIMEOUT","message":"Upstream stream stalled"}}\n\n' + if done_seen and not guarded_chunks.timed_out: yield b"data: [DONE]\n\n" except httpx.RemoteProtocolError as stream_error: @@ -2170,6 +2205,7 @@ class BaseUpstreamProvider: provider_fee: float | None, reservation_snapshot: ReservationSnapshot, finalizer: PersistentStreamFinalizer | None = None, + guarded_chunks: GuardedStream | None = None, ) -> AsyncGenerator[bytes, None]: """Relay an opaque stream and settle it even if the caller disconnects.""" if finalizer is None: @@ -2187,12 +2223,22 @@ class BaseUpstreamProvider: ) ) try: - async for chunk in response.aiter_bytes(): + if guarded_chunks is None: + guarded_chunks = await self._guard_stream( + response, model_obj, sse=False + ) + async for chunk in guarded_chunks: yield chunk + if guarded_chunks.timed_out: + raise UpstreamError( + "Upstream stream stalled", + status_code=UPSTREAM_ERROR_STATUS, + code="UPSTREAM_TIMEOUT", + ) finally: await finalizer.run() - def _generic_streaming_response( + async def _generic_streaming_response( self, response: httpx.Response, key_hash: str, @@ -2203,6 +2249,7 @@ class BaseUpstreamProvider: reservation_snapshot: ReservationSnapshot, terminal_outcome: TerminalOutcomeContext | None = None, ) -> ClosingStreamingResponse: + guarded_chunks = await self._guard_stream(response, model_obj, sse=False) finalizer = PersistentStreamFinalizer( lambda: finalize_and_close_stream( lambda: self._finalize_generic_streaming_payment( @@ -2226,6 +2273,7 @@ class BaseUpstreamProvider: provider_fee, reservation_snapshot, finalizer, + guarded_chunks, ) return ClosingStreamingResponse( stream, @@ -2245,6 +2293,8 @@ class BaseUpstreamProvider: request_body: bytes | None = None, terminal_outcome: TerminalOutcomeContext | None = None, ) -> StreamingResponse: + guarded_chunks = await self._guard_stream(response, model_obj, sse=True) + usage_estimator = MissingUsageEstimator(request_body, model_obj) usage_finalized = False last_model_seen: str | None = None @@ -2344,7 +2394,7 @@ class BaseUpstreamProvider: total_cost = max(total_cost, _coerce_usd(usage_or_root.get(field))) try: - async for chunk in response.aiter_bytes(): + async for chunk in guarded_chunks: stored_chunks.append(chunk) terminal_sse_buffer = observe_terminal_sse_bytes( outcome_state, terminal_sse_buffer, chunk @@ -2535,6 +2585,8 @@ class BaseUpstreamProvider: maybe_cost_event = await finalize_without_usage() if maybe_cost_event is not None: yield maybe_cost_event + if guarded_chunks.timed_out: + yield b'event: error\ndata: {"error":{"code":"UPSTREAM_TIMEOUT","message":"Upstream stream stalled"}}\n\n' except httpx.ReadError: if not usage_finalized: @@ -3099,28 +3151,43 @@ class BaseUpstreamProvider: output_cost = 0.0 terminal_outcome = terminal_outcome_context(request_id, model_obj) - async for annotated in messages_dispatch.stream_annotated_events( - iterator, requested_model - ): - usage_presence = usage_presence.merged( - event_usage_presence(annotated.event) + try: + annotated_events = messages_dispatch.stream_annotated_events( + iterator, requested_model ) - if annotated.model: - last_model_seen = annotated.model - # See _stream_litellm_messages for why this is max() not +=. - input_tokens = max(input_tokens, annotated.input_tokens) - output_tokens = max(output_tokens, annotated.output_tokens) - cache_read_input_tokens = max( - cache_read_input_tokens, annotated.cache_read_input_tokens - ) - cache_creation_input_tokens = max( - cache_creation_input_tokens, - annotated.cache_creation_input_tokens, - ) - total_cost = max(total_cost, annotated.total_cost) - input_cost = max(input_cost, annotated.input_cost) - output_cost = max(output_cost, annotated.output_cost) - buffered.append(annotated) + async for annotated in annotated_events: + usage_presence = usage_presence.merged( + event_usage_presence(annotated.event) + ) + if annotated.model: + last_model_seen = annotated.model + # See _stream_litellm_messages for why this is max() not +=. + input_tokens = max(input_tokens, annotated.input_tokens) + output_tokens = max(output_tokens, annotated.output_tokens) + cache_read_input_tokens = max( + cache_read_input_tokens, annotated.cache_read_input_tokens + ) + cache_creation_input_tokens = max( + cache_creation_input_tokens, + annotated.cache_creation_input_tokens, + ) + total_cost = max(total_cost, annotated.total_cost) + input_cost = max(input_cost, annotated.input_cost) + output_cost = max(output_cost, annotated.output_cost) + buffered.append(annotated) + except Exception as exc: + # Buffering lets us return an HTTP error before sending headers. + if messages_dispatch.is_provider_exception(exc): + raise messages_dispatch.upstream_error_from_exception( + exc, + log_message="Upstream stream failed mid-flight", + log_extra={ + "model": last_model_seen or requested_model or "unknown", + "provider": self.provider_type or self.base_url, + "request_id": request_id, + }, + ) from exc + raise response_headers: dict[str, str] = { "Cache-Control": "no-cache", @@ -3560,7 +3627,7 @@ class BaseUpstreamProvider: }, ) - result = self._generic_streaming_response( + result = await self._generic_streaming_response( response, key.hashed_key, max_cost_for_model, @@ -3851,7 +3918,7 @@ class BaseUpstreamProvider: }, ) - result = self._generic_streaming_response( + result = await self._generic_streaming_response( response, key.hashed_key, max_cost_for_model, diff --git a/routstr/upstream/cooldown.py b/routstr/upstream/cooldown.py new file mode 100644 index 00000000..7031817c --- /dev/null +++ b/routstr/upstream/cooldown.py @@ -0,0 +1,79 @@ +"""In-memory circuit breaker for a failing (provider, model) pair. + +Process-local by design: each node observes its own upstream failures, and a +cooldown that outlives a restart would hide a provider that has recovered. +""" + +from __future__ import annotations + +import time +from typing import Any + +from ..core import get_logger +from ..core.settings import settings + +logger = get_logger(__name__) + +_FAILURE_WINDOW_SECONDS = 60.0 + +_failures: dict[tuple[str, str], list[float]] = {} +_cooling_until: dict[tuple[str, str], float] = {} + + +def provider_identity(upstream: Any) -> str: + db_id = getattr(upstream, "db_id", None) + if isinstance(db_id, int): + return f"db:{db_id}" + return f"{upstream.provider_type.lower()}|{upstream.base_url.lower()}" + + +def model_identity(model_id: str) -> str: + return model_id.lower() + + +def candidate_model_identity(model: Any, requested_model_id: str) -> str: + model_id = getattr(model, "id", None) + return model_identity( + model_id if isinstance(model_id, str) and model_id else requested_model_id + ) + + +def record_failure(provider_id: str, model_id: str) -> None: + """Count a timeout or 5xx, opening a cooldown once too many land in a minute.""" + if settings.upstream_cooldown_seconds <= 0: + return + + pair = (provider_id, model_id) + now = time.monotonic() + recent = [t for t in _failures.get(pair, []) if now - t < _FAILURE_WINDOW_SECONDS] + recent.append(now) + + if len(recent) >= settings.upstream_allowed_fails: + _failures.pop(pair, None) + _cooling_until[pair] = now + settings.upstream_cooldown_seconds + logger.warning( + "Upstream cooling down after repeated failures", + extra={ + "provider": provider_id, + "model": model_id, + "cooldown_seconds": settings.upstream_cooldown_seconds, + }, + ) + else: + _failures[pair] = recent + + +def is_cooling_down(provider_id: str, model_id: str) -> bool: + pair = (provider_id, model_id) + until = _cooling_until.get(pair) + if until is None: + return False + if time.monotonic() >= until: + del _cooling_until[pair] + return False + return True + + +def reset_cooldowns() -> None: + _failures.clear() + _cooling_until.clear() diff --git a/routstr/upstream/deepseek.py b/routstr/upstream/deepseek.py new file mode 100644 index 00000000..a4eed7b3 --- /dev/null +++ b/routstr/upstream/deepseek.py @@ -0,0 +1,106 @@ +"""First-class upstream for the DeepSeek API. + +Pricing comes from ``_PEAK_RATES`` below, not from litellm or OpenRouter: +litellm's bundled ``deepseek-v4-flash`` entry is stale (input, output and cache +rates alike), the OpenRouter feed carries resale prices below DeepSeek's own +peak rate, and neither the bundled map nor OpenRouter knows the current +``deepseek-flash`` id. A model DeepSeek lists that the table does not +cover is imported disabled rather than priced from those sources. + +DeepSeek bills peak hours at twice the off-peak rate. The node has one flat +price per model, so the table holds the PEAK rates: a client may overpay +off-peak but the node never bills below its own cost. + +Rates: https://api-docs.deepseek.com/quick_start/pricing (checked 2026-09-30). +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +from .base import BaseUpstreamProvider +from .generic import GenericUpstreamProvider +from .pricing_resolver import ResolvedPricing + +if TYPE_CHECKING: + from ..core.db import UpstreamProviderRow + +_CONTEXT_LENGTH = 1_000_000 +_MAX_OUTPUT_TOKENS = 384_000 + +# USD per 1M tokens at DeepSeek's peak rate: (input cache miss, output, input +# cache hit). DeepSeek has no cache-write charge. +_FLASH = (0.30, 1.20, 0.006) +_PRO = (1.32, 3.96, 0.044) + +_PEAK_RATES: dict[str, tuple[float, float, float]] = { + "deepseek-flash": _FLASH, + # Retired ids DeepSeek still accepts, served and billed as deepseek-flash. + "deepseek-v4-flash": _FLASH, + "deepseek-v4-flash-vision-exp": _FLASH, + "deepseek-v4-pro": _PRO, +} + +# Pro is the only current model without vision support. +_TEXT_ONLY = {"deepseek-v4-pro"} + + +class DeepSeekUpstreamProvider(GenericUpstreamProvider): + """Upstream provider specifically configured for the DeepSeek API.""" + + provider_type = "deepseek" + default_base_url = "https://api.deepseek.com" + platform_url = "https://platform.deepseek.com/api_keys" + litellm_provider_prefix = "deepseek/" + use_fallback_pricing = False + + def __init__(self, api_key: str, provider_fee: float = 1.01): + super().__init__( + base_url=self.default_base_url, + api_key=api_key, + provider_fee=provider_fee, + upstream_name="DeepSeek", + ) + + @classmethod + def _build_from_row( + cls, provider_row: "UpstreamProviderRow" + ) -> "DeepSeekUpstreamProvider": + return cls(api_key=provider_row.api_key, provider_fee=provider_row.provider_fee) + + @classmethod + def get_provider_metadata(cls) -> dict[str, object]: + return { + "id": cls.provider_type, + "name": "DeepSeek", + "default_base_url": cls.default_base_url, + "fixed_base_url": True, + "platform_url": cls.platform_url, + } + + def _apply_provider_field(self, response_json: object) -> None: + # A first-party upstream: stamp "deepseek", not Generic's hostname. + BaseUpstreamProvider._apply_provider_field(self, response_json) + + def transform_model_name(self, model_id: str) -> str: + """Strip the 'deepseek/' prefix for DeepSeek API compatibility.""" + return model_id.removeprefix("deepseek/") + + def _native_pricing( + self, model_id: str, model_spec: dict + ) -> ResolvedPricing | None: + """Price ``model_id`` from the peak-rate table; ``None`` if absent.""" + rates = _PEAK_RATES.get(model_id) + if rates is None: + return None + input_usd, output_usd, cache_hit_usd = rates + input_modalities = ["text"] if model_id in _TEXT_ONLY else ["text", "image"] + return ResolvedPricing( + prompt=input_usd / 1_000_000, + completion=output_usd / 1_000_000, + context_length=_CONTEXT_LENGTH, + source="native", + max_completion_tokens=_MAX_OUTPUT_TOKENS, + input_cache_read=cache_hit_usd / 1_000_000, + input_modalities=input_modalities, + ) diff --git a/routstr/upstream/deepseek_v4_pricing_shim.py b/routstr/upstream/deepseek_v4_pricing_shim.py deleted file mode 100644 index ba0c392d..00000000 --- a/routstr/upstream/deepseek_v4_pricing_shim.py +++ /dev/null @@ -1,73 +0,0 @@ -"""TEMPORARY: local DeepSeek V4 pricing shim. - -litellm's bundled cost map does not yet ship ``deepseek-v4-flash`` / -``deepseek-v4-pro``. Without an entry, ``backfill_cache_pricing`` cannot find a -``cache_read_input_token_cost`` and cache reads fall back to the full input -rate — a large overcharge on cache hits (DeepSeek V4 hits are ~0.008-0.02x -input, i.e. cached tokens cost 50-120x less than regular input). - -This module injects the missing entries into ``litellm.model_cost`` at startup -so the existing backfill path resolves them. Rates mirror the canonical -``deepseek`` provider entries now in litellm's ``model_prices`` map -(``input_cost_per_token`` is the cache-*miss* rate; -``cache_read_input_token_cost`` is the cache-*hit* rate), sourced from -https://api-docs.deepseek.com/quick_start/pricing via -https://github.com/BerriAI/litellm/pull/26380 (issue -https://github.com/BerriAI/litellm/issues/30430). - -=== REMOVAL (once litellm ships these models) === -Delete this file and the single ``register_deepseek_v4_pricing()`` call in -``routstr/core/main.py``. Nothing else depends on it. Entries are only added -when absent, so a stale shim is harmless after upstream lands — but remove it. -""" - -import litellm - -from ..core import get_logger - -logger = get_logger(__name__) - -# USD per token. Mirrors the canonical ``deepseek`` provider entries in -# litellm's model_prices map (source: DeepSeek API pricing docs). Keep these in -# sync with ``litellm.model_cost["deepseek/deepseek-v4-*"]``. -_DEEPSEEK_V4_RATES: dict[str, dict[str, float]] = { - "deepseek-v4-flash": { - "input_cost_per_token": 1.4e-07, - "output_cost_per_token": 2.8e-07, - "cache_read_input_token_cost": 2.8e-09, - "cache_creation_input_token_cost": 0.0, - "input_cost_per_token_cache_hit": 2.8e-09, - }, - "deepseek-v4-pro": { - "input_cost_per_token": 4.35e-07, - "output_cost_per_token": 8.7e-07, - "cache_read_input_token_cost": 3.625e-09, - "cache_creation_input_token_cost": 0.0, - "input_cost_per_token_cache_hit": 3.625e-09, - }, -} - - -def register_deepseek_v4_pricing() -> None: - """Inject DeepSeek V4 pricing into ``litellm.model_cost`` if absent. - - Idempotent and non-destructive: a key already present in the cost map - (e.g. once litellm ships it) is left untouched. Registers both the bare - (``deepseek-v4-flash``) and prefixed (``deepseek/deepseek-v4-flash``) - spellings since ``backfill_cache_pricing`` tries both. - """ - added = [] - for bare, rates in _DEEPSEEK_V4_RATES.items(): - for key in (bare, f"deepseek/{bare}"): - if key in litellm.model_cost: - continue - entry: dict[str, object] = dict(rates) - entry["litellm_provider"] = "deepseek" - entry["mode"] = "chat" - litellm.model_cost[key] = entry - added.append(key) - if added: - logger.info( - "Registered temporary DeepSeek V4 pricing shim", - extra={"models": added}, - ) diff --git a/routstr/upstream/generic.py b/routstr/upstream/generic.py index c9edf109..1032e85d 100644 --- a/routstr/upstream/generic.py +++ b/routstr/upstream/generic.py @@ -28,7 +28,11 @@ class GenericUpstreamProvider(BaseUpstreamProvider): provider_type = "generic" default_base_url = "http://localhost:8888" - platform_url = None + platform_url: str | None = None + # Subclasses that own an authoritative price table set this False so a model + # the table misses imports disabled instead of taking a litellm/OpenRouter + # price that may undercut the upstream's own rate. + use_fallback_pricing = True def __init__( self, @@ -162,7 +166,7 @@ class GenericUpstreamProvider(BaseUpstreamProvider): model_spec = model_data.get("model_spec", {}) resolved = self._native_pricing(model_id, model_spec) - if resolved is None: + if resolved is None and self.use_fallback_pricing: resolved = await resolver.resolve(model_id) if resolved is None: diff --git a/routstr/upstream/helpers.py b/routstr/upstream/helpers.py index dc544b3a..c94733bc 100644 --- a/routstr/upstream/helpers.py +++ b/routstr/upstream/helpers.py @@ -272,6 +272,7 @@ async def _seed_providers_from_settings( ("PERPLEXITY_API_KEY", "perplexity", None, None), ("FIREWORKS_API_KEY", "fireworks", None, None), ("XAI_API_KEY", "xai", None, None), + ("DEEPSEEK_API_KEY", "deepseek", None, None), ("TINFOIL_API_KEY", "tinfoil", None, None), ("TYPESAFE_API_KEY", "typesafe", None, None), ] diff --git a/routstr/upstream/messages_dispatch.py b/routstr/upstream/messages_dispatch.py index d9df0277..da7c6f9f 100644 --- a/routstr/upstream/messages_dispatch.py +++ b/routstr/upstream/messages_dispatch.py @@ -485,6 +485,76 @@ def compute_refund(amount: int, unit: str, cost_msats: int) -> int: raise ValueError(f"Invalid unit: {unit}") +_MAX_UPSTREAM_MESSAGE_CHARS = 300 + + +def collapse_litellm_message(message: str) -> str: + """Keep the innermost provider message and cap its length.""" + tail = message.rsplit("Original exception:", 1)[-1].strip() + while True: + stripped = tail.removeprefix("litellm.") + head, _, rest = stripped.partition(": ") + if rest and head.endswith(("Error", "Exception")): + stripped = rest.strip() + if stripped == tail: + break + tail = stripped + if len(tail) > _MAX_UPSTREAM_MESSAGE_CHARS: + tail = tail[: _MAX_UPSTREAM_MESSAGE_CHARS - 1].rstrip() + "…" + return tail + + +def is_provider_exception(exc: BaseException) -> bool: + """Distinguish SDK failures from bugs in our stream handling.""" + return type(exc).__module__.split(".", 1)[0] in {"litellm", "openai"} + + +def upstream_error_from_exception( + exc: Exception, + *, + log_message: str, + log_extra: dict[str, Any] | None = None, +) -> UpstreamError: + """Redact and classify provider failures, including mid-stream errors.""" + raw_message = getattr(exc, "message", None) or str(exc) or repr(exc) + # Redact provider account ids before the message reaches logs or the client. + exc_message = collapse_litellm_message(redact_org_ids(raw_message)) + exc_status = getattr(exc, "status_code", None) + exc_response = getattr(exc, "response", None) + response_text = None + if exc_response is not None: + try: + response_text = redact_org_ids( + getattr(exc_response, "text", str(exc_response)) + ) + except Exception: + response_text = "" + status_for_classify = exc_status if isinstance(exc_status, int) else 502 + rate_limit = classify_rate_limit( + status_for_classify, exc_message, getattr(exc, "headers", None) + ) + logger.error( + log_message, + extra={ + "error": exc_message, + "error_type": type(exc).__name__, + "status_code": exc_status, + "error_code": rate_limit.code if rate_limit else None, + "llm_provider": getattr(exc, "llm_provider", None), + "body": redact_org_ids(str(getattr(exc, "body", "") or "")) or None, + "response_text": response_text, + **(log_extra or {}), + }, + ) + return UpstreamError( + f"Upstream error via litellm: {exc_message}", + status_code=status_for_classify, + code=rate_limit.code if rate_limit else None, + details=rate_limit.as_details() if rate_limit else None, + from_upstream_response=True, + ) + + async def dispatch_anthropic_messages( *, request_body: bytes | None, @@ -606,44 +676,10 @@ async def dispatch_anthropic_messages( try: result = await litellm.anthropic.messages.acreate(**kwargs) except Exception as exc: - raw_message = getattr(exc, "message", None) or str(exc) or repr(exc) - # Redact provider account identifiers before the message reaches logs - # or the surfaced error. - exc_message = redact_org_ids(raw_message) - exc_status = getattr(exc, "status_code", None) - exc_response = getattr(exc, "response", None) - response_text = None - if exc_response is not None: - try: - response_text = redact_org_ids( - getattr(exc_response, "text", str(exc_response)) - ) - except Exception: - response_text = "" - status_for_classify = exc_status if isinstance(exc_status, int) else 502 - rate_limit = classify_rate_limit( - status_for_classify, exc_message, getattr(exc, "headers", None) - ) - logger.error( - "litellm dispatch failed", - extra={ - "error": exc_message, - "error_type": type(exc).__name__, - "status_code": exc_status, - "error_code": rate_limit.code if rate_limit else None, - "llm_provider": getattr(exc, "llm_provider", None), - "body": redact_org_ids(str(getattr(exc, "body", "") or "")) or None, - "response_text": response_text, - "model": litellm_model, - "api_base": base_url, - }, - ) - raise UpstreamError( - f"Upstream error via litellm: {exc_message}", - status_code=status_for_classify, - code=rate_limit.code if rate_limit else None, - details=rate_limit.as_details() if rate_limit else None, - from_upstream_response=True, + raise upstream_error_from_exception( + exc, + log_message="litellm dispatch failed", + log_extra={"model": litellm_model, "api_base": base_url}, ) from exc if transform_stream is not None and hasattr(result, "__aiter__"): @@ -661,6 +697,13 @@ async def dispatch_anthropic_messages( cast(AsyncIterator[Any], result) ) except Exception as exc: + if is_provider_exception(exc): + # Upstream failed part-way through, not an aggregation bug. + raise upstream_error_from_exception( + exc, + log_message="Upstream stream failed mid-flight", + log_extra={"model": litellm_model, "api_base": base_url}, + ) from exc logger.error( "Failed to aggregate streamed events into message", extra={ diff --git a/routstr/upstream/stream_timeout.py b/routstr/upstream/stream_timeout.py new file mode 100644 index 00000000..d6256da4 --- /dev/null +++ b/routstr/upstream/stream_timeout.py @@ -0,0 +1,117 @@ +"""Timeout guards for upstream streaming responses.""" + +from __future__ import annotations + +import asyncio +from collections.abc import AsyncIterator, Callable + +import httpx + +from ..core import get_logger +from ..core.error_scope import UPSTREAM_ERROR_STATUS +from ..core.exceptions import UpstreamError +from ..core.settings import settings +from .sse_splitter import SSEEventSplitter + +logger = get_logger(__name__) + + +class GuardedStream(AsyncIterator[bytes]): + def __init__( + self, + first: bytes | None, + chunks: AsyncIterator[bytes], + provider_type: str, + on_idle_timeout: Callable[[], None] | None, + ) -> None: + self.timed_out = False + self._chunks = self._resume(first, chunks, provider_type, on_idle_timeout) + + def __aiter__(self) -> GuardedStream: + return self + + async def __anext__(self) -> bytes: + return await anext(self._chunks) + + async def _resume( + self, + first: bytes | None, + chunks: AsyncIterator[bytes], + provider_type: str, + on_idle_timeout: Callable[[], None] | None, + ) -> AsyncIterator[bytes]: + chunk = first + while chunk is not None: + yield chunk + try: + chunk = await _next_chunk( + chunks, settings.upstream_stream_idle_timeout_seconds + ) + except TimeoutError: + self.timed_out = True + logger.warning( + "Upstream stream stalled; aborting and billing actual usage", + extra={ + "provider": provider_type, + "idle_timeout_seconds": settings.upstream_stream_idle_timeout_seconds, + }, + ) + if on_idle_timeout is not None: + on_idle_timeout() + return + + +def _has_data(event: bytes) -> bool: + return any( + line.startswith(b"data:") and line[5:].strip() for line in event.split(b"\n") + ) + + +async def _sse_events(chunks: AsyncIterator[bytes]) -> AsyncIterator[bytes]: + """Yield only deliverable SSE data events; comments cannot reset deadlines.""" + splitter = SSEEventSplitter() + async for chunk in chunks: + for event in splitter.feed(chunk): + if _has_data(event): + yield event + b"\n\n" + tail = splitter.flush() + if _has_data(tail): + # Keep an unterminated tail unterminated: the caller's final flush must + # not mistake truncated JSON for a complete SSE frame. + yield tail + + +async def open_guarded_stream( + response: httpx.Response, + provider_type: str, + *, + sse: bool = False, + on_idle_timeout: Callable[[], None] | None = None, +) -> GuardedStream: + """Prefetch a deliverable event before handing a response to the client. + + Once the first event is sent, a stall cannot fail over; the stream ends and + the caller's finalizer settles usage observed before the interruption. + """ + chunks = response.aiter_bytes().__aiter__() + guarded_chunks = _sse_events(chunks) if sse else chunks + timeout = settings.upstream_first_token_timeout_seconds + try: + first = await _next_chunk(guarded_chunks, timeout) + except TimeoutError: + await response.aclose() + raise UpstreamError( + f"Upstream {provider_type} sent no first chunk within {timeout}s", + status_code=UPSTREAM_ERROR_STATUS, + code="UPSTREAM_TIMEOUT", + ) from None + return GuardedStream(first, guarded_chunks, provider_type, on_idle_timeout) + + +async def _next_chunk(chunks: AsyncIterator[bytes], timeout: float) -> bytes | None: + """Next chunk, or ``None`` at end of stream. ``timeout <= 0`` disables it.""" + step = anext(chunks) + try: + return await (asyncio.wait_for(step, timeout) if timeout > 0 else step) + except StopAsyncIteration: + return None diff --git a/tests/conftest.py b/tests/conftest.py index d1bfa919..0e0ccbb3 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -31,3 +31,13 @@ def _isolate_redemption_negative_cache() -> Iterator[None]: redemption_negative_cache.clear() yield redemption_negative_cache.clear() + + +@pytest.fixture(autouse=True) +def _isolate_upstream_cooldowns() -> Iterator[None]: + """Clear process-wide upstream cooldowns so one test's failures can't skip providers in the next.""" + from routstr.upstream.cooldown import reset_cooldowns + + reset_cooldowns() + yield + reset_cooldowns() diff --git a/tests/integration/conftest.py b/tests/integration/conftest.py index f8babb17..badaf549 100644 --- a/tests/integration/conftest.py +++ b/tests/integration/conftest.py @@ -365,6 +365,10 @@ async def test_database_url(tmp_path: Any) -> str: @pytest_asyncio.fixture async def integration_engine(test_database_url: str) -> AsyncGenerator[Any, None]: """Create an async engine for integration tests""" + from routstr.core.settings import settings + + # Match the production engine's busy timeout; sqlite3's 5s default makes + # concurrency tests flake with "database is locked" on slow CI runners. engine = create_async_engine( test_database_url, echo=False, @@ -372,6 +376,7 @@ async def integration_engine(test_database_url: str) -> AsyncGenerator[Any, None pool_pre_ping=True, pool_size=5, max_overflow=10, + connect_args={"timeout": settings.database_busy_timeout}, ) # Initialize database schema diff --git a/tests/integration/test_secret_bootstrap.py b/tests/integration/test_secret_bootstrap.py index 12f0d9a1..6dcdb990 100644 --- a/tests/integration/test_secret_bootstrap.py +++ b/tests/integration/test_secret_bootstrap.py @@ -470,7 +470,6 @@ async def test_startup_runs_bootstrap_before_settings_initialize( return None monkeypatch.setattr(main, "configure_litellm", lambda: None) - monkeypatch.setattr(main, "register_deepseek_v4_pricing", lambda: None) monkeypatch.setattr(main, "run_migrations", lambda: None) monkeypatch.setattr(main, "init_db", noop_init_db) monkeypatch.setattr(main, "create_session", fake_create_session) diff --git a/tests/unit/test_cache_pricing.py b/tests/unit/test_cache_pricing.py index cc4a4dbc..cfb7e9a6 100644 --- a/tests/unit/test_cache_pricing.py +++ b/tests/unit/test_cache_pricing.py @@ -31,15 +31,6 @@ from routstr.payment.models import ( backfill_cache_pricing, ) from routstr.upstream import GenericUpstreamProvider -from routstr.upstream.deepseek_v4_pricing_shim import register_deepseek_v4_pricing - - -@pytest.fixture(autouse=True) -def _deepseek_v4_pricing() -> None: - # litellm's bundled cost map lacks the DeepSeek V4 entries (they only - # appear when its remote map is reachable); production injects them at - # startup via this same shim. - register_deepseek_v4_pricing() def _make_model(model_id: str, pricing: Pricing) -> Model: diff --git a/tests/unit/test_messages_upstream_errors.py b/tests/unit/test_messages_upstream_errors.py new file mode 100644 index 00000000..62699f69 --- /dev/null +++ b/tests/unit/test_messages_upstream_errors.py @@ -0,0 +1,152 @@ +import os +from typing import Any, AsyncIterator +from unittest.mock import AsyncMock, patch + +import litellm +import pytest +from litellm.exceptions import MidStreamFallbackError + +os.environ.setdefault("UPSTREAM_BASE_URL", "http://test") +os.environ.setdefault("UPSTREAM_API_KEY", "test") + +from routstr.core.exceptions import UpstreamError # noqa: E402 +from routstr.payment.models import Architecture, Model, Pricing # noqa: E402 +from routstr.upstream.base import BaseUpstreamProvider # noqa: E402 +from routstr.upstream.messages_dispatch import ( # noqa: E402 + collapse_litellm_message, +) + +_MIDSTREAM_FAILURE = MidStreamFallbackError( + message="No credits.", + model="x", + llm_provider="openai", + original_exception=litellm.APIError( + status_code=500, message="No credits.", llm_provider="openai", model="x" + ), +) + + +@pytest.mark.parametrize( + ("message", "expected"), + [ + ("You have no credits remaining.", "You have no credits remaining."), + # upstream_error_from_exception reads `.message`, which omits the + # "Original exception:" chain that only `str()` appends. + (_MIDSTREAM_FAILURE.message, "No credits."), + (str(_MIDSTREAM_FAILURE), "No credits."), + ("x" * 301, "x" * 299 + "…"), + ], +) +def test_collapse_litellm_message(message: str, expected: str) -> None: + assert collapse_litellm_message(message) == expected + + +_RATE_LIMIT = litellm.RateLimitError( + message=( + "Rate limit reached for gpt-4o on tokens per min (TPM): Limit 30000, " + "Used 29000, Requested 2000. Please try again in 1.2s." + ), + llm_provider="openai", + model="gpt-4o", +) +_BAD_REQUEST = litellm.BadRequestError( + message="context length exceeded", model="gpt-4o", llm_provider="openai" +) + +_MID_STREAM_CASES = [ + pytest.param(_RATE_LIMIT, 429, "UPSTREAM_RATE_LIMIT", id="rate-limit"), + pytest.param(_BAD_REQUEST, 400, None, id="bad-request"), + pytest.param(_MIDSTREAM_FAILURE, 500, None, id="midstream-fallback"), +] + + +def _make_model() -> Model: + return Model( + id="gpt-4o", + name="gpt-4o", + created=0, + description="", + context_length=8192, + architecture=Architecture( + modality="text", + input_modalities=["text"], + output_modalities=["text"], + tokenizer="x", + instruct_type=None, + ), + pricing=Pricing( + prompt=0.0, + completion=0.0, + request=0.0, + image=0.0, + web_search=0.0, + internal_reasoning=0.0, + max_cost=0.0, + ), + ) + + +def _failing_stream(exc: Exception) -> AsyncIterator[dict]: + async def gen() -> AsyncIterator[dict]: + yield { + "type": "message_start", + "message": {"id": "msg_1", "model": "gpt-4o", "usage": {}}, + } + raise exc + + return gen() + + +def _assert_upstream_error( + err: UpstreamError, status_code: int, code: str | None +) -> None: + assert err.status_code == status_code + assert err.code == code + assert err.from_upstream_response is True + assert "litellm." not in str(err) + + +@pytest.mark.asyncio +@pytest.mark.parametrize(("exc", "status_code", "code"), _MID_STREAM_CASES) +async def test_non_streaming_aggregation_surfaces_mid_stream_failure( + exc: Exception, status_code: int, code: str | None +) -> None: + async def fake_acreate(**kwargs: Any) -> AsyncIterator[dict]: + return _failing_stream(exc) + + with ( + patch( + "litellm.anthropic.messages.acreate", + new=AsyncMock(side_effect=fake_acreate), + ), + pytest.raises(UpstreamError) as exc_info, + ): + await BaseUpstreamProvider( + base_url="http://test", api_key="k" + )._dispatch_anthropic_messages( + request_body=b'{"messages": [], "max_tokens": 8, "stream": false}', + model_obj=_make_model(), + ) + + _assert_upstream_error(exc_info.value, status_code, code) + + +@pytest.mark.asyncio +@pytest.mark.parametrize(("exc", "status_code", "code"), _MID_STREAM_CASES) +async def test_x_cashu_buffered_stream_surfaces_mid_stream_failure( + exc: Exception, status_code: int, code: str | None +) -> None: + provider = BaseUpstreamProvider(base_url="http://test", api_key="k") + + with pytest.raises(UpstreamError) as exc_info: + await provider._stream_x_cashu_litellm_messages( + _failing_stream(exc), + amount=5_000, + unit="sat", + max_cost_for_model=10_000, + requested_model="gpt-4o", + mint=None, + request_id="req-test", + ) + + _assert_upstream_error(exc_info.value, status_code, code) diff --git a/tests/unit/test_proxy_path_allowlist.py b/tests/unit/test_proxy_path_allowlist.py index 4cd1c67f..1019dbad 100644 --- a/tests/unit/test_proxy_path_allowlist.py +++ b/tests/unit/test_proxy_path_allowlist.py @@ -51,6 +51,8 @@ def test_ambiguous_paths_are_rejected(path: str) -> None: "v1/chat/completions", "chat/completions", "v1/responses", + "v1/messages", + "v1/messages/count_tokens", "v1/embeddings", "models", "v1/models/gpt-4", @@ -138,6 +140,7 @@ def test_known_prefix_does_not_carry_an_unknown_endpoint(path: str) -> None: ("completions", "POST"), ("v1/responses", "POST"), ("v1/messages", "POST"), + ("v1/messages/count_tokens", "POST"), ("v1/embeddings", "POST"), ("models", "GET"), ("attestation", "GET"), @@ -163,6 +166,24 @@ def test_method_must_match_the_endpoint(path: str, method: str) -> None: assert _forwarding_allowed(path, method) is False +@pytest.mark.parametrize( + "path", + [ + "messages/count_tokens", + "v1/messages/count_tokens", + "v1/messages/count_tokens/", + ], +) +def test_count_tokens_endpoint_stays_allowed(path: str) -> None: + # Regression guard: /v1/messages/count_tokens is supported end-to-end + # (local handler when the upstream lacks native Anthropic support, plain + # forward otherwise), but the exact-match allowlist once omitted it, so + # Claude Code and the Anthropic SDKs were 404'd on every request. It must + # always be reachable, on POST only. + assert _forwarding_allowed(path, "POST") is True + assert _forwarding_allowed(path, "GET") is False + + def test_operator_additions_are_parsed_per_endpoint() -> None: parsed = _parse_extra_allowed_endpoints("POST:v1/rerank, GET:batches ,post:audio/x") assert parsed == { diff --git a/tests/unit/test_streaming_billing_finalization.py b/tests/unit/test_streaming_billing_finalization.py index 89714d28..8a0e82f5 100644 --- a/tests/unit/test_streaming_billing_finalization.py +++ b/tests/unit/test_streaming_billing_finalization.py @@ -364,7 +364,7 @@ async def test_aborted_generic_stream_still_records_its_settlement() -> None: patch("routstr.upstream.base.adjust_payment_for_tokens", adjust), patch("routstr.upstream.base.create_session", return_value=session_context), ): - response = provider._generic_streaming_response( + response = await provider._generic_streaming_response( upstream_response, "key-hash", 500, @@ -396,7 +396,7 @@ async def test_streaming_response_closes_iterator_when_downstream_send_is_cancel reservation = MagicMock(spec=ReservationSnapshot) upstream_response.status_code = 201 upstream_response.headers = {"x-upstream": "preserved"} - response = provider._generic_streaming_response( + response = await provider._generic_streaming_response( upstream_response, "key-hash", 500, @@ -458,7 +458,7 @@ async def test_generic_stream_settles_when_response_start_fails() -> None: upstream_response.status_code = 201 upstream_response.headers = {"x-upstream": "preserved"} reservation = MagicMock(spec=ReservationSnapshot) - response = provider._generic_streaming_response( + response = await provider._generic_streaming_response( upstream_response, "key-hash", 500, @@ -788,6 +788,11 @@ async def test_stream_closed_before_first_chunk_still_records_its_settlement( status_code=200, headers={"content-type": "text/event-stream"} ) upstream_response.aclose = AsyncMock() + + async def aiter_bytes() -> AsyncGenerator[bytes, None]: + yield b"data: {}\n\n" + + upstream_response.aiter_bytes = aiter_bytes key = MagicMock(spec=ApiKey) key.hashed_key = f"{api}-never-started" key.balance = 10_000 @@ -1198,12 +1203,12 @@ async def test_native_messages_error_event_still_records_its_settlement() -> Non True, {"input_tokens": 10, "output_tokens": 0}, {"output_tokens": 5}, - 3, + 15, (10, 5), ("reported", "reported"), ), (False, {}, {}, 3, (3, 0), ("estimated", "estimated")), - (True, {}, {"output_tokens": 5}, 3, (3, 5), ("estimated", "reported")), + (True, {}, {"output_tokens": 5}, 5, (0, 5), ("missing", "reported")), ], ) async def test_native_messages_stats_ignore_network_chunk_boundaries( @@ -1282,7 +1287,8 @@ async def test_native_messages_stats_ignore_network_chunk_boundaries( async for _ in response.body_iterator: pass - # Billing keeps its existing parser and charge; stats retain reported usage. + # The stream guard hands billing whole SSE events, so neither billing + # nor stats depend on where the network split them. async with sessions() as session: stored_key = await session.get(ApiKey, key.hashed_key) assert stored_key is not None diff --git a/tests/unit/test_upstream_deepseek.py b/tests/unit/test_upstream_deepseek.py new file mode 100644 index 00000000..96ab14f9 --- /dev/null +++ b/tests/unit/test_upstream_deepseek.py @@ -0,0 +1,297 @@ +"""Unit tests for ``DeepSeekUpstreamProvider``. + +DeepSeek is priced from the provider's own peak-rate table, never from litellm +or OpenRouter: litellm's ``deepseek-v4-flash`` entry is stale and OpenRouter +resells below DeepSeek's peak rate, so either would bill under cost. These +tests pin the table prices (including the cache-hit rate), that a model the +table misses imports disabled without consulting the fallback chain, and that +``reasoning_content`` in history reaches DeepSeek untouched — thinking mode +with ``tools`` answers 400 when it is stripped. +""" + +from __future__ import annotations + +import json +import threading +from collections.abc import Iterator +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from typing import Any +from unittest.mock import AsyncMock, Mock, patch + +import litellm +import pytest + +from routstr.upstream import upstream_provider_classes +from routstr.upstream.deepseek import DeepSeekUpstreamProvider + + +class _FakeResponse: + def __init__(self, payload: dict[str, Any]) -> None: + self._payload = payload + + def raise_for_status(self) -> None: + return None + + def json(self) -> dict[str, Any]: + return self._payload + + +class _FakeAsyncClient: + def __init__(self, payload: dict[str, Any], calls: list[dict[str, Any]]) -> None: + self._payload = payload + self._calls = calls + + async def __aenter__(self) -> "_FakeAsyncClient": + return self + + async def __aexit__(self, *exc: object) -> bool: + return False + + async def get( + self, url: str, headers: dict[str, str] | None = None + ) -> _FakeResponse: + self._calls.append({"url": url, "headers": headers}) + return _FakeResponse(self._payload) + + +# Shape of DeepSeek's ``GET /models``: bare ids, no pricing. +CATALOG: dict[str, Any] = { + "object": "list", + "data": [ + {"id": "deepseek-flash", "object": "model", "owned_by": "deepseek"}, + {"id": "deepseek-v4-pro", "object": "model", "owned_by": "deepseek"}, + {"id": "deepseek-v4-flash", "object": "model", "owned_by": "deepseek"}, + {"id": "deepseek-chat", "object": "model", "owned_by": "deepseek"}, + ], +} + + +async def _fetch( + catalog: dict[str, Any] = CATALOG, +) -> tuple[dict[str, Any], list[dict[str, Any]], AsyncMock]: + calls: list[dict[str, Any]] = [] + fallback = AsyncMock(return_value=None) + provider = DeepSeekUpstreamProvider(api_key="sk-test") + with ( + patch( + "routstr.upstream.generic.httpx.AsyncClient", + lambda *args, **kwargs: _FakeAsyncClient(catalog, calls), + ), + patch("routstr.upstream.generic.FallbackPricingResolver.resolve", fallback), + ): + models = await provider.fetch_models() + return {m.id: m for m in models}, calls, fallback + + +def test_metadata_and_registration() -> None: + assert DeepSeekUpstreamProvider in upstream_provider_classes + assert DeepSeekUpstreamProvider.get_provider_metadata() == { + "id": "deepseek", + "name": "DeepSeek", + "default_base_url": "https://api.deepseek.com", + "fixed_base_url": True, + "platform_url": "https://platform.deepseek.com/api_keys", + } + + +def test_build_from_row_ignores_row_base_url() -> None: + row = Mock( + api_key="sk-row", provider_fee=1.05, base_url="https://elsewhere.example" + ) + provider = DeepSeekUpstreamProvider._build_from_row(row) + assert provider.api_key == "sk-row" + assert provider.provider_fee == 1.05 + assert provider.base_url == "https://api.deepseek.com" + + +def test_litellm_prefix_is_deepseek() -> None: + provider = DeepSeekUpstreamProvider(api_key="sk-test") + assert provider.get_litellm_provider_prefix() == "deepseek/" + + +@pytest.mark.parametrize( + "model_id,expected", + [ + ("deepseek/deepseek-v4-flash", "deepseek-v4-flash"), + ("deepseek-v4-flash", "deepseek-v4-flash"), + ("deepseek/deepseek-flash", "deepseek-flash"), + ], +) +def test_transform_model_name(model_id: str, expected: str) -> None: + provider = DeepSeekUpstreamProvider(api_key="sk-test") + assert provider.transform_model_name(model_id) == expected + + +def test_provider_field_names_deepseek_not_host() -> None: + provider = DeepSeekUpstreamProvider(api_key="sk-test") + payload: dict[str, Any] = {"id": "chatcmpl-1"} + provider._apply_provider_field(payload) + assert payload["provider"] == "deepseek" + + +@pytest.mark.asyncio +async def test_fetch_models_calls_deepseek_models_endpoint_with_key() -> None: + _, calls, _ = await _fetch() + assert calls == [ + { + "url": "https://api.deepseek.com/models", + "headers": {"Authorization": "Bearer sk-test"}, + } + ] + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "model_id,prompt,completion,cache_read", + [ + ("deepseek-flash", 0.30, 1.20, 0.006), + # Retired alias DeepSeek serves and bills as deepseek-flash. + ("deepseek-v4-flash", 0.30, 1.20, 0.006), + ("deepseek-v4-pro", 1.32, 3.96, 0.044), + ], +) +async def test_table_models_priced_at_peak_rate( + model_id: str, prompt: float, completion: float, cache_read: float +) -> None: + models, _, _ = await _fetch() + model = models[model_id] + assert model.enabled is True + assert model.pricing.prompt == pytest.approx(prompt / 1_000_000) + assert model.pricing.completion == pytest.approx(completion / 1_000_000) + assert model.pricing.input_cache_read == pytest.approx(cache_read / 1_000_000) + assert model.context_length == 1_000_000 + + +@pytest.mark.asyncio +async def test_vision_follows_the_model() -> None: + models, _, _ = await _fetch() + assert "image" in models["deepseek-flash"].architecture.input_modalities + assert models["deepseek-v4-pro"].architecture.input_modalities == ["text"] + + +@pytest.mark.asyncio +async def test_unlisted_model_imports_disabled_without_fallback() -> None: + """litellm prices ``deepseek-chat``; the provider must not take that price.""" + models, _, fallback = await _fetch() + model = models["deepseek-chat"] + assert model.enabled is False + assert model.pricing.prompt == 0.0 + assert model.pricing.completion == 0.0 + fallback.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_cache_rate_survives_fee_and_is_not_replaced_by_litellm() -> None: + """litellm's stale ``deepseek-v4-flash`` cache rate (1.4e-08 in the bundled + map) must not replace the table's; backfill only fills an absent rate. The + fee applies to the cache rate like every other component. + + The litellm entry is pinned here because the remote cost map already + carries the table's rate, which would let an overwrite go unnoticed.""" + models, _, _ = await _fetch() + provider = DeepSeekUpstreamProvider(api_key="sk-test", provider_fee=1.05) + stale = {"cache_read_input_token_cost": 1.4e-08} + with patch("routstr.payment.models.litellm_cost_entry", return_value=stale): + priced = provider._apply_provider_fee_to_model(models["deepseek-v4-flash"]) + assert priced.pricing.input_cache_read == pytest.approx(0.006e-6 * 1.05) + assert priced.pricing.prompt == pytest.approx(0.30e-6 * 1.05) + # A cache hit costs 2% of a miss, not the full input rate. + assert priced.pricing.input_cache_read / priced.pricing.prompt == pytest.approx( + 0.02 + ) + + +@pytest.mark.asyncio +async def test_reasoning_content_in_history_reaches_upstream() -> None: + models, _, _ = await _fetch() + provider = DeepSeekUpstreamProvider(api_key="sk-test") + messages = [ + {"role": "user", "content": "weather in Paris?"}, + { + "role": "assistant", + "content": "", + "reasoning_content": "Need the weather tool.", + "tool_calls": [ + { + "id": "call_1", + "type": "function", + "function": {"name": "get_weather", "arguments": "{}"}, + } + ], + }, + {"role": "tool", "tool_call_id": "call_1", "content": "18C"}, + ] + body = json.dumps( + { + "model": "deepseek/deepseek-flash", + "messages": messages, + "tools": [{"type": "function", "function": {"name": "get_weather"}}], + } + ).encode() + out = provider.prepare_request_body(body, models["deepseek-flash"]) + + assert out is not None + sent = json.loads(out) + assert sent["model"] == "deepseek-flash" + assert sent["messages"] == messages + + +_ANTHROPIC_SSE = ( + b"event: message_start\n" + b'data: {"type":"message_start","message":{"id":"msg_1","type":"message",' + b'"role":"assistant","model":"deepseek-flash","content":[],' + b'"stop_reason":null,"usage":{"input_tokens":3,"output_tokens":0}}}\n\n' + b"event: message_stop\n" + b'data: {"type":"message_stop"}\n\n' +) + + +@pytest.fixture +def anthropic_stub() -> Iterator[tuple[str, list[tuple[str, dict[str, Any]]]]]: + """Loopback stand-in for DeepSeek's Anthropic-format endpoint.""" + seen: list[tuple[str, dict[str, Any]]] = [] + + class Handler(BaseHTTPRequestHandler): + def do_POST(self) -> None: + length = int(self.headers["Content-Length"]) + seen.append((self.path, json.loads(self.rfile.read(length)))) + self.send_response(200) + self.send_header("Content-Type", "text/event-stream") + self.send_header("Content-Length", str(len(_ANTHROPIC_SSE))) + self.end_headers() + self.wfile.write(_ANTHROPIC_SSE) + + def log_message(self, *args: Any) -> None: + return None + + server = ThreadingHTTPServer(("127.0.0.1", 0), Handler) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + try: + yield f"http://127.0.0.1:{server.server_address[1]}", seen + finally: + server.shutdown() + server.server_close() + + +@pytest.mark.asyncio +async def test_messages_stream_reaches_deepseek_anthropic_endpoint( + anthropic_stub: tuple[str, list[tuple[str, dict[str, Any]]]], +) -> None: + # litellm sends deepseek/ Messages calls to DeepSeek's /anthropic endpoint; + # its stream iterator imports litellm.proxy, which needs ``backoff``. + api_base, seen = anthropic_stub + stream = await litellm.anthropic.messages.acreate( + model=DeepSeekUpstreamProvider.litellm_provider_prefix + "deepseek-flash", + messages=[{"role": "user", "content": "hi"}], + max_tokens=8, + stream=True, + api_key="sk-test", + api_base=api_base, + ) + chunks = [chunk async for chunk in stream] # type: ignore[union-attr] + + assert b"message_stop" in b"".join(chunks) + assert len(seen) == 1 + assert seen[0][0] == "/anthropic/v1/messages" + assert seen[0][1]["model"] == "deepseek-flash" diff --git a/tests/unit/test_upstream_stream_timeout.py b/tests/unit/test_upstream_stream_timeout.py new file mode 100644 index 00000000..7942337c --- /dev/null +++ b/tests/unit/test_upstream_stream_timeout.py @@ -0,0 +1,506 @@ +"""First-token / idle stream guards and the per-(provider, model) cooldown.""" + +import asyncio +import json +from collections.abc import AsyncIterator +from typing import Any +from unittest.mock import AsyncMock, MagicMock, patch + +import httpx +import pytest + +from routstr.core.error_scope import ( + ERROR_SCOPE_HEADER, + ERROR_SCOPE_NODE, + ERROR_SCOPE_UPSTREAM, +) +from routstr.core.exceptions import UpstreamError +from routstr.core.settings import Settings, settings +from routstr.upstream.base import BaseUpstreamProvider +from routstr.upstream.cooldown import is_cooling_down, record_failure +from routstr.upstream.stream_timeout import open_guarded_stream + + +def _response(chunks: AsyncIterator[bytes]) -> MagicMock: + response = MagicMock(spec=httpx.Response) + response.aiter_bytes = MagicMock(return_value=chunks) + response.aclose = AsyncMock() + return response + + +async def _never() -> AsyncIterator[bytes]: + await asyncio.sleep(10) + yield b"late" + + +async def _stalls_after_first() -> AsyncIterator[bytes]: + yield b"first" + await asyncio.sleep(10) + yield b"never delivered" + + +async def _heartbeat_only(frame: bytes = b": keepalive\n\n") -> AsyncIterator[bytes]: + while True: + yield frame + await asyncio.sleep(0.002) + + +@pytest.fixture +def fast_timeouts(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(settings, "upstream_first_token_timeout_seconds", 0.01) + monkeypatch.setattr(settings, "upstream_stream_idle_timeout_seconds", 0.01) + + +@pytest.mark.asyncio +async def test_first_token_timeout_closes_response_and_raises( + fast_timeouts: None, +) -> None: + response = _response(_never()) + + with pytest.raises(UpstreamError) as exc_info: + await open_guarded_stream(response, "test") + + assert exc_info.value.code == "UPSTREAM_TIMEOUT" + assert exc_info.value.from_upstream_response is False + response.aclose.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_generic_stream_times_out_before_response_is_handed_off( + fast_timeouts: None, +) -> None: + provider = BaseUpstreamProvider(base_url="https://slow.example", api_key="test") + response = _response(_never()) + + with pytest.raises(UpstreamError, match="no first chunk"): + await provider._generic_streaming_response( + response, "key-hash", 100, "audio/speech", None, None, MagicMock() + ) + + response.aclose.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_generic_stream_idle_abort_settles_without_clean_completion( + fast_timeouts: None, +) -> None: + provider = BaseUpstreamProvider(base_url="https://slow.example", api_key="test") + finalize = AsyncMock() + provider._finalize_generic_streaming_payment = finalize # type: ignore[method-assign] + upstream = _response(_stalls_after_first()) + upstream.status_code = 200 + upstream.headers = {} + response = await provider._generic_streaming_response( + upstream, "key-hash", 100, "audio/speech", None, None, MagicMock() + ) + chunks = [] + with pytest.raises(UpstreamError, match="stream stalled"): + async for chunk in response.body_iterator: + chunks.append(chunk) + + assert chunks == [b"first"] + finalize.assert_awaited_once() + upstream.aclose.assert_awaited_once() + + +@pytest.mark.asyncio +@pytest.mark.parametrize("frame", [b": keepalive\n\n", b"data: \n\n"]) +async def test_sse_heartbeats_do_not_satisfy_first_token_timeout( + fast_timeouts: None, frame: bytes +) -> None: + response = _response(_heartbeat_only(frame)) + with pytest.raises(UpstreamError, match="no first chunk"): + await open_guarded_stream(response, "test", sse=True) + response.aclose.assert_awaited_once() + + +@pytest.mark.asyncio +@pytest.mark.parametrize("frame", [b": keepalive\n\n", b"data: \n\n"]) +async def test_sse_heartbeats_do_not_reset_idle_timeout( + fast_timeouts: None, frame: bytes +) -> None: + async def chunks() -> AsyncIterator[bytes]: + yield b'data: {"delta":"first"}\n\n' + async for chunk in _heartbeat_only(frame): + yield chunk + + failures = MagicMock() + stream = await open_guarded_stream( + _response(chunks()), "test", sse=True, on_idle_timeout=failures + ) + assert [chunk async for chunk in stream] == [b'data: {"delta":"first"}\n\n'] + assert stream.timed_out is True + failures.assert_called_once() + + +@pytest.mark.asyncio +async def test_zero_first_token_timeout_disables_the_guard( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr(settings, "upstream_first_token_timeout_seconds", 0) + monkeypatch.setattr(settings, "upstream_stream_idle_timeout_seconds", 0) + + async def _slow() -> AsyncIterator[bytes]: + await asyncio.sleep(0.02) + yield b"first" + + stream = await open_guarded_stream(_response(_slow()), "test") + + assert [chunk async for chunk in stream] == [b"first"] + + +def test_stream_guards_are_off_by_default() -> None: + # Reasoning models can think silently for minutes; on by default, the + # guards would fail requests that succeed without them. + fields = Settings.__fields__ + assert fields["upstream_first_token_timeout_seconds"].default == 0 + assert fields["upstream_stream_idle_timeout_seconds"].default == 0 + + +@pytest.mark.asyncio +async def test_idle_timeout_cools_down_the_serving_provider( + fast_timeouts: None, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setattr(settings, "upstream_allowed_fails", 1) + provider = BaseUpstreamProvider(base_url="https://slow.example", api_key="test") + provider.db_id = 17 + model = MagicMock(id="test-model") + guarded = await provider._guard_stream( + _response(_stalls_after_first()), model, sse=False + ) + + assert [chunk async for chunk in guarded] == [b"first"] + assert is_cooling_down("db:17", "test-model") + + +@pytest.mark.asyncio +@pytest.mark.parametrize("terminal_before_stall", [False, True]) +async def test_responses_idle_timeout_does_not_emit_completed( + fast_timeouts: None, terminal_before_stall: bool +) -> None: + async def chunks() -> AsyncIterator[bytes]: + event = ( + b'data: {"type":"response.completed","response":{"model":"test","usage":{"input_tokens":0,"output_tokens":1}}}\n\n' + if terminal_before_stall + else b'data: {"type":"response.created","response":{"model":"test"}}\n\n' + ) + yield event + await asyncio.sleep(10) + + response = _response(chunks()) + response.status_code = 200 + response.headers = {"content-type": "text/event-stream"} + key = MagicMock() + key.hashed_key = "test-key" + key.balance = 1000 + session = MagicMock() + session.get = AsyncMock(return_value=key) + session_context = MagicMock() + session_context.__aenter__ = AsyncMock(return_value=session) + session_context.__aexit__ = AsyncMock(return_value=None) + provider = BaseUpstreamProvider(base_url="https://slow.example", api_key="test") + + with ( + patch("routstr.upstream.base.create_session", return_value=session_context), + patch( + "routstr.upstream.base.adjust_payment_for_tokens", + AsyncMock(return_value={"input_tokens": 0, "output_tokens": 1}), + ), + ): + result = await provider.handle_streaming_responses_completion( + response, key, 100, reservation_snapshot=MagicMock() + ) + emitted = b"".join( + [ + chunk.encode() if isinstance(chunk, str) else bytes(chunk) + async for chunk in result.body_iterator + ] + ) + + assert b'"type": "response.failed"' in emitted + assert b'"code": "UPSTREAM_TIMEOUT"' in emitted + assert b'"type": "response.completed"' not in emitted + response.aclose.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_guarded_stream_passes_every_chunk_through() -> None: + async def _chunks() -> AsyncIterator[bytes]: + yield b"a" + yield b"b" + yield b"c" + + stream = await open_guarded_stream(_response(_chunks()), "test") + + assert [chunk async for chunk in stream] == [b"a", b"b", b"c"] + + +def test_cooldown_opens_after_allowed_fails_and_expires( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr(settings, "upstream_allowed_fails", 3) + monkeypatch.setattr(settings, "upstream_cooldown_seconds", 30) + + for _ in range(2): + record_failure("https://a.example", "m") + assert is_cooling_down("https://a.example", "m") is False + + record_failure("https://a.example", "m") + assert is_cooling_down("https://a.example", "m") is True + # Scoped to the exact pair. + assert is_cooling_down("https://b.example", "m") is False + assert is_cooling_down("https://a.example", "other") is False + + with patch("routstr.upstream.cooldown.time.monotonic", return_value=1e6): + assert is_cooling_down("https://a.example", "m") is False + + +def test_zero_cooldown_disables_skipping(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(settings, "upstream_allowed_fails", 1) + monkeypatch.setattr(settings, "upstream_cooldown_seconds", 0) + + record_failure("https://a.example", "m") + + assert is_cooling_down("https://a.example", "m") is False + + +def _proxy_request() -> MagicMock: + request = MagicMock() + request.method = "POST" + request.headers = {"authorization": "Bearer sk-key"} + request.body = AsyncMock(return_value=b'{"model": "test-model", "stream": true}') + request.state = MagicMock() + request.state.request_id = "req-1" + return request + + +def _upstream(base_url: str, forward: AsyncMock) -> MagicMock: + upstream = MagicMock() + upstream.provider_type = "test" + upstream.base_url = base_url + upstream.db_id = None + upstream.prepare_headers = MagicMock(side_effect=lambda h: h) + upstream.forward_request = forward + return upstream + + +async def _run_proxy( + candidates: list[tuple[MagicMock, MagicMock]], + revert_mock: AsyncMock, + request: MagicMock | None = None, +) -> Any: + from routstr import proxy as proxy_module + from routstr.auth import ReservationSnapshot + from routstr.core.db import ApiKey + + key = ApiKey(hashed_key="streamkey", balance=10_000) + reservation = ReservationSnapshot( + release_id="release", + key_hash=key.hashed_key, + billing_key_hash=key.hashed_key, + reserved_msats=1_000, + ) + + with ( + patch.object(proxy_module, "get_candidates", return_value=candidates), + patch.object( + proxy_module, "get_max_cost_for_model", AsyncMock(return_value=1_000) + ), + patch.object( + proxy_module, + "calculate_discounted_max_cost", + AsyncMock(return_value=1_000), + ), + patch.object(proxy_module, "check_token_balance", MagicMock()), + patch.object(proxy_module, "get_bearer_token_key", AsyncMock(return_value=key)), + patch.object( + proxy_module, "pay_for_request", AsyncMock(return_value=reservation) + ), + patch.object(proxy_module, "revert_pay_for_request", revert_mock), + ): + request = request or _proxy_request() + return await proxy_module._proxy( + request, "v1/chat/completions", MagicMock(), await request.body() + ) + + +@pytest.mark.asyncio +async def test_first_token_timeout_fails_over_to_the_next_candidate( + fast_timeouts: None, +) -> None: + async def _timing_out(*args: Any, **kwargs: Any) -> Any: + return await open_guarded_stream(_response(_never()), "test") + + served = MagicMock() + served.status_code = 200 + slow = _upstream("https://slow.example", AsyncMock(side_effect=_timing_out)) + fast = _upstream("https://fast.example", AsyncMock(return_value=served)) + revert_mock = AsyncMock(return_value=True) + + response = await _run_proxy([(MagicMock(), slow), (MagicMock(), fast)], revert_mock) + + assert response is served + fast.forward_request.assert_awaited_once() + # The reservation carries over to the candidate that served the request. + revert_mock.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_first_token_timeout_on_last_candidate_reverts_reservation( + fast_timeouts: None, +) -> None: + async def _timing_out(*args: Any, **kwargs: Any) -> Any: + return await open_guarded_stream(_response(_never()), "test") + + slow = _upstream("https://slow.example", AsyncMock(side_effect=_timing_out)) + revert_mock = AsyncMock(return_value=True) + + response = await _run_proxy([(MagicMock(), slow)], revert_mock) + + assert response.status_code == 424 + assert json.loads(bytes(response.body))["error"]["code"] == "UPSTREAM_TIMEOUT" + revert_mock.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_cooling_down_candidate_is_skipped_then_recovers( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr(settings, "upstream_allowed_fails", 1) + monkeypatch.setattr(settings, "upstream_cooldown_seconds", 30) + + sick_response = MagicMock() + sick_response.status_code = 200 + healthy_response = MagicMock() + healthy_response.status_code = 200 + sick = _upstream("https://sick.example", AsyncMock(return_value=sick_response)) + healthy = _upstream("https://ok.example", AsyncMock(return_value=healthy_response)) + candidates = [(MagicMock(), sick), (MagicMock(), healthy)] + + record_failure("test|https://sick.example", "test-model") + assert await _run_proxy(candidates, AsyncMock()) is healthy_response + sick.forward_request.assert_not_awaited() + + with patch("routstr.upstream.cooldown.time.monotonic", return_value=1e6): + assert await _run_proxy(candidates, AsyncMock()) is sick_response + + +@pytest.mark.asyncio +async def test_cooldown_never_empties_the_candidate_list( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr(settings, "upstream_allowed_fails", 1) + monkeypatch.setattr(settings, "upstream_cooldown_seconds", 30) + + only_response = MagicMock() + only_response.status_code = 200 + only = _upstream("https://only.example", AsyncMock(return_value=only_response)) + + record_failure("test|https://only.example", "test-model") + + assert await _run_proxy([(MagicMock(), only)], AsyncMock()) is only_response + + +@pytest.mark.asyncio +async def test_cooldown_distinguishes_credentials_at_same_url( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr(settings, "upstream_allowed_fails", 1) + bad = _upstream("https://same.example", AsyncMock()) + bad.db_id = 1 + good_response = MagicMock(status_code=200) + good = _upstream("https://same.example", AsyncMock(return_value=good_response)) + good.db_id = 2 + other = _upstream("https://other.example", AsyncMock()) + record_failure("db:1", "test-model") + + assert ( + await _run_proxy( + [(MagicMock(), bad), (MagicMock(), good), (MagicMock(), other)], AsyncMock() + ) + is good_response + ) + bad.forward_request.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_cooldown_normalizes_model_spelling( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr(settings, "upstream_allowed_fails", 1) + bad = _upstream("https://bad.example", AsyncMock()) + good_response = MagicMock(status_code=200) + good = _upstream("https://good.example", AsyncMock(return_value=good_response)) + record_failure("test|https://bad.example", "test-model") + request = _proxy_request() + request.body = AsyncMock( + return_value=b'{"model":"TEST-MODEL-20251222","stream":true}' + ) + + assert ( + await _run_proxy( + [(MagicMock(id="test-model"), bad), (MagicMock(id="test-model"), good)], + AsyncMock(), + request, + ) + is good_response + ) + bad.forward_request.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_x_cashu_upstream_failure_opens_cooldown( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr(settings, "upstream_allowed_fails", 1) + upstream = _upstream("https://cashu.example", AsyncMock()) + upstream.handle_x_cashu = AsyncMock( + return_value=MagicMock( + status_code=503, headers={ERROR_SCOPE_HEADER: ERROR_SCOPE_UPSTREAM} + ) + ) + request = _proxy_request() + request.headers = {"x-cashu": "token"} + + response = await _run_proxy([(MagicMock(), upstream)], AsyncMock(), request) + + assert response.status_code == 503 + assert is_cooling_down("test|https://cashu.example", "test-model") + + +@pytest.mark.asyncio +async def test_x_cashu_local_mint_failure_does_not_cool_provider( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr(settings, "upstream_allowed_fails", 1) + upstream = _upstream("https://cashu.example", AsyncMock()) + upstream.handle_x_cashu = AsyncMock( + return_value=MagicMock(status_code=503, headers={}) + ) + request = _proxy_request() + request.headers = {"x-cashu": "token"} + + response = await _run_proxy([(MagicMock(), upstream)], AsyncMock(), request) + + assert response.status_code == 503 + assert not is_cooling_down("test|https://cashu.example", "test-model") + + +@pytest.mark.asyncio +async def test_node_scoped_upstream_exception_does_not_cool_provider( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr(settings, "upstream_allowed_fails", 1) + upstream = _upstream( + "https://healthy.example", + AsyncMock( + side_effect=UpstreamError( + "local fault", status_code=500, scope=ERROR_SCOPE_NODE + ) + ), + ) + + response = await _run_proxy([(MagicMock(), upstream)], AsyncMock()) + + assert response.status_code == 500 + assert not is_cooling_down("test|https://healthy.example", "test-model") diff --git a/ui/package.json b/ui/package.json index 1d34f131..209c4fad 100644 --- a/ui/package.json +++ b/ui/package.json @@ -39,7 +39,7 @@ "@radix-ui/react-toggle-group": "^1.1.11", "@radix-ui/react-tooltip": "^1.2.8", "@tanstack/react-query": "^5.90.21", - "axios": "^1.16.0", + "axios": "^1.20.0", "class-variance-authority": "^0.7.1", "clsx": "^2.1.1", "cmdk": "^1.1.1", @@ -48,7 +48,7 @@ "geist": "^1.7.0", "input-otp": "^1.4.2", "lucide-react": "^0.575.0", - "next": "16.3.4", + "next": "16.3.6", "next-themes": "^0.4.6", "qrcode": "^1.5.4", "radix-ui": "^1.4.3", @@ -74,7 +74,7 @@ "@types/react": "^19.2.14", "@types/react-dom": "^19.2.3", "eslint": "^9.7.0", - "eslint-config-next": "16.3.4", + "eslint-config-next": "16.3.6", "eslint-config-prettier": "^10.1.8", "eslint-plugin-prettier": "^5.5.5", "eslint-plugin-react": "^7.37.5", diff --git a/ui/pnpm-lock.yaml b/ui/pnpm-lock.yaml index 097f3490..836a06f6 100644 --- a/ui/pnpm-lock.yaml +++ b/ui/pnpm-lock.yaml @@ -7,8 +7,8 @@ settings: overrides: '@babel/core': 7.29.6 ajv@6: 6.14.0 - brace-expansion@1: 1.1.18 - brace-expansion@5: 5.0.9 + brace-expansion@1: 1.1.21 + brace-expansion@5: 5.0.12 flatted: 3.4.2 follow-redirects: 1.16.0 form-data: 4.0.6 @@ -103,8 +103,8 @@ importers: specifier: ^5.90.21 version: 5.90.21(react@19.2.4) axios: - specifier: ^1.16.0 - version: 1.18.1 + specifier: ^1.20.0 + version: 1.20.0 class-variance-authority: specifier: ^0.7.1 version: 0.7.1 @@ -122,7 +122,7 @@ importers: version: 8.6.0(react@19.2.4) geist: specifier: ^1.7.0 - version: 1.7.0(next@16.3.4(@babel/core@7.29.6)(@types/node@25.4.0)(react-dom@19.2.4(react@19.2.4))(react@19.2.4)) + version: 1.7.0(next@16.3.6(@babel/core@7.29.6)(@types/node@25.4.0)(react-dom@19.2.4(react@19.2.4))(react@19.2.4)) input-otp: specifier: ^1.4.2 version: 1.4.2(react-dom@19.2.4(react@19.2.4))(react@19.2.4) @@ -130,8 +130,8 @@ importers: specifier: ^0.575.0 version: 0.575.0(react@19.2.4) next: - specifier: 16.3.4 - version: 16.3.4(@babel/core@7.29.6)(@types/node@25.4.0)(react-dom@19.2.4(react@19.2.4))(react@19.2.4) + specifier: 16.3.6 + version: 16.3.6(@babel/core@7.29.6)(@types/node@25.4.0)(react-dom@19.2.4(react@19.2.4))(react@19.2.4) next-themes: specifier: ^0.4.6 version: 0.4.6(react-dom@19.2.4(react@19.2.4))(react@19.2.4) @@ -203,8 +203,8 @@ importers: specifier: ^9.7.0 version: 9.38.0(jiti@2.6.1) eslint-config-next: - specifier: 16.3.4 - version: 16.3.4(@typescript-eslint/parser@8.57.0(eslint@9.38.0(jiti@2.6.1))(typescript@5.9.3))(eslint@9.38.0(jiti@2.6.1))(typescript@5.9.3) + specifier: 16.3.6 + version: 16.3.6(@typescript-eslint/parser@8.57.0(eslint@9.38.0(jiti@2.6.1))(typescript@5.9.3))(eslint@9.38.0(jiti@2.6.1))(typescript@5.9.3) eslint-config-prettier: specifier: ^10.1.8 version: 10.1.8(eslint@9.38.0(jiti@2.6.1)) @@ -580,56 +580,56 @@ packages: '@napi-rs/wasm-runtime@0.2.12': resolution: {integrity: sha512-ZVWUcfwY4E/yPitQJl481FjFo3K22D6qF0DuFH6Y/nbnE11GY5uguDxZMGXPQ8WQ0128MXQD7TnfHyK4oWoIJQ==} - '@next/env@16.3.4': - resolution: {integrity: sha512-cjWZnUUa6jZq2kFaNe/ZyJdZonOZ/QoN0Zka2nz/FLOrfx14pQuM9c5RaSVkWMqgdt4ksgPAMWPyHSs/CyV48Q==} + '@next/env@16.3.6': + resolution: {integrity: sha512-x9Vblze1EbtltQYnNH38xCPWU3TVfBd1eXqA3+w9+BTpedkkdNpAaltXlGQ/nsc1+E0mVTNrtcbX3GoO09zeLQ==} - '@next/eslint-plugin-next@16.3.4': - resolution: {integrity: sha512-szW9y2Aumu4z88YXfTzcFsgUAg2k64uzbtcO5L9f1AKS4w/GUKJcbFllRflROVyNPgJtGOnvNxiyp3v6b+prIA==} + '@next/eslint-plugin-next@16.3.6': + resolution: {integrity: sha512-jowwDX+7DOlDIjJLgTMxudw+k37QnWu1JkZLkSi9MaJBfDYcfhAPMKBhXL0idYzFN/AGg//axnOR4cLkHX/Rng==} - '@next/swc-darwin-arm64@16.3.4': - resolution: {integrity: sha512-iBr3I5LZNk5/bgl5//iTgD2tcym14MX0Xo7fD//u9dYAEgGzza1y9oywluPtf74YnOswVdH1908aK9xVz7zQTw==} + '@next/swc-darwin-arm64@16.3.6': + resolution: {integrity: sha512-E/7GEqaUkt8mk/T8v9lAnrhzR06kdq1ZBkC12F8tAMkdIadwNp3H1KqHynDHrpcTlGCUdq/qu6vUL2aYVyYBdw==} engines: {node: '>= 10'} cpu: [arm64] os: [darwin] - '@next/swc-darwin-x64@16.3.4': - resolution: {integrity: sha512-2dpiSyl2Jw/NrBPaU2MAKGSa+2MR82pJIn4Sm5Rjr+gxAeuh0z158Su3Z2O8zn7UNNq+ej4bToed6RcRN/Lydg==} + '@next/swc-darwin-x64@16.3.6': + resolution: {integrity: sha512-yBE893/nDWTlaiBD1p+qgt7NUen4U5R6FXyH0s67Npq1S3E0cVSef1WIXC2xBRgQvwAvJq6DnS6Y6PrY0cy4Ew==} engines: {node: '>= 10'} cpu: [x64] os: [darwin] - '@next/swc-linux-arm64-gnu@16.3.4': - resolution: {integrity: sha512-+t+U8HZT+fApePCS5h89CSH3datz29MkzyfCn+6fpsZBG/oiEOhINcb9rtkv6sdpToLGFn2e6146NzaKCXkqrA==} + '@next/swc-linux-arm64-gnu@16.3.6': + resolution: {integrity: sha512-KJDpjBqBPYlvkivmyrp+Qys6k/7ksbqGQvRVc6ZEGfR+cjQxx+nUkJaWmNZJsmoOrqYNbaXByF8wa0lBwDhB3Q==} engines: {node: '>= 10'} cpu: [arm64] os: [linux] - '@next/swc-linux-arm64-musl@16.3.4': - resolution: {integrity: sha512-mx03GNs1ocQA5JQ4FxDMmIsNkdrZh8cuezKCrId28e5/gIPU/l7Kcy2+vmCCzdjnnmXJy+iOAu+7K0QppO6Urg==} + '@next/swc-linux-arm64-musl@16.3.6': + resolution: {integrity: sha512-mqNg2K+hvWskSRb/QM+Ix412DvBsuSF0XV+frTSw5vmoucNnIlynFwKYew8D01bfATErMOM7Bujrf0BA5DRKFA==} engines: {node: '>= 10'} cpu: [arm64] os: [linux] - '@next/swc-linux-x64-gnu@16.3.4': - resolution: {integrity: sha512-YIhGY6fSMfha52bnVxnzc9zaVBzJg+cqQTOD8tXIBSx4fuv0pVMxQTE0PaS59YhnMOiYiG09IMwxJAf/CFm/Dw==} + '@next/swc-linux-x64-gnu@16.3.6': + resolution: {integrity: sha512-nFncBNGAYouRHjRVaITs9beZRfhX4ssVwpnvPIAbkZVH6LtGoAVlH4bJ8Cnf9SOo9bsXgPFer/GdHtEE3JNOkw==} engines: {node: '>= 10'} cpu: [x64] os: [linux] - '@next/swc-linux-x64-musl@16.3.4': - resolution: {integrity: sha512-+eaaX6axpDb0yF1GCpiERe6njplvdC+nks/fKfcHu3XPGRrald8P3/X7yv7QLdjA51knnxwl9pxdIJsg+w1L+Q==} + '@next/swc-linux-x64-musl@16.3.6': + resolution: {integrity: sha512-5Mf3cHDGR/Iz0ng2Bj3zUR3p5QS9YK3Hn2QiAfavFmyF48zwThAjpFoiTKNIcOHLYS4zEk+gzyJ/9deQ2ZB8yQ==} engines: {node: '>= 10'} cpu: [x64] os: [linux] - '@next/swc-win32-arm64-msvc@16.3.4': - resolution: {integrity: sha512-0jcXW7Xs/uzICrmgV3MhDYDeRy++1CqnpDIerlPIqYO4bhzB4WNbX/aRnQclustsAyTkFKB0z6rbcjmNg5tR8A==} + '@next/swc-win32-arm64-msvc@16.3.6': + resolution: {integrity: sha512-0jkJy0C2kbrJWTk4YLa3xk80pVBpx8FCHJym7CnUfDAXe/FWv5qT7SQJbR0KuemyxaEDlEx5WT4VQJoTW+/9Qw==} engines: {node: '>= 10'} cpu: [arm64] os: [win32] - '@next/swc-win32-x64-msvc@16.3.4': - resolution: {integrity: sha512-vvBzwu1pYQCp92maZCFCIw/XgOTMR5tur9GjakwIo2cmwRTMKajRZZDS9+e4KsUZWKu1E007WUeAFXRRjZeuzw==} + '@next/swc-win32-x64-msvc@16.3.6': + resolution: {integrity: sha512-/YXjI1e5OXcZ7YpxRwgP/1jAV/SBKTzeVKqN2mk7mLpcICsyn3Gl5+dIfDTJp70M0ccMhyMMRso4v6mPDCGepg==} engines: {node: '>= 10'} cpu: [x64] os: [win32] @@ -1842,8 +1842,8 @@ packages: resolution: {integrity: sha512-BASOg+YwO2C+346x3LZOeoovTIoTrRqEsqMa6fmfAV0P+U9mFr9NsyOEpiYvFjbc64NMrSswhV50WdXzdb/Z5A==} engines: {node: '>=4'} - axios@1.18.1: - resolution: {integrity: sha512-3nTvFlvpn9Zu/RkHUqtc7/+al4UpRW5az71ap5zccp6e8RAYEzhMTecX8Dz1wWDYrPpUoB1HAQEGEAEvUr7S9g==} + axios@1.20.0: + resolution: {integrity: sha512-r8aOh8j9cGKpgQAqpzrUHnSIc6a59Y3Xf/cv8sy1DrHCkZHzQGEuoq1tARk6qSyDdtQGSDgpb9kFlruzPvrgwg==} axobject-query@4.1.0: resolution: {integrity: sha512-qIj0G9wZbMGNLjLmg1PT6v2mE9AH2zlnADJD/2tC6E00hgmhUOfEB6greHPAfLRSufHqROIUTkw6E+M3lH0PTQ==} @@ -1861,11 +1861,11 @@ packages: engines: {node: '>=6.0.0'} hasBin: true - brace-expansion@1.1.18: - resolution: {integrity: sha512-Edep/X9fGqVNmzKBVsDYIOtD+z1tuezV70LBjdCst9Tqu76lsnvRiZ6oTic1n+/BIwX6QDGAO94PN4N2SADvtw==} + brace-expansion@1.1.21: + resolution: {integrity: sha512-9zeA+KLZNNzglF2TPKRQEDyx6Yby7daAkuy8MiPzpXPsYDWi/DRM8jmwUDxokQjYqBpv5DgPiwD4h4ZZSy1Ujw==} - brace-expansion@5.0.9: - resolution: {integrity: sha512-ScQ4IuvIEF1TMlP7Zt+vjJ//9zlPb2SDcxWxM3bk8s6t6GGdJ7KO1dCcTidOPJKePW30LE/2cT7wCyPho9/Wxg==} + brace-expansion@5.0.12: + resolution: {integrity: sha512-YovQ3rzhaLMIrDjNDMkNS01tea93qhEhG5xy8f6+R0l+dw3Ki+5sCoIoI942iuLZTHWogWktgwVDhU09iNEimQ==} engines: {node: 20 || >=22} braces@3.0.3: @@ -2142,8 +2142,8 @@ packages: resolution: {integrity: sha512-TtpcNJ3XAzx3Gq8sWRzJaVajRs0uVxA2YAkdb1jm2YkPz4G6egUFAyA3n5vtEIZefPk5Wa4UXbKuS5fKkJWdgA==} engines: {node: '>=10'} - eslint-config-next@16.3.4: - resolution: {integrity: sha512-35/8RM10huEL9vlr8hUZMERMENHBrnyHN3ZZkF9efSgzGaqK34jIqry44A956//zriUhUAUW0XSkcolhrryqAA==} + eslint-config-next@16.3.6: + resolution: {integrity: sha512-1Upt3U7BDwU+ilpe2byZjAfts9oNq4d4fv/zXEvs8/4yS+cwOQW/WCxUNy8gCDquX67SzeehDvKblVC6ZBMocQ==} peerDependencies: eslint: '>=9.0.0' typescript: '>=3.3.1' @@ -2821,8 +2821,8 @@ packages: react: ^16.8 || ^17 || ^18 || ^19 || ^19.0.0-rc react-dom: ^16.8 || ^17 || ^18 || ^19 || ^19.0.0-rc - next@16.3.4: - resolution: {integrity: sha512-/Ztf6CeRH+ejEXUrYtqI4gkS66eFIHuSwqi60RgcpWKodxFZx2/dqVCMKBwILfAHXQ+F1b1vAudgj3mnxqtoIA==} + next@16.3.6: + resolution: {integrity: sha512-L+otWM/aQbYTx98aZhgEoMb4bZAXx1YVW4UMA/vuCyCoWG5HJyZUili8QAkqzrcC+5///tsz3s0M+SlyB5bLMw==} engines: {node: '>=20.9.0'} hasBin: true peerDependencies: @@ -3899,37 +3899,37 @@ snapshots: '@tybys/wasm-util': 0.10.1 optional: true - '@next/env@16.3.4': {} + '@next/env@16.3.6': {} - '@next/eslint-plugin-next@16.3.4(eslint@9.38.0(jiti@2.6.1))': + '@next/eslint-plugin-next@16.3.6(eslint@9.38.0(jiti@2.6.1))': dependencies: '@eslint-community/eslint-utils': 4.9.1(eslint@9.38.0(jiti@2.6.1)) fast-glob: 3.3.1 transitivePeerDependencies: - eslint - '@next/swc-darwin-arm64@16.3.4': + '@next/swc-darwin-arm64@16.3.6': optional: true - '@next/swc-darwin-x64@16.3.4': + '@next/swc-darwin-x64@16.3.6': optional: true - '@next/swc-linux-arm64-gnu@16.3.4': + '@next/swc-linux-arm64-gnu@16.3.6': optional: true - '@next/swc-linux-arm64-musl@16.3.4': + '@next/swc-linux-arm64-musl@16.3.6': optional: true - '@next/swc-linux-x64-gnu@16.3.4': + '@next/swc-linux-x64-gnu@16.3.6': optional: true - '@next/swc-linux-x64-musl@16.3.4': + '@next/swc-linux-x64-musl@16.3.6': optional: true - '@next/swc-win32-arm64-msvc@16.3.4': + '@next/swc-win32-arm64-msvc@16.3.6': optional: true - '@next/swc-win32-x64-msvc@16.3.4': + '@next/swc-win32-x64-msvc@16.3.6': optional: true '@nodelib/fs.scandir@2.1.5': @@ -5168,7 +5168,7 @@ snapshots: axe-core@4.11.1: {} - axios@1.18.1: + axios@1.20.0: dependencies: follow-redirects: 1.16.0 form-data: 4.0.6 @@ -5186,12 +5186,12 @@ snapshots: baseline-browser-mapping@2.11.21: {} - brace-expansion@1.1.18: + brace-expansion@1.1.21: dependencies: balanced-match: 1.0.2 concat-map: 0.0.1 - brace-expansion@5.0.9: + brace-expansion@5.0.12: dependencies: balanced-match: 4.0.4 @@ -5576,9 +5576,9 @@ snapshots: escape-string-regexp@4.0.0: {} - eslint-config-next@16.3.4(@typescript-eslint/parser@8.57.0(eslint@9.38.0(jiti@2.6.1))(typescript@5.9.3))(eslint@9.38.0(jiti@2.6.1))(typescript@5.9.3): + eslint-config-next@16.3.6(@typescript-eslint/parser@8.57.0(eslint@9.38.0(jiti@2.6.1))(typescript@5.9.3))(eslint@9.38.0(jiti@2.6.1))(typescript@5.9.3): dependencies: - '@next/eslint-plugin-next': 16.3.4(eslint@9.38.0(jiti@2.6.1)) + '@next/eslint-plugin-next': 16.3.6(eslint@9.38.0(jiti@2.6.1)) eslint: 9.38.0(jiti@2.6.1) eslint-import-resolver-node: 0.3.9 eslint-import-resolver-typescript: 3.10.1(eslint-plugin-import@2.32.0)(eslint@9.38.0(jiti@2.6.1)) @@ -5872,9 +5872,9 @@ snapshots: functions-have-names@1.2.3: {} - geist@1.7.0(next@16.3.4(@babel/core@7.29.6)(@types/node@25.4.0)(react-dom@19.2.4(react@19.2.4))(react@19.2.4)): + geist@1.7.0(next@16.3.6(@babel/core@7.29.6)(@types/node@25.4.0)(react-dom@19.2.4(react@19.2.4))(react@19.2.4)): dependencies: - next: 16.3.4(@babel/core@7.29.6)(@types/node@25.4.0)(react-dom@19.2.4(react@19.2.4))(react@19.2.4) + next: 16.3.6(@babel/core@7.29.6)(@types/node@25.4.0)(react-dom@19.2.4(react@19.2.4))(react@19.2.4) generator-function@2.0.1: {} @@ -6263,11 +6263,11 @@ snapshots: minimatch@10.2.4: dependencies: - brace-expansion: 5.0.9 + brace-expansion: 5.0.12 minimatch@3.1.4: dependencies: - brace-expansion: 1.1.18 + brace-expansion: 1.1.21 minimist@1.2.8: {} @@ -6284,9 +6284,9 @@ snapshots: react: 19.2.4 react-dom: 19.2.4(react@19.2.4) - next@16.3.4(@babel/core@7.29.6)(@types/node@25.4.0)(react-dom@19.2.4(react@19.2.4))(react@19.2.4): + next@16.3.6(@babel/core@7.29.6)(@types/node@25.4.0)(react-dom@19.2.4(react@19.2.4))(react@19.2.4): dependencies: - '@next/env': 16.3.4 + '@next/env': 16.3.6 '@swc/helpers': 0.5.23 baseline-browser-mapping: 2.11.21 caniuse-lite: 1.0.30001810 @@ -6295,14 +6295,14 @@ snapshots: react-dom: 19.2.4(react@19.2.4) styled-jsx: 5.1.6(@babel/core@7.29.6)(react@19.2.4) optionalDependencies: - '@next/swc-darwin-arm64': 16.3.4 - '@next/swc-darwin-x64': 16.3.4 - '@next/swc-linux-arm64-gnu': 16.3.4 - '@next/swc-linux-arm64-musl': 16.3.4 - '@next/swc-linux-x64-gnu': 16.3.4 - '@next/swc-linux-x64-musl': 16.3.4 - '@next/swc-win32-arm64-msvc': 16.3.4 - '@next/swc-win32-x64-msvc': 16.3.4 + '@next/swc-darwin-arm64': 16.3.6 + '@next/swc-darwin-x64': 16.3.6 + '@next/swc-linux-arm64-gnu': 16.3.6 + '@next/swc-linux-arm64-musl': 16.3.6 + '@next/swc-linux-x64-gnu': 16.3.6 + '@next/swc-linux-x64-musl': 16.3.6 + '@next/swc-win32-arm64-msvc': 16.3.6 + '@next/swc-win32-x64-msvc': 16.3.6 sharp: 0.35.4(@types/node@25.4.0) transitivePeerDependencies: - '@babel/core' diff --git a/ui/pnpm-workspace.yaml b/ui/pnpm-workspace.yaml index 317f28e0..5eb49756 100644 --- a/ui/pnpm-workspace.yaml +++ b/ui/pnpm-workspace.yaml @@ -5,8 +5,8 @@ onlyBuiltDependencies: overrides: '@babel/core': 7.29.6 ajv@6: 6.14.0 - brace-expansion@1: 1.1.18 - brace-expansion@5: 5.0.9 + brace-expansion@1: 1.1.21 + brace-expansion@5: 5.0.12 flatted: 3.4.2 follow-redirects: 1.16.0 form-data: 4.0.6 diff --git a/uv.lock b/uv.lock index 86d2be5d..1764054b 100644 --- a/uv.lock +++ b/uv.lock @@ -9,10 +9,13 @@ resolution-markers = [ [manifest] constraints = [ + { name = "anyio", specifier = ">=4.14.2" }, { name = "grpcio", specifier = ">=1.76.0,<2.0.0" }, { name = "grpcio-tools", specifier = ">=1.76.0,<2.0.0" }, { name = "httpcore", specifier = ">=1.0.9" }, + { name = "pyjwt", specifier = ">=2.15.0" }, { name = "starlette", specifier = ">=1.3.1" }, + { name = "urllib3", specifier = ">=2.8.0" }, ] overrides = [ { name = "cryptography", specifier = ">=49.0.0" }, @@ -211,16 +214,15 @@ wheels = [ [[package]] name = "anyio" -version = "4.9.0" +version = "4.14.2" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "idna" }, - { name = "sniffio" }, { name = "typing-extensions", marker = "python_full_version < '3.13'" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/95/7d/4c1bd541d4dffa1b52bd83fb8527089e097a106fc90b467a7313b105f840/anyio-4.9.0.tar.gz", hash = "sha256:673c0c244e15788651a4ff38710fea9675823028a6f08a5eda409e0c9840a028", size = 190949, upload-time = "2025-03-17T00:02:54.77Z" } +sdist = { url = "https://files.pythonhosted.org/packages/61/cc/a381afa6efea9f496eff839d4a6a1aed3bfafc7b3ab4b0d1b243a12573dd/anyio-4.14.2.tar.gz", hash = "sha256:cfa139f3ed1a23ee8f88a145ddb5ac7605b8bbfd8592baacd7ce3d8bb4313c7f", size = 260176, upload-time = "2026-07-12T20:29:07.082Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/a1/ee/48ca1a7c89ffec8b6a0c5d02b89c305671d5ffd8d3c94acf8b8c408575bb/anyio-4.9.0-py3-none-any.whl", hash = "sha256:9f76d541cad6e36af7beb62e978876f3b41e3e04f2c1fbf0884604c0a9c4d93c", size = 100916, upload-time = "2025-03-17T00:02:52.713Z" }, + { url = "https://files.pythonhosted.org/packages/da/35/f2287558c17e29fafc8ef3daf819bb9834061cfa43bff8014f7df7f63bdc/anyio-4.14.2-py3-none-any.whl", hash = "sha256:9f505dda5ac9f0c8309b5e8bd445a8c2bf7246f3ce950121e45ea15bc41d1494", size = 125813, upload-time = "2026-07-12T20:29:05.763Z" }, ] [[package]] @@ -282,6 +284,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/77/06/bb80f5f86020c4551da315d78b3ab75e8228f89f0162f2c3a819e407941a/attrs-25.3.0-py3-none-any.whl", hash = "sha256:427318ce031701fea540783410126f03899a97ffc6f61596ad581ac2e40e3bc3", size = 63815, upload-time = "2025-03-13T11:10:21.14Z" }, ] +[[package]] +name = "backoff" +version = "2.2.1" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/47/d7/5bbeb12c44d7c4f2fb5b56abce497eb5ed9f34d85701de869acedd602619/backoff-2.2.1.tar.gz", hash = "sha256:03f829f5bb1923180821643f8753b0502c3b682293992485b0eef2807afa5cba", size = 17001, upload-time = "2022-10-05T19:19:32.061Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/df/73/b6e24bd22e6720ca8ee9a85a0c4a2971af8497d8f3193fa05390cbd46e09/backoff-2.2.1-py3-none-any.whl", hash = "sha256:63579f9a0628e06278f7e47b7d7d5b6ce20dc65c5e96a6f3ca99a6adca0396e8", size = 15148, upload-time = "2022-10-05T19:19:30.546Z" }, +] + [[package]] name = "base58" version = "2.1.1" @@ -2394,11 +2405,11 @@ wheels = [ [[package]] name = "pyjwt" -version = "2.13.0" +version = "2.15.1" source = { registry = "https://pypi.org/simple" } -sdist = { url = "https://files.pythonhosted.org/packages/3b/81/58d0ac84e1ef3a3843791d6954d94c0b33d526c75eeb1efbce9d0a4c4077/pyjwt-2.13.0.tar.gz", hash = "sha256:41571c89ca91598c79e8ef18a2d07367d4810fbbd6f637794879baf1b7703423", size = 107515, upload-time = "2026-05-21T19:54:36.618Z" } +sdist = { url = "https://files.pythonhosted.org/packages/43/ea/5194e52748b0da83d71e082d75496eaec6e58f419f5e184786ded517e6a9/pyjwt-2.15.1.tar.gz", hash = "sha256:4f259e80cdfb6b3fc18a7de51fd1ef9ec79652f25019bae68975ca2468a34df8", size = 121252, upload-time = "2026-09-28T18:40:42.598Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/a3/5e/ecf12fdb62546d64385c158514e9b2b671f7832108ef2ecd2020ce0af2d1/pyjwt-2.13.0-py3-none-any.whl", hash = "sha256:66adcc2aff09b3f1bbd95fc1e1577df8ac8723c978552fd43304c8a290ac5728", size = 31274, upload-time = "2026-05-21T19:54:35.362Z" }, + { url = "https://files.pythonhosted.org/packages/50/ca/44de4e75f8aadc457f0634be3b542815078ded46dca30efb960edeecad6e/pyjwt-2.15.1-py3-none-any.whl", hash = "sha256:42d59d631f7768a1028a64c7ff581a9bf7519804daf91fc5b6c56e30eec5e193", size = 33860, upload-time = "2026-09-28T18:40:41.429Z" }, ] [[package]] @@ -2711,6 +2722,7 @@ source = { editable = "." } dependencies = [ { name = "aiosqlite" }, { name = "alembic" }, + { name = "backoff" }, { name = "cashu" }, { name = "fastapi", extra = ["standard-no-fastapi-cloud-cli"] }, { name = "greenlet" }, @@ -2747,6 +2759,7 @@ dev = [ requires-dist = [ { name = "aiosqlite", specifier = ">=0.20" }, { name = "alembic", specifier = ">=1.13" }, + { name = "backoff", specifier = ">=2.2" }, { name = "cashu", specifier = ">=0.20" }, { name = "fastapi", extras = ["standard-no-fastapi-cloud-cli"], specifier = ">=0.141" }, { name = "greenlet", specifier = ">=3.2.1" }, @@ -3229,11 +3242,11 @@ wheels = [ [[package]] name = "urllib3" -version = "2.7.0" +version = "2.8.0" source = { registry = "https://pypi.org/simple" } -sdist = { url = "https://files.pythonhosted.org/packages/53/0c/06f8b233b8fd13b9e5ee11424ef85419ba0d8ba0b3138bf360be2ff56953/urllib3-2.7.0.tar.gz", hash = "sha256:231e0ec3b63ceb14667c67be60f2f2c40a518cb38b03af60abc813da26505f4c", size = 433602, upload-time = "2026-05-07T16:13:18.596Z" } +sdist = { url = "https://files.pythonhosted.org/packages/e3/05/b17359e1cefb4f909b5e40b1b90a496d987258916dbbf88e842c729f510e/urllib3-2.8.0.tar.gz", hash = "sha256:63bf2ead4c879426ebf22ef2a781eeb4aa3b4ae798a0435506f8687fd5bb9b63", size = 458972, upload-time = "2026-09-15T19:29:36.253Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/7f/3e/5db95bcf282c52709639744ca2a8b149baccf648e39c8cc87553df9eae0c/urllib3-2.7.0-py3-none-any.whl", hash = "sha256:9fb4c81ebbb1ce9531cce37674bbc6f1360472bc18ca9a553ede278ef7276897", size = 131087, upload-time = "2026-05-07T16:13:17.151Z" }, + { url = "https://files.pythonhosted.org/packages/92/9d/c4e665119135114480843e7ab388fa94d8480650450e6f8e26b70d323a4c/urllib3-2.8.0-py3-none-any.whl", hash = "sha256:0cf3cae568d36aa9576b28dfb35f11328f1cb974ca7647d9475ebb86c75ac6e3", size = 135717, upload-time = "2026-09-15T19:29:34.577Z" }, ] [[package]]