add url and specific model infos to paht

Replace untyped lambdas with typed handler/provider functions; CI runs
mypy over tests as well.

docs/api/endpoints.md

@@ -327,6 +327,62 @@ GET /v1/models
 }
 ```
 
+### List Model Paths
+
+Get the selectable upstream routes for each advertised model. This endpoint is
+discovery-only; request-side selection will be added separately.
+
+```http
+GET /v1/models/paths
+```
+
+**Response:**
+
+```json
+{
+  "data": [
+    {
+      "id": "anthropic/claude-sonnet-4",
+      "paths": [
+        {
+          "path": "url=https%3A%2F%2Fapi.anthropic.com%2Fv1&provider-id=12&model-id=anthropic%2Fclaude-sonnet-4",
+          "provider": {"id": 12, "slug": "anthropic-primary", "type": "anthropic"},
+          "endpoint": null
+        },
+        {
+          "path": "url=https%3A%2F%2Fopenrouter.ai%2Fapi%2Fv1&provider-id=42&model-id=anthropic%2Fclaude-sonnet-4&endpoint=google-vertex%2Fus",
+          "provider": {"id": 42, "slug": "openrouter-main", "type": "openrouter"},
+          "endpoint": {"tag": "google-vertex/us", "name": "Google"}
+        }
+      ]
+    }
+  ],
+  "updated_at": 1753500000
+}
+```
+
+`path` is an opaque, percent-encoded selector. Clients must store and return it
+unchanged rather than parsing or reconstructing it. It identifies the exact
+configured route with `url`, `provider-id`, and `model-id`. To avoid exposing
+private network details, a configured private IP address or any URL with an
+explicit port is advertised as `http://localhost`. OpenRouter routes additionally
+preserve the exact machine-readable endpoint `tag`. Provider slugs/types and
+endpoint names remain display data. When request-side selection is implemented,
+an endpoint tag must not silently fall back to another backend.
+
+### List Paths for One Model
+
+Use the exact model ID advertised by `/v1/models`. The query parameter safely
+supports IDs containing `/`.
+
+```http
+GET /v1/models/paths/model?model_id=anthropic/claude-sonnet-4
+```
+
+The response uses the same path objects and `updated_at` field as the collection
+endpoint. An unknown model returns `404 Model not found`. A known model whose
+paths have not been discovered yet returns `200` with an empty `data` array.
+
 ## 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

@@ -142,6 +142,8 @@ 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 pause the refresh (previously discovered paths keep being served) | `600` |
+| `ENABLE_MODEL_PATHS_REFRESH` | Kill switch for the background model-path refresh (OpenRouter endpoint fan-out) | `true` |
 
 ### Priority
 
@@ -175,3 +177,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/4e0c3d195a49_add_model_paths_table.py

@@ -0,0 +1,54 @@
+"""add model paths table
+
+Revision ID: 4e0c3d195a49
+Revises: 9c4d8e2f1a6b
+Create Date: 2026-07-24 21:14:39.062179
+"""
+
+import sqlalchemy as sa
+import sqlmodel
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "4e0c3d195a49"
+down_revision = "9c4d8e2f1a6b"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "model_paths",
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("model_id", sqlmodel.sql.sqltypes.AutoString(), nullable=False),
+        sa.Column("path", sqlmodel.sql.sqltypes.AutoString(), nullable=False),
+        sa.Column("provider_slug", sqlmodel.sql.sqltypes.AutoString(), nullable=False),
+        sa.Column("provider_type", sqlmodel.sql.sqltypes.AutoString(), nullable=False),
+        sa.Column("endpoint_tag", sqlmodel.sql.sqltypes.AutoString(), nullable=True),
+        sa.Column("endpoint_name", sqlmodel.sql.sqltypes.AutoString(), nullable=True),
+        sa.Column("upstream_provider_id", sa.Integer(), nullable=False),
+        sa.Column("updated_at", sa.Integer(), nullable=False, server_default="0"),
+        sa.ForeignKeyConstraint(
+            ["upstream_provider_id"], ["upstream_providers.id"], ondelete="CASCADE"
+        ),
+        sa.PrimaryKeyConstraint("id"),
+        sa.UniqueConstraint(
+            "model_id",
+            "path",
+            "upstream_provider_id",
+            name="uq_model_paths_model_path_provider",
+        ),
+    )
+    # No standalone index on model_id: the unique constraint's autoindex already
+    # leads on model_id.
+    op.create_index(
+        op.f("ix_model_paths_upstream_provider_id"),
+        "model_paths",
+        ["upstream_provider_id"],
+        unique=False,
+    )
+
+
+def downgrade() -> None:
+    op.drop_index(op.f("ix_model_paths_upstream_provider_id"), table_name="model_paths")
+    op.drop_table("model_paths")

routstr/core/admin.py

@@ -51,6 +51,13 @@ ADMIN_SESSION_DURATION = 3600
 MAX_USAGE_ANALYTICS_HOURS = 365 * 24
 
 
+async def _refresh_provider_model_paths(upstream_provider_id: int) -> None:
+    """Queue discovery sync without blocking the committed admin mutation."""
+    from ..upstream.model_paths import schedule_model_paths_refresh_for_provider
+
+    await schedule_model_paths_refresh_for_provider(upstream_provider_id)
+
+
 async def require_admin_api(request: Request) -> None:
     auth_header = request.headers.get("Authorization")
     if not auth_header or not auth_header.startswith("Bearer "):
