From c5da73f1e97bbd670c48ea48a5c81fee27b1677e Mon Sep 17 00:00:00 2001 From: 9qeklajc Date: Sun, 26 Jul 2026 20:12:12 +0200 Subject: [PATCH] revert: remove unrelated repository formatting --- routstr/algorithm.py | 20 +- routstr/balance.py | 29 +-- routstr/core/admin.py | 26 +- routstr/core/log_manager.py | 21 +- routstr/core/usage_analytics_store.py | 29 +-- routstr/nostr/analytics.py | 8 +- routstr/payment/cost_calculation.py | 19 +- routstr/payment/usage.py | 4 +- routstr/upstream/azure.py | 4 +- routstr/upstream/ehbp.py | 32 ++- routstr/upstream/gemini.py | 4 +- routstr/upstream/gemini_messages.py | 4 +- routstr/upstream/groq.py | 4 +- routstr/upstream/litellm_routing.py | 4 +- routstr/upstream/messages_dispatch.py | 12 +- routstr/upstream/model_paths.py | 322 +++++++++++++++---------- routstr/upstream/ollama.py | 8 +- routstr/upstream/rate_limit.py | 4 +- routstr/upstream/request_correction.py | 4 +- routstr/upstream/routstr.py | 3 +- routstr/upstream/xai.py | 4 +- routstr/wallet.py | 13 +- 22 files changed, 319 insertions(+), 259 deletions(-) diff --git a/routstr/algorithm.py b/routstr/algorithm.py index ef4b8574..fbc5388e 100644 --- a/routstr/algorithm.py +++ b/routstr/algorithm.py @@ -232,10 +232,7 @@ def create_model_mappings( aliases.append(prefixed_id) # Register forwarded_model_id as a routable alias - if ( - model_to_use.forwarded_model_id - and model_to_use.forwarded_model_id not in aliases - ): + if model_to_use.forwarded_model_id and model_to_use.forwarded_model_id not in aliases: aliases.append(model_to_use.forwarded_model_id) # Try to set each alias @@ -325,10 +322,7 @@ def create_model_mappings( aliases.append(prefixed_id) # Register forwarded_model_id as a routable alias - if ( - model_to_use.forwarded_model_id - and model_to_use.forwarded_model_id not in aliases - ): + if model_to_use.forwarded_model_id and model_to_use.forwarded_model_id not in aliases: aliases.append(model_to_use.forwarded_model_id) for alias in aliases: @@ -348,10 +342,16 @@ def create_model_mappings( forwarded_model_ids, the one whose forwarded_model_id equals the requested alias wins. """ - if model.forwarded_model_id and model.forwarded_model_id.lower() == alias: + if ( + model.forwarded_model_id + and model.forwarded_model_id.lower() == alias + ): return 5 - if model.id and model.id.lower() == alias: + if ( + model.id + and model.id.lower() == alias + ): return 4 model_base = get_base_model_id(model.id) diff --git a/routstr/balance.py b/routstr/balance.py index 03dc4d33..91b19ce5 100644 --- a/routstr/balance.py +++ b/routstr/balance.py @@ -260,11 +260,7 @@ async def _lookup_key_no_create( async def _restore_balance( - session: AsyncSession, - hashed_key: str, - balance: int, - reserved_balance: int, - mint_url: str, + session: AsyncSession, hashed_key: str, balance: int, reserved_balance: int, mint_url: str ) -> None: """Restore balance after a failed refund mint attempt.""" restore_stmt = ( @@ -279,11 +275,7 @@ async def _restore_balance( await session.commit() logger.info( "refund_wallet_endpoint: balance restored after mint failure", - extra={ - "hashed_key": hashed_key, - "restored_balance": balance, - "mint_url": mint_url, - }, + extra={"hashed_key": hashed_key, "restored_balance": balance, "mint_url": mint_url}, ) @@ -468,23 +460,11 @@ async def refund_wallet_endpoint( except HTTPException: # Minting failed — restore the debited balance - await _restore_balance( - session, - key.hashed_key, - pre_debit_balance, - pre_debit_reserved, - key.refund_mint_url or "", - ) + await _restore_balance(session, key.hashed_key, pre_debit_balance, pre_debit_reserved, key.refund_mint_url or "") raise except Exception as e: # Minting failed — restore the debited balance - await _restore_balance( - session, - key.hashed_key, - pre_debit_balance, - pre_debit_reserved, - key.refund_mint_url or "", - ) + await _restore_balance(session, key.hashed_key, pre_debit_balance, pre_debit_reserved, key.refund_mint_url or "") error_msg = str(e) logger.error( "refund_wallet_endpoint: mint/send failed", @@ -705,6 +685,7 @@ async def reset_child_key_spent( return {"success": True, "message": "Child key balance reset successfully."} + @router.api_route( "/{path:path}", methods=["GET", "POST", "PUT", "DELETE"], diff --git a/routstr/core/admin.py b/routstr/core/admin.py index 1521510e..66a1d288 100644 --- a/routstr/core/admin.py +++ b/routstr/core/admin.py @@ -68,9 +68,7 @@ async def require_admin_api(request: Request) -> None: async with create_session() as session: result = await session.exec(select(CliToken).where(CliToken.token == token)) cli_token = result.first() - if cli_token and ( - cli_token.expires_at is None or cli_token.expires_at > now_ts - ): + if cli_token and (cli_token.expires_at is None or cli_token.expires_at > now_ts): cli_token.last_used_at = now_ts session.add(cli_token) await session.commit() @@ -257,12 +255,16 @@ async def update_password(request: Request, password_update: PasswordUpdate) -> secret = await get_secret(session) if not secret.admin_password_hash: - raise HTTPException(status_code=500, detail="Admin password not configured") + raise HTTPException( + status_code=500, detail="Admin password not configured" + ) if not vault.verify_password( password_update.current_password, secret.admin_password_hash ): - raise HTTPException(status_code=401, detail="Current password is incorrect") + raise HTTPException( + status_code=401, detail="Current password is incorrect" + ) # Validate new password new_password = password_update.new_password.strip() @@ -978,7 +980,9 @@ async def update_upstream_provider_by_slug( lookup = _validate_slug(payload.slug) async with create_session() as session: result = await session.exec( - select(UpstreamProviderRow).where(UpstreamProviderRow.slug == lookup) + select(UpstreamProviderRow).where( + UpstreamProviderRow.slug == lookup + ) ) provider = result.first() if not provider: @@ -1665,11 +1669,7 @@ async def get_transactions_api( ) total = count_result.one() - stmt = ( - base.order_by(col(CashuTransaction.created_at).desc()) - .offset(offset) - .limit(limit) - ) + stmt = base.order_by(col(CashuTransaction.created_at).desc()).offset(offset).limit(limit) results = await session.exec(stmt) transactions = results.all() @@ -1679,7 +1679,9 @@ async def get_transactions_api( } -@admin_router.get("/api/lightning-invoices", dependencies=[Depends(require_admin_api)]) +@admin_router.get( + "/api/lightning-invoices", dependencies=[Depends(require_admin_api)] +) async def get_lightning_invoices_api( status: str | None = None, purpose: str | None = None, diff --git a/routstr/core/log_manager.py b/routstr/core/log_manager.py index b111f68a..0444dcbf 100644 --- a/routstr/core/log_manager.py +++ b/routstr/core/log_manager.py @@ -408,9 +408,7 @@ class LogManager: def get_error_details(self, hours: int = 24, limit: int = 100) -> dict: def compute() -> dict: try: - return self._usage_store.get_error_details( - hours_back=hours, limit=limit - ) + return self._usage_store.get_error_details(hours_back=hours, limit=limit) except Exception as e: logger.error( f"Usage analytics index failed, falling back to log scan: {e}" @@ -630,7 +628,8 @@ class LogManager: stats["total_tokens"] += input_tokens + output_tokens failed = ( - "upstream request failed" in message or "revert payment" in message + "upstream request failed" in message + or "revert payment" in message ) if failed: stats["total_requests"] += 1 @@ -788,9 +787,7 @@ class LogManager: if bucket_key: model_mix_buckets[bucket_key][model] += 1 if revenue_msats > 0: - model_mix_revenue_buckets[bucket_key][model] += ( - revenue_msats - ) + model_mix_revenue_buckets[bucket_key][model] += revenue_msats model_mix_revenue_totals[model] += revenue_msats if input_tokens > 0 or output_tokens > 0: token_total = input_tokens + output_tokens @@ -804,7 +801,8 @@ class LogManager: bucket["revenue_msats"] += revenue_msats failed = ( - "upstream request failed" in message or "revert payment" in message + "upstream request failed" in message + or "revert payment" in message ) if failed: summary_stats["total_requests"] += 1 @@ -874,7 +872,9 @@ class LogManager: models.sort(key=lambda x: float(x["net_revenue_sats"]), reverse=True) latest_errors = [ item - for _, item in sorted(latest_errors_heap, key=lambda x: x[0], reverse=True) + for _, item in sorted( + latest_errors_heap, key=lambda x: x[0], reverse=True + ) ] top_model_limit = max(1, min(model_limit, 20)) top_models_requests = [ @@ -1051,7 +1051,8 @@ class LogManager: bucket["warnings"] += 1 failed = ( - "upstream request failed" in message or "revert payment" in message + "upstream request failed" in message + or "revert payment" in message ) if failed: bucket["total_requests"] += 1 diff --git a/routstr/core/usage_analytics_store.py b/routstr/core/usage_analytics_store.py index 36fa4bcb..7ba90e24 100644 --- a/routstr/core/usage_analytics_store.py +++ b/routstr/core/usage_analytics_store.py @@ -314,7 +314,9 @@ class UsageAnalyticsStore: if column in existing_columns: return - conn.execute(f"ALTER TABLE {table} ADD COLUMN {column} {column_definition}") + conn.execute( + f"ALTER TABLE {table} ADD COLUMN {column} {column_definition}" + ) logger.info(f"Migrated analytics schema: added {table}.{column}") def _drop_index_tables_locked(self, conn: sqlite3.Connection) -> None: @@ -362,11 +364,7 @@ class UsageAnalyticsStore: self._drop_index_tables_locked(conn) self._initialize_schema_locked(conn) - files = ( - log_files - if log_files is not None - else sorted(self.logs_dir.glob("app_*.log")) - ) + files = log_files if log_files is not None else sorted(self.logs_dir.glob("app_*.log")) for log_file in files: try: self._process_log_file_locked(conn, log_file, force_full_read=True) @@ -570,7 +568,8 @@ class UsageAnalyticsStore: model_bucket["revenue_msats"] += revenue_msats failed = ( - "upstream request failed" in message or "revert payment" in message + "upstream request failed" in message + or "revert payment" in message ) if failed: bucket["total_requests"] += 1 @@ -593,9 +592,9 @@ class UsageAnalyticsStore: if isinstance(max_cost, (int, float)) and max_cost > 0: max_cost_float = float(max_cost) bucket["refunds_msats"] += max_cost_float - model_updates[(minute_key, model)]["refunds_msats"] += ( - max_cost_float - ) + model_updates[(minute_key, model)][ + "refunds_msats" + ] += max_cost_float return ( end_offset, @@ -1033,9 +1032,7 @@ class UsageAnalyticsStore: """, (cutoff_timestamp,), ).fetchone() - total_error_count = ( - int(total_error_count_row[0]) if total_error_count_row else 0 - ) + total_error_count = int(total_error_count_row[0]) if total_error_count_row else 0 return { "errors": [ @@ -1207,7 +1204,11 @@ class UsageAnalyticsStore: total_successful = int(row["total_successful"]) total_revenue_msats = float(row["total_revenue_msats"]) total_tokens = int(row["total_tokens"]) - if total_successful <= 0 and total_revenue_msats <= 0 and total_tokens <= 0: + if ( + total_successful <= 0 + and total_revenue_msats <= 0 + and total_tokens <= 0 + ): continue bucket_ts = str(row["bucket_ts"]) diff --git a/routstr/nostr/analytics.py b/routstr/nostr/analytics.py index 8b6da590..e568b5e0 100644 --- a/routstr/nostr/analytics.py +++ b/routstr/nostr/analytics.py @@ -215,9 +215,7 @@ def _build_window_payload( summary = dashboard.get("summary", {}) model_usage_mix = dashboard.get("model_usage_mix", {}) - summary_payload = _build_summary_payload( - summary if isinstance(summary, dict) else {} - ) + summary_payload = _build_summary_payload(summary if isinstance(summary, dict) else {}) usage_mix_payload = model_usage_mix if isinstance(model_usage_mix, dict) else {} top_model_usage, others_usage = _aggregate_top_model_usage(usage_mix_payload) @@ -340,9 +338,7 @@ async def publish_usage_analytics() -> None: nsec = (settings.nsec or "").strip() if not nsec: if not warned_missing_nsec: - logger.info( - "NSEC is not configured; skipping analytics sharing to Nostr" - ) + logger.info("NSEC is not configured; skipping analytics sharing to Nostr") warned_missing_nsec = True await asyncio.sleep(DISABLED_POLL_SECONDS) continue diff --git a/routstr/payment/cost_calculation.py b/routstr/payment/cost_calculation.py index e7cee8ca..37ac15d3 100644 --- a/routstr/payment/cost_calculation.py +++ b/routstr/payment/cost_calculation.py @@ -224,7 +224,9 @@ async def calculate_cost( "Token counts %s in the upstream response but cannot be " "priced; the request will appear in dashboards with the " "raw counts and a fixed max-cost charge.", - "are present" if (input_tokens > 0 or output_tokens > 0) else "are zero", + "are present" + if (input_tokens > 0 or output_tokens > 0) + else "are zero", extra={ "base_cost_msats": max_cost, "model": response_data.get("model", "unknown"), @@ -301,7 +303,9 @@ def _resolve_usd_cost(usage_data: dict, response_data: dict) -> float: # actually deducts from the balance. For non-BYOK providers (e.g. # OpenRouter) usage.cost already equals upstream_inference_cost, so we # fall through to the normal ``cost`` lookup below. - upstream_cost = _coerce_usd(cost_details.get("upstream_inference_cost")) + upstream_cost = _coerce_usd( + cost_details.get("upstream_inference_cost") + ) if upstream_cost > 0 and usage_data.get("is_byok"): byok_fee = _coerce_usd(usage_data.get("cost")) return upstream_cost + byok_fee @@ -332,7 +336,8 @@ def _get_pricing_rates( ``None`` means configured fixed pricing should be used by the caller. """ if settings.fixed_pricing and ( - settings.fixed_per_1k_input_tokens or settings.fixed_per_1k_output_tokens + settings.fixed_per_1k_input_tokens + or settings.fixed_per_1k_output_tokens ): return None @@ -388,8 +393,12 @@ def _get_pricing_rates( usd_per_sat = sats_usd_price() mspp_1k = input_usd * provider_fee * 1_000_000.0 / usd_per_sat mspc_1k = output_usd * provider_fee * 1_000_000.0 / usd_per_sat - cache_read_usd = _coerce_usd(pricing.get("cache_read_input_token_cost")) - cache_write_usd = _coerce_usd(pricing.get("cache_creation_input_token_cost")) + cache_read_usd = _coerce_usd( + pricing.get("cache_read_input_token_cost") + ) + cache_write_usd = _coerce_usd( + pricing.get("cache_creation_input_token_cost") + ) mscr_1k = ( cache_read_usd * provider_fee * 1_000_000.0 / usd_per_sat if cache_read_usd > 0 diff --git a/routstr/payment/usage.py b/routstr/payment/usage.py index 11d5c01e..02c90055 100644 --- a/routstr/payment/usage.py +++ b/routstr/payment/usage.py @@ -110,7 +110,9 @@ def normalize_usage(usage_data: object) -> NormalizedUsage | None: if not isinstance(usage_data, dict): return None - output_tokens = _first_token_count(usage_data, "completion_tokens", "output_tokens") + output_tokens = _first_token_count( + usage_data, "completion_tokens", "output_tokens" + ) cache_read, cache_write = _extract_cache_tokens(usage_data) # ``prompt_tokens`` is the inclusive grand total; ``input_tokens`` (Anthropic diff --git a/routstr/upstream/azure.py b/routstr/upstream/azure.py index 985bcfd2..a693b763 100644 --- a/routstr/upstream/azure.py +++ b/routstr/upstream/azure.py @@ -94,7 +94,9 @@ class AzureUpstreamProvider(BaseUpstreamProvider): deployment_id = deployment_id.split("/")[-1] return f"openai/deployments/{deployment_id}/{clean_path}" - def get_request_base_url(self, path: str, model_obj: "Model | None" = None) -> str: + def get_request_base_url( + self, path: str, model_obj: "Model | None" = None + ) -> str: """Use endpoint root, stripping accidental /openai/v1 suffix if present.""" base_url = self.base_url.rstrip("/") marker = "/openai/v1" diff --git a/routstr/upstream/ehbp.py b/routstr/upstream/ehbp.py index a3d1505a..96955492 100644 --- a/routstr/upstream/ehbp.py +++ b/routstr/upstream/ehbp.py @@ -191,9 +191,7 @@ def _resolve_ehbp_target_url( otherwise the header is ignored so callers cannot redirect other providers or leak upstream API keys. """ - override_header = ( - profile.client_target_url_header if profile else _ENCLAVE_URL_HEADER - ) + override_header = profile.client_target_url_header if profile else _ENCLAVE_URL_HEADER if not override_header: return target_url enclave_url = _get_header_case_insensitive(headers, override_header) @@ -297,7 +295,9 @@ def _build_cost_info( return result -def _inject_cost_response_headers(headers: dict[str, str], cost_info: dict) -> None: +def _inject_cost_response_headers( + headers: dict[str, str], cost_info: dict +) -> None: """Add per-request cost headers to an EHBP response. Since EHBP response bodies are opaque encrypted blobs, cost cannot be @@ -375,7 +375,9 @@ async def _compute_ehbp_actual_cost( resolved_upstream_model = ( actual_model_obj.forwarded_model_id or actual_model_obj.id ) - resolved_identity = _normalize_upstream_model_id(resolved_upstream_model) + resolved_identity = _normalize_upstream_model_id( + resolved_upstream_model + ) if resolved_identity != expected_identity: logger.info( "EHBP served model differs from requested, using actual " @@ -515,9 +517,7 @@ async def finalize_ehbp_actual_cost_payment( billing_key = await get_billing_key(key, session) key_hash = key.hashed_key billing_key_hash = billing_key.hashed_key - total_cost_msats = max( - 0, int(cost_info.get("total_msats", reserved_cost_for_model)) - ) + total_cost_msats = max(0, int(cost_info.get("total_msats", reserved_cost_for_model))) now = int(time.time()) safe_reserved = case( @@ -560,9 +560,7 @@ async def finalize_ehbp_actual_cost_payment( ) child_result = await session.exec(child_stmt) # type: ignore[call-overload] - if result.rowcount == 0 or ( - child_result is not None and child_result.rowcount == 0 - ): + if result.rowcount == 0 or (child_result is not None and child_result.rowcount == 0): await session.rollback() logger.error( "Failed to finalize EHBP usage-based payment", @@ -692,9 +690,7 @@ async def finalize_ehbp_max_cost_payment( else: child_result = None - if result.rowcount == 0 or ( - child_result is not None and child_result.rowcount == 0 - ): + if result.rowcount == 0 or (child_result is not None and child_result.rowcount == 0): await session.rollback() logger.error( "Failed to finalize EHBP max-cost payment", @@ -1038,9 +1034,7 @@ async def forward_ehbp_x_cashu_request( target_url = _resolve_ehbp_target_url( target.url, path, headers, provider_type, profile ) - upstream_headers = _prepare_ehbp_upstream_headers( - headers, target.headers, profile - ) + upstream_headers = _prepare_ehbp_upstream_headers(headers, target.headers, profile) request_body = await request.body() # Merge query params into the target URL @@ -1088,7 +1082,9 @@ async def forward_ehbp_x_cashu_request( usage_source = ( "header" if usage_header_name - and any(k.lower() == usage_header_name.lower() for k, _ in resp.headers) + and any( + k.lower() == usage_header_name.lower() for k, _ in resp.headers + ) else ("trailer" if usage_header else "none") ) diff --git a/routstr/upstream/gemini.py b/routstr/upstream/gemini.py index d58199c2..54de41a3 100644 --- a/routstr/upstream/gemini.py +++ b/routstr/upstream/gemini.py @@ -94,7 +94,9 @@ class GeminiUpstreamProvider(BaseUpstreamProvider): """ return self.base_url.rstrip("/").removesuffix("/openai") + "/openai" - def get_request_base_url(self, path: str, model_obj: "Model | None" = None) -> str: + def get_request_base_url( + self, path: str, model_obj: "Model | None" = None + ) -> str: """Route every proxied request to the OpenAI-compat surface. Required because the stored ``base_url`` typically points at the diff --git a/routstr/upstream/gemini_messages.py b/routstr/upstream/gemini_messages.py index a7e41c9b..11440b91 100644 --- a/routstr/upstream/gemini_messages.py +++ b/routstr/upstream/gemini_messages.py @@ -371,7 +371,9 @@ async def dispatch_gemini_messages( aggregates). """ if not request_body: - raise UpstreamError("Missing request body for /v1/messages", status_code=400) + raise UpstreamError( + "Missing request body for /v1/messages", status_code=400 + ) try: body: dict = json.loads(request_body) diff --git a/routstr/upstream/groq.py b/routstr/upstream/groq.py index a0c9475e..17103c35 100644 --- a/routstr/upstream/groq.py +++ b/routstr/upstream/groq.py @@ -20,9 +20,7 @@ class GroqUpstreamProvider(BaseUpstreamProvider): ) @classmethod - def _build_from_row( - cls, provider_row: "UpstreamProviderRow" - ) -> "GroqUpstreamProvider": + def _build_from_row(cls, provider_row: "UpstreamProviderRow") -> "GroqUpstreamProvider": return cls( api_key=provider_row.api_key, provider_fee=provider_row.provider_fee, diff --git a/routstr/upstream/litellm_routing.py b/routstr/upstream/litellm_routing.py index b7790394..0b2a92a3 100644 --- a/routstr/upstream/litellm_routing.py +++ b/routstr/upstream/litellm_routing.py @@ -91,7 +91,9 @@ OLLAMA_HOST_HINTS: tuple[str, ...] = ( ) -def detect_litellm_prefix(base_url: str | None, default: str = DEFAULT_PREFIX) -> str: +def detect_litellm_prefix( + base_url: str | None, default: str = DEFAULT_PREFIX +) -> str: """Return the litellm provider prefix (`"/"`) for `base_url`. Falls back to `default` when the host doesn't match any known provider. diff --git a/routstr/upstream/messages_dispatch.py b/routstr/upstream/messages_dispatch.py index efcf591b..3d689922 100644 --- a/routstr/upstream/messages_dispatch.py +++ b/routstr/upstream/messages_dispatch.py @@ -108,7 +108,9 @@ def parse_sse_blocks(buffer: bytes) -> tuple[list[dict], bytes]: return events, buffer -def events_from_chunk(chunk: object, sse_buffer: bytes) -> tuple[list[dict], bytes]: +def events_from_chunk( + chunk: object, sse_buffer: bytes +) -> tuple[list[dict], bytes]: """Normalize a stream chunk into one or more event dicts. ``litellm.anthropic.messages.acreate(stream=True)`` yields raw SSE @@ -199,7 +201,9 @@ async def aggregate_anthropic_events_to_message( raw_json = partial_json.pop(idx, None) if raw_json is not None and idx < len(blocks): try: - blocks[idx]["input"] = json.loads(raw_json) if raw_json else {} + blocks[idx]["input"] = ( + json.loads(raw_json) if raw_json else {} + ) except json.JSONDecodeError: blocks[idx]["input"] = raw_json elif etype == "message_delta": @@ -441,7 +445,9 @@ async def dispatch_anthropic_messages( on bad input or upstream failure. """ if not request_body: - raise UpstreamError("Missing request body for /v1/messages", status_code=400) + raise UpstreamError( + "Missing request body for /v1/messages", status_code=400 + ) try: body: dict = json.loads(request_body) diff --git a/routstr/upstream/model_paths.py b/routstr/upstream/model_paths.py index beaafa6a..f3bb9971 100644 --- a/routstr/upstream/model_paths.py +++ b/routstr/upstream/model_paths.py @@ -1,19 +1,15 @@ """Model-path discovery service. -Exposes every upstream provider path a Routstr model is reachable through. -This is discovery/visibility data only — routing still selects the cheapest or -best provider separately. +Exposes every selectable upstream route a Routstr model is reachable through. +This PR remains discovery-only: request-side routing will consume the opaque +selectors in a follow-up. -A *path* is the provider string that may appear in Routstr chat completion -responses. The strings emitted here are produced by the provider's own -``discovery_path_for_subprovider`` / ``discovery_base_paths`` hooks, which -mirror ``_apply_provider_field`` so discovery and response stamping cannot -drift: +A path is a standard percent-encoded query string containing the normalized +upstream URL and, for an exact OpenRouter endpoint, its machine-readable tag. +Display names never participate in identity:: -- Direct upstream -> ```` e.g. ``anthropic`` -- Generic/custom OpenRouter-compatible upstream -> ``generic:`` -- Native OpenRouter routing to a sub-provider -> ``openrouter:`` -- Native OpenRouter with no usable sub-provider -> ``unknown`` + url=https%3A%2F%2Fapi.anthropic.com%2Fv1 + url=https%3A%2F%2Fopenrouter.ai%2Fapi%2Fv1&provider=google-vertex%2Fus """ from __future__ import annotations @@ -21,10 +17,12 @@ from __future__ import annotations import asyncio import random import time -from typing import TYPE_CHECKING, Callable +from dataclasses import dataclass +from typing import TYPE_CHECKING, Any, Callable +from urllib.parse import urlencode import httpx -from sqlalchemy import insert, or_ +from sqlalchemy import insert from sqlalchemy.orm import selectinload from sqlmodel import col, delete, select @@ -51,6 +49,46 @@ _PERSIST_CHUNK_SIZE = 500 ModelKey = tuple[str, int] +@dataclass(frozen=True) +class EndpointIdentity: + """Exact OpenRouter endpoint identity returned by ``/endpoints``.""" + + tag: str + provider_name: str | None + + +@dataclass(frozen=True) +class DiscoveredPath: + """One model route ready for persistence and API serialization.""" + + model_id: str + path: str + upstream_url: str + provider_tag: str | None = None + provider_name: str | None = None + + +@dataclass(frozen=True) +class ProviderPathSnapshot: + """Refresh result plus model IDs whose prior rows must survive degradation.""" + + paths: tuple[DiscoveredPath, ...] + preserve_model_ids: frozenset[str] = frozenset() + + +def normalize_upstream_url(base_url: str) -> str: + """Normalize route identity without changing URL semantics.""" + return base_url.rstrip("/") + + +def encode_model_path(base_url: str, provider_tag: str | None = None) -> str: + """Encode a stable opaque selector for future request-side routing.""" + components = [("url", normalize_upstream_url(base_url))] + if provider_tag: + components.append(("provider", provider_tag)) + return urlencode(components) + + def _make_http_client() -> httpx.AsyncClient: """Client factory, separated so tests can substitute a mock transport.""" return httpx.AsyncClient() @@ -69,9 +107,15 @@ def is_openrouter_base_url(base_url: str | None) -> bool: def exposed_model_id(model: object) -> str: - """Client-visible ``/v1/models`` id for a cached model.""" + """Return exactly the ID advertised by ``/v1/models``. + + A forwarded ID is already a public routable alias and must remain intact, + including any slash. Without one, ``/v1/models`` exposes the base ID. + """ forwarded = getattr(model, "forwarded_model_id", None) - return forwarded or getattr(model, "id") + if forwarded: + return forwarded + return public_model_id(getattr(model, "id")) def public_model_id(model_id: str) -> str: @@ -116,7 +160,7 @@ class _RefreshCycleState: """ def __init__(self) -> None: - self.endpoint_cache: dict[tuple[str, str], list[str] | None] = {} + self.endpoint_cache: dict[tuple[str, str], list[EndpointIdentity] | None] = {} self.rate_limited = False @@ -127,8 +171,8 @@ async def _fetch_openrouter_endpoint_subproviders( author_slug: str, semaphore: asyncio.Semaphore, cycle: _RefreshCycleState, -) -> list[str] | None: - """Return sub-provider names for one model, or ``None`` when unknown. +) -> list[EndpointIdentity] | None: + """Return exact endpoint identities for one model, or ``None`` when unknown. ``None`` (not ``[]``) signals a degraded fetch — network failure, rate limit, non-200, or an unparseable payload — so callers can distinguish @@ -143,7 +187,7 @@ async def _fetch_openrouter_endpoint_subproviders( url = f"{base_url.rstrip('/')}/models/{author_slug}/endpoints" headers = {"Authorization": f"Bearer {api_key}"} if api_key else {} - result: list[str] | None + result: list[EndpointIdentity] | None async with semaphore: try: resp = await client.get( @@ -177,14 +221,24 @@ async def _fetch_openrouter_endpoint_subproviders( endpoints = resp.json().get("data", {}).get("endpoints", []) if not isinstance(endpoints, list): endpoints = [] - names: list[str] = [] + identities: dict[str, EndpointIdentity] = {} for endpoint in endpoints: - provider_name = ( - endpoint.get("provider_name") if isinstance(endpoint, dict) else None + if not isinstance(endpoint, dict): + continue + tag = endpoint.get("tag") + if not isinstance(tag, str) or not tag.strip(): + continue + provider_name = endpoint.get("provider_name") + identities.setdefault( + tag, + EndpointIdentity( + tag=tag, + provider_name=provider_name + if isinstance(provider_name, str) and provider_name + else None, + ), ) - if provider_name: - names.append(provider_name) - result = list(dict.fromkeys(names)) + result = list(identities.values()) except Exception as e: # noqa: BLE001 logger.warning( "OpenRouter endpoint discovery bad payload", @@ -287,46 +341,41 @@ async def _collect_provider_paths( overrides_by_key: dict[ModelKey, ModelRow] | None = None, disabled_model_keys: set[ModelKey] | None = None, cycle: _RefreshCycleState | None = None, -) -> list[tuple[str, str]] | None: - """Collect ``(model_id, path)`` pairs for one provider instance. +) -> ProviderPathSnapshot: + """Collect selectable routes while marking model-level degraded fetches. - Emits the provider's ``discovery_base_paths`` for normal upstreams. For - OpenRouter-compatible providers, additionally emits one path per OpenRouter - sub-provider endpoint via ``discovery_path_for_subprovider`` so the strings - match response stamping exactly. - - Returns ``None`` when the provider's path set could not be determined this - cycle (every endpoint fetch degraded); callers must then keep previously - persisted rows instead of wiping them. + A failed OpenRouter lookup preserves only that model's prior rows. Other + models in the same provider still refresh, so a partial outage cannot erase + valid discovery data or freeze the entire provider snapshot. """ cycle = cycle or _RefreshCycleState() models = _apply_model_visibility(upstream, overrides_by_key, disabled_model_keys) - base_paths = upstream.discovery_base_paths() + upstream_url = normalize_upstream_url(upstream.base_url) + + def _base_path(model: object) -> DiscoveredPath: + return DiscoveredPath( + model_id=exposed_model_id(model), + path=encode_model_path(upstream_url), + upstream_url=upstream_url, + ) if not is_openrouter_base_url(upstream.base_url): - return [ - (exposed_model_id(model), path) for model in models for path in base_paths - ] + return ProviderPathSnapshot(paths=tuple(_base_path(model) for model in models)) if not (upstream.provider_type or "").strip(): - return [] + return ProviderPathSnapshot(paths=()) - any_fetch_succeeded = False - any_fetch_attempted = False semaphore = asyncio.Semaphore(_OPENROUTER_CONCURRENCY) async with _make_http_client() as client: - async def _for_model(model: object) -> list[tuple[str, str]]: - nonlocal any_fetch_succeeded, any_fetch_attempted + async def _for_model( + model: object, + ) -> tuple[list[DiscoveredPath], str | None]: model_id = exposed_model_id(model) - # Base paths always apply: responses whose upstream payload lacks a - # provider field are stamped with them (see _apply_provider_field). - pairs = [(model_id, path) for path in base_paths] author_slug = openrouter_author_slug(model) if not author_slug: - return pairs - any_fetch_attempted = True - sub_providers = await _fetch_openrouter_endpoint_subproviders( + return [_base_path(model)], None + endpoints = await _fetch_openrouter_endpoint_subproviders( client, upstream.base_url, upstream.api_key, @@ -334,67 +383,78 @@ async def _collect_provider_paths( semaphore, cycle, ) - if sub_providers is None: - return [] - any_fetch_succeeded = True - paths = [ - upstream.discovery_path_for_subprovider(name) for name in sub_providers - ] - pairs.extend((model_id, path) for path in paths if path) - return list(dict.fromkeys(pairs)) + if endpoints is None: + return [], model_id + paths = [_base_path(model)] + paths.extend( + DiscoveredPath( + model_id=model_id, + path=encode_model_path(upstream_url, endpoint.tag), + upstream_url=upstream_url, + provider_tag=endpoint.tag, + provider_name=endpoint.provider_name, + ) + for endpoint in endpoints + ) + return paths, None results = await asyncio.gather( - *(_for_model(m) for m in models), return_exceptions=True + *(_for_model(model) for model in models), return_exceptions=True ) - if any_fetch_attempted and not any_fetch_succeeded: - # Every endpoint lookup degraded (offline, throttled, bad payloads): - # the true path set is unknown, not empty. - return None - - pairs: list[tuple[str, str]] = [] - for result in results: + paths: list[DiscoveredPath] = [] + preserve_model_ids: set[str] = set() + for model, result in zip(models, results): if isinstance(result, BaseException): + model_id = exposed_model_id(model) + preserve_model_ids.add(model_id) logger.warning( "OpenRouter endpoint discovery task errored", extra={"provider": upstream.provider_type, "error": str(result)}, ) continue - pairs.extend(result) + model_paths, preserved_model_id = result + paths.extend(model_paths) + if preserved_model_id: + preserve_model_ids.add(preserved_model_id) - return pairs + return ProviderPathSnapshot( + paths=tuple(paths), preserve_model_ids=frozenset(preserve_model_ids) + ) async def _persist_provider_paths( - upstream_provider_id: int, pairs: list[tuple[str, str]] + upstream_provider_id: int, snapshot: ProviderPathSnapshot ) -> None: - """Replace all rows for ``upstream_provider_id`` with ``pairs``. - - Replacement (not upsert) so stale paths disappear when provider config or - upstream availability changes. Rows are written with chunked bulk INSERTs - so the transaction holds SQLite's write lock briefly — billing writes share - this database file. - """ - unique_pairs = list(dict.fromkeys(pairs)) + """Replace refreshed rows while retaining model-level degraded snapshots.""" + unique_paths = list( + {(path.model_id, path.path): path for path in snapshot.paths}.values() + ) now = int(time.time()) async with create_session() as session: - await session.exec( # type: ignore[call-overload] - delete(ModelPathRow).where( - col(ModelPathRow.upstream_provider_id) == upstream_provider_id - ) + delete_stmt = delete(ModelPathRow).where( + col(ModelPathRow.upstream_provider_id) == upstream_provider_id ) - for start in range(0, len(unique_pairs), _PERSIST_CHUNK_SIZE): - chunk = unique_pairs[start : start + _PERSIST_CHUNK_SIZE] + if snapshot.preserve_model_ids: + delete_stmt = delete_stmt.where( + col(ModelPathRow.model_id).not_in(sorted(snapshot.preserve_model_ids)) + ) + await session.exec(delete_stmt) # type: ignore[call-overload] + for start in range(0, len(unique_paths), _PERSIST_CHUNK_SIZE): + chunk = unique_paths[start : start + _PERSIST_CHUNK_SIZE] await session.execute( insert(ModelPathRow), [ { - "model_id": model_id, - "path": path, + "model_id": discovered.model_id, + "path": discovered.path, + "upstream_url": discovered.upstream_url, + "provider_tag": discovered.provider_tag, + "provider_name": discovered.provider_name, "upstream_provider_id": upstream_provider_id, "updated_at": now, } - for model_id, path in chunk + for discovered in chunk ], ) await session.commit() @@ -454,22 +514,22 @@ async def refresh_model_paths( if upstream.db_id is None or upstream.db_id not in enabled_provider_ids: continue try: - pairs = await _collect_provider_paths( + snapshot = await _collect_provider_paths( upstream, overrides_by_key=overrides_by_key, disabled_model_keys=disabled_model_keys, cycle=cycle, ) - if pairs is None: + if snapshot.preserve_model_ids: logger.warning( - "Model paths unknown this cycle; keeping previous rows", + "Some model paths are unknown; keeping their previous rows", extra={ "provider": upstream.provider_type or upstream.base_url, "db_id": upstream.db_id, + "preserved_models": len(snapshot.preserve_model_ids), }, ) - continue - await _persist_provider_paths(upstream.db_id, pairs) + await _persist_provider_paths(upstream.db_id, snapshot) except Exception as e: # noqa: BLE001 - isolate per-provider failures logger.error( "Failed to refresh model paths for provider", @@ -482,6 +542,21 @@ async def refresh_model_paths( ) +async def refresh_model_paths_for_provider(upstream_provider_id: int) -> None: + """Immediately synchronize discovery after an admin provider/model mutation.""" + from ..proxy import get_upstreams + + matching = [ + upstream + for upstream in get_upstreams() + if upstream.db_id == upstream_provider_id + ] + if matching: + await refresh_model_paths(matching) + else: + await prune_model_paths_for_inactive_providers() + + def _refresh_interval_seconds() -> int: """Current interval, re-read every loop so runtime setting changes apply.""" from ..core.settings import settings @@ -535,8 +610,19 @@ async def refresh_model_paths_periodically( break +def _serialize_path(row: ModelPathRow) -> dict[str, Any]: + provider = None + if row.provider_tag or row.provider_name: + provider = {"name": row.provider_name, "slug": row.provider_tag} + return { + "path": row.path, + "upstream_url": row.upstream_url, + "provider": provider, + } + + async def get_all_model_paths() -> dict: - """All models with their paths, shaped for ``GET /v1/models/paths``.""" + """All models with their exact selectable routes.""" async with create_session() as session: rows = ( await session.exec( @@ -548,49 +634,37 @@ async def get_all_model_paths() -> dict: ) ).all() - grouped: dict[str, list[dict]] = {} + grouped: dict[str, list[dict[str, Any]]] = {} seen_paths: dict[str, set[str]] = {} updated_at = 0 for row in rows: updated_at = max(updated_at, row.updated_at) - model_id = public_model_id(row.model_id) - if row.path in seen_paths.setdefault(model_id, set()): + if row.path in seen_paths.setdefault(row.model_id, set()): continue - seen_paths[model_id].add(row.path) - grouped.setdefault(model_id, []).append({"path": row.path}) - # Deterministic output: models sorted by public id, paths sorted within. - data: list[dict] = [] - for grouped_model_id in sorted(grouped): - model_paths = sorted(grouped[grouped_model_id], key=lambda p: str(p["path"])) - data.append({"id": grouped_model_id, "paths": model_paths}) + seen_paths[row.model_id].add(row.path) + grouped.setdefault(row.model_id, []).append(_serialize_path(row)) + data = [ + { + "id": grouped_model_id, + "paths": sorted( + grouped[grouped_model_id], key=lambda item: str(item["path"]) + ), + } + for grouped_model_id in sorted(grouped) + ] return {"data": data, "updated_at": updated_at or None} async def get_paths_for_model(model_id: str) -> dict: - """Paths for a single model, shaped for ``GET /v1/models/paths/model``. - - Match by the public, unqualified model id, mirroring the model cache alias - behavior. Both ``deepseek-v4-pro`` and ``deepseek/deepseek-v4-pro`` resolve - every row whose stored id has the same base model id. The candidate set is - narrowed in SQL (exact id or ``%/`` suffix) so the route does not - materialize the whole table per request. - """ - # The request may be a full stored id ("z-ai/glm-5v-turbo") or an - # already-stripped public id ("fireworks/models/glm-5"); accept both. - accepted_ids = {model_id, public_model_id(model_id)} + """Return paths only for the exact model ID advertised by ``/v1/models``.""" async with create_session() as session: - conditions = [] - for candidate in accepted_ids: - conditions.append(col(ModelPathRow.model_id) == candidate) - conditions.append(col(ModelPathRow.model_id).endswith(f"/{candidate}")) rows = ( await session.exec( select(ModelPathRow) - .where(or_(*conditions)) + .where(col(ModelPathRow.model_id) == model_id) .order_by( col(ModelPathRow.path), col(ModelPathRow.upstream_provider_id), - col(ModelPathRow.model_id), ) ) ).all() @@ -599,15 +673,9 @@ async def get_paths_for_model(model_id: str) -> dict: paths: list[dict] = [] updated_at = 0 for row in rows: - # The SQL suffix match is a prefilter; enforce the exact public-id rule. - if ( - row.model_id not in accepted_ids - and public_model_id(row.model_id) not in accepted_ids - ): - continue updated_at = max(updated_at, row.updated_at) if row.path in seen: continue seen.add(row.path) - paths.append({"path": row.path}) + paths.append(_serialize_path(row)) return {"data": paths, "updated_at": updated_at or None} diff --git a/routstr/upstream/ollama.py b/routstr/upstream/ollama.py index c4873ea0..9fed0154 100644 --- a/routstr/upstream/ollama.py +++ b/routstr/upstream/ollama.py @@ -66,7 +66,9 @@ class OllamaUpstreamProvider(BaseUpstreamProvider): """Strip 'ollama/' prefix for Ollama API compatibility.""" return model_id.removeprefix("ollama/") - def get_request_base_url(self, path: str, model_obj: Model | None = None) -> str: + def get_request_base_url( + self, path: str, model_obj: Model | None = None + ) -> str: """Route proxy traffic through Ollama's OpenAI-compatible /v1 endpoint.""" return f"{self.base_url.rstrip('/')}/v1" @@ -183,9 +185,7 @@ class OllamaUpstreamProvider(BaseUpstreamProvider): except Exception: self._models_cache = models_with_fees - self._models_by_id = { - m.forwarded_model_id or m.id: m for m in self._models_cache - } + self._models_by_id = {m.forwarded_model_id or m.id: m for m in self._models_cache} logger.info( f"Refreshed models cache for {self.base_url}", extra={"model_count": len(models)}, diff --git a/routstr/upstream/rate_limit.py b/routstr/upstream/rate_limit.py index ac1eff78..dca1ba5b 100644 --- a/routstr/upstream/rate_limit.py +++ b/routstr/upstream/rate_limit.py @@ -119,9 +119,7 @@ def classify_rate_limit( retry_match = _RETRY_RE.search(redacted) if retry_match is not None: value = float(retry_match.group(1)) - retry_after = ( - value / 1000.0 if retry_match.group(2).lower() == "ms" else value - ) + retry_after = value / 1000.0 if retry_match.group(2).lower() == "ms" else value limit_name_match = _LIMIT_NAME_RE.search(redacted) diff --git a/routstr/upstream/request_correction.py b/routstr/upstream/request_correction.py index 8e4379a0..c2ea5b1d 100644 --- a/routstr/upstream/request_correction.py +++ b/routstr/upstream/request_correction.py @@ -84,7 +84,9 @@ def extract_error_message(response: Response) -> str: return "" -def strip_unsupported_param(body: dict, error_message: str) -> tuple[dict, str] | None: +def strip_unsupported_param( + body: dict, error_message: str +) -> tuple[dict, str] | None: """Drop a top-level param the upstream named as unsupported/deprecated. Returns ``(new_body, param)`` (a new dict, original untouched) when the diff --git a/routstr/upstream/routstr.py b/routstr/upstream/routstr.py index de1aa3bd..0371946a 100644 --- a/routstr/upstream/routstr.py +++ b/routstr/upstream/routstr.py @@ -50,7 +50,8 @@ class RoutstrUpstreamProvider(BaseUpstreamProvider): def normalize_request_path( self, path: str, model_obj: "Model | None" = None ) -> str: - """Preserve the ``v1/`` prefix when forwarding to an upstream Routstr.""" + """Preserve the ``v1/`` prefix when forwarding to an upstream Routstr. + """ return path.lstrip("/") @classmethod diff --git a/routstr/upstream/xai.py b/routstr/upstream/xai.py index 12e3dd93..58caaba0 100644 --- a/routstr/upstream/xai.py +++ b/routstr/upstream/xai.py @@ -21,9 +21,7 @@ class XAIUpstreamProvider(BaseUpstreamProvider): ) @classmethod - def _build_from_row( - cls, provider_row: "UpstreamProviderRow" - ) -> "XAIUpstreamProvider": + def _build_from_row(cls, provider_row: "UpstreamProviderRow") -> "XAIUpstreamProvider": return cls( api_key=provider_row.api_key, provider_fee=provider_row.provider_fee, diff --git a/routstr/wallet.py b/routstr/wallet.py index cef6902d..dd92d913 100644 --- a/routstr/wallet.py +++ b/routstr/wallet.py @@ -243,9 +243,7 @@ async def send(amount: int, unit: str, mint_url: str | None = None) -> tuple[int all_mint_urls = list({k.mint_url for k in wallet.keysets.values()}) proof_summary = { - f"{k.mint_url}/{k.unit.name}": sum( - p.amount for p in wallet.proofs if p.id == k.id - ) + f"{k.mint_url}/{k.unit.name}": sum(p.amount for p in wallet.proofs if p.id == k.id) for k in wallet.keysets.values() } # Show ALL proofs in DB by keyset_id, regardless of whether the loaded wallet @@ -600,16 +598,11 @@ async def swap_to_primary_mint( # advance the counter so the next request derives fresh secrets. logger.warning( "swap_to_primary_mint: outputs already signed — recovering orphaned proofs", - extra={ - "mint_quote_id": mint_quote.quote, - "minted_amount": minted_amount, - }, + extra={"mint_quote_id": mint_quote.quote, "minted_amount": minted_amount}, ) try: for keyset_id in primary_wallet.keysets: - await primary_wallet.restore_tokens_for_keyset( - keyset_id, to=1, batch=25 - ) + await primary_wallet.restore_tokens_for_keyset(keyset_id, to=1, batch=25) await primary_wallet.load_proofs(reload=True) post_recovery_balance = primary_wallet.available_balance.amount balance_gained = post_recovery_balance - pre_mint_balance