only activ model should be visible

make cache prices configurable

docs/api/endpoints.md

@@ -327,6 +327,58 @@ GET /v1/models
 }
 ```
 
+### List Model Paths
+
+Get the upstream provider paths each advertised model can be reached through.
+This is discovery data only; routing still chooses the provider per request.
+
+```http
+GET /v1/models/paths
+```
+
+**Response:**
+
+```json
+{
+  "data": [
+    {
+      "id": "claude-sonnet-4",
+      "paths": [
+        {"path": "anthropic"},
+        {"path": "openrouter:Anthropic"}
+      ]
+    }
+  ]
+}
+```
+
+### List Paths for One Model
+
+Use a query parameter so model IDs containing `/` are handled safely. Lookup is
+by the public, unqualified model ID: `glm-5v-turbo` resolves
+`z-ai/glm-5v-turbo`, and `deepseek-v4-pro` and `deepseek/deepseek-v4-pro`
+return the same merged path set.
+
+```http
+GET /v1/models/paths/model?model_id=anthropic/claude-sonnet-4
+```
+
+**Response:**
+
+```json
+{
+  "data": [
+    {"path": "anthropic"},
+    {"path": "openrouter:Anthropic"}
+  ]
+}
+```
+
+Model IDs in responses are unqualified display IDs: provider prefixes such as
+`z-ai/` or `openai/` are stripped. Path values match the provider string stamped
+on chat-completion responses, such as `anthropic`, `generic:my-upstream`, or
+`openrouter:Anthropic`.
+
 ## Wallet Management
 
 ### Create Wallet (Coming Soon)

docs/api/overview.md

@@ -100,6 +100,7 @@ All errors follow a consistent format:
 Standard OpenAI-compatible endpoints:
 
 - **Models**: `/v1/models`
+- **Model paths**: `/v1/models/paths`, `/v1/models/paths/model?model_id=...`
 - **Responses**: `/v1/responses`
 - **Chat Completions**: `/v1/chat/completions`
 - **Embeddings**: `/v1/embeddings`
@@ -302,7 +303,7 @@ Get node metadata:
 GET /v1/info
 ```
 
-Supported models and pricing are available at `/v1/models`.
+Supported models and pricing are available at `/v1/models`. Upstream provider path discovery is available at `/v1/models/paths` and `/v1/models/paths/model?model_id=...`.
 
 ## Next Steps
 

docs/provider/configuration.md

@@ -137,6 +137,7 @@ Use environment variables for:
 | `TOR_PROXY_URL`      | SOCKS5 proxy for Tor              | `socks5://127.0.0.1:9050`            |
 | `CORS_ORIGINS`       | Allowed CORS origins              | `*`                                  |
 | `RELAYS`             | Nostr relays (comma-separated)    | (default set)                        |
+| `MODEL_PATHS_REFRESH_INTERVAL_SECONDS` | How often to refresh `/v1/models/paths` discovery data; set `0` to disable | `600` |
 
 ### Priority
 
@@ -156,3 +157,8 @@ Manage which AI models you offer:
    - **Create aliases** — friendly names for models
 
 See [Pricing](pricing.md) for per-model pricing strategies.
+
+Model path discovery is refreshed in the background and exposed through
+`/v1/models/paths`. The response groups each client-visible model ID with the
+provider paths that may appear in chat-completion response metadata. Tune the
+refresh cadence with `MODEL_PATHS_REFRESH_INTERVAL_SECONDS`.

migrations/versions/c6d7e8f9a0b1_add_slug_to_upstream_providers.py

@@ -0,0 +1,88 @@
+"""add slug to upstream_providers
+
+Revision ID: c6d7e8f9a0b1
+Revises: b5e7c9d1f3a2
+Create Date: 2026-06-29 00:00:00.000000
+"""
+
+from __future__ import annotations
+
+import sqlalchemy as sa
+from alembic import op
+
+from routstr.core.provider_slugs import provider_slug_base, provider_slug_candidate
+
+revision = "c6d7e8f9a0b1"
+down_revision = "b5e7c9d1f3a2"
+branch_labels = None
+depends_on = None
+
+
+def _allocate_backfill_slug(provider_type: str, reserved_slugs: set[str]) -> str:
+    base = provider_slug_base(provider_type)
+    suffix_number = 1
+    while True:
+        candidate = provider_slug_candidate(base, suffix_number)
+        if candidate not in reserved_slugs:
+            reserved_slugs.add(candidate)
+            return candidate
+        suffix_number += 1
+
+
+def _backfill_provider_slugs(conn: sa.Connection) -> None:
+    existing_rows = conn.execute(
+        sa.text(
+            "SELECT slug FROM upstream_providers "
+            "WHERE slug IS NOT NULL AND slug != ''"
+        )
+    )
+    reserved_slugs = {str(row.slug).lower() for row in existing_rows}
+
+    rows_to_backfill = conn.execute(
+        sa.text(
+            "SELECT id, provider_type FROM upstream_providers "
+            "WHERE slug IS NULL OR slug = '' "
+            "ORDER BY id"
+        )
+    )
+    for row in rows_to_backfill:
+        slug = _allocate_backfill_slug(str(row.provider_type), reserved_slugs)
+        conn.execute(
+            sa.text("UPDATE upstream_providers SET slug = :slug WHERE id = :id"),
+            {"slug": slug, "id": row.id},
+        )
+
+
+def upgrade() -> None:
+    conn = op.get_bind()
+    inspector = sa.inspect(conn)
+    columns = {c["name"] for c in inspector.get_columns("upstream_providers")}
+
+    if "slug" not in columns:
+        op.add_column(
+            "upstream_providers",
+            sa.Column("slug", sa.String(), nullable=True),
+        )
+
+    _backfill_provider_slugs(conn)
+
+    existing_indexes = {idx["name"] for idx in inspector.get_indexes("upstream_providers")}
+    if "ix_upstream_providers_slug" not in existing_indexes:
+        op.create_index(
+            "ix_upstream_providers_slug",
+            "upstream_providers",
+            ["slug"],
+            unique=True,
+        )
+
+
+def downgrade() -> None:
+    conn = op.get_bind()
+    inspector = sa.inspect(conn)
+    existing_indexes = {idx["name"] for idx in inspector.get_indexes("upstream_providers")}
+    if "ix_upstream_providers_slug" in existing_indexes:
+        op.drop_index("ix_upstream_providers_slug", table_name="upstream_providers")
+
+    columns = {c["name"] for c in inspector.get_columns("upstream_providers")}
+    if "slug" in columns:
+        op.drop_column("upstream_providers", "slug")

migrations/versions/d7e8f9a0b1c2_add_model_paths_table.py

@@ -0,0 +1,66 @@
+"""add model_paths table
+
+Revision ID: d7e8f9a0b1c2
+Revises: c6d7e8f9a0b1
+Create Date: 2026-07-05 00:00:00.000000
+"""
+
+from __future__ import annotations
+
+import sqlalchemy as sa
+from alembic import op
+
+revision = "d7e8f9a0b1c2"
+down_revision = "c6d7e8f9a0b1"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    conn = op.get_bind()
+    inspector = sa.inspect(conn)
+    if "model_paths" in inspector.get_table_names():
+        return
+
+    op.create_table(
+        "model_paths",
+        sa.Column("id", sa.Integer(), primary_key=True, autoincrement=True),
+        sa.Column("model_id", sa.String(), nullable=False),
+        sa.Column("path", sa.String(), nullable=False),
+        sa.Column("upstream_provider_id", sa.Integer(), nullable=False),
+        sa.ForeignKeyConstraint(
+            ["upstream_provider_id"],
+            ["upstream_providers.id"],
+            ondelete="CASCADE",
+        ),
+        sa.UniqueConstraint(
+            "model_id",
+            "path",
+            "upstream_provider_id",
+            name="uq_model_paths_model_path_provider",
+        ),
+    )
+    op.create_index(
+        "ix_model_paths_model_id",
+        "model_paths",
+        ["model_id"],
+    )
+    op.create_index(
+        "ix_model_paths_upstream_provider_id",
+        "model_paths",
+        ["upstream_provider_id"],
+    )
+
+
+def downgrade() -> None:
+    conn = op.get_bind()
+    inspector = sa.inspect(conn)
+    if "model_paths" not in inspector.get_table_names():
+        return
+
+    existing_indexes = {idx["name"] for idx in inspector.get_indexes("model_paths")}
+    if "ix_model_paths_upstream_provider_id" in existing_indexes:
+        op.drop_index("ix_model_paths_upstream_provider_id", table_name="model_paths")
+    if "ix_model_paths_model_id" in existing_indexes:
+        op.drop_index("ix_model_paths_model_id", table_name="model_paths")
+    op.drop_table("model_paths")

routstr/core/admin.py

@@ -1,5 +1,6 @@
 import asyncio
 import json
+import re
 import secrets
 from datetime import datetime, timezone
 from pathlib import Path
@@ -8,6 +9,7 @@ from fastapi import APIRouter, Depends, HTTPException, Query, Request
 from pydantic import BaseModel, RootModel
 from pydantic.v1 import ValidationError as PydanticValidationError
 from sqlmodel import select
+from sqlmodel.ext.asyncio.session import AsyncSession
 
 from ..payment.models import _row_to_model, list_models
 from ..proxy import refresh_model_maps, reinitialize_upstreams
@@ -29,6 +31,7 @@ from .db import (
 )
 from .log_manager import log_manager
 from .logging import get_logger
+from .provider_slugs import allocate_unique_provider_slug
 from .settings import SettingsService, settings
 
 logger = get_logger(__name__)
@@ -456,19 +459,18 @@ class ModelCreate(BaseModel):
     dependencies=[Depends(require_admin_api)],
 )
 async def upsert_provider_model(
-    provider_id: int, payload: ModelCreate
+    provider_id: str, payload: ModelCreate
 ) -> dict[str, object]:
     print(payload)
     logger.info(
         f"UPSERT_PROVIDER_MODEL called: provider_id={provider_id}, model_id={payload.id}"
     )
     async with create_session() as session:
-        provider = await session.get(UpstreamProviderRow, provider_id)
-        if not provider:
-            raise HTTPException(status_code=404, detail="Provider not found")
+        provider = await _get_upstream_provider_by_ref(session, provider_id)
+        provider_pk = _provider_pk(provider)
 
         # Try to get existing model
-        existing_row = await session.get(ModelRow, (payload.id, provider_id))
+        existing_row = await session.get(ModelRow, (payload.id, provider_pk))
 
         if existing_row:
             # Update existing model
@@ -524,7 +526,7 @@ async def upsert_provider_model(
                 alias_ids=(
                     json.dumps(payload.alias_ids) if payload.alias_ids else None
                 ),
-                upstream_provider_id=provider_id,
+                upstream_provider_id=provider_pk,
                 enabled=payload.enabled,
                 forwarded_model_id=payload.forwarded_model_id or payload.id,
             )