@@ -579,6 +586,7 @@ async def upsert_provider_model(
             await session.refresh(row)
 
     await refresh_model_maps()
+    await _refresh_provider_model_paths(provider_pk)
     return _row_to_model(
         row, apply_provider_fee=True, provider_fee=provider.provider_fee
     ).dict()  # type: ignore
@@ -633,6 +641,7 @@ async def delete_provider_model(provider_id: str, model_id: str) -> dict[str, ob
         await session.delete(row)
         await session.commit()
     await refresh_model_maps()
+    await _refresh_provider_model_paths(provider_pk)
     return {"ok": True, "deleted_id": model_id}
 
 
@@ -652,6 +661,7 @@ async def delete_all_provider_models(provider_id: str) -> dict[str, object]:
             await session.delete(row)  # type: ignore
         await session.commit()
     await refresh_model_maps()
+    await _refresh_provider_model_paths(provider_pk)
     return {"ok": True, "deleted": len(rows)}
 
 
@@ -743,6 +753,7 @@ async def batch_override_provider_models(
         await session.commit()
 
     await refresh_model_maps()
+    await _refresh_provider_model_paths(provider_pk)
     return {
         "ok": True,
         "count": overridden_count,
@@ -943,6 +954,7 @@ async def create_upstream_provider(
 
     await reinitialize_upstreams()
     await refresh_model_maps()
+    await _refresh_provider_model_paths(_provider_pk(provider))
     return _serialize_provider(provider)
 
 
@@ -968,6 +980,7 @@ async def update_upstream_provider(
 
     await reinitialize_upstreams()
     await refresh_model_maps()
+    await _refresh_provider_model_paths(_provider_pk(provider))
     return _serialize_provider(provider)
 
 
@@ -1003,6 +1016,7 @@ async def update_upstream_provider_by_slug(
 
     await reinitialize_upstreams()
     await refresh_model_maps()
+    await _refresh_provider_model_paths(_provider_pk(provider))
     return _serialize_provider(provider)
 
 

routstr/core/db.py

@@ -222,7 +222,10 @@ async def release_stale_reservations(
     if released:
         logger.warning(
             "Released stale reservations",
-            extra={"released_reservations": released, "max_age_seconds": max_age_seconds},
+            extra={
+                "released_reservations": released,
+                "max_age_seconds": max_age_seconds,
+            },
         )
     return released
 
@@ -255,9 +258,7 @@ async def prune_dead_api_keys(session: AsyncSession, min_age_seconds: int) -> in
         .where(col(ApiKey.total_spent) == 0)
         .where(col(ApiKey.total_requests) == 0)
         .where(col(ApiKey.parent_key_hash).is_(None))
-        .where(
-            (col(ApiKey.created_at).is_(None)) | (col(ApiKey.created_at) < cutoff)
-        )
+        .where((col(ApiKey.created_at).is_(None)) | (col(ApiKey.created_at) < cutoff))
         .where(~pending_invoice)
         .where(~has_children)
     )
@@ -311,6 +312,60 @@ 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)
+    # No standalone index on model_id: the unique constraint's autoindex already
+    # leads on model_id, so a second index only adds write amplification.
+    model_id: str = Field(
+        description="Client-visible /v1/models id (forwarded_model_id or id)"
+    )
+    path: str = Field(
+        description=(
+            "Opaque selector containing upstream URL, provider ID, model ID, "
+            "and optional endpoint tag"
+        )
+    )
+    provider_slug: str = Field(
+        description="Public slug of the configured upstream provider"
+    )
+    provider_type: str = Field(description="Configured upstream provider type")
+    endpoint_tag: str | None = Field(
+        default=None,
+        description="Exact OpenRouter endpoint tag used for request-side selection",
+    )
+    endpoint_name: str | None = Field(
+        default=None, description="Human-readable endpoint display name"
+    )
+    upstream_provider_id: int = Field(
+        index=True,
+        foreign_key="upstream_providers.id",
+        ondelete="CASCADE",
+        description="upstream_providers.id this path was discovered from",
+    )
+    updated_at: int = Field(
+        default=0,
+        description="Unix timestamp of the refresh cycle that wrote this row",
+    )
+
+
 class LightningInvoice(SQLModel, table=True):  # type: ignore
     __tablename__ = "lightning_invoices"
 
@@ -596,9 +651,7 @@ class CliToken(SQLModel, table=True):  # type: ignore
     """Long-lived authorization token for CLI/agent use against admin endpoints."""
 
     __tablename__ = "cli_tokens"
-    id: str = Field(
-        primary_key=True, default_factory=lambda: uuid.uuid4().hex
-    )
+    id: str = Field(primary_key=True, default_factory=lambda: uuid.uuid4().hex)
     token: str = Field(unique=True, index=True, description="Bearer token value")
     name: str = Field(description="Human-readable label for this token")
     created_at: int = Field(default_factory=lambda: int(time.time()))
@@ -697,9 +750,7 @@ async def reset_routstr_fee(session: AsyncSession, paid_msats: int) -> bool:
     return result.rowcount == 1
 
 
-async def complete_routstr_fee_payout(
-    session: AsyncSession, paid_msats: int
-) -> bool:
+async def complete_routstr_fee_payout(session: AsyncSession, paid_msats: int) -> bool:
     """Mark a checkpointed payout complete after the external payment succeeds."""
     stmt = (
         update(RoutstrFee)

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
@@ -130,6 +131,13 @@ async def lifespan(_: FastAPI) -> AsyncGenerator[None, None]:
                 refresh_upstreams_models_periodically(get_upstreams)
             )
         model_maps_refresh_task = asyncio.create_task(refresh_model_maps_periodically())
+        # Always started: the loop re-reads the enable flag and interval every
+        # iteration, so 0 -> N (or re-enabling) takes effect without a restart.
+        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())
@@ -137,9 +145,7 @@ async def lifespan(_: FastAPI) -> AsyncGenerator[None, None]:
         if global_settings.providers_refresh_interval_seconds > 0:
             providers_task = asyncio.create_task(providers_cache_refresher())
         key_reset_task = asyncio.create_task(periodic_key_reset())
-        stale_reservation_task = asyncio.create_task(
-            periodic_stale_reservation_sweep()
-        )
+        stale_reservation_task = asyncio.create_task(periodic_stale_reservation_sweep())
         dead_key_prune_task = asyncio.create_task(periodic_dead_key_prune())
         auto_topup_task = asyncio.create_task(periodic_auto_topup())
         refund_sweep_task = asyncio.create_task(periodic_refund_sweep())
@@ -176,6 +182,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:
@@ -209,6 +217,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:
@@ -245,9 +255,7 @@ class _ImmutableStaticFiles(StaticFiles):
     async def get_response(self, path: str, scope: Scope) -> StarletteResponse:
         response = await super().get_response(path, scope)
         if response.status_code == 200:
-            response.headers["Cache-Control"] = (
-                "public, max-age=31536000, immutable"
-            )
+            response.headers["Cache-Control"] = "public, max-age=31536000, immutable"
         return response
 
 
@@ -321,9 +329,7 @@ if UI_DIST_PATH.exists() and UI_DIST_PATH.is_dir():
     # Serve the App Router RSC payload for the home page.
     @app.get("/index.txt", include_in_schema=False)
     async def serve_root_rsc() -> FileResponse:
-        return FileResponse(
-            UI_DIST_PATH / "index.txt", media_type="text/x-component"
-        )
+        return FileResponse(UI_DIST_PATH / "index.txt", media_type="text/x-component")
 
     # Next.js is built with `trailingSlash: true`, so all UI page URLs end
     # with a slash (e.g. `/login/`). The proxy router catches `/{path:path}`

routstr/core/settings.py

@@ -95,10 +95,18 @@ 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")
+    enable_model_paths_refresh: bool = Field(
+        default=True, env="ENABLE_MODEL_PATHS_REFRESH"
+    )
     refund_cache_ttl_seconds: int = Field(default=3600, env="REFUND_CACHE_TTL_SECONDS")
-    refund_sweep_ttl_seconds: int = Field(default=604800, env="REFUND_SWEEP_TTL_SECONDS")
+    refund_sweep_ttl_seconds: int = Field(
+        default=604800, env="REFUND_SWEEP_TTL_SECONDS"
+    )
 
     # Logging
     log_level: str = Field(default="INFO", env="LOG_LEVEL")
@@ -117,9 +125,8 @@ class Settings(BaseSettings):
 
     # Discovery
     relays: list[str] = Field(default_factory=list, env="RELAYS")
-    enable_analytics_sharing: bool = Field(
-        default=True, env="ENABLE_ANALYTICS_SHARING"
-    )
+    enable_analytics_sharing: bool = Field(default=True, env="ENABLE_ANALYTICS_SHARING")
+
 
 def _normalize_settings_data(data: dict[str, Any]) -> dict[str, Any]:
     """Discard unknown keys from persisted settings."""
@@ -327,7 +334,11 @@ class SettingsService:
             valid_fields = set(env_resolved.dict().keys())
             merged_dict: dict[str, Any] = dict(env_resolved.dict())
             merged_dict.update(
-                {k: v for k, v in db_json.items() if v not in (None, "", [], {}) and k in valid_fields}
+                {
+                    k: v
+                    for k, v in db_json.items()
+                    if v not in (None, "", [], {}) and k in valid_fields
+                }
             )
             merged_dict = Settings(**merged_dict).dict()
 

routstr/payment/models.py

@@ -455,7 +455,9 @@ async def _update_sats_pricing_once() -> None:
             for m in upstream.get_cached_models()
         ]
         upstream._models_cache = updated_models
-        upstream._models_by_id = {m.forwarded_model_id or m.id: m for m in updated_models}
+        upstream._models_by_id = {
+            m.forwarded_model_id or m.id: m for m in updated_models
+        }
         updated_count += len(updated_models)
 
     if updated_count > 0:
@@ -510,9 +512,7 @@ class ModelTestRequest(V2BaseModel):
     request_data: dict
 
 
-@models_router.post(
-    "/api/models/test", dependencies=[Depends(_require_admin_api)]
-)
+@models_router.post("/api/models/test", dependencies=[Depends(_require_admin_api)])
 async def test_model(
     payload: ModelTestRequest,
     session: AsyncSession = Depends(get_session),
@@ -595,6 +595,37 @@ 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 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 ..proxy import get_unique_models
+    from ..upstream.model_paths import get_paths_for_model
+
+    result = await get_paths_for_model(model_id)
+    if not result["data"]:
+        advertised_ids = {
+            model.forwarded_model_id or model.id for model in get_unique_models()
+        }
+        if model_id not in advertised_ids:
+            raise HTTPException(status_code=404, detail="Model not found")
+    return result
+
+
 @models_router.get("/v1/models")
 @models_router.get("/v1/models/", include_in_schema=False)
 @models_router.get("/models")

routstr/proxy.py

@@ -183,6 +183,19 @@ async def refresh_model_maps() -> None:
         disabled_model_keys=disabled_model_keys,
     )
 
+    # Keep model-path discovery in sync with admin mutations: disabling or
+    # deleting a provider must stop advertising its paths immediately rather
+    # than after the next timed refresh.
+    from .upstream.model_paths import prune_model_paths_for_inactive_providers
+
+    try:
+        await prune_model_paths_for_inactive_providers()
+    except Exception as e:  # noqa: BLE001 - discovery sync must not break routing
+        logger.warning(
+            "Failed to prune model paths for inactive providers",
+            extra={"error": str(e), "error_type": type(e).__name__},
+        )
+
 
 async def refresh_model_maps_periodically() -> None:
     """Background task to refresh model maps every minute."""

routstr/upstream/model_paths.py

@@ -0,0 +1,807 @@
+"""Model-path discovery service.
+
+Exposes every selectable upstream route a Routstr model is reachable through.
+This PR remains discovery-only: request-side routing will consume the opaque
+selectors in a follow-up.
+
+A path is a standard percent-encoded query string containing the configured
+upstream URL, provider ID, client-visible model ID and, for an exact OpenRouter
+endpoint, its machine-readable tag::
+
+    url=https%3A%2F%2Fapi.anthropic.com%2Fv1&provider-id=12&model-id=claude-sonnet-4
+    url=https%3A%2F%2Fopenrouter.ai%2Fapi%2Fv1&provider-id=42&model-id=claude-sonnet-4&endpoint=google-vertex%2Fus
+"""
+
+from __future__ import annotations
+
+import asyncio
+import ipaddress
+import random
+import time
+from dataclasses import dataclass
+from typing import TYPE_CHECKING, Any, Callable
+from urllib.parse import urlencode, urlsplit
+
+import httpx
+from sqlalchemy.dialects.sqlite import insert
+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
+
+if TYPE_CHECKING:
+    from .base import BaseUpstreamProvider
+
+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
+
+# Rows inserted per statement during persist. Keeps each INSERT bounded while
+# avoiding per-row round-trips that hold SQLite's write lock for ~1s per cycle.
+_PERSIST_CHUNK_SIZE = 500
+
+# Admin mutations enqueue provider IDs here instead of running OpenRouter's
+# per-model endpoint fan-out inside the request. One worker serializes refreshes
+# and coalesces repeated mutations for the same provider.
+_scheduled_provider_refresh_ids: set[int] = set()
+_scheduled_provider_refresh_task: asyncio.Task[None] | None = None
+
+# Visibility key used across this module: routing carries the provider
+# dimension everywhere (ModelRow's primary key is (id, upstream_provider_id)),
+# so all model-id keyed maps here do too, lowercased like proxy.refresh_model_maps.
+ModelKey = tuple[str, int]
+
+
+@dataclass(frozen=True)
+class EndpointIdentity:
+    """Exact OpenRouter endpoint identity returned by ``/endpoints``."""
+
+    tag: str
+    provider_name: str | None
+
+
+@dataclass(frozen=True)
+class ConfiguredProviderIdentity:
+    """Public identity of one configured upstream provider."""
+
+    id: int
+    slug: str
+    provider_type: str
+    base_url: str
+
+
+@dataclass(frozen=True)
+class DiscoveredPath:
+    """One model route ready for persistence and API serialization."""
+
+    model_id: str
+    path: str
+    provider: ConfiguredProviderIdentity
+    endpoint_tag: str | None = None
+    endpoint_name: str | None = None
+
+
+@dataclass(frozen=True)
+class ProviderPathSnapshot:
+    """Refresh result plus model IDs whose prior rows must survive degradation."""
+
+    paths: tuple[DiscoveredPath, ...]
+    preserve_model_ids: frozenset[str] = frozenset()
+
+
+def public_provider_url(base_url: str) -> str:
+    """Mask private IP addresses and URLs with explicit ports."""
+    parsed = urlsplit(base_url)
+    try:
+        if parsed.port is not None:
+            return "http://localhost"
+    except ValueError:
+        # An invalid explicit port must not accidentally leak through.
+        return "http://localhost"
+
+    hostname = parsed.hostname
+    if hostname is None:
+        return base_url
+    try:
+        address = ipaddress.ip_address(hostname)
+    except ValueError:
+        return base_url
+    return "http://localhost" if address.is_private else base_url
+
+
+def encode_model_path(
+    base_url: str,
+    provider_id: int,
+    model_id: str,
+    endpoint_tag: str | None = None,
+) -> str:
+    """Encode the complete upstream route selector advertised to clients."""
+    components: list[tuple[str, str | int]] = [
+        ("url", base_url),
+        ("provider-id", provider_id),
+        ("model-id", model_id),
+    ]
+    if endpoint_tag:
+        components.append(("endpoint", endpoint_tag))
+    return urlencode(components)
+
+
+def _make_http_client() -> httpx.AsyncClient:
+    """Client factory, separated so tests can substitute a mock transport."""
+    return httpx.AsyncClient()
+
+
+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:
+    """Return exactly the ID advertised by ``/v1/models``.
+
+    A forwarded ID is already a public routable alias and must remain intact,
+    including any slash. Without one, ``/v1/models`` exposes the base ID.
+    """
+    forwarded = getattr(model, "forwarded_model_id", None)
+    if forwarded:
+        return forwarded
+    return public_model_id(getattr(model, "id"))
+
+
+def public_model_id(model_id: str) -> str:
+    """Model id exposed by model-path API responses.
+
+    Uses the same rule as ``create_model_mappings.get_base_model_id`` and
+    ``resolve_model_alias`` — strip everything before the *first* slash — so
+    the id shown here can be sent back to ``/v1/chat/completions`` verbatim.
+    """
+    return model_id.split("/", 1)[1] if "/" in model_id else model_id
+
+
+def openrouter_author_slug(model: object) -> str | None:
+    """Return a canonical ``author/slug`` for the OpenRouter endpoints API.
+
+    Prefer ``canonical_slug``, then a slash-containing ``id``, then a
+    slash-containing ``forwarded_model_id``. The forwarded id is exactly what
+    the proxy sends upstream for admin-created alias rows (``base.py`` forwards
+    ``forwarded_model_id or id``), so it is a valid OpenRouter id when the
+    bare ``id`` is a local alias with no slash.
+    """
+    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
+    forwarded = getattr(model, "forwarded_model_id", None)
+    if forwarded and "/" in forwarded:
+        return forwarded
+    return None
+
+
+class _RefreshCycleState:
+    """Per-refresh shared state: fetch dedupe cache and rate-limit latch.
+
+    ``endpoint_cache`` dedupes byte-identical ``/endpoints`` fetches when two
+    providers point at the same OpenRouter base URL. ``rate_limited`` latches
+    on the first 429 so the rest of the cycle stops hammering a throttled API;
+    the whole provider result then degrades to "unknown" instead of an empty
+    list, which preserves previously persisted rows.
+    """
+
+    def __init__(self) -> None:
+        self.endpoint_cache: dict[tuple[str, str], list[EndpointIdentity] | None] = {}
+        self.rate_limited = False
+
+
+async def _fetch_openrouter_endpoint_subproviders(
+    client: httpx.AsyncClient,
+    base_url: str,
+    api_key: str,
+    author_slug: str,
+    semaphore: asyncio.Semaphore,
+    cycle: _RefreshCycleState,
+) -> list[EndpointIdentity] | None:
+    """Return exact endpoint identities for one model, or ``None`` when unknown.
+
+    ``None`` (not ``[]``) signals a degraded fetch — network failure, rate
+    limit, non-200, or an unparseable payload — so callers can distinguish
+    "this model has no endpoints" from "we could not find out". Failures are
+    logged and swallowed so one model never breaks the whole refresh.
+    """
+    cache_key = (base_url, author_slug)
+    if cache_key in cycle.endpoint_cache:
+        return cycle.endpoint_cache[cache_key]
+    if cycle.rate_limited:
+        return None
+
+    url = f"{base_url.rstrip('/')}/models/{author_slug}/endpoints"
+    headers = {"Authorization": f"Bearer {api_key}"} if api_key else {}
+    result: list[EndpointIdentity] | None
+    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)},
+            )
+            cycle.endpoint_cache[cache_key] = None
+            return None
+
+    if resp.status_code == 429:
+        logger.warning(
+            "OpenRouter endpoint discovery rate-limited; aborting cycle",
+            extra={"author_slug": author_slug},
+        )
+        cycle.rate_limited = True
+        cycle.endpoint_cache[cache_key] = None
+        return None
+    if resp.status_code != 200:
+        logger.warning(
+            "OpenRouter endpoint discovery non-200",
+            extra={"author_slug": author_slug, "status_code": resp.status_code},
+        )
+        cycle.endpoint_cache[cache_key] = None
+        return None
+
+    try:
+        payload = resp.json()
+        data = payload.get("data") if isinstance(payload, dict) else None
+        endpoints = data.get("endpoints") if isinstance(data, dict) else None
+        if not isinstance(endpoints, list):
+            raise ValueError("endpoints must be a list")
+        identities: dict[str, EndpointIdentity] = {}
+        for endpoint in endpoints:
+            if not isinstance(endpoint, dict):
+                continue
+            tag = endpoint.get("tag")
+            if not isinstance(tag, str) or not tag.strip():
+                continue
+            provider_name = endpoint.get("provider_name")
+            identities.setdefault(
+                tag,
+                EndpointIdentity(
+                    tag=tag,
+                    provider_name=provider_name
+                    if isinstance(provider_name, str) and provider_name
+                    else None,
+                ),
+            )
+        if endpoints and not identities:
+            raise ValueError("endpoints contain no usable tags")
+        result = list(identities.values())
+    except Exception as e:  # noqa: BLE001
+        logger.warning(
+            "OpenRouter endpoint discovery bad payload",
+            extra={"author_slug": author_slug, "error": str(e)},
+        )
+        result = None
+
+    cycle.endpoint_cache[cache_key] = result
+    return result
+
+
+async def _load_model_visibility() -> tuple[
+    dict[ModelKey, ModelRow],
+    set[ModelKey],
+    dict[int, ConfiguredProviderIdentity],
+]:
+    """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 keys — all keyed on
+    ``(model_id.lower(), upstream_provider_id)`` because ``ModelRow``'s primary
+    key is composite and the same id legitimately exists on several providers.
+    Model-path discovery uses the same keying so disabling a model on one
+    provider never hides it on another, and one provider's
+    ``forwarded_model_id`` alias is never applied to a different provider.
+    """
+    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_key: dict[ModelKey, ModelRow] = {}
+    disabled_model_keys: set[ModelKey] = set()
+    provider_identities: dict[int, ConfiguredProviderIdentity] = {}
+
+    for provider in provider_rows:
+        if not provider.enabled or provider.id is None:
+            continue
+        provider_identities[provider.id] = ConfiguredProviderIdentity(
+            id=provider.id,
+            slug=provider.slug or f"provider-{provider.id}",
+            provider_type=provider.provider_type,
+            base_url=public_provider_url(provider.base_url),
+        )
+        for model in provider.models:
+            key = (model.id.lower(), provider.id)
+            if model.enabled:
+                overrides_by_key[key] = model
+            else:
+                disabled_model_keys.add(key)
+
+    return overrides_by_key, disabled_model_keys, provider_identities
+
+
+def _apply_model_visibility(
+    upstream: BaseUpstreamProvider,
+    overrides_by_key: dict[ModelKey, ModelRow] | None,
+    disabled_model_keys: set[ModelKey] | None,
+) -> list[object]:
+    """Return provider models after DB disabled/override state is applied.
+
+    Only the identity fields (``id``, ``forwarded_model_id``,
+    ``canonical_slug``) matter for path discovery, so DB override rows are used
+    directly rather than rebuilt into fully priced ``Model`` objects — the
+    pricing pipeline costs ~0.7ms of event-loop CPU per row for data this
+    module immediately discards.
+    """
+    overrides_by_key = overrides_by_key or {}
+    disabled_model_keys = disabled_model_keys or set()
+    upstream_provider_id = getattr(upstream, "db_id", None)
+    if not isinstance(upstream_provider_id, int):
+        return [
+            model
+            for model in upstream.get_cached_models()
+            if getattr(model, "enabled", True)
+        ]
+
+    visible_models: list[object] = []
+    seen_model_ids: set[str] = set()
+
+    for model in upstream.get_cached_models():
+        model_id = getattr(model, "id", "")
+        key = (model_id.lower(), upstream_provider_id)
+        if not getattr(model, "enabled", True) or key in disabled_model_keys:
+            continue
+        # Apply overrides only for this provider's own model row.
+        override_row = overrides_by_key.get(key)
+        visible: object = model if override_row is None else override_row
+        visible_models.append(visible)
+        seen_model_ids.add(model_id.lower())
+
+    # DB-only override rows for this provider with no cached counterpart.
+    for (model_id_lower, provider_id), override_row in overrides_by_key.items():
+        if provider_id != upstream_provider_id:
+            continue
+        if model_id_lower in seen_model_ids:
+            continue
+        visible_models.append(override_row)
+        seen_model_ids.add(model_id_lower)
+
+    return visible_models
+
+
+async def _collect_provider_paths(
+    upstream: BaseUpstreamProvider,
+    provider_identity: ConfiguredProviderIdentity,
+    overrides_by_key: dict[ModelKey, ModelRow] | None = None,
+    disabled_model_keys: set[ModelKey] | None = None,
+    cycle: _RefreshCycleState | None = None,
+) -> ProviderPathSnapshot:
+    """Collect selectable routes while marking model-level degraded fetches.
+
+    A failed OpenRouter lookup preserves only that model's prior rows. Other
+    models in the same provider still refresh, so a partial outage cannot erase
+    valid discovery data or freeze the entire provider snapshot.
+    """
+    cycle = cycle or _RefreshCycleState()
+    models = _apply_model_visibility(upstream, overrides_by_key, disabled_model_keys)
+
+    def _base_path(model: object) -> DiscoveredPath:
+        model_id = exposed_model_id(model)
+        return DiscoveredPath(
+            model_id=model_id,
+            path=encode_model_path(
+                provider_identity.base_url, provider_identity.id, model_id
+            ),
+            provider=provider_identity,
+        )
+
+    if not is_openrouter_base_url(upstream.base_url):
+        return ProviderPathSnapshot(paths=tuple(_base_path(model) for model in models))
+
+    if not (upstream.provider_type or "").strip():
+        return ProviderPathSnapshot(paths=())
+
+    semaphore = asyncio.Semaphore(_OPENROUTER_CONCURRENCY)
+    async with _make_http_client() as client:
+
+        async def _for_model(
+            model: object,
+        ) -> tuple[list[DiscoveredPath], str | None]:
+            model_id = exposed_model_id(model)
+            author_slug = openrouter_author_slug(model)
+            if not author_slug:
+                return [_base_path(model)], None
+            endpoints = await _fetch_openrouter_endpoint_subproviders(
+                client,
+                upstream.base_url,
+                upstream.api_key,
+                author_slug,
+                semaphore,
+                cycle,
+            )
+            if endpoints is None:
+                return [], model_id
+            paths = [_base_path(model)]
+            paths.extend(
+                DiscoveredPath(
+                    model_id=model_id,
+                    path=encode_model_path(
+                        provider_identity.base_url,
+                        provider_identity.id,
+                        model_id,
+                        endpoint.tag,
+                    ),
+                    provider=provider_identity,
+                    endpoint_tag=endpoint.tag,
+                    endpoint_name=endpoint.provider_name,
+                )
+                for endpoint in endpoints
+            )
+            return paths, None
+
+        results = await asyncio.gather(
+            *(_for_model(model) for model in models), return_exceptions=True
+        )
+
+    paths: list[DiscoveredPath] = []
+    preserve_model_ids: set[str] = set()
+    for model, result in zip(models, results):
+        if isinstance(result, BaseException):
+            model_id = exposed_model_id(model)
+            preserve_model_ids.add(model_id)
+            logger.warning(
+                "OpenRouter endpoint discovery task errored",
+                extra={"provider": upstream.provider_type, "error": str(result)},
+            )
+            continue
+        model_paths, preserved_model_id = result
+        paths.extend(model_paths)
+        if preserved_model_id:
+            preserve_model_ids.add(preserved_model_id)
+
+    return ProviderPathSnapshot(
+        paths=tuple(paths), preserve_model_ids=frozenset(preserve_model_ids)
+    )
+
+
+async def _persist_provider_paths(
+    upstream_provider_id: int, snapshot: ProviderPathSnapshot
+) -> None:
+    """Replace refreshed rows while retaining model-level degraded snapshots."""
+    unique_paths = list(
+        {(path.model_id, path.path): path for path in snapshot.paths}.values()
+    )
+    now = int(time.time())
+    async with create_session() as session:
+        delete_stmt = delete(ModelPathRow).where(
+            col(ModelPathRow.upstream_provider_id) == upstream_provider_id
+        )
+        if snapshot.preserve_model_ids:
+            delete_stmt = delete_stmt.where(
+                col(ModelPathRow.model_id).not_in(sorted(snapshot.preserve_model_ids))
+            )
+        await session.exec(delete_stmt)  # type: ignore[call-overload]
+        for start in range(0, len(unique_paths), _PERSIST_CHUNK_SIZE):
+            chunk = unique_paths[start : start + _PERSIST_CHUNK_SIZE]
+            values = [
+                {
+                    "model_id": discovered.model_id,
+                    "path": discovered.path,
+                    "provider_slug": discovered.provider.slug,
+                    "provider_type": discovered.provider.provider_type,
+                    "endpoint_tag": discovered.endpoint_tag,
+                    "endpoint_name": discovered.endpoint_name,
+                    "upstream_provider_id": upstream_provider_id,
+                    "updated_at": now,
+                }
+                for discovered in chunk
+            ]
+            insert_stmt = insert(ModelPathRow).values(values)
+            await session.execute(
+                insert_stmt.on_conflict_do_update(
+                    index_elements=["model_id", "path", "upstream_provider_id"],
+                    set_={
+                        "provider_slug": insert_stmt.excluded.provider_slug,
+                        "provider_type": insert_stmt.excluded.provider_type,
+                        "endpoint_tag": insert_stmt.excluded.endpoint_tag,
+                        "endpoint_name": insert_stmt.excluded.endpoint_name,
+                        "updated_at": insert_stmt.excluded.updated_at,
+                    },
+                )
+            )
+        await session.commit()
+
+
+async def prune_model_paths_for_inactive_providers() -> None:
+    """Delete paths whose provider is no longer enabled in the database.
+
+    Called from ``refresh_model_maps`` so admin mutations (disable/delete
+    provider) stop advertising a provider's paths immediately instead of
+    waiting for the next timed refresh. Uses the DB as the source of truth, so
+    it is safe at boot even before upstreams initialize.
+    """
+    async with create_session() as session:
+        enabled_ids = (
+            await session.exec(
+                select(UpstreamProviderRow.id).where(
+                    col(UpstreamProviderRow.enabled).is_(True)
+                )
+            )
+        ).all()
+        stmt = delete(ModelPathRow)
+        if enabled_ids:
+            stmt = stmt.where(
+                col(ModelPathRow.upstream_provider_id).not_in(
+                    [pid for pid in enabled_ids if pid is not None]
+                )
+            )
+        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.
+    A provider whose paths could not be determined this cycle keeps its
+    previously persisted rows. An empty ``upstreams`` list (e.g. a failed
+    ``initialize_upstreams`` at boot) is treated as "unknown" and touches
+    nothing.
+    """
+    if not upstreams:
+        logger.warning("Skipping model paths refresh: no live upstreams")
+        return
+
+    (
+        overrides_by_key,
+        disabled_model_keys,
+        provider_identities,
+    ) = await _load_model_visibility()
+    await prune_model_paths_for_inactive_providers()
+
+    cycle = _RefreshCycleState()
+    for upstream in upstreams:
+        if upstream.db_id is None or upstream.db_id not in provider_identities:
+            continue
+        try:
+            snapshot = await _collect_provider_paths(
+                upstream,
+                provider_identity=provider_identities[upstream.db_id],
+                overrides_by_key=overrides_by_key,
+                disabled_model_keys=disabled_model_keys,
+                cycle=cycle,
+            )
+            if snapshot.preserve_model_ids:
+                logger.warning(
+                    "Some model paths are unknown; keeping their previous rows",
+                    extra={
+                        "provider": upstream.provider_type or upstream.base_url,
+                        "db_id": upstream.db_id,
+                        "preserved_models": len(snapshot.preserve_model_ids),
+                    },
+                )
+            await _persist_provider_paths(upstream.db_id, snapshot)
+        except Exception as e:  # noqa: BLE001 - isolate per-provider failures
+            logger.error(
+                "Failed to refresh model paths for provider",
+                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_for_provider(upstream_provider_id: int) -> None:
+    """Synchronize one provider when model-path discovery is enabled."""
+    if _refresh_interval_seconds() <= 0:
+        return
+
+    from ..proxy import get_upstreams
+
+    matching = [
+        upstream
+        for upstream in get_upstreams()
+        if upstream.db_id == upstream_provider_id
+    ]
+    if matching:
+        await refresh_model_paths(matching)
+    else:
+        await prune_model_paths_for_inactive_providers()
+
+
+async def _drain_scheduled_provider_refreshes() -> None:
+    """Serialize and coalesce model-path refreshes scheduled by admin writes."""
+    global _scheduled_provider_refresh_task
+
+    try:
+        # Let mutations in the same event-loop turn collapse into one refresh.
+        await asyncio.sleep(0)
+        while _scheduled_provider_refresh_ids:
+            if _refresh_interval_seconds() <= 0:
+                _scheduled_provider_refresh_ids.clear()
+                return
+            provider_id = min(_scheduled_provider_refresh_ids)
+            _scheduled_provider_refresh_ids.remove(provider_id)
+            try:
+                await refresh_model_paths_for_provider(provider_id)
+            except asyncio.CancelledError:
+                raise
+            except Exception as exc:  # noqa: BLE001 - background best effort
+                logger.warning(
+                    "Failed to refresh model paths after admin mutation",
+                    extra={
+                        "upstream_provider_id": provider_id,
+                        "error": str(exc),
+                        "error_type": type(exc).__name__,
+                    },
+                )
+    finally:
+        _scheduled_provider_refresh_task = None
+
+
+async def schedule_model_paths_refresh_for_provider(
+    upstream_provider_id: int,
+) -> None:
+    """Queue a non-blocking, coalesced refresh after an admin mutation."""
+    global _scheduled_provider_refresh_task
+
+    if _refresh_interval_seconds() <= 0:
+        return
+    _scheduled_provider_refresh_ids.add(upstream_provider_id)
+    if (
+        _scheduled_provider_refresh_task is None
+        or _scheduled_provider_refresh_task.done()
+    ):
+        _scheduled_provider_refresh_task = asyncio.create_task(
+            _drain_scheduled_provider_refreshes(),
+            name="model-path-admin-refresh",
+        )
+
+
+def _refresh_interval_seconds() -> int:
+    """Current interval, re-read every loop so runtime setting changes apply."""
+    from ..core.settings import settings
+
+    if not getattr(settings, "enable_model_paths_refresh", True):
+        return 0
+    return int(getattr(settings, "model_paths_refresh_interval_seconds", 0) or 0)
+
+
+async def refresh_model_paths_periodically(
+    upstreams_provider: (
+        Callable[[], list[BaseUpstreamProvider]] | list[BaseUpstreamProvider]
+    ),
+) -> None:
+    """Background task mirroring ``refresh_upstreams_models_periodically``.
+
+    The interval and enable flag are re-read every iteration, so the refresh
+    can be turned off (or on) and retuned without a restart. While disabled the
+    task idles instead of exiting, so re-enabling takes effect.
+    """
+    _DISABLED_POLL_SECONDS = 60.0
+
+    def _resolve_upstreams() -> list[BaseUpstreamProvider]:
+        if callable(upstreams_provider):
+            return upstreams_provider()
+        return upstreams_provider
+
+    while True:
+        interval = _refresh_interval_seconds()
+        if interval <= 0:
+            try:
+                await asyncio.sleep(_DISABLED_POLL_SECONDS)
+            except asyncio.CancelledError:
+                break
+            continue
+
+        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
+
+
+def _serialize_path(row: ModelPathRow) -> dict[str, Any]:
+    endpoint = None
+    if row.endpoint_tag or row.endpoint_name:
+        endpoint = {"tag": row.endpoint_tag, "name": row.endpoint_name}
+    return {
+        "path": row.path,
+        "provider": {
+            "id": row.upstream_provider_id,
+            "slug": row.provider_slug,
+            "type": row.provider_type,
+        },
+        "endpoint": endpoint,
+    }
+
+
+async def get_all_model_paths() -> dict:
+    """All models with their exact selectable routes."""
+    async with create_session() as session:
+        rows = (
+            await session.exec(
+                select(ModelPathRow).order_by(
+                    col(ModelPathRow.model_id),
+                    col(ModelPathRow.path),
+                    col(ModelPathRow.upstream_provider_id),
+                )
+            )
+        ).all()
+
+    grouped: dict[str, list[dict[str, Any]]] = {}
+    seen_paths: dict[str, set[str]] = {}
+    updated_at = 0
+    for row in rows:
+        updated_at = max(updated_at, row.updated_at)
+        if row.path in seen_paths.setdefault(row.model_id, set()):
+            continue
+        seen_paths[row.model_id].add(row.path)
+        grouped.setdefault(row.model_id, []).append(_serialize_path(row))
+    data = [
+        {
+            "id": grouped_model_id,
+            "paths": grouped[grouped_model_id],
+        }
+        for grouped_model_id in sorted(grouped)
+    ]
+    return {"data": data, "updated_at": updated_at or None}
+
+
+async def get_paths_for_model(model_id: str) -> dict:
+    """Return paths only for the exact model ID advertised by ``/v1/models``."""
+    async with create_session() as session:
+        rows = (
+            await session.exec(
+                select(ModelPathRow)
+                .where(col(ModelPathRow.model_id) == model_id)
+                .order_by(
+                    col(ModelPathRow.path),
+                    col(ModelPathRow.upstream_provider_id),
+                )
+            )
+        ).all()
+
+    seen: set[str] = set()
+    paths: list[dict] = []
+    updated_at = 0
+    for row in rows:
+        updated_at = max(updated_at, row.updated_at)
+        if row.path in seen:
+            continue
+        seen.add(row.path)
+        paths.append(_serialize_path(row))
+    return {"data": paths, "updated_at": updated_at or None}

tests/unit/test_fee_payout_migration.py

@@ -4,6 +4,9 @@ import subprocess
 import sys
 from pathlib import Path
 
+from alembic.config import Config
+from alembic.script import ScriptDirectory
+
 
 def _run_alembic(root: Path, database_url: str, revision: str) -> None:
     env = os.environ.copy()
@@ -37,7 +40,10 @@ def test_fresh_node_migrates_fee_payout_schema_to_head(tmp_path: Path) -> None:
             "payout_in_progress_msats, payout_started_at FROM routstr_fees"
         ).fetchone()
 
-    assert version == ("9c4d8e2f1a6b",)
+    migration_config = Config(str(root / "alembic.ini"))
+    assert version == (
+        ScriptDirectory.from_config(migration_config).get_current_head(),
+    )
     assert {
         "id",
         "accumulated_msats",