only activ model should be visible

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/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/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"
 

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/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/models.py

@@ -596,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/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

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"]