@@ -543,7 +545,7 @@ async def upsert_provider_model(
     dependencies=[Depends(require_admin_api)],
 )
 async def update_provider_model_legacy(
-    provider_id: int, model_id: str, payload: ModelCreate
+    provider_id: str, model_id: str, payload: ModelCreate
 ) -> dict[str, object]:
     """Legacy PATCH endpoint - redirects to upsert POST endpoint for backward compatibility."""
     logger.info(
@@ -556,13 +558,12 @@ async def update_provider_model_legacy(
     "/api/upstream-providers/{provider_id}/models/{model_id:path}",
     dependencies=[Depends(require_admin_api)],
 )
-async def get_provider_model(provider_id: int, model_id: str) -> dict[str, object]:
+async def get_provider_model(provider_id: str, model_id: str) -> dict[str, object]:
     async with create_session() as session:
-        provider = await session.get(UpstreamProviderRow, provider_id)
-        if not provider:
-            raise HTTPException(status_code=404, detail="Provider not found")
+        provider = await _get_upstream_provider_by_ref(session, provider_id)
+        provider_pk = _provider_pk(provider)
 
-        row = await session.get(ModelRow, (model_id, provider_id))
+        row = await session.get(ModelRow, (model_id, provider_pk))
         if not row:
             raise HTTPException(
                 status_code=404, detail="Model not found for this provider"
@@ -576,9 +577,11 @@ async def get_provider_model(provider_id: int, model_id: str) -> dict[str, objec
     "/api/upstream-providers/{provider_id}/models/{model_id:path}",
     dependencies=[Depends(require_admin_api)],
 )
-async def delete_provider_model(provider_id: int, model_id: str) -> dict[str, object]:
+async def delete_provider_model(provider_id: str, model_id: str) -> dict[str, object]:
     async with create_session() as session:
-        row = await session.get(ModelRow, (model_id, provider_id))
+        provider = await _get_upstream_provider_by_ref(session, provider_id)
+        provider_pk = _provider_pk(provider)
+        row = await session.get(ModelRow, (model_id, provider_pk))
         if not row:
             raise HTTPException(
                 status_code=404, detail="Model not found for this provider"
@@ -593,10 +596,12 @@ async def delete_provider_model(provider_id: int, model_id: str) -> dict[str, ob
     "/api/upstream-providers/{provider_id}/models",
     dependencies=[Depends(require_admin_api)],
 )
-async def delete_all_provider_models(provider_id: int) -> dict[str, object]:
+async def delete_all_provider_models(provider_id: str) -> dict[str, object]:
     async with create_session() as session:
+        provider = await _get_upstream_provider_by_ref(session, provider_id)
+        provider_pk = _provider_pk(provider)
         result = await session.exec(
-            select(ModelRow).where(ModelRow.upstream_provider_id == provider_id)
+            select(ModelRow).where(ModelRow.upstream_provider_id == provider_pk)
         )  # type: ignore
         rows = result.all()
         for row in rows:
@@ -615,7 +620,7 @@ class BatchOverrideRequest(BaseModel):
     dependencies=[Depends(require_admin_api)],
 )
 async def batch_override_provider_models(
-    provider_id: int, payload: BatchOverrideRequest
+    provider_id: str, payload: BatchOverrideRequest
 ) -> dict[str, object]:
     """Batch override models for a specific provider."""
     logger.info(
@@ -623,15 +628,14 @@ async def batch_override_provider_models(
     )
 
     async with create_session() as session:
-        provider = await session.get(UpstreamProviderRow, provider_id)
-        if not provider:
-            raise HTTPException(status_code=404, detail="Provider not found")
+        provider = await _get_upstream_provider_by_ref(session, provider_id)
+        provider_pk = _provider_pk(provider)
 
         overridden_count = 0
 
         for model_data in payload.models:
             # Try to get existing model regardless of whether it's enabled or not
-            existing_row = await session.get(ModelRow, (model_data.id, provider_id))
+            existing_row = await session.get(ModelRow, (model_data.id, provider_pk))
 
             if existing_row:
                 # Update existing
@@ -685,7 +689,7 @@ async def batch_override_provider_models(
                         if model_data.alias_ids
                         else None
                     ),
-                    upstream_provider_id=provider_id,
+                    upstream_provider_id=provider_pk,
                     enabled=model_data.enabled,
                 )
                 session.add(row)
@@ -702,6 +706,85 @@ async def batch_override_provider_models(
     }
 
 
+_SLUG_PATTERN = re.compile(r"^[a-z0-9][a-z0-9-]{1,62}[a-z0-9]$")
+
+
+def _validate_slug(value: str) -> str:
+    candidate = value.strip().lower()
+    if not _SLUG_PATTERN.fullmatch(candidate):
+        raise HTTPException(
+            status_code=400,
+            detail=(
+                "slug must be 3-64 chars, lowercase letters/digits/hyphens, "
+                "and may not start or end with a hyphen"
+            ),
+        )
+    if candidate.isdigit():
+        raise HTTPException(
+            status_code=400,
+            detail="slug must not be all digits",
+        )
+    return candidate
+
+
+async def _ensure_unique_slug(
+    session: AsyncSession, slug: str, exclude_id: int | None = None
+) -> None:
+    stmt = select(UpstreamProviderRow).where(UpstreamProviderRow.slug == slug)
+    result = await session.exec(stmt)
+    existing = result.first()
+    if existing and existing.id != exclude_id:
+        raise HTTPException(
+            status_code=409,
+            detail="Provider with this slug already exists",
+        )
+
+
+async def _get_upstream_provider_by_ref(
+    session: AsyncSession, provider_ref: str
+) -> UpstreamProviderRow:
+    if provider_ref.isdigit():
+        provider = await session.get(UpstreamProviderRow, int(provider_ref))
+    else:
+        slug = _validate_slug(provider_ref)
+        result = await session.exec(
+            select(UpstreamProviderRow).where(UpstreamProviderRow.slug == slug)
+        )
+        provider = result.first()
+
+    if not provider:
+        raise HTTPException(status_code=404, detail="Provider not found")
+    return provider
+
+
+def _provider_pk(provider: UpstreamProviderRow) -> int:
+    if provider.id is None:
+        raise HTTPException(status_code=500, detail="Provider has no database id")
+    return provider.id
+
+
+def _serialize_provider(
+    provider: UpstreamProviderRow, redact_api_key: bool = True
+) -> dict[str, object]:
+    return {
+        "id": provider.id,
+        "slug": provider.slug,
+        "provider_type": provider.provider_type,
+        "base_url": provider.base_url,
+        "api_key": "[REDACTED]"
+        if (redact_api_key and provider.api_key)
+        else provider.api_key
+        if not redact_api_key
+        else "",
+        "api_version": provider.api_version,
+        "enabled": provider.enabled,
+        "provider_fee": provider.provider_fee,
+        "provider_settings": json.loads(provider.provider_settings)
+        if provider.provider_settings
+        else None,
+    }
+
+
 class UpstreamProviderCreate(BaseModel):
     provider_type: str
     base_url: str
@@ -710,6 +793,7 @@ class UpstreamProviderCreate(BaseModel):
     enabled: bool = True
     provider_fee: float = 1.01
     provider_settings: dict | None = None
+    slug: str | None = None
 
 
 class UpstreamProviderUpdate(BaseModel):
@@ -720,6 +804,50 @@ class UpstreamProviderUpdate(BaseModel):
     enabled: bool | None = None
     provider_fee: float | None = None
     provider_settings: dict | None = None
+    slug: str | None = None
+
+
+class UpstreamProviderUpdateBySlug(BaseModel):
+    slug: str
+    new_slug: str | None = None
+    provider_type: str | None = None
+    base_url: str | None = None
+    api_key: str | None = None
+    api_version: str | None = None
+    enabled: bool | None = None
+    provider_fee: float | None = None
+    provider_settings: dict | None = None
+
+
+async def _apply_provider_update(
+    session: AsyncSession,
+    provider: UpstreamProviderRow,
+    payload: UpstreamProviderUpdate,
+    new_slug: str | None = None,
+) -> None:
+    if new_slug is not None:
+        validated = _validate_slug(new_slug)
+        await _ensure_unique_slug(session, validated, exclude_id=provider.id)
+        provider.slug = validated
+
+    if payload.provider_type is not None:
+        provider.provider_type = payload.provider_type
+    if payload.base_url is not None:
+        provider.base_url = payload.base_url
+    if payload.api_key is not None:
+        provider.api_key = payload.api_key
+    if payload.api_version is not None:
+        provider.api_version = payload.api_version
+    if payload.enabled is not None:
+        provider.enabled = payload.enabled
+    if payload.provider_fee is not None:
+        provider.provider_fee = payload.provider_fee
+    if payload.provider_settings is not None:
+        provider.provider_settings = json.dumps(payload.provider_settings)
+
+    session.add(provider)
+    await session.commit()
+    await session.refresh(provider)
 
 
 @admin_router.get("/api/upstream-providers", dependencies=[Depends(require_admin_api)])
@@ -727,21 +855,7 @@ async def get_upstream_providers() -> list[dict[str, object]]:
     async with create_session() as session:
         result = await session.exec(select(UpstreamProviderRow))
         providers = result.all()
-        return [
-            {
-                "id": p.id,
-                "provider_type": p.provider_type,
-                "base_url": p.base_url,
-                "api_key": "[REDACTED]" if p.api_key else "",
-                "api_version": p.api_version,
-                "enabled": p.enabled,
-                "provider_fee": p.provider_fee,
-                "provider_settings": json.loads(p.provider_settings)
-                if p.provider_settings
-                else None,
-            }
-            for p in providers
-        ]
+        return [_serialize_provider(p) for p in providers]
 
 
 @admin_router.post("/api/upstream-providers", dependencies=[Depends(require_admin_api)])
@@ -761,7 +875,14 @@ async def create_upstream_provider(
                 detail="Provider with this base URL and API key already exists",
             )
 
+        if payload.slug:
+            slug = _validate_slug(payload.slug)
+            await _ensure_unique_slug(session, slug)
+        else:
+            slug = await allocate_unique_provider_slug(session, payload.provider_type)
+
         provider = UpstreamProviderRow(
+            slug=slug,
             provider_type=payload.provider_type,
             base_url=payload.base_url,
             api_key=payload.api_key,
@@ -778,99 +899,81 @@ async def create_upstream_provider(
 
     await reinitialize_upstreams()
     await refresh_model_maps()
-    return {
-        "id": provider.id,
-        "provider_type": provider.provider_type,
-        "base_url": provider.base_url,
-        "api_key": "[REDACTED]",
-        "api_version": provider.api_version,
-        "enabled": provider.enabled,
-        "provider_fee": provider.provider_fee,
-        "provider_settings": payload.provider_settings,
-    }
+    return _serialize_provider(provider)
 
 
 @admin_router.get(
     "/api/upstream-providers/{provider_id}", dependencies=[Depends(require_admin_api)]
 )
-async def get_upstream_provider(provider_id: int) -> dict[str, object]:
+async def get_upstream_provider(provider_id: str) -> dict[str, object]:
     async with create_session() as session:
-        provider = await session.get(UpstreamProviderRow, provider_id)
-        if not provider:
-            raise HTTPException(status_code=404, detail="Provider not found")
-        return {
-            "id": provider.id,
-            "provider_type": provider.provider_type,
-            "base_url": provider.base_url,
-            "api_key": "[REDACTED]" if provider.api_key else "",
-            "api_version": provider.api_version,
-            "enabled": provider.enabled,
-            "provider_fee": provider.provider_fee,
-            "provider_settings": json.loads(provider.provider_settings)
-            if provider.provider_settings
-            else None,
-        }
+        provider = await _get_upstream_provider_by_ref(session, provider_id)
+        return _serialize_provider(provider)
 
 
 @admin_router.patch(
     "/api/upstream-providers/{provider_id}", dependencies=[Depends(require_admin_api)]
 )
 async def update_upstream_provider(
-    provider_id: int, payload: UpstreamProviderUpdate
+    provider_id: str, payload: UpstreamProviderUpdate
 ) -> dict[str, object]:
     async with create_session() as session:
-        provider = await session.get(UpstreamProviderRow, provider_id)
-        if not provider:
-            raise HTTPException(status_code=404, detail="Provider not found")
+        provider = await _get_upstream_provider_by_ref(session, provider_id)
 
-        if payload.provider_type is not None:
-            provider.provider_type = payload.provider_type
-        if payload.base_url is not None:
-            provider.base_url = payload.base_url
-        if payload.api_key is not None:
-            provider.api_key = payload.api_key
-        if payload.api_version is not None:
-            provider.api_version = payload.api_version
-        if payload.enabled is not None:
-            provider.enabled = payload.enabled
-        if payload.provider_fee is not None:
-            provider.provider_fee = payload.provider_fee
-        if payload.provider_settings is not None:
-            provider.provider_settings = json.dumps(payload.provider_settings)
-
-        session.add(provider)
-        await session.commit()
-        await session.refresh(provider)
+        await _apply_provider_update(session, provider, payload, new_slug=payload.slug)
 
     await reinitialize_upstreams()
     await refresh_model_maps()
-    return {
-        "id": provider.id,
-        "provider_type": provider.provider_type,
-        "base_url": provider.base_url,
-        "api_key": "[REDACTED]",
-        "api_version": provider.api_version,
-        "enabled": provider.enabled,
-        "provider_fee": provider.provider_fee,
-        "provider_settings": json.loads(provider.provider_settings)
-        if provider.provider_settings
-        else None,
-    }
+    return _serialize_provider(provider)
+
+
+@admin_router.patch(
+    "/api/upstream-providers", dependencies=[Depends(require_admin_api)]
+)
+async def update_upstream_provider_by_slug(
+    payload: UpstreamProviderUpdateBySlug,
+) -> dict[str, object]:
+    lookup = _validate_slug(payload.slug)
+    async with create_session() as session:
+        result = await session.exec(
+            select(UpstreamProviderRow).where(
+                UpstreamProviderRow.slug == lookup
+            )
+        )
+        provider = result.first()
+        if not provider:
+            raise HTTPException(status_code=404, detail="Provider not found")
+
+        update_payload = UpstreamProviderUpdate(
+            provider_type=payload.provider_type,
+            base_url=payload.base_url,
+            api_key=payload.api_key,
+            api_version=payload.api_version,
+            enabled=payload.enabled,
+            provider_fee=payload.provider_fee,
+            provider_settings=payload.provider_settings,
+        )
+        await _apply_provider_update(
+            session, provider, update_payload, new_slug=payload.new_slug
+        )
+
+    await reinitialize_upstreams()
+    await refresh_model_maps()
+    return _serialize_provider(provider)
 
 
 @admin_router.delete(
     "/api/upstream-providers/{provider_id}", dependencies=[Depends(require_admin_api)]
 )
-async def delete_upstream_provider(provider_id: int) -> dict[str, object]:
+async def delete_upstream_provider(provider_id: str) -> dict[str, object]:
     async with create_session() as session:
-        provider = await session.get(UpstreamProviderRow, provider_id)
-        if not provider:
-            raise HTTPException(status_code=404, detail="Provider not found")
+        provider = await _get_upstream_provider_by_ref(session, provider_id)
+        deleted_id = _provider_pk(provider)
         await session.delete(provider)
         await session.commit()
     await reinitialize_upstreams()
     await refresh_model_maps()
-    return {"ok": True, "deleted_id": provider_id}
+    return {"ok": True, "deleted_id": deleted_id}
 
 
 @admin_router.get("/api/provider-types", dependencies=[Depends(require_admin_api)])
@@ -885,17 +988,16 @@ async def get_provider_types() -> list[dict[str, object]]:
     "/api/upstream-providers/{provider_id}/models",
     dependencies=[Depends(require_admin_api)],
 )
-async def get_provider_models(provider_id: int) -> dict[str, object]:
+async def get_provider_models(provider_id: str) -> dict[str, object]:
     from ..upstream.helpers import _instantiate_provider
 
     async with create_session() as session:
-        provider = await session.get(UpstreamProviderRow, provider_id)
-        if not provider:
-            raise HTTPException(status_code=404, detail="Provider not found")
+        provider = await _get_upstream_provider_by_ref(session, provider_id)
+        provider_pk = _provider_pk(provider)
 
         db_models = await list_models(
             session=session,
-            upstream_id=provider_id,
+            upstream_id=provider_pk,
             include_disabled=True,
             apply_fees=False,
         )
@@ -985,13 +1087,11 @@ class TopupTokenRequest(BaseModel):
     dependencies=[Depends(require_admin_api)],
 )
 async def topup_provider_with_token(
-    provider_id: int, payload: TopupTokenRequest
+    provider_id: str, payload: TopupTokenRequest
 ) -> dict:
     """Redeem a Cashu token for an upstream provider."""
     async with create_session() as session:
-        provider = await session.get(UpstreamProviderRow, provider_id)
-        if not provider:
-            raise HTTPException(status_code=404, detail="Provider not found")
+        provider = await _get_upstream_provider_by_ref(session, provider_id)
 
         import httpx
 
@@ -1022,15 +1122,13 @@ async def topup_provider_with_token(
     dependencies=[Depends(require_admin_api)],
 )
 async def initiate_provider_topup(
-    provider_id: int, payload: TopupRequest
+    provider_id: str, payload: TopupRequest
 ) -> dict[str, object]:
     """Initiate a Lightning Network top-up for the upstream provider account."""
     from ..upstream.helpers import _instantiate_provider
 
     async with create_session() as session:
-        provider = await session.get(UpstreamProviderRow, provider_id)
-        if not provider:
-            raise HTTPException(status_code=404, detail="Provider not found")
+        provider = await _get_upstream_provider_by_ref(session, provider_id)
 
         try:
             logger.info(
@@ -1150,15 +1248,13 @@ async def initiate_provider_topup(
     "/api/upstream-providers/{provider_id}/topup/{invoice_id}/status",
     dependencies=[Depends(require_admin_api)],
 )
-async def check_topup_status(provider_id: int, invoice_id: str) -> dict[str, object]:
+async def check_topup_status(provider_id: str, invoice_id: str) -> dict[str, object]:
     """Check the status of a Lightning Network top-up invoice."""
     from ..upstream.helpers import _instantiate_provider
     from ..upstream.ppqai import PPQAIUpstreamProvider
 
     async with create_session() as session:
-        provider = await session.get(UpstreamProviderRow, provider_id)
-        if not provider:
-            raise HTTPException(status_code=404, detail="Provider not found")
+        provider = await _get_upstream_provider_by_ref(session, provider_id)
 
         # For Routstr providers, proxy the status check
         if provider.provider_type == "routstr":
@@ -1205,14 +1301,12 @@ async def check_topup_status(provider_id: int, invoice_id: str) -> dict[str, obj
     "/api/upstream-providers/{provider_id}/balance",
     dependencies=[Depends(require_admin_api)],
 )
-async def get_provider_balance(provider_id: int) -> dict[str, object]:
+async def get_provider_balance(provider_id: str) -> dict[str, object]:
     """Get the current balance for an upstream provider account."""
     from ..upstream.helpers import _instantiate_provider
 
     async with create_session() as session:
-        provider = await session.get(UpstreamProviderRow, provider_id)
-        if not provider:
-            raise HTTPException(status_code=404, detail="Provider not found")
+        provider = await _get_upstream_provider_by_ref(session, provider_id)
 
         # For Routstr providers, proxy the balance check
         if provider.provider_type == "routstr":
@@ -1591,15 +1685,13 @@ async def get_lightning_invoices_api(
     "/api/upstream-providers/{provider_id}/routstr/refund",
     dependencies=[Depends(require_admin_api)],
 )
-async def refund_routstr_provider_balance(provider_id: int) -> dict[str, object]:
+async def refund_routstr_provider_balance(provider_id: str) -> dict[str, object]:
     """Refund balance from an upstream Routstr provider back to the local wallet."""
     from ..upstream.helpers import _instantiate_provider
     from ..upstream.routstr import RoutstrUpstreamProvider
 
     async with create_session() as session:
-        provider_row = await session.get(UpstreamProviderRow, provider_id)
-        if not provider_row:
-            raise HTTPException(status_code=404, detail="Provider not found")
+        provider_row = await _get_upstream_provider_by_ref(session, provider_id)
 
         if provider_row.provider_type != "routstr":
             raise HTTPException(

routstr/core/db.py

@@ -210,6 +210,41 @@ class ModelRow(SQLModel, table=True):  # type: ignore
     upstream_provider: "UpstreamProviderRow" = Relationship(back_populates="models")
 
 
+class ModelPathRow(SQLModel, table=True):  # type: ignore
+    """Upstream provider path a model is reachable through.
+
+    Discovery/visibility data only. ``model_id`` is intentionally NOT globally
+    unique: it is the client-visible ``/v1/models`` id (``forwarded_model_id or
+    id``) grouped across every provider that exposes the model. A single model
+    can therefore have several rows — one per direct provider path plus one per
+    OpenRouter sub-provider endpoint.
+    """
+
+    __tablename__ = "model_paths"
+    __table_args__ = (
+        UniqueConstraint(
+            "model_id",
+            "path",
+            "upstream_provider_id",
+            name="uq_model_paths_model_path_provider",
+        ),
+    )
+    id: int | None = Field(default=None, primary_key=True)
+    model_id: str = Field(
+        index=True, description="Client-visible /v1/models id (forwarded_model_id or id)"
+    )
+    path: str = Field(
+        description="Provider path stamped on chat completion responses, e.g. "
+        "'anthropic' or 'openrouter:Anthropic'"
+    )
+    upstream_provider_id: int = Field(
+        index=True,
+        foreign_key="upstream_providers.id",
+        ondelete="CASCADE",
+        description="upstream_providers.id this path was discovered from",
+    )
+
+
 class LightningInvoice(SQLModel, table=True):  # type: ignore
     __tablename__ = "lightning_invoices"
 
@@ -319,6 +354,12 @@ class UpstreamProviderRow(SQLModel, table=True):  # type: ignore
         ),
     )
     id: int | None = Field(default=None, primary_key=True)
+    slug: str | None = Field(
+        default=None,
+        unique=True,
+        index=True,
+        description="Stable external slug used for updates via API key.",
+    )
     provider_type: str = Field(
         description="Provider type: custom, openai, anthropic, azure, openrouter, etc."
     )

routstr/core/main.py

@@ -58,6 +58,7 @@ async def lifespan(_: FastAPI) -> AsyncGenerator[None, None]:
     providers_task = None
     models_refresh_task = None
     model_maps_refresh_task = None
+    model_paths_refresh_task = None
     key_reset_task = None
     stale_reservation_task = None
     dead_key_prune_task = None
@@ -127,6 +128,12 @@ async def lifespan(_: FastAPI) -> AsyncGenerator[None, None]:
                 refresh_upstreams_models_periodically(get_upstreams)
             )
         model_maps_refresh_task = asyncio.create_task(refresh_model_maps_periodically())
+        if global_settings.model_paths_refresh_interval_seconds > 0:
+            from ..upstream.model_paths import refresh_model_paths_periodically
+
+            model_paths_refresh_task = asyncio.create_task(
+                refresh_model_paths_periodically(get_upstreams)
+            )
         payout_task = asyncio.create_task(periodic_payout())
         if global_settings.nsec:
             nip91_task = asyncio.create_task(announce_provider())
@@ -173,6 +180,8 @@ async def lifespan(_: FastAPI) -> AsyncGenerator[None, None]:
             models_refresh_task.cancel()
         if model_maps_refresh_task is not None:
             model_maps_refresh_task.cancel()
+        if model_paths_refresh_task is not None:
+            model_paths_refresh_task.cancel()
         if key_reset_task is not None:
             key_reset_task.cancel()
         if stale_reservation_task is not None:
@@ -206,6 +215,8 @@ async def lifespan(_: FastAPI) -> AsyncGenerator[None, None]:
                 tasks_to_wait.append(models_refresh_task)
             if model_maps_refresh_task is not None:
                 tasks_to_wait.append(model_maps_refresh_task)
+            if model_paths_refresh_task is not None:
+                tasks_to_wait.append(model_paths_refresh_task)
             if key_reset_task is not None:
                 tasks_to_wait.append(key_reset_task)
             if stale_reservation_task is not None:

routstr/core/provider_slugs.py

@@ -0,0 +1,65 @@
+from __future__ import annotations
+
+import re
+from itertools import count
+from typing import Collection
+
+from sqlmodel import select
+from sqlmodel.ext.asyncio.session import AsyncSession
+
+from .db import UpstreamProviderRow
+
+_SLUG_BASE_PATTERN = re.compile(r"[^a-z0-9]+")
+_MAX_SLUG_LENGTH = 64
+
+
+def provider_slug_base(provider_type: str) -> str:
+    """Return a deterministic slug base for a provider type."""
+    base = _SLUG_BASE_PATTERN.sub("-", provider_type.lower()).strip("-")
+    if not base:
+        base = "provider"
+    elif base.isdigit():
+        base = f"provider-{base}"
+    elif len(base) < 3:
+        base = f"{base}-provider"
+
+    if len(base) > _MAX_SLUG_LENGTH:
+        base = base[:_MAX_SLUG_LENGTH].rstrip("-") or "provider"
+    return base
+
+
+def provider_slug_candidate(base: str, suffix_number: int) -> str:
+    if suffix_number == 1:
+        return base
+
+    suffix = f"-{suffix_number}"
+    max_base_length = _MAX_SLUG_LENGTH - len(suffix)
+    return f"{base[:max_base_length].rstrip('-')}{suffix}"
+
+
+async def allocate_unique_provider_slug(
+    session: AsyncSession,
+    provider_type: str,
+    reserved_slugs: Collection[str] = (),
+) -> str:
+    """Allocate a stable, deterministic provider slug.
+
+    The first provider of a type gets ``openai``; later collisions get
+    ``openai-2``, ``openai-3``, etc. ``reserved_slugs`` covers rows staged in
+    memory but not flushed yet, such as settings/env seeding.
+    """
+    base = provider_slug_base(provider_type)
+    reserved = {slug.lower() for slug in reserved_slugs}
+
+    for suffix_number in count(1):
+        candidate = provider_slug_candidate(base, suffix_number)
+        if candidate in reserved:
+            continue
+
+        result = await session.exec(
+            select(UpstreamProviderRow).where(UpstreamProviderRow.slug == candidate)
+        )
+        if result.first() is None:
+            return candidate
+
+    raise RuntimeError("unreachable")

routstr/core/settings.py

@@ -94,6 +94,9 @@ class Settings(BaseSettings):
     models_refresh_interval_seconds: int = Field(
         default=360, env="MODELS_REFRESH_INTERVAL_SECONDS"
     )
+    model_paths_refresh_interval_seconds: int = Field(
+        default=600, env="MODEL_PATHS_REFRESH_INTERVAL_SECONDS"
+    )
     enable_pricing_refresh: bool = Field(default=True, env="ENABLE_PRICING_REFRESH")
     enable_models_refresh: bool = Field(default=True, env="ENABLE_MODELS_REFRESH")
     refund_cache_ttl_seconds: int = Field(default=3600, env="REFUND_CACHE_TTL_SECONDS")

routstr/payment/cost_calculation.py

@@ -41,6 +41,28 @@ class CostDataError(BaseModel):
     code: str
 
 
+def _empty_cost(cls: type[CostData] = CostData) -> CostData:
+    """Build an all-zero cost object — a full refund for an empty response.
+
+    Shared by the two paths that must not bill: an upstream response with no
+    usage data at all, and one that reports a USD cost but carries zero tokens
+    in every bucket.
+    """
+    return cls(
+        base_msats=0,
+        input_msats=0,
+        output_msats=0,
+        total_msats=0,
+        total_usd=0.0,
+        input_tokens=0,
+        output_tokens=0,
+        cache_read_input_tokens=0,
+        cache_creation_input_tokens=0,
+        cache_read_msats=0,
+        cache_creation_msats=0,
+    )
+
+
 async def calculate_cost(
     response_data: dict,
     max_cost: int,
@@ -83,19 +105,7 @@ async def calculate_cost(
                 else None,
             },
         )
-        return MaxCostData(
-            base_msats=0,
-            input_msats=0,
-            output_msats=0,
-            total_msats=0,
-            total_usd=0.0,
-            input_tokens=0,
-            output_tokens=0,
-            cache_read_input_tokens=0,
-            cache_creation_input_tokens=0,
-            cache_read_msats=0,
-            cache_creation_msats=0,
-        )
+        return _empty_cost(MaxCostData)
 
     usage_data = response_data.get("usage") or {}
     if not isinstance(usage_data, dict):
@@ -109,6 +119,27 @@ async def calculate_cost(
     # Try USD cost first
     usd_cost = _resolve_usd_cost(usage_data, response_data)
     if usd_cost > 0:
+        truly_empty = (
+            input_tokens == 0
+            and output_tokens == 0
+            and cache_read_tokens == 0
+            and cache_creation_tokens == 0
+        )
+        if truly_empty:
+            logger.warning(
+                "Upstream reported a USD cost but the response carries no "
+                "tokens at all (input, output, cache-read and cache-creation "
+                "are all zero) — refunding in full rather than billing the "
+                "USD-derived cost for an empty response.",
+                extra={
+                    "model": response_data.get("model", "unknown"),
+                    "usd_cost": usd_cost,
+                    "usage_keys": sorted(usage_data.keys())
+                    if isinstance(usage_data, dict)
+                    else None,
+                },
+            )
+            return _empty_cost()
         if input_tokens == 0 and output_tokens == 0:
             logger.warning(
                 "Upstream reported a USD cost but no token counts — "

routstr/payment/models.py

@@ -94,7 +94,10 @@ def backfill_cache_pricing(model_id: str, pricing: Pricing) -> Pricing:
     cache reads (DeepSeek hits are 10x cheaper) and undercharges Anthropic
     cache writes (1.25x). litellm ships per-model USD rates keyed by the exact
     OpenRouter id (deepseek/deepseek-chat) or by the bare model name
-    (gpt-4o, claude-sonnet-4-5), so both spellings are tried.
+    (gpt-4o, claude-sonnet-4-5), so both spellings are tried. litellm keys are
+    lowercase, but a generic upstream may report a mixed-case id
+    (``deepseek-ai/DeepSeek-V4-Flash``); an exact match is attempted first, then
+    a case-insensitive fallback so such ids still resolve.
 
     Rates already present (e.g. provided by OpenRouter) are authoritative and
     never overwritten. Unknown models are returned unchanged.
@@ -106,12 +109,26 @@ def backfill_cache_pricing(model_id: str, pricing: Pricing) -> Pricing:
 
     import litellm
 
+    candidates = (model_id, model_id.split("/", 1)[-1])
     info: dict | None = None
-    for key in (model_id, model_id.split("/", 1)[-1]):
+    for key in candidates:
         candidate = litellm.model_cost.get(key)
         if isinstance(candidate, dict):
             info = candidate
             break
+    if info is None:
+        # Case-insensitive fallback: a mixed-case upstream id (e.g.
+        # ``deepseek-ai/DeepSeek-V4-Flash``) won't match litellm's lowercase
+        # keys exactly. Build a lowercased index once and retry.
+        lowered = {c.lower() for c in candidates}
+        for key, candidate in litellm.model_cost.items():
+            if (
+                isinstance(key, str)
+                and key.lower() in lowered
+                and isinstance(candidate, dict)
+            ):
+                info = candidate
+                break
     if info is None:
         return pricing
 
@@ -215,13 +232,29 @@ def _row_to_model(
     )
     top_provider_dict = json.loads(row.top_provider) if row.top_provider else None
 
-    if apply_provider_fee and isinstance(pricing, dict):
-        pricing = {k: float(v) * provider_fee for k, v in pricing.items()}
-
     if isinstance(pricing, dict) and float(pricing.get("request", 0.0)) <= 0.0:
         pricing["request"] = max(pricing.get("request", 0.0), 0.0)
 
     parsed_pricing = Pricing.parse_obj(pricing)
+
+    # Fill missing cache-read/write rates from litellm's cost map BEFORE applying
+    # the provider fee, so they carry the same markup as every other component.
+    # DB-stored override pricing (e.g. generic providers) omits cache rates;
+    # without this, ``_row_to_model`` bills cache reads at the full input rate —
+    # the ``_apply_provider_fee_to_model`` path backfills, but the override path
+    # used for admin-configured providers did not.
+    #
+    # Key on ``forwarded_model_id`` (the actual upstream model name litellm
+    # prices) when set: an alias row (id="local-alias",
+    # forwarded_model_id="deepseek-v4-flash") would otherwise look up the alias
+    # and miss the cache rate.
+    pricing_model_id = getattr(row, "forwarded_model_id", None) or row.id
+    parsed_pricing = backfill_cache_pricing(pricing_model_id, parsed_pricing)
+
+    if apply_provider_fee:
+        parsed_pricing = Pricing.parse_obj(
+            {k: float(v) * provider_fee for k, v in parsed_pricing.dict().items()}
+        )
     model = Model(
         id=row.id,
         name=row.name,
@@ -563,6 +596,29 @@ async def test_model(
         }
 
 
+@models_router.get("/v1/models/paths")
+@models_router.get("/v1/models/paths/", include_in_schema=False)
+async def model_paths() -> dict:
+    """All models with every upstream provider path they are reachable through."""
+    from ..upstream.model_paths import get_all_model_paths
+
+    return {"data": await get_all_model_paths()}
+
+
+@models_router.get("/v1/models/paths/model")
+@models_router.get("/v1/models/paths/model/", include_in_schema=False)
+async def model_paths_for_model(model_id: str) -> dict:
+    """Paths for a single model.
+
+    Uses a query parameter (``?model_id=...``) under a fully static route so
+    model ids containing ``/`` (e.g. ``anthropic/claude-opus-4.6``) need no URL
+    encoding and there is no dynamic-route ambiguity.
+    """
+    from ..upstream.model_paths import get_paths_for_model
+
+    return {"data": await get_paths_for_model(model_id)}
+
+
 @models_router.get("/v1/models")
 @models_router.get("/v1/models/", include_in_schema=False)
 @models_router.get("/models")

routstr/upstream/anthropic.py

@@ -24,7 +24,7 @@ class AnthropicUpstreamProvider(BaseUpstreamProvider):
         )
 
     @classmethod
-    def from_db_row(
+    def _build_from_row(
         cls, provider_row: "UpstreamProviderRow"
     ) -> "AnthropicUpstreamProvider":
         return cls(

routstr/upstream/auto_topup.py

@@ -97,6 +97,8 @@ async def _check_and_topup(row: UpstreamProviderRow) -> None:
 
     # Instantiate provider and check balance
     provider = RoutstrUpstreamProvider.from_db_row(row)
+    if provider is None:
+        return
     balance = await provider.get_balance()
 
     if balance is None:

routstr/upstream/azure.py

@@ -38,7 +38,7 @@ class AzureUpstreamProvider(BaseUpstreamProvider):
         self.api_version = api_version
 
     @classmethod
-    def from_db_row(
+    def _build_from_row(
         cls, provider_row: "UpstreamProviderRow"
     ) -> "AzureUpstreamProvider | None":
         if not provider_row.api_version:

routstr/upstream/base.py

@@ -6,13 +6,12 @@ import math
 import traceback
 import uuid
 from collections.abc import AsyncGenerator, AsyncIterator, Iterator
-from typing import Any, Mapping, cast
+from typing import Any, Mapping, Self, cast
 
 import httpx
 from fastapi import BackgroundTasks, HTTPException, Request
 from fastapi.responses import Response, StreamingResponse
 from pydantic.v1 import BaseModel
-from sqlmodel import select
 
 from ..auth import adjust_payment_for_tokens
 from ..core import get_logger
@@ -92,6 +91,11 @@ class BaseUpstreamProvider:
     base_url: str
     api_key: str
     provider_fee: float = 1.05
+    # Primary key of the ``upstream_providers`` row this instance was built
+    # from. Set by ``from_db_row`` so a live provider can re-find its own row by
+    # stable identity instead of its rotatable ``api_key``. ``None`` for
+    # instances not sourced from a row.
+    db_id: int | None = None
     _models_cache: list[Model] = []
     _models_by_id: dict[str, Model] = {}
 
@@ -106,6 +110,7 @@ class BaseUpstreamProvider:
         self.base_url = base_url
         self.api_key = api_key
         self.provider_fee = provider_fee
+        self.db_id = None
         self._models_cache = []
         self._models_by_id = {}
 
@@ -123,10 +128,13 @@ class BaseUpstreamProvider:
         return detect_litellm_prefix(self.base_url)
 
     @classmethod
-    def from_db_row(
-        cls, provider_row: "UpstreamProviderRow"
-    ) -> "BaseUpstreamProvider | None":
-        """Factory method to instantiate provider from database row.
+    def from_db_row(cls, provider_row: "UpstreamProviderRow") -> "Self | None":
+        """Instantiate a provider from a database row, carrying its identity.
+
+        Construction itself is delegated to the ``_build_from_row`` hook (which
+        subclasses override to match their constructor); this wrapper stamps the
+        row's primary key onto the instance as ``db_id`` so the provider can
+        later re-find its own row by identity rather than by its ``api_key``.
 
         Args:
             provider_row: Database row containing provider configuration
@@ -134,6 +142,19 @@ class BaseUpstreamProvider:
         Returns:
             Instantiated provider or None if instantiation fails
         """
+        provider = cls._build_from_row(provider_row)
+        if provider is not None:
+            provider.db_id = provider_row.id
+        return provider
+
+    @classmethod
+    def _build_from_row(cls, provider_row: "UpstreamProviderRow") -> "Self | None":
+        """Construct the provider instance from a row (no identity stamping).
+
+        Overridden by subclasses whose constructors differ from the base
+        ``(base_url, api_key, provider_fee)`` shape. Callers should use
+        ``from_db_row`` instead, which also attaches ``db_id``.
+        """
         return cls(
             base_url=provider_row.base_url,
             api_key=provider_row.api_key,
@@ -4878,14 +4899,11 @@ class BaseUpstreamProvider:
         """Refresh the in-memory models cache from upstream API."""
         try:
             async with create_session() as session:
-                stmt = select(UpstreamProviderRow).where(
-                    UpstreamProviderRow.base_url == self.base_url,
-                    UpstreamProviderRow.api_key == self.api_key,
+                provider = (
+                    await session.get(UpstreamProviderRow, self.db_id)
+                    if self.db_id is not None
+                    else None
                 )
-                result = await session.exec(stmt)
-
-                # .first() returns the object or None if not found
-                provider = result.first()
                 if not provider or not provider.id:
                     raise HTTPException(status_code=404, detail="Provider not found")
 

routstr/upstream/deepseek_v4_pricing_shim.py

@@ -3,10 +3,15 @@
 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 ~60% overcharge on cache hits (DeepSeek hits are ~0.2x 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 open upstream PR
+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).
 
@@ -22,21 +27,23 @@ from ..core import get_logger
 
 logger = get_logger(__name__)
 
-# USD per token. Source: BerriAI/litellm PR #26380.
+# 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-08,
+        "cache_read_input_token_cost": 2.8e-09,
         "cache_creation_input_token_cost": 0.0,
-        "input_cost_per_token_cache_hit": 2.8e-08,
+        "input_cost_per_token_cache_hit": 2.8e-09,
     },
     "deepseek-v4-pro": {
-        "input_cost_per_token": 1.74e-06,
-        "output_cost_per_token": 3.48e-06,
-        "cache_read_input_token_cost": 1.4e-07,
+        "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": 1.4e-07,
+        "input_cost_per_token_cache_hit": 3.625e-09,
     },
 }
 

routstr/upstream/fireworks.py

@@ -20,7 +20,7 @@ class FireworksUpstreamProvider(BaseUpstreamProvider):
         )
 
     @classmethod
-    def from_db_row(
+    def _build_from_row(
         cls, provider_row: "UpstreamProviderRow"
     ) -> "FireworksUpstreamProvider":
         return cls(

routstr/upstream/gemini.py

@@ -51,7 +51,7 @@ class GeminiUpstreamProvider(BaseUpstreamProvider):
         return self._client
 
     @classmethod
-    def from_db_row(
+    def _build_from_row(
         cls, provider_row: "UpstreamProviderRow"
     ) -> "GeminiUpstreamProvider":
         return cls(

routstr/upstream/generic.py

@@ -45,7 +45,7 @@ class GenericUpstreamProvider(BaseUpstreamProvider):
         )
 
     @classmethod
-    def from_db_row(
+    def _build_from_row(
         cls, provider_row: "UpstreamProviderRow"
     ) -> "GenericUpstreamProvider":
         return cls(

routstr/upstream/groq.py

@@ -20,7 +20,7 @@ class GroqUpstreamProvider(BaseUpstreamProvider):
         )
 
     @classmethod
-    def from_db_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,

routstr/upstream/helpers.py

@@ -12,6 +12,7 @@ from sqlmodel import select
 
 from ..core import get_logger
 from ..core.db import AsyncSession, ModelRow, UpstreamProviderRow, create_session
+from ..core.provider_slugs import allocate_unique_provider_slug
 from ..payment.models import Model
 from .base import BaseUpstreamProvider
 
@@ -215,9 +216,6 @@ async def init_upstreams() -> list[BaseUpstreamProvider]:
 
             provider = _instantiate_provider(provider_row)
             if provider:
-                # Keep provider DB id on runtime instance so model mapping can
-                # bind DB overrides to the correct upstream.
-                setattr(provider, "db_id", provider_row.id)
                 await provider.refresh_models_cache()
                 logger.debug(
                     f"Initialized {provider_row.provider_type} provider",
@@ -250,6 +248,7 @@ async def _seed_providers_from_settings(
 
     providers_to_add: list[UpstreamProviderRow] = []
     seeded_provider_keys: set[tuple[str, str]] = set()
+    reserved_slugs: set[str] = set()
 
     provider_classes_by_type = {
         cls.provider_type: cls
@@ -279,8 +278,13 @@ async def _seed_providers_from_settings(
                     )
                 )
                 if not result.first():
+                    slug = await allocate_unique_provider_slug(
+                        session, provider_type, reserved_slugs
+                    )
+                    reserved_slugs.add(slug)
                     providers_to_add.append(
                         UpstreamProviderRow(
+                            slug=slug,
                             provider_type=provider_type,
                             base_url=base_url,
                             api_key=api_key,
@@ -299,8 +303,13 @@ async def _seed_providers_from_settings(
             )
         )
         if not result.first():
+            slug = await allocate_unique_provider_slug(
+                session, "ollama", reserved_slugs
+            )
+            reserved_slugs.add(slug)
             providers_to_add.append(
                 UpstreamProviderRow(
+                    slug=slug,
                     provider_type="ollama",
                     base_url=ollama_base_url,
                     api_key=ollama_api_key,
@@ -320,8 +329,13 @@ async def _seed_providers_from_settings(
                 )
             )
             if not result.first():
+                slug = await allocate_unique_provider_slug(
+                    session, "azure", reserved_slugs
+                )
+                reserved_slugs.add(slug)
                 providers_to_add.append(
                     UpstreamProviderRow(
+                        slug=slug,
                         provider_type="azure",
                         base_url=base_url,
                         api_key=api_key,
@@ -342,8 +356,13 @@ async def _seed_providers_from_settings(
                 )
             )
             if not result.first():
+                slug = await allocate_unique_provider_slug(
+                    session, "custom", reserved_slugs
+                )
+                reserved_slugs.add(slug)
                 providers_to_add.append(
                     UpstreamProviderRow(
+                        slug=slug,
                         provider_type="custom",
                         base_url=base_url,
                         api_key=api_key,
@@ -356,7 +375,7 @@ async def _seed_providers_from_settings(
         session.add(provider)
         logger.info(
             f"Seeding {provider.provider_type} provider",  # type: ignore[str-format]
-            extra={"base_url": provider.base_url},
+            extra={"base_url": provider.base_url, "slug": provider.slug},
         )
 
 
@@ -391,9 +410,7 @@ def _instantiate_provider(
             return provider
 
         if provider_row.provider_type == "custom":
-            return BaseUpstreamProvider(
-                provider_row.base_url, provider_row.api_key, provider_row.provider_fee
-            )
+            return BaseUpstreamProvider.from_db_row(provider_row)
 
         logger.error(
             f"Unknown provider type: {provider_row.provider_type}",

routstr/upstream/model_paths.py

@@ -0,0 +1,472 @@
+"""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.
+
+A *path* is the provider string that may appear in Routstr chat completion
+responses (see ``BaseUpstreamProvider._apply_provider_field``):
+
+- Direct upstream -> ``<provider_type>`` e.g. ``anthropic``
+- Generic/custom OpenRouter-compatible upstream -> ``generic:<name>``
+- Native OpenRouter routing to a sub-provider -> ``openrouter:<name>``
+
+Native OpenRouter does not emit a useful bare ``openrouter`` path when no
+sub-provider is present; it reports ``unknown`` instead.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import random
+from typing import TYPE_CHECKING, Callable
+
+import httpx
+from sqlalchemy.orm import selectinload
+from sqlmodel import col, delete, select
+
+from ..core.db import ModelPathRow, ModelRow, UpstreamProviderRow, create_session
+from ..core.logging import get_logger
+from .base import BaseUpstreamProvider
+
+if TYPE_CHECKING:
+    from ..payment.models import Model
+
+logger = get_logger(__name__)
+
+# Bound the per-model OpenRouter /endpoints fan-out so a provider with hundreds
+# of models does not open hundreds of concurrent requests every refresh.
+_OPENROUTER_CONCURRENCY = 5
+_OPENROUTER_TIMEOUT_SECONDS = 10.0
+
+
+def is_openrouter_base_url(base_url: str | None) -> bool:
+    """True when ``base_url`` points at OpenRouter.
+
+    Deliberately separate from ``BaseUpstreamProvider._upstream_accepts_cache_control``:
+    that predicate also returns True for native Anthropic (correct for
+    cache-control, wrong for OpenRouter endpoint discovery). This one keys only
+    on the URL so a ``GenericUpstreamProvider`` aimed at OpenRouter is matched
+    while native Anthropic is not.
+    """
+    return "openrouter.ai" in (base_url or "")
+
+
+def exposed_model_id(model: object) -> str:
+    """Client-visible ``/v1/models`` id for a cached model."""
+    forwarded = getattr(model, "forwarded_model_id", None)
+    return forwarded or getattr(model, "id")
+
+
+def public_model_id(model_id: str) -> str:
+    """Model id exposed by model-path API responses.
+
+    Provider-prefixed ids such as ``z-ai/glm-5v-turbo`` are returned as
+    ``glm-5v-turbo`` so clients can search and display the same unqualified id
+    they pass to ``/v1/models/paths/model``.
+    """
+    return model_id.rsplit("/", 1)[-1]
+
+
+def openrouter_author_slug(model: object) -> str | None:
+    """Return a canonical ``author/slug`` for the OpenRouter endpoints API.
+
+    OpenRouter requires the canonical id, never ``forwarded_model_id``. Prefer
+    ``canonical_slug``, then a slash-containing ``id``; otherwise there is no
+    usable form and endpoint discovery is skipped for this model.
+    """
+    canonical = getattr(model, "canonical_slug", None)
+    if canonical and "/" in canonical:
+        return canonical
+    model_id = getattr(model, "id", None)
+    if model_id and "/" in model_id:
+        return model_id
+    return None
+
+
+async def _fetch_openrouter_endpoint_paths(
+    client: httpx.AsyncClient,
+    base_url: str,
+    api_key: str,
+    author_slug: str,
+    path_prefix: str,
+    semaphore: asyncio.Semaphore,
+) -> list[str]:
+    """Return ``<path_prefix>:<provider_name>`` paths for one model, or ``[]``.
+
+    Failures (network, rate limit, bad payload) are logged and swallowed so one
+    model never breaks the whole refresh.
+    """
+    url = f"{base_url.rstrip('/')}/models/{author_slug}/endpoints"
+    headers = {"Authorization": f"Bearer {api_key}"} if api_key else {}
+    async with semaphore:
+        try:
+            resp = await client.get(
+                url, headers=headers, timeout=_OPENROUTER_TIMEOUT_SECONDS
+            )
+        except Exception as e:  # noqa: BLE001 - isolate per-model failures
+            logger.warning(
+                "OpenRouter endpoint discovery request failed",
+                extra={"author_slug": author_slug, "error": str(e)},
+            )
+            return []
+
+    if resp.status_code == 429:
+        logger.warning(
+            "OpenRouter endpoint discovery rate-limited",
+            extra={"author_slug": author_slug},
+        )
+        return []
+    if resp.status_code != 200:
+        logger.warning(
+            "OpenRouter endpoint discovery non-200",
+            extra={"author_slug": author_slug, "status_code": resp.status_code},
+        )
+        return []
+
+    try:
+        endpoints = resp.json().get("data", {}).get("endpoints", [])
+    except Exception as e:  # noqa: BLE001
+        logger.warning(
+            "OpenRouter endpoint discovery bad payload",
+            extra={"author_slug": author_slug, "error": str(e)},
+        )
+        return []
+
+    paths: list[str] = []
+    for endpoint in endpoints:
+        provider_name = (endpoint or {}).get("provider_name")
+        if provider_name:
+            paths.append(f"{path_prefix}:{provider_name}")
+    # De-duplicate while preserving order.
+    return list(dict.fromkeys(paths))
+
+
+async def _load_model_visibility() -> tuple[
+    dict[str, tuple[ModelRow, float]], set[str], set[int]
+]:
+    """Load the same DB model visibility inputs used by routing.
+
+    ``refresh_model_maps`` builds routing from enabled providers, enabled DB
+    override rows, and disabled model ids. Model-path discovery uses the same
+    view so the discovery API does not advertise models routing would hide and
+    reports forwarded aliases from DB overrides consistently with ``/v1/models``.
+    """
+    async with create_session() as session:
+        query = select(UpstreamProviderRow).options(
+            selectinload(UpstreamProviderRow.models)  # type: ignore[arg-type]
+        )
+        provider_rows = (await session.exec(query)).all()
+
+    overrides_by_id: dict[str, tuple[ModelRow, float]] = {}
+    disabled_model_ids: set[str] = set()
+    enabled_provider_ids: set[int] = set()
+
+    for provider in provider_rows:
+        if not provider.enabled:
+            continue
+        if provider.id is not None:
+            enabled_provider_ids.add(provider.id)
+        for model in provider.models:
+            if model.enabled:
+                overrides_by_id[model.id] = (model, provider.provider_fee)
+            else:
+                disabled_model_ids.add(model.id)
+
+    return overrides_by_id, disabled_model_ids, enabled_provider_ids
+
+
+def _row_to_visible_model(
+    model_id: str,
+    row: ModelRow,
+    provider_fee: float,
+) -> Model | None:
+    """Convert an enabled DB override row into a routed model object."""
+    from ..payment.models import _row_to_model
+
+    try:
+        return _row_to_model(row, apply_provider_fee=True, provider_fee=provider_fee)
+    except Exception as exc:  # noqa: BLE001 - skip invalid override row
+        logger.warning(
+            "Skipping invalid model override while collecting model paths",
+            extra={
+                "model_id": model_id,
+                "upstream_provider_id": getattr(row, "upstream_provider_id", None),
+                "error": str(exc),
+                "error_type": type(exc).__name__,
+            },
+        )
+        return None
+
+
+def _apply_model_visibility(
+    upstream: BaseUpstreamProvider,
+    overrides_by_id: dict[str, tuple[ModelRow, float]] | None,
+    disabled_model_ids: set[str] | None,
+) -> list[object]:
+    """Return provider models after DB disabled/override state is applied."""
+    overrides_by_id = overrides_by_id or {}
+    disabled_model_ids = disabled_model_ids or set()
+    visible_models: list[object] = []
+    seen_model_ids: set[str] = set()
+
+    for model in upstream.get_cached_models():
+        model_id = getattr(model, "id", "")
+        if not getattr(model, "enabled", True) or model_id in disabled_model_ids:
+            continue
+
+        if model_id in overrides_by_id:
+            override_row, provider_fee = overrides_by_id[model_id]
+            visible_model = _row_to_visible_model(model_id, override_row, provider_fee)
+            if visible_model is None:
+                continue
+            model = visible_model
+
+        if not getattr(model, "enabled", True):
+            continue
+        visible_models.append(model)
+        seen_model_ids.add(model_id.lower())
+
+    upstream_provider_id = getattr(upstream, "db_id", None)
+    if isinstance(upstream_provider_id, int):
+        for model_id, (override_row, provider_fee) in overrides_by_id.items():
+            if model_id in disabled_model_ids:
+                continue
+            if (
+                getattr(override_row, "upstream_provider_id", None)
+                != upstream_provider_id
+            ):
+                continue
+            if model_id.lower() in seen_model_ids:
+                continue
+            override_model = _row_to_visible_model(model_id, override_row, provider_fee)
+            if override_model is None:
+                continue
+            if getattr(override_model, "enabled", True):
+                visible_models.append(override_model)
+                seen_model_ids.add(model_id.lower())
+
+    return visible_models
+
+
+async def _collect_provider_paths(
+    upstream: BaseUpstreamProvider,
+    overrides_by_id: dict[str, tuple[ModelRow, float]] | None = None,
+    disabled_model_ids: set[str] | None = None,
+) -> list[tuple[str, str]]:
+    """Collect ``(model_id, path)`` pairs for one provider instance.
+
+    Emits the direct ``<provider_type>`` path for normal upstreams. For
+    OpenRouter-compatible providers, emits one path per OpenRouter sub-provider
+    endpoint, prefixed the same way response stamping prefixes it.
+    """
+    provider_type = (upstream.provider_type or "").strip()
+    models = _apply_model_visibility(upstream, overrides_by_id, disabled_model_ids)
+    is_openrouter = is_openrouter_base_url(upstream.base_url)
+
+    pairs: list[tuple[str, str]] = []
+    if not is_openrouter:
+        for model in models:
+            if provider_type:
+                pairs.append((exposed_model_id(model), provider_type))
+        return pairs
+
+    if not provider_type:
+        return pairs
+
+    semaphore = asyncio.Semaphore(_OPENROUTER_CONCURRENCY)
+    async with httpx.AsyncClient() as client:
+
+        async def _for_model(model: object) -> list[tuple[str, str]]:
+            author_slug = openrouter_author_slug(model)
+            if not author_slug:
+                return []
+            paths = await _fetch_openrouter_endpoint_paths(
+                client,
+                upstream.base_url,
+                upstream.api_key,
+                author_slug,
+                provider_type,
+                semaphore,
+            )
+            model_id = exposed_model_id(model)
+            return [(model_id, path) for path in paths]
+
+        results = await asyncio.gather(
+            *(_for_model(m) for m in models), return_exceptions=True
+        )
+
+    for result in results:
+        if isinstance(result, BaseException):
+            logger.warning(
+                "OpenRouter endpoint discovery task errored",
+                extra={"provider": provider_type, "error": str(result)},
+            )
+            continue
+        pairs.extend(result)
+
+    return pairs
+
+
+async def _persist_provider_paths(
+    upstream_provider_id: int, pairs: list[tuple[str, str]]
+) -> None:
+    """Replace all rows for ``upstream_provider_id`` with ``pairs``.
+
+    Replacement (not upsert) so stale paths disappear when provider config or
+    upstream availability changes.
+    """
+    unique_pairs = list(dict.fromkeys(pairs))
+    async with create_session() as session:
+        await session.exec(  # type: ignore[call-overload]
+            delete(ModelPathRow).where(
+                col(ModelPathRow.upstream_provider_id) == upstream_provider_id
+            )
+        )
+        for model_id, path in unique_pairs:
+            session.add(
+                ModelPathRow(
+                    model_id=model_id,
+                    path=path,
+                    upstream_provider_id=upstream_provider_id,
+                )
+            )
+        await session.commit()
+
+
+async def _prune_inactive_provider_paths(active_provider_ids: set[int]) -> None:
+    """Delete paths for providers no longer present in the live upstream set."""
+    async with create_session() as session:
+        stmt = delete(ModelPathRow)
+        if active_provider_ids:
+            stmt = stmt.where(
+                col(ModelPathRow.upstream_provider_id).not_in(active_provider_ids)
+            )
+        await session.exec(stmt)  # type: ignore[call-overload]
+        await session.commit()
+
+
+async def refresh_model_paths(
+    upstreams: list[BaseUpstreamProvider],
+) -> None:
+    """Recompute and persist model paths for every enabled provider.
+
+    One provider's failure is logged and isolated; it must not break the rest.
+    """
+    (
+        overrides_by_id,
+        disabled_model_ids,
+        enabled_provider_ids,
+    ) = await _load_model_visibility()
+    active_provider_ids = {
+        upstream.db_id
+        for upstream in upstreams
+        if upstream.db_id is not None and upstream.db_id in enabled_provider_ids
+    }
+    await _prune_inactive_provider_paths(active_provider_ids)
+
+    for upstream in upstreams:
+        if upstream.db_id is None or upstream.db_id not in enabled_provider_ids:
+            continue
+        try:
+            pairs = await _collect_provider_paths(
+                upstream,
+                overrides_by_id=overrides_by_id,
+                disabled_model_ids=disabled_model_ids,
+            )
+            await _persist_provider_paths(upstream.db_id, pairs)
+        except Exception as e:  # noqa: BLE001 - isolate per-provider failures
+            logger.error(
+                "Failed to refresh model paths for provider",
+                extra={
+                    "provider": upstream.provider_type or upstream.base_url,
+                    "db_id": upstream.db_id,
+                    "error": str(e),
+                    "error_type": type(e).__name__,
+                },
+            )
+
+
+async def refresh_model_paths_periodically(
+    upstreams_provider: (
+        Callable[[], list[BaseUpstreamProvider]] | list[BaseUpstreamProvider]
+    ),
+) -> None:
+    """Background task mirroring ``refresh_upstreams_models_periodically``."""
+    from ..core.settings import settings
+
+    interval = getattr(settings, "model_paths_refresh_interval_seconds", 0)
+    if not interval or interval <= 0:
+        logger.info("Model paths refresh disabled (interval <= 0)")
+        return
+
+    def _resolve_upstreams() -> list[BaseUpstreamProvider]:
+        if callable(upstreams_provider):
+            return upstreams_provider()
+        return upstreams_provider
+
+    while True:
+        try:
+            await refresh_model_paths(_resolve_upstreams())
+        except asyncio.CancelledError:
+            break
+        except Exception as e:  # noqa: BLE001
+            logger.error(
+                "Error in model paths refresh loop",
+                extra={"error": str(e), "error_type": type(e).__name__},
+            )
+
+        try:
+            jitter = max(0.0, float(interval) * 0.1)
+            await asyncio.sleep(interval + random.uniform(0, jitter))
+        except asyncio.CancelledError:
+            break
+
+
+async def get_all_model_paths() -> list[dict]:
+    """All models with their paths, shaped for ``GET /v1/models/paths``."""
+    async with create_session() as session:
+        rows = (
+            await session.exec(select(ModelPathRow).order_by(ModelPathRow.model_id))
+        ).all()
+
+    grouped: dict[str, list[dict]] = {}
+    seen_paths: dict[str, set[str]] = {}
+    for row in rows:
+        model_id = public_model_id(row.model_id)
+        if row.path in seen_paths.setdefault(model_id, set()):
+            continue
+        seen_paths[model_id].add(row.path)
+        grouped.setdefault(model_id, []).append({"path": row.path})
+    return [{"id": model_id, "paths": paths} for model_id, paths in grouped.items()]
+
+
+async def get_paths_for_model(model_id: str) -> list[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.
+    """
+    requested_id = public_model_id(model_id)
+    async with create_session() as session:
+        rows = (
+            await session.exec(
+                select(ModelPathRow).order_by(
+                    ModelPathRow.path,
+                    col(ModelPathRow.upstream_provider_id),
+                    ModelPathRow.model_id,
+                )
+            )
+        ).all()
+
+    seen: set[str] = set()
+    paths: list[dict] = []
+    for row in rows:
+        if public_model_id(row.model_id) != requested_id:
+            continue
+        if row.path in seen:
+            continue
+        seen.add(row.path)
+        paths.append({"path": row.path})
+    return paths

routstr/upstream/ollama.py

@@ -43,7 +43,7 @@ class OllamaUpstreamProvider(BaseUpstreamProvider):
         )
 
     @classmethod
-    def from_db_row(
+    def _build_from_row(
         cls, provider_row: "UpstreamProviderRow"
     ) -> "OllamaUpstreamProvider":
         return cls(

routstr/upstream/openai.py

@@ -20,7 +20,7 @@ class OpenAIUpstreamProvider(BaseUpstreamProvider):
         )
 
     @classmethod
-    def from_db_row(
+    def _build_from_row(
         cls, provider_row: "UpstreamProviderRow"
     ) -> "OpenAIUpstreamProvider":
         return cls(

routstr/upstream/openrouter.py

@@ -90,7 +90,7 @@ class OpenRouterUpstreamProvider(BaseUpstreamProvider):
         )
 
     @classmethod
-    def from_db_row(
+    def _build_from_row(
         cls, provider_row: "UpstreamProviderRow"
     ) -> "OpenRouterUpstreamProvider":
         return cls(

routstr/upstream/perplexity.py

@@ -23,7 +23,7 @@ class PerplexityUpstreamProvider(BaseUpstreamProvider):
         )
 
     @classmethod
-    def from_db_row(
+    def _build_from_row(
         cls, provider_row: "UpstreamProviderRow"
     ) -> "PerplexityUpstreamProvider":
         return cls(

routstr/upstream/ppqai.py

@@ -46,7 +46,7 @@ class PPQAIUpstreamProvider(BaseUpstreamProvider):
         )
 
     @classmethod
-    def from_db_row(
+    def _build_from_row(
         cls, provider_row: "UpstreamProviderRow"
     ) -> "PPQAIUpstreamProvider":
         return cls(
@@ -229,17 +229,14 @@ class PPQAIUpstreamProvider(BaseUpstreamProvider):
                 f"Disabling PPQ.AI provider ({self.base_url}) due to insufficient balance",
                 extra={"error": error_message},
             )
-            from sqlmodel import select
-
             from ..core.db import UpstreamProviderRow, create_session
 
             async with create_session() as session:
-                statement = select(UpstreamProviderRow).where(
-                    UpstreamProviderRow.base_url == self.base_url,
-                    UpstreamProviderRow.api_key == self.api_key,
+                provider = (
+                    await session.get(UpstreamProviderRow, self.db_id)
+                    if self.db_id is not None
+                    else None
                 )
-                result = await session.exec(statement)
-                provider = result.first()
 
                 if provider:
                     provider.enabled = False

routstr/upstream/routstr.py

@@ -55,7 +55,7 @@ class RoutstrUpstreamProvider(BaseUpstreamProvider):
         return path.lstrip("/")
 
     @classmethod
-    def from_db_row(
+    def _build_from_row(
         cls, provider_row: "UpstreamProviderRow"
     ) -> "RoutstrUpstreamProvider":
         import json

routstr/upstream/xai.py

@@ -21,7 +21,7 @@ class XAIUpstreamProvider(BaseUpstreamProvider):
         )
 
     @classmethod
-    def from_db_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,

tests/integration/test_admin_model_cache_pricing.py

@@ -0,0 +1,229 @@
+import json
+from datetime import datetime, timedelta, timezone
+from unittest.mock import patch
+
+import pytest
+from httpx import AsyncClient
+from sqlmodel.ext.asyncio.session import AsyncSession
+
+from routstr.core.admin import admin_sessions
+from routstr.core.db import ModelRow, UpstreamProviderRow
+from routstr.payment.cost_calculation import CostData, calculate_cost
+from routstr.proxy import get_model_instance, reinitialize_upstreams
+
+
+def _admin_headers() -> dict[str, str]:
+    token = "test-admin-cache-pricing-token"
+    admin_sessions[token] = int(
+        (datetime.now(timezone.utc) + timedelta(minutes=5)).timestamp()
+    )
+    return {"Authorization": f"Bearer {token}"}
+
+
+def _model_payload(
+    provider_id: int,
+    *,
+    cache_read: float,
+    cache_write: float,
+) -> dict[str, object]:
+    return {
+        "id": "custom-cache-model",
+        "name": "Custom Cache Model",
+        "description": "custom model with explicit cache pricing",
+        "created": 0,
+        "context_length": 128000,
+        "architecture": {
+            "modality": "text",
+            "input_modalities": ["text"],
+            "output_modalities": ["text"],
+            "tokenizer": "unknown",
+            "instruct_type": None,
+        },
+        "pricing": {
+            "prompt": 1.4e-7,
+            "completion": 2.8e-7,
+            "input_cache_read": cache_read,
+            "input_cache_write": cache_write,
+            "request": 0.0,
+            "image": 0.0,
+            "web_search": 0.0,
+            "internal_reasoning": 0.0,
+        },
+        "per_request_limits": None,
+        "top_provider": None,
+        "upstream_provider_id": provider_id,
+        "canonical_slug": None,
+        "alias_ids": [],
+        "enabled": True,
+        "forwarded_model_id": "custom-cache-model",
+    }
+
+
+@pytest.mark.integration
+@pytest.mark.asyncio
+async def test_admin_provider_model_api_persists_cache_pricing_on_create_and_update(
+    integration_client: AsyncClient,
+    integration_session: AsyncSession,
+) -> None:
+    provider = UpstreamProviderRow(
+        provider_type="generic",
+        base_url="https://custom-upstream.example/v1",
+        api_key="test-key",
+        provider_fee=1.0,
+    )
+    integration_session.add(provider)
+    await integration_session.commit()
+    await integration_session.refresh(provider)
+    assert provider.id is not None
+    await reinitialize_upstreams()
+
+    headers = _admin_headers()
+    create_payload = _model_payload(
+        provider.id,
+        cache_read=2.8e-9,
+        cache_write=3.5e-9,
+    )
+
+    with patch("routstr.payment.models.sats_usd_price", return_value=1e-6):
+        create_response = await integration_client.post(
+            f"/admin/api/upstream-providers/{provider.id}/models",
+            headers=headers,
+            json=create_payload,
+        )
+
+    assert create_response.status_code == 200
+    create_body = create_response.json()
+    assert create_body["pricing"]["input_cache_read"] == pytest.approx(2.8e-9)
+    assert create_body["pricing"]["input_cache_write"] == pytest.approx(3.5e-9)
+
+    row = await integration_session.get(ModelRow, ("custom-cache-model", provider.id))
+    assert row is not None
+    stored_pricing = json.loads(row.pricing)
+    assert stored_pricing["input_cache_read"] == pytest.approx(2.8e-9)
+    assert stored_pricing["input_cache_write"] == pytest.approx(3.5e-9)
+
+    update_payload = _model_payload(
+        provider.id,
+        cache_read=1.25e-9,
+        cache_write=4.5e-9,
+    )
+
+    with patch("routstr.payment.models.sats_usd_price", return_value=1e-6):
+        update_response = await integration_client.post(
+            f"/admin/api/upstream-providers/{provider.id}/models",
+            headers=headers,
+            json=update_payload,
+        )
+
+    assert update_response.status_code == 200
+    update_body = update_response.json()
+    assert update_body["pricing"]["input_cache_read"] == pytest.approx(1.25e-9)
+    assert update_body["pricing"]["input_cache_write"] == pytest.approx(4.5e-9)
+
+    await integration_session.refresh(row)
+    updated_pricing = json.loads(row.pricing)
+    assert updated_pricing["input_cache_read"] == pytest.approx(1.25e-9)
+    assert updated_pricing["input_cache_write"] == pytest.approx(4.5e-9)
+
+    model = get_model_instance("custom-cache-model")
+    assert model is not None
+    assert model.sats_pricing is not None
+    assert model.sats_pricing.input_cache_read == pytest.approx(0.00125)
+    assert model.sats_pricing.input_cache_write == pytest.approx(0.0045)
+
+    with patch("routstr.payment.cost_calculation.sats_usd_price", return_value=1e-6):
+        cost = await calculate_cost(
+            {
+                "model": "custom-cache-model",
+                "usage": {
+                    "prompt_tokens": 1000,
+                    "completion_tokens": 100,
+                    "prompt_tokens_details": {"cached_tokens": 800},
+                },
+            },
+            max_cost=1_000_000,
+        )
+
+    assert isinstance(cost, CostData)
+    assert cost.input_tokens == 200
+    assert cost.cache_read_input_tokens == 800
+    assert cost.cache_read_msats == 1000
+    assert cost.output_msats == 28000
+    assert cost.input_msats == 29000
+    assert cost.total_msats == 57000
+
+
+@pytest.mark.integration
+@pytest.mark.asyncio
+async def test_upstream_response_cost_uses_model_cache_pricing(
+    integration_session: AsyncSession,
+    patched_db_engine: None,
+) -> None:
+    """A model's configured cache price must discount upstream cached-token usage."""
+    provider = UpstreamProviderRow(
+        provider_type="generic",
+        base_url="https://cache-priced-upstream.example/v1",
+        api_key="test-key",
+        provider_fee=1.0,
+    )
+    integration_session.add(provider)
+    await integration_session.commit()
+    await integration_session.refresh(provider)
+    assert provider.id is not None
+
+    row = ModelRow(
+        id="cache-priced-model",
+        name="Cache Priced Model",
+        description="model seeded with explicit cache pricing",
+        created=0,
+        context_length=128000,
+        architecture=json.dumps(
+            {
+                "modality": "text",
+                "input_modalities": ["text"],
+                "output_modalities": ["text"],
+                "tokenizer": "unknown",
+                "instruct_type": None,
+            }
+        ),
+        pricing=json.dumps(
+            {
+                "prompt": 1.4e-7,
+                "completion": 2.8e-7,
+                "input_cache_read": 1.25e-9,
+                "input_cache_write": 4.5e-9,
+            }
+        ),
+        upstream_provider_id=provider.id,
+        enabled=True,
+        forwarded_model_id="cache-priced-model",
+    )
+    integration_session.add(row)
+    await integration_session.commit()
+
+    with patch("routstr.payment.models.sats_usd_price", return_value=1e-6):
+        await reinitialize_upstreams()
+
+    with patch("routstr.payment.cost_calculation.sats_usd_price", return_value=1e-6):
+        cost = await calculate_cost(
+            {
+                "model": "cache-priced-model",
+                "usage": {
+                    "prompt_tokens": 1000,
+                    "completion_tokens": 100,
+                    "prompt_tokens_details": {"cached_tokens": 800},
+                },
+            },
+            max_cost=1_000_000,
+        )
+
+    assert isinstance(cost, CostData)
+    # prompt 1.4e-7 USD/token -> 0.14 sats/token -> 140 msats/token
+    # cache 1.25e-9 USD/token -> 0.00125 sats/token -> 1.25 msats/token
+    # completion 2.8e-7 USD/token -> 0.28 sats/token -> 280 msats/token
+    assert cost.input_tokens == 200
+    assert cost.cache_read_input_tokens == 800
+    assert cost.cache_read_msats == 1000
+    assert cost.input_msats == 29000
+    assert cost.output_msats == 28000
+    assert cost.total_msats == 57000

tests/integration/test_provider_self_lookup.py

@@ -0,0 +1,128 @@
+"""A live upstream provider resolves its OWN database row by stable identity
+(its primary key), not by its mutable/secret ``api_key``.
+
+Today ``from_db_row`` drops ``provider_row.id`` and the two self-referential
+paths — PPQ.AI's insufficient-balance self-disable and the base
+``refresh_models_cache`` — re-find their own row with
+``WHERE base_url == self.base_url AND api_key == self.api_key``. That uses a
+rotatable secret as a self-handle: the moment the row's key changes underneath a
+live object (a rotation racing an in-flight request), the object can no longer
+find itself. These tests pin the invariant that a provider looks itself up by
+identity, so the lookup survives a key change (and, later, key encryption).
+"""
+
+from unittest.mock import AsyncMock, patch
+
+import pytest
+
+from routstr.core.db import UpstreamProviderRow
+from routstr.upstream.ppqai import PPQAIUpstreamProvider
+
+
+@pytest.mark.integration
+@pytest.mark.asyncio
+async def test_provider_object_carries_its_persistent_identity(
+    integration_session: object,
+    patched_db_engine: None,
+) -> None:
+    """``from_db_row`` gives the in-memory object its row's identity (``db_id``)."""
+    row = UpstreamProviderRow(
+        provider_type="ppqai",
+        base_url="https://api.ppq.ai",
+        api_key="sk-original",
+        enabled=True,
+        provider_fee=1.0,
+    )
+    integration_session.add(row)  # type: ignore[attr-defined]
+    await integration_session.commit()  # type: ignore[attr-defined]
+    await integration_session.refresh(row)  # type: ignore[attr-defined]
+
+    provider = PPQAIUpstreamProvider.from_db_row(row)
+    assert provider is not None
+
+    assert provider.db_id == row.id
+
+
+@pytest.mark.integration
+@pytest.mark.asyncio
+async def test_self_disable_targets_own_row_after_key_rotation(
+    integration_session: object,
+    patched_db_engine: None,
+) -> None:
+    """PPQ.AI self-disable must disable *its* row even after the key rotated.
+
+    RED (current): the object holds the pre-rotation key, so the
+    ``(base_url, api_key)`` lookup misses the row → the provider is never
+    disabled. GREEN: lookup by ``id`` finds it and disables it.
+    """
+    row = UpstreamProviderRow(
+        provider_type="ppqai",
+        base_url="https://api.ppq.ai",
+        api_key="sk-original",
+        enabled=True,
+        provider_fee=1.0,
+    )
+    integration_session.add(row)  # type: ignore[attr-defined]
+    await integration_session.commit()  # type: ignore[attr-defined]
+    await integration_session.refresh(row)  # type: ignore[attr-defined]
+
+    provider = PPQAIUpstreamProvider.from_db_row(row)  # captures sk-original
+    assert provider is not None
+
+    # Key is rotated in the DB while `provider` is still live.
+    row.api_key = "sk-rotated"
+    integration_session.add(row)  # type: ignore[attr-defined]
+    await integration_session.commit()  # type: ignore[attr-defined]
+
+    with patch("routstr.proxy.reinitialize_upstreams", new=AsyncMock()):
+        await provider.on_upstream_error_redirect(402, "Insufficient balance")
+
+    await integration_session.refresh(row)  # type: ignore[attr-defined]
+    assert row.enabled is False
+
+
+@pytest.mark.integration
+@pytest.mark.asyncio
+async def test_refresh_models_cache_finds_own_row_after_key_rotation(
+    integration_session: object,
+    patched_db_engine: None,
+) -> None:
+    """``refresh_models_cache`` must resolve its own row after a key rotation.
+
+    ``refresh_models_cache`` swallows every exception (it only logs), so the
+    observable proof it found its row is that it reaches ``list_models`` — which
+    is called with the row's ``id`` only *after* the row is resolved. RED
+    (current): the stale-key ``(base_url, api_key)`` lookup returns nothing, the
+    method raises ``404`` internally and returns before ``list_models`` is ever
+    called. GREEN: lookup by ``id`` finds the row and ``list_models`` runs for
+    that ``id``.
+    """
+    row = UpstreamProviderRow(
+        provider_type="ppqai",
+        base_url="https://api.ppq.ai",
+        api_key="sk-original",
+        enabled=True,
+        provider_fee=1.0,
+    )
+    integration_session.add(row)  # type: ignore[attr-defined]
+    await integration_session.commit()  # type: ignore[attr-defined]
+    await integration_session.refresh(row)  # type: ignore[attr-defined]
+    row_id = row.id
+
+    provider = PPQAIUpstreamProvider.from_db_row(row)  # captures sk-original
+    assert provider is not None
+
+    row.api_key = "sk-rotated"
+    integration_session.add(row)  # type: ignore[attr-defined]
+    await integration_session.commit()  # type: ignore[attr-defined]
+
+    list_models_mock = AsyncMock(return_value=[])
+    with (
+        patch.object(provider, "fetch_models", new=AsyncMock(return_value=[])),
+        patch("routstr.upstream.base.list_models", new=list_models_mock),
+    ):
+        await provider.refresh_models_cache()
+
+    list_models_mock.assert_awaited_once()
+    assert list_models_mock.await_args is not None
+    assert list_models_mock.await_args.kwargs["upstream_id"] == row_id

tests/unit/test_cache_pricing.py

@@ -81,6 +81,21 @@ def test_backfill_strips_vendor_prefix_for_litellm_lookup() -> None:
     assert result.input_cache_read == expected
 
 
+def test_backfill_case_insensitive_lookup() -> None:
+    """A generic upstream may report a mixed-case id
+    (deepseek-ai/DeepSeek-V4-Flash); litellm keys are lowercase. The
+    case-insensitive fallback still resolves the cache rate."""
+    pricing = Pricing(prompt=1.4e-07, completion=2.8e-07)
+
+    result = backfill_cache_pricing("deepseek-ai/DeepSeek-V4-Flash", pricing)
+
+    expected = litellm.model_cost["deepseek-v4-flash"][
+        "cache_read_input_token_cost"
+    ]
+    assert result.input_cache_read == expected
+    assert result.input_cache_read < pricing.prompt  # sanity: it's a discount
+
+
 def test_backfill_fills_cache_write_rate() -> None:
     """Anthropic cache writes cost more than input (1.25x); billing them at
     the input rate undercharges. litellm carries the write rate."""
@@ -136,6 +151,93 @@ def test_provider_fee_applies_to_backfilled_cache_rates() -> None:
     assert adjusted.pricing.prompt == pytest.approx(2.8e-07 * 2.0)
 
 
+def test_row_to_model_backfills_cache_rate() -> None:
+    """The DB-override path (admin-configured providers, e.g. a generic
+    upstream) stores pricing without cache rates. ``_row_to_model`` must
+    backfill them from litellm just like ``_apply_provider_fee_to_model``,
+    otherwise cache reads bill at the full input rate."""
+    import json
+
+    from routstr.core.db import ModelRow
+    from routstr.payment.models import _row_to_model
+
+    row = ModelRow(
+        id="deepseek-v4-flash",
+        name="deepseek-v4-flash",
+        created=0,
+        description="",
+        context_length=1000000,
+        architecture=json.dumps(
+            {
+                "modality": "text",
+                "input_modalities": ["text"],
+                "output_modalities": ["text"],
+                "tokenizer": "unknown",
+                "instruct_type": None,
+            }
+        ),
+        # Stored pricing omits input_cache_read (generic provider never sets it).
+        pricing=json.dumps({"prompt": 1.4e-07, "completion": 2.8e-07}),
+        enabled=True,
+        upstream_provider_id=1,
+    )
+
+    with patch(
+        "routstr.payment.models.sats_usd_price", return_value=5.0e-5
+    ):
+        model = _row_to_model(row, apply_provider_fee=True, provider_fee=1.0)
+
+    litellm_read = litellm.model_cost["deepseek-v4-flash"][
+        "cache_read_input_token_cost"
+    ]
+    assert model.pricing.input_cache_read == pytest.approx(litellm_read)
+    assert model.pricing.input_cache_read < model.pricing.prompt  # a discount
+    assert model.sats_pricing is not None
+    assert model.sats_pricing.input_cache_read > 0
+
+
+def test_row_to_model_backfills_via_forwarded_model_id() -> None:
+    """An alias row (id != forwarded_model_id) must backfill cache rates from
+    the *forwarded* model name — the real upstream model litellm prices —
+    not the alias id, which litellm doesn't know."""
+    import json
+
+    from routstr.core.db import ModelRow
+    from routstr.payment.models import _row_to_model
+
+    row = ModelRow(
+        id="local-alias",  # litellm has no such key
+        name="local-alias",
+        created=0,
+        description="",
+        context_length=1000000,
+        architecture=json.dumps(
+            {
+                "modality": "text",
+                "input_modalities": ["text"],
+                "output_modalities": ["text"],
+                "tokenizer": "unknown",
+                "instruct_type": None,
+            }
+        ),
+        pricing=json.dumps({"prompt": 1.4e-07, "completion": 2.8e-07}),
+        enabled=True,
+        upstream_provider_id=1,
+        forwarded_model_id="deepseek-v4-flash",
+    )
+
+    with patch(
+        "routstr.payment.models.sats_usd_price", return_value=5.0e-5
+    ):
+        model = _row_to_model(row, apply_provider_fee=True, provider_fee=1.0)
+
+    litellm_read = litellm.model_cost["deepseek-v4-flash"][
+        "cache_read_input_token_cost"
+    ]
+    assert model.pricing.input_cache_read == pytest.approx(litellm_read)
+    assert model.pricing.input_cache_read < model.pricing.prompt
+
+
 # ============================================================================
 # calculate_cost — cached tokens billed at cache rates
 # ============================================================================

tests/unit/test_cost_calculation_caching.py

@@ -404,6 +404,67 @@ async def test_deepseek_malformed_hit_tokens_coerce_to_zero() -> None:
     assert result.cache_read_input_tokens == 0
 
 
+# ============================================================================
+# Truly-empty response with a non-zero USD cost → full refund
+#
+# When an upstream reports a USD cost but the response carries NO tokens at all
+# (input, output, cache-read and cache-creation all zero), billing the
+# USD-derived cost charges the user for nothing. Refund in full. The gate is
+# tightened relative to PR #489: a cache-read/-creation-only turn legitimately
+# reports zero prompt/completion tokens with a real cost and must still bill.
+# ============================================================================
+@pytest.mark.asyncio
+async def test_truly_empty_usd_cost_response_is_refunded(
+    mock_fixed_pricing: None,
+) -> None:
+    """0 input + 0 output + 0 cache tokens with a non-zero USD cost → refund."""
+    response = {
+        "model": "gpt-4",
+        "usage": {
+            "prompt_tokens": 0,
+            "completion_tokens": 0,
+            "total_cost": 0.01,  # non-zero USD cost despite no tokens
+        },
+    }
+    result = await calculate_cost(response, max_cost=100000)
+
+    assert isinstance(result, CostData)
+    assert result.total_msats == 0  # full refund
+    assert result.input_msats == 0
+    assert result.output_msats == 0
+    assert result.total_usd == 0.0
+    assert result.input_tokens == 0
+    assert result.output_tokens == 0
+    assert result.cache_read_input_tokens == 0
+    assert result.cache_creation_input_tokens == 0
+
+
+@pytest.mark.asyncio
+async def test_cache_read_only_usd_cost_response_is_billed(
+    mock_fixed_pricing: None,
+) -> None:
+    """Cache-read-only turn (0 prompt/completion, non-zero cost) still bills."""
+    response = {
+        "model": "claude-3-5-sonnet",
+        "usage": {
+            "input_tokens": 0,
+            "output_tokens": 0,
+            "cache_read_input_tokens": 1000,  # real cached usage
+            "cache_creation_input_tokens": 0,
+            "total_cost": 0.01,  # non-zero USD cost
+        },
+    }
+    result = await calculate_cost(response, max_cost=100000)
+
+    assert isinstance(result, CostData)
+    # NOT refunded — the USD cost is billed in full. Pinning the exact value
+    # guards against any future regression that would over-refund a cache-only
+    # turn (the bug in PR #489, which refunded whenever prompt+completion == 0).
+    assert result.total_msats == 200000
+    assert result.total_usd == 0.01
+    assert result.cache_read_input_tokens == 1000
+
+
 # ============================================================================
 # Test 13: Missing Usage Block
 # ============================================================================

tests/unit/test_model_paths.py

@@ -0,0 +1,712 @@
+"""Tests for the model-path discovery service and endpoints."""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import os
+from contextlib import asynccontextmanager
+from types import SimpleNamespace
+from typing import Any, AsyncGenerator
+
+import pytest
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+from sqlalchemy.ext.asyncio import AsyncEngine, create_async_engine
+from sqlmodel import SQLModel
+from sqlmodel.ext.asyncio.session import AsyncSession
+
+os.environ.setdefault("UPSTREAM_BASE_URL", "http://test")
+os.environ.setdefault("UPSTREAM_API_KEY", "test")
+
+from routstr.core.db import ModelRow, UpstreamProviderRow  # noqa: E402
+from routstr.payment.models import models_router  # noqa: E402
+from routstr.upstream import model_paths as mp  # noqa: E402
+
+# --------------------------------------------------------------------------- #
+# Fakes
+# --------------------------------------------------------------------------- #
+
+
+def _model(
+    id: str,
+    *,
+    forwarded_model_id: str | None = None,
+    canonical_slug: str | None = None,
+    enabled: bool = True,
+) -> SimpleNamespace:
+    return SimpleNamespace(
+        id=id,
+        forwarded_model_id=forwarded_model_id,
+        canonical_slug=canonical_slug,
+        enabled=enabled,
+    )
+
+
+def _model_row(
+    id: str,
+    *,
+    upstream_provider_id: int = 1,
+    forwarded_model_id: str | None = None,
+    canonical_slug: str | None = None,
+    enabled: bool = True,
+) -> ModelRow:
+    return ModelRow(
+        id=id,
+        upstream_provider_id=upstream_provider_id,
+        name=id,
+        created=0,
+        description="test model",
+        context_length=8192,
+        architecture=json.dumps(
+            {
+                "modality": "text",
+                "input_modalities": ["text"],
+                "output_modalities": ["text"],
+                "tokenizer": "test",
+                "instruct_type": None,
+            }
+        ),
+        pricing=json.dumps({"prompt": 0.000001, "completion": 0.000002}),
+        enabled=enabled,
+        forwarded_model_id=forwarded_model_id,
+        canonical_slug=canonical_slug,
+    )
+
+
+class _FakeProvider:
+    def __init__(
+        self,
+        *,
+        provider_type: str,
+        base_url: str,
+        models: list[SimpleNamespace],
+        db_id: int | None = 1,
+        api_key: str = "sk-test",
+    ) -> None:
+        self.provider_type = provider_type
+        self.base_url = base_url
+        self.api_key = api_key
+        self.db_id = db_id
+        self._models = models
+
+    def get_cached_models(self) -> list[SimpleNamespace]:
+        return self._models
+
+
+class _FakeResponse:
+    def __init__(self, status_code: int, payload: Any) -> None:
+        self.status_code = status_code
+        self._payload = payload
+
+    def json(self) -> Any:
+        return self._payload
+
+
+@pytest.fixture
+async def patched_session(
+    monkeypatch: pytest.MonkeyPatch,
+) -> AsyncGenerator[AsyncEngine, None]:
+    """Bind the service's ``create_session`` to a fresh in-memory engine."""
+    engine = create_async_engine("sqlite+aiosqlite:///:memory:")
+    async with engine.begin() as conn:
+        await conn.run_sync(SQLModel.metadata.create_all)
+
+    # Seed the FK target so ModelPathRow inserts satisfy the constraint.
+    async with AsyncSession(engine) as session:
+        for pid in (1, 2):
+            session.add(
+                UpstreamProviderRow(
+                    id=pid,
+                    slug=f"p{pid}",
+                    provider_type="anthropic" if pid == 1 else "openrouter",
+                    base_url=f"https://provider-{pid}",
+                    api_key=f"k{pid}",
+                )
+            )
+        await session.commit()
+
+    @asynccontextmanager
+    async def _factory() -> AsyncGenerator[AsyncSession, None]:
+        async with AsyncSession(engine, expire_on_commit=False) as session:
+            yield session
+
+    monkeypatch.setattr(mp, "create_session", _factory)
+    yield engine
+    await engine.dispose()
+
+
+# --------------------------------------------------------------------------- #
+# Predicates / pure helpers
+# --------------------------------------------------------------------------- #
+
+
+def test_is_openrouter_base_url() -> None:
+    assert mp.is_openrouter_base_url("https://openrouter.ai/api/v1") is True
+    assert mp.is_openrouter_base_url("https://api.anthropic.com") is False
+    assert mp.is_openrouter_base_url(None) is False
+
+
+def test_native_anthropic_not_openrouter() -> None:
+    """Native Anthropic must not be treated as OpenRouter-compatible even though
+    ``_upstream_accepts_cache_control`` returns True for it."""
+    assert mp.is_openrouter_base_url("https://api.anthropic.com/v1") is False
+
+
+def test_exposed_model_id_prefers_forwarded() -> None:
+    assert (
+        mp.exposed_model_id(_model("claude-x", forwarded_model_id="fwd-claude"))
+        == "fwd-claude"
+    )
+    assert mp.exposed_model_id(_model("claude-x")) == "claude-x"
+
+
+def test_public_model_id_strips_provider_prefix() -> None:
+    assert mp.public_model_id("z-ai/glm-5v-turbo") == "glm-5v-turbo"
+    assert mp.public_model_id("gpt-4o-mini") == "gpt-4o-mini"
+
+
+def test_openrouter_author_slug_uses_canonical_not_forwarded() -> None:
+    m = _model(
+        "claude-opus-4.6",
+        forwarded_model_id="forwarded-only",
+        canonical_slug="anthropic/claude-opus-4.6",
+    )
+    assert mp.openrouter_author_slug(m) == "anthropic/claude-opus-4.6"
+
+
+def test_openrouter_author_slug_falls_back_to_slash_id() -> None:
+    m = _model("anthropic/claude-opus-4.6", canonical_slug="claude-opus-4.6")
+    assert mp.openrouter_author_slug(m) == "anthropic/claude-opus-4.6"
+
+
+def test_openrouter_author_slug_none_when_no_slash() -> None:
+    m = _model("claude-opus-4.6", canonical_slug="claude-opus-4.6")
+    assert mp.openrouter_author_slug(m) is None
+
+
+# --------------------------------------------------------------------------- #
+# Collection
+# --------------------------------------------------------------------------- #
+
+
+@pytest.mark.asyncio
+async def test_direct_provider_single_path_uses_provider_type() -> None:
+    provider = _FakeProvider(
+        provider_type="anthropic",
+        base_url="https://api.anthropic.com/v1",
+        models=[_model("claude-opus-4.6")],
+    )
+    pairs = await mp._collect_provider_paths(provider)  # type: ignore[arg-type]
+    assert pairs == [("claude-opus-4.6", "anthropic")]
+
+
+@pytest.mark.asyncio
+async def test_direct_path_stores_exposed_model_id() -> None:
+    provider = _FakeProvider(
+        provider_type="anthropic",
+        base_url="https://api.anthropic.com/v1",
+        models=[_model("internal-id", forwarded_model_id="claude-opus-4.6")],
+    )
+    pairs = await mp._collect_provider_paths(provider)  # type: ignore[arg-type]
+    assert pairs == [("claude-opus-4.6", "anthropic")]
+
+
+@pytest.mark.asyncio
+async def test_disabled_models_excluded() -> None:
+    provider = _FakeProvider(
+        provider_type="anthropic",
+        base_url="https://api.anthropic.com/v1",
+        models=[
+            _model("enabled-model"),
+            _model("disabled-model", enabled=False),
+        ],
+    )
+    pairs = await mp._collect_provider_paths(provider)  # type: ignore[arg-type]
+    assert pairs == [("enabled-model", "anthropic")]
+
+
+@pytest.mark.asyncio
+async def test_openrouter_provider_adds_endpoint_paths(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    provider = _FakeProvider(
+        provider_type="openrouter",
+        base_url="https://openrouter.ai/api/v1",
+        models=[_model("claude-opus-4.6", canonical_slug="anthropic/claude-opus-4.6")],
+    )
+
+    async def _fake_get(
+        self: object,
+        url: str,
+        headers: object = None,
+        timeout: object = None,
+    ) -> _FakeResponse:
+        return _FakeResponse(
+            200,
+            {
+                "data": {
+                    "endpoints": [
+                        {"provider_name": "Anthropic"},
+                        {"provider_name": "Amazon Bedrock"},
+                    ]
+                }
+            },
+        )
+
+    monkeypatch.setattr("httpx.AsyncClient.get", _fake_get)
+
+    pairs = await mp._collect_provider_paths(provider)  # type: ignore[arg-type]
+    assert ("claude-opus-4.6", "openrouter:Anthropic") in pairs
+    assert ("claude-opus-4.6", "openrouter:Amazon Bedrock") in pairs
+    assert ("claude-opus-4.6", "openrouter") not in pairs
+    assert len(pairs) == 2
+
+
+@pytest.mark.asyncio
+async def test_generic_provider_with_openrouter_base_url_discovers(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """A generic provider pointed at OpenRouter exposes the response-stamped
+    ``generic:<upstream>`` path, not a native ``openrouter:`` path."""
+    provider = _FakeProvider(
+        provider_type="generic",
+        base_url="https://openrouter.ai/api/v1",
+        models=[_model("claude-opus-4.6", canonical_slug="anthropic/claude-opus-4.6")],
+    )
+
+    async def _fake_get(
+        self: object,
+        url: str,
+        headers: object = None,
+        timeout: object = None,
+    ) -> _FakeResponse:
+        return _FakeResponse(
+            200, {"data": {"endpoints": [{"provider_name": "Anthropic"}]}}
+        )
+
+    monkeypatch.setattr("httpx.AsyncClient.get", _fake_get)
+
+    pairs = await mp._collect_provider_paths(provider)  # type: ignore[arg-type]
+    assert pairs == [("claude-opus-4.6", "generic:Anthropic")]
+
+
+@pytest.mark.asyncio
+async def test_openrouter_failure_degrades_gracefully(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    provider = _FakeProvider(
+        provider_type="openrouter",
+        base_url="https://openrouter.ai/api/v1",
+        models=[_model("claude-opus-4.6", canonical_slug="anthropic/claude-opus-4.6")],
+    )
+
+    async def _boom(
+        self: object,
+        url: str,
+        headers: object = None,
+        timeout: object = None,
+    ) -> _FakeResponse:
+        raise RuntimeError("network down")
+
+    monkeypatch.setattr("httpx.AsyncClient.get", _boom)
+
+    pairs = await mp._collect_provider_paths(provider)  # type: ignore[arg-type]
+    assert pairs == []
+
+
+@pytest.mark.asyncio
+async def test_openrouter_rate_limit_skips_model(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    provider = _FakeProvider(
+        provider_type="openrouter",
+        base_url="https://openrouter.ai/api/v1",
+        models=[_model("claude-opus-4.6", canonical_slug="anthropic/claude-opus-4.6")],
+    )
+
+    async def _rate_limited(
+        self: object,
+        url: str,
+        headers: object = None,
+        timeout: object = None,
+    ) -> _FakeResponse:
+        return _FakeResponse(429, {})
+
+    monkeypatch.setattr("httpx.AsyncClient.get", _rate_limited)
+
+    pairs = await mp._collect_provider_paths(provider)  # type: ignore[arg-type]
+    assert pairs == []
+
+
+@pytest.mark.asyncio
+async def test_openrouter_fanout_is_bounded(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setattr(mp, "_OPENROUTER_CONCURRENCY", 3)
+    models = [
+        _model(f"m{i}", canonical_slug=f"author/m{i}") for i in range(20)
+    ]
+    provider = _FakeProvider(
+        provider_type="openrouter",
+        base_url="https://openrouter.ai/api/v1",
+        models=models,
+    )
+
+    state = {"current": 0, "max": 0}
+
+    async def _slow_get(
+        self: object,
+        url: str,
+        headers: object = None,
+        timeout: object = None,
+    ) -> _FakeResponse:
+        state["current"] += 1
+        state["max"] = max(state["max"], state["current"])
+        await asyncio.sleep(0.02)
+        state["current"] -= 1
+        return _FakeResponse(200, {"data": {"endpoints": [{"provider_name": "X"}]}})
+
+    monkeypatch.setattr("httpx.AsyncClient.get", _slow_get)
+
+    await mp._collect_provider_paths(provider)  # type: ignore[arg-type]
+    assert state["max"] <= 3, f"concurrency exceeded bound: {state['max']}"
+
+
+# --------------------------------------------------------------------------- #
+# Persistence + query
+# --------------------------------------------------------------------------- #
+
+
+@pytest.mark.asyncio
+async def test_refresh_replaces_stale_rows(
+    patched_session: AsyncEngine,
+) -> None:
+    await mp._persist_provider_paths(1, [("m1", "anthropic"), ("m2", "anthropic")])
+    first = await mp.get_all_model_paths()
+    assert {row["id"] for row in first} == {"m1", "m2"}
+
+    # Second refresh with a different set — stale m2 must disappear.
+    await mp._persist_provider_paths(1, [("m1", "anthropic")])
+    second = await mp.get_all_model_paths()
+    assert {row["id"] for row in second} == {"m1"}
+
+
+@pytest.mark.asyncio
+async def test_refresh_model_paths_excludes_db_disabled_override(
+    patched_session: AsyncEngine,
+) -> None:
+    async with AsyncSession(patched_session) as session:
+        session.add(_model_row("disabled-by-db", enabled=False))
+        await session.commit()
+
+    provider = _FakeProvider(
+        provider_type="anthropic",
+        base_url="https://api.anthropic.com/v1",
+        models=[_model("disabled-by-db")],
+        db_id=1,
+    )
+
+    await mp.refresh_model_paths([provider])  # type: ignore[list-item]
+
+    assert await mp.get_all_model_paths() == []
+
+
+@pytest.mark.asyncio
+async def test_refresh_model_paths_uses_db_forwarded_alias(
+    patched_session: AsyncEngine,
+) -> None:
+    async with AsyncSession(patched_session) as session:
+        session.add(_model_row("internal-id", forwarded_model_id="public-alias"))
+        await session.commit()
+
+    provider = _FakeProvider(
+        provider_type="anthropic",
+        base_url="https://api.anthropic.com/v1",
+        models=[_model("internal-id")],
+        db_id=1,
+    )
+
+    await mp.refresh_model_paths([provider])  # type: ignore[list-item]
+
+    assert await mp.get_all_model_paths() == [
+        {"id": "public-alias", "paths": [{"path": "anthropic"}]}
+    ]
+
+
+@pytest.mark.asyncio
+async def test_refresh_model_paths_includes_enabled_db_override_missing_from_cache(
+    patched_session: AsyncEngine,
+) -> None:
+    async with AsyncSession(patched_session) as session:
+        session.add(_model_row("deployment-id", forwarded_model_id="public-deployment"))
+        await session.commit()
+
+    provider = _FakeProvider(
+        provider_type="generic",
+        base_url="https://custom-provider/v1",
+        models=[],
+        db_id=1,
+    )
+
+    await mp.refresh_model_paths([provider])  # type: ignore[list-item]
+
+    assert await mp.get_all_model_paths() == [
+        {"id": "public-deployment", "paths": [{"path": "generic"}]}
+    ]
+
+
+@pytest.mark.asyncio
+async def test_refresh_model_paths_prunes_inactive_provider_rows(
+    patched_session: AsyncEngine,
+) -> None:
+    await mp._persist_provider_paths(1, [("m1", "anthropic")])
+    await mp._persist_provider_paths(2, [("m2", "openrouter:Anthropic")])
+
+    provider = _FakeProvider(
+        provider_type="anthropic",
+        base_url="https://api.anthropic.com/v1",
+        models=[_model("m1")],
+        db_id=1,
+    )
+
+    await mp.refresh_model_paths([provider])  # type: ignore[list-item]
+    active_only = await mp.get_all_model_paths()
+    assert active_only == [{"id": "m1", "paths": [{"path": "anthropic"}]}]
+
+    await mp.refresh_model_paths([])
+    assert await mp.get_all_model_paths() == []
+
+
+@pytest.mark.asyncio
+async def test_refresh_model_paths_skips_disabled_db_provider(
+    patched_session: AsyncEngine,
+) -> None:
+    await mp._persist_provider_paths(1, [("stale-model", "anthropic")])
+    async with AsyncSession(patched_session) as session:
+        provider_row = await session.get(UpstreamProviderRow, 1)
+        assert provider_row is not None
+        provider_row.enabled = False
+        await session.commit()
+
+    provider = _FakeProvider(
+        provider_type="anthropic",
+        base_url="https://api.anthropic.com/v1",
+        models=[_model("fresh-model")],
+        db_id=1,
+    )
+
+    await mp.refresh_model_paths([provider])  # type: ignore[list-item]
+
+    assert await mp.get_all_model_paths() == []
+
+
+@pytest.mark.asyncio
+async def test_same_model_two_providers_two_paths(
+    patched_session: AsyncEngine,
+) -> None:
+    await mp._persist_provider_paths(1, [("claude-opus-4.6", "anthropic")])
+    await mp._persist_provider_paths(2, [("claude-opus-4.6", "openrouter:Anthropic")])
+
+    data = await mp.get_all_model_paths()
+    assert len(data) == 1
+    entry = data[0]
+    assert entry["id"] == "claude-opus-4.6"
+    paths = {p["path"] for p in entry["paths"]}
+    assert paths == {"anthropic", "openrouter:Anthropic"}
+    # No canonical_id anywhere.
+    assert "canonical_id" not in entry
+    assert all("canonical_id" not in p for p in entry["paths"])
+
+
+@pytest.mark.asyncio
+async def test_get_all_model_paths_deduplicates_visible_paths(
+    patched_session: AsyncEngine,
+) -> None:
+    await mp._persist_provider_paths(1, [("anthropic/claude-opus-4.6", "anthropic")])
+    await mp._persist_provider_paths(2, [("claude-opus-4.6", "anthropic")])
+
+    assert await mp.get_all_model_paths() == [
+        {"id": "claude-opus-4.6", "paths": [{"path": "anthropic"}]}
+    ]
+
+
+@pytest.mark.asyncio
+async def test_get_all_model_paths_returns_unqualified_model_ids(
+    patched_session: AsyncEngine,
+) -> None:
+    await mp._persist_provider_paths(4, [("z-ai/glm-5v-turbo", "openrouter:Z.AI")])
+    await mp._persist_provider_paths(5, [("openai/gpt-4o-mini", "openrouter:OpenAI")])
+
+    data = await mp.get_all_model_paths()
+
+    assert {row["id"] for row in data} == {"glm-5v-turbo", "gpt-4o-mini"}
+
+
+@pytest.mark.asyncio
+async def test_get_paths_for_model_returns_only_paths(
+    patched_session: AsyncEngine,
+) -> None:
+    await mp._persist_provider_paths(1, [("claude-opus-4.6", "anthropic")])
+    await mp._persist_provider_paths(2, [("claude-opus-4.6", "openrouter:Anthropic")])
+
+    paths = await mp.get_paths_for_model("claude-opus-4.6")
+    assert {p["path"] for p in paths} == {"anthropic", "openrouter:Anthropic"}
+    assert all(set(p.keys()) == {"path"} for p in paths)
+    assert await mp.get_paths_for_model("does-not-exist") == []
+
+
+@pytest.mark.asyncio
+async def test_get_paths_for_model_falls_back_to_provider_prefixed_id(
+    patched_session: AsyncEngine,
+) -> None:
+    await mp._persist_provider_paths(4, [("z-ai/glm-5v-turbo", "openrouter:Z.AI")])
+
+    paths = await mp.get_paths_for_model("glm-5v-turbo")
+
+    assert paths == [{"path": "openrouter:Z.AI"}]
+
+
+@pytest.mark.asyncio
+async def test_get_paths_for_model_deduplicates_visible_paths(
+    patched_session: AsyncEngine,
+) -> None:
+    await mp._persist_provider_paths(1, [("anthropic/claude-opus-4.6", "anthropic")])
+    await mp._persist_provider_paths(2, [("claude-opus-4.6", "anthropic")])
+
+    assert await mp.get_paths_for_model("claude-opus-4.6") == [
+        {"path": "anthropic"}
+    ]
+    assert await mp.get_paths_for_model("anthropic/claude-opus-4.6") == [
+        {"path": "anthropic"}
+    ]
+
+
+@pytest.mark.asyncio
+async def test_get_paths_for_model_merges_prefixed_and_unprefixed_aliases(
+    patched_session: AsyncEngine,
+) -> None:
+    await mp._persist_provider_paths(7, [("deepseek-v4-pro", "generic")])
+    await mp._persist_provider_paths(
+        4, [("deepseek/deepseek-v4-pro", "openrouter:DeepSeek")]
+    )
+
+    short_paths = await mp.get_paths_for_model("deepseek-v4-pro")
+    prefixed_paths = await mp.get_paths_for_model("deepseek/deepseek-v4-pro")
+
+    assert short_paths == [
+        {"path": "generic"},
+        {"path": "openrouter:DeepSeek"},
+    ]
+    assert prefixed_paths == short_paths
+
+
+@pytest.mark.asyncio
+async def test_refresh_model_paths_skips_provider_without_db_id(
+    patched_session: AsyncEngine,
+) -> None:
+    provider = _FakeProvider(
+        provider_type="anthropic",
+        base_url="https://api.anthropic.com/v1",
+        models=[_model("claude-opus-4.6")],
+        db_id=None,
+    )
+    await mp.refresh_model_paths([provider])  # type: ignore[list-item]
+    assert await mp.get_all_model_paths() == []
+
+
+@pytest.mark.asyncio
+async def test_refresh_model_paths_isolates_provider_failure(
+    patched_session: AsyncEngine, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    good = _FakeProvider(
+        provider_type="anthropic",
+        base_url="https://api.anthropic.com/v1",
+        models=[_model("claude-opus-4.6")],
+        db_id=1,
+    )
+    bad = _FakeProvider(
+        provider_type="openrouter",
+        base_url="https://openrouter.ai/api/v1",
+        models=[_model("m", canonical_slug="a/m")],
+        db_id=2,
+    )
+
+    original = mp._collect_provider_paths
+
+    async def _maybe_fail(
+        upstream: Any, *args: Any, **kwargs: Any
+    ) -> list[tuple[str, str]]:
+        if upstream is bad:
+            raise RuntimeError("boom")
+        return await original(upstream, *args, **kwargs)  # type: ignore[arg-type]
+
+    monkeypatch.setattr(mp, "_collect_provider_paths", _maybe_fail)
+
+    await mp.refresh_model_paths([good, bad])  # type: ignore[list-item]
+    data = await mp.get_all_model_paths()
+    assert {row["id"] for row in data} == {"claude-opus-4.6"}
+
+
+# --------------------------------------------------------------------------- #
+# Endpoints
+# --------------------------------------------------------------------------- #
+
+
+def _make_model_paths_app() -> FastAPI:
+    app = FastAPI()
+    app.include_router(models_router)
+    return app
+
+
+def test_model_paths_endpoint_returns_all_paths(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    async def _fake_get_all_model_paths() -> list[dict[str, Any]]:
+        return [
+            {
+                "id": "claude-opus-4.6",
+                "paths": [
+                    {"path": "anthropic"},
+                    {"path": "openrouter:Anthropic"},
+                ],
+            }
+        ]
+
+    monkeypatch.setattr(mp, "get_all_model_paths", _fake_get_all_model_paths)
+
+    response = TestClient(_make_model_paths_app()).get("/v1/models/paths")
+
+    assert response.status_code == 200
+    assert response.json() == {
+        "data": [
+            {
+                "id": "claude-opus-4.6",
+                "paths": [
+                    {"path": "anthropic"},
+                    {"path": "openrouter:Anthropic"},
+                ],
+            }
+        ]
+    }
+
+
+def test_model_paths_for_model_endpoint_accepts_slash_model_id(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    calls: list[str] = []
+
+    async def _fake_get_paths_for_model(model_id: str) -> list[dict[str, Any]]:
+        calls.append(model_id)
+        return [{"path": "generic:Anthropic"}]
+
+    monkeypatch.setattr(mp, "get_paths_for_model", _fake_get_paths_for_model)
+
+    response = TestClient(_make_model_paths_app()).get(
+        "/v1/models/paths/model",
+        params={"model_id": "anthropic/claude-opus-4.6"},
+    )
+
+    assert response.status_code == 200
+    assert response.json() == {"data": [{"path": "generic:Anthropic"}]}
+    assert calls == ["anthropic/claude-opus-4.6"]

tests/unit/test_provider_slug_migration.py

@@ -0,0 +1,55 @@
+from __future__ import annotations
+
+import importlib.util
+from pathlib import Path
+
+import sqlalchemy as sa
+
+_MIGRATION_PATH = (
+    Path(__file__).resolve().parents[2]
+    / "migrations"
+    / "versions"
+    / "c6d7e8f9a0b1_add_slug_to_upstream_providers.py"
+)
+_spec = importlib.util.spec_from_file_location("provider_slug_migration", _MIGRATION_PATH)
+assert _spec is not None and _spec.loader is not None
+migration = importlib.util.module_from_spec(_spec)
+_spec.loader.exec_module(migration)
+
+
+def test_slug_migration_backfill_uses_api_safe_deterministic_slugs() -> None:
+    engine = sa.create_engine("sqlite:///:memory:")
+    with engine.begin() as conn:
+        conn.execute(
+            sa.text(
+                "CREATE TABLE upstream_providers ("
+                "id INTEGER PRIMARY KEY, "
+                "provider_type VARCHAR NOT NULL, "
+                "slug VARCHAR NULL"
+                ")"
+            )
+        )
+        conn.execute(
+            sa.text(
+                "INSERT INTO upstream_providers (id, provider_type, slug) VALUES "
+                "(1, 'OpenAI Compatible', NULL), "
+                "(2, 'OpenAI Compatible', ''), "
+                "(3, '123', NULL), "
+                "(4, 'x', NULL), "
+                "(5, 'anthropic', 'anthropic')"
+            )
+        )
+
+        migration._backfill_provider_slugs(conn)
+
+        rows = conn.execute(
+            sa.text("SELECT id, slug FROM upstream_providers ORDER BY id")
+        ).all()
+
+    assert rows == [
+        (1, "openai-compatible"),
+        (2, "openai-compatible-2"),
+        (3, "provider-123"),
+        (4, "x-provider"),
+        (5, "anthropic"),
+    ]

tests/unit/test_provider_slugs.py

@@ -0,0 +1,144 @@
+from __future__ import annotations
+
+import pytest
+from sqlalchemy.ext.asyncio import create_async_engine
+from sqlmodel import SQLModel, select
+from sqlmodel.ext.asyncio.session import AsyncSession
+
+from routstr.core.admin import _get_upstream_provider_by_ref
+from routstr.core.db import UpstreamProviderRow
+from routstr.core.provider_slugs import (
+    allocate_unique_provider_slug,
+    provider_slug_base,
+)
+from routstr.upstream.helpers import _seed_providers_from_settings
+
+
+@pytest.mark.asyncio
+async def test_allocate_unique_provider_slug_is_deterministic_with_suffixes() -> None:
+    engine = create_async_engine("sqlite+aiosqlite:///:memory:")
+    async with engine.begin() as conn:
+        await conn.run_sync(SQLModel.metadata.create_all)
+
+    async with AsyncSession(engine) as session:
+        session.add(
+            UpstreamProviderRow(
+                slug="openai",
+                provider_type="openai",
+                base_url="https://api.openai.com/v1",
+                api_key="key-1",
+            )
+        )
+        await session.commit()
+
+        assert await allocate_unique_provider_slug(session, "openai") == "openai-2"
+        assert (
+            await allocate_unique_provider_slug(session, "openai", {"openai-2"})
+            == "openai-3"
+        )
+
+    await engine.dispose()
+
+
+def test_provider_slug_base_sanitizes_provider_type() -> None:
+    assert provider_slug_base("OpenAI Compatible") == "openai-compatible"
+    assert provider_slug_base("!!!") == "provider"
+    assert provider_slug_base("AI") == "ai-provider"
+    assert provider_slug_base("123") == "provider-123"
+
+
+@pytest.mark.asyncio
+async def test_provider_ref_lookup_accepts_existing_numeric_ids_and_slugs() -> None:
+    engine = create_async_engine("sqlite+aiosqlite:///:memory:")
+    async with engine.begin() as conn:
+        await conn.run_sync(SQLModel.metadata.create_all)
+
+    async with AsyncSession(engine) as session:
+        provider = UpstreamProviderRow(
+            slug="openai",
+            provider_type="openai",
+            base_url="https://api.openai.com/v1",
+            api_key="key-1",
+        )
+        session.add(provider)
+        await session.commit()
+        await session.refresh(provider)
+
+        by_id = await _get_upstream_provider_by_ref(session, str(provider.id))
+        by_slug = await _get_upstream_provider_by_ref(session, "openai")
+
+    assert by_id.id == provider.id
+    assert by_slug.id == provider.id
+
+    await engine.dispose()
+
+
+@pytest.mark.asyncio
+async def test_seed_providers_from_settings_sets_deterministic_slug(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    engine = create_async_engine("sqlite+aiosqlite:///:memory:")
+    async with engine.begin() as conn:
+        await conn.run_sync(SQLModel.metadata.create_all)
+
+    monkeypatch.setenv("OPENAI_API_KEY", "seeded-openai-key")
+
+    class SettingsStub:
+        chat_completions_api_version: str | None = None
+        upstream_base_url: str | None = None
+        upstream_api_key: str = ""
+
+    async with AsyncSession(engine) as session:
+        await _seed_providers_from_settings(session, SettingsStub())  # type: ignore[arg-type]
+        await session.commit()
+
+        result = await session.exec(select(UpstreamProviderRow))
+        providers: list[UpstreamProviderRow] = list(result.all())
+
+    assert [(p.provider_type, p.slug) for p in providers] == [("openai", "openai")]
+
+    await engine.dispose()
+
+
+@pytest.mark.asyncio
+async def test_seed_providers_from_settings_keeps_slug_stable_on_reseed(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    engine = create_async_engine("sqlite+aiosqlite:///:memory:")
+    async with engine.begin() as conn:
+        await conn.run_sync(SQLModel.metadata.create_all)
+
+    monkeypatch.setenv("OPENAI_API_KEY", "seeded-openai-key")
+
+    class SettingsStub:
+        chat_completions_api_version: str | None = None
+        upstream_base_url: str | None = None
+        upstream_api_key: str = ""
+
+    async with AsyncSession(engine) as session:
+        session.add(
+            UpstreamProviderRow(
+                slug="openai",
+                provider_type="openai",
+                base_url="https://example.invalid/v1",
+                api_key="other-key",
+            )
+        )
+        await session.commit()
+
+        await _seed_providers_from_settings(session, SettingsStub())  # type: ignore[arg-type]
+        await session.commit()
+        await _seed_providers_from_settings(session, SettingsStub())  # type: ignore[arg-type]
+        await session.commit()
+
+        result = await session.exec(
+            select(UpstreamProviderRow).order_by(UpstreamProviderRow.slug)
+        )
+        providers: list[UpstreamProviderRow] = list(result.all())
+
+    assert [(p.provider_type, p.slug) for p in providers] == [
+        ("openai", "openai"),
+        ("openai", "openai-2"),
+    ]
+
+    await engine.dispose()

ui/components/add-provider-model-dialog.tsx

@@ -68,6 +68,8 @@ const FormSchema = z.object({
   upstream_provider_id: z.string().default(''),
   input_cost: z.coerce.number().min(0).default(0),
   output_cost: z.coerce.number().min(0).default(0),
+  cache_read_cost: z.coerce.number().min(0).default(0),
+  cache_write_cost: z.coerce.number().min(0).default(0),
   request_cost: z.coerce.number().min(0).default(0),
   image_cost: z.coerce.number().min(0).default(0),
   web_search_cost: z.coerce.number().min(0).default(0),
@@ -125,6 +127,8 @@ export function AddProviderModelDialog({
       upstream_provider_id: '',
       input_cost: 0,
       output_cost: 0,
+      cache_read_cost: 0,
+      cache_write_cost: 0,
       request_cost: 0,
       image_cost: 0,
       web_search_cost: 0,
@@ -190,6 +194,8 @@ export function AddProviderModelDialog({
             : initialData.upstream_provider_id?.toString() || '',
         input_cost: pricing?.prompt ?? 0,
         output_cost: pricing?.completion ?? 0,
+        cache_read_cost: pricing?.input_cache_read ?? 0,
+        cache_write_cost: pricing?.input_cache_write ?? 0,
         request_cost: pricing?.request ?? 0,
         image_cost: pricing?.image ?? 0,
         web_search_cost: pricing?.web_search ?? 0,
@@ -231,6 +237,8 @@ export function AddProviderModelDialog({
         upstream_provider_id: '',
         input_cost: 0,
         output_cost: 0,
+        cache_read_cost: 0,
+        cache_write_cost: 0,
         request_cost: 0,
         image_cost: 0,
         web_search_cost: 0,
@@ -294,6 +302,8 @@ export function AddProviderModelDialog({
     );
     form.setValue('input_cost', pricing?.prompt ?? 0);
     form.setValue('output_cost', pricing?.completion ?? 0);
+    form.setValue('cache_read_cost', pricing?.input_cache_read ?? 0);
+    form.setValue('cache_write_cost', pricing?.input_cache_write ?? 0);
     form.setValue('request_cost', pricing?.request ?? 0);
     form.setValue('image_cost', pricing?.image ?? 0);
     form.setValue('web_search_cost', pricing?.web_search ?? 0);
@@ -365,6 +375,8 @@ export function AddProviderModelDialog({
         pricing: {
           prompt: data.input_cost,
           completion: data.output_cost,
+          input_cache_read: data.cache_read_cost,
+          input_cache_write: data.cache_write_cost,
           request: data.request_cost,
           image: data.image_cost,
           web_search: data.web_search_cost,
@@ -810,6 +822,38 @@ export function AddProviderModelDialog({
                     </FormItem>
                   )}
                 />
+                <FormField
+                  control={form.control}
+                  name='cache_read_cost'
+                  render={({ field }) => (
+                    <FormItem>
+                      <FormLabel>Cache Read Cost</FormLabel>
+                      <FormControl>
+                        <Input type='number' step='0.000001' {...field} />
+                      </FormControl>
+                      <FormDescription>
+                        Discounted cached-input read price per 1M tokens.
+                      </FormDescription>
+                      <FormMessage />
+                    </FormItem>
+                  )}
+                />
+                <FormField
+                  control={form.control}
+                  name='cache_write_cost'
+                  render={({ field }) => (
+                    <FormItem>
+                      <FormLabel>Cache Write Cost</FormLabel>
+                      <FormControl>
+                        <Input type='number' step='0.000001' {...field} />
+                      </FormControl>
+                      <FormDescription>
+                        Cached-input creation price per 1M tokens.
+                      </FormDescription>
+                      <FormMessage />
+                    </FormItem>
+                  )}
+                />
                 <FormField
                   control={form.control}
                   name='request_cost'

ui/components/provider-form-fields.tsx

@@ -118,6 +118,26 @@ export function ProviderFormFields({
         />
       )}
 
+      <div className='grid gap-2'>
+        <Label htmlFor={`${idPrefix}slug`}>
+          Slug {mode === 'create' ? '(optional, auto-generated)' : ''}
+        </Label>
+        <Input
+          id={`${idPrefix}slug`}
+          value={formData.slug || ''}
+          onChange={(e) =>
+            setFormData((prev) => ({
+              ...prev,
+              slug: e.target.value || undefined,
+            }))
+          }
+          placeholder='e.g. openai-prod'
+        />
+        <p className='text-muted-foreground text-xs'>
+          Stable external key used to update this provider via the admin API.
+        </p>
+      </div>
+
       <div className='grid gap-2'>
         <Label htmlFor={`${idPrefix}base_url`}>Base URL</Label>
         <Input

ui/lib/api/services/admin.ts

@@ -14,6 +14,7 @@ export const ProviderTypeSchema = z.object({
 
 export const UpstreamProviderSchema = z.object({
   id: z.number(),
+  slug: z.string().nullable().optional(),
   provider_type: z.string(),
   base_url: z.string(),
   api_key: z.string().optional(),
@@ -31,6 +32,7 @@ export const CreateUpstreamProviderSchema = z.object({
   enabled: z.boolean().default(true),
   provider_fee: z.number().optional(),
   provider_settings: z.record(z.string(), z.any()).nullable().optional(),
+  slug: z.string().optional(),
 });
 
 export const UpdateUpstreamProviderSchema = z.object({
@@ -41,6 +43,7 @@ export const UpdateUpstreamProviderSchema = z.object({
   enabled: z.boolean().optional(),
   provider_fee: z.number().optional(),
   provider_settings: z.record(z.string(), z.any()).nullable().optional(),
+  slug: z.string().optional(),
 });
 
 export const AdminModelPricingSchema = z.object({
@@ -50,6 +53,8 @@ export const AdminModelPricingSchema = z.object({
   image: z.number().optional(),
   web_search: z.number().optional(),
   internal_reasoning: z.number().optional(),
+  input_cache_read: z.number().optional(),
+  input_cache_write: z.number().optional(),
 });
 
 export const AdminModelArchitectureSchema = z.object({
@@ -146,7 +151,7 @@ export class AdminService {
     if (!pricing) return pricing;
     const result = { ...pricing };
 
-    // Only prompt and completion are per-token and need scaling to per-1M
+    // Token-priced fields are stored per-token by the API and shown per-1M in the UI.
     const convertField = (field: string) => {
       const val = result[field];
       if (val !== undefined && val !== null) {
@@ -161,6 +166,8 @@ export class AdminService {
 
     convertField('prompt');
     convertField('completion');
+    convertField('input_cache_read');
+    convertField('input_cache_write');
 
     // Other fields (request, image, etc.) are already flat fees (per item)
     // so we do NOT scale them.
@@ -174,7 +181,7 @@ export class AdminService {
     if (!pricing) return pricing;
     const result = { ...pricing };
 
-    // Only prompt and completion are per-1M in UI and need scaling down to per-token
+    // Token-priced fields are per-1M in the UI and need scaling down to per-token.
     const convertField = (field: string) => {
       const val = result[field];
       if (val !== undefined && val !== null) {
@@ -187,6 +194,8 @@ export class AdminService {
 
     convertField('prompt');
     convertField('completion');
+    convertField('input_cache_read');
+    convertField('input_cache_write');
 
     // Other fields stay as flat fees