Refactor provider runtime ownership (#925)

## Problem

Provider construction, model discovery, validation, and cleanup lived in
one registry module. API and admin routes depended on registry-shaped
app state and legacy process-level provider helpers.

## Changes

| Before | After |
| --- | --- |
| `providers.registry` mixed provider factories, config, cache,
discovery, validation, and cleanup. | `providers.runtime` splits
factories, config, cache, model cache, discovery, validation, and
runtime orchestration. |
| API and admin routes read `app.state.provider_registry` and sometimes
created registries ad hoc. | API and admin routes use app-scoped
`ProviderRuntime` through `app.state.provider_runtime`. |
| `api.dependencies` kept process-global provider cache helpers. |
`api.dependencies` resolves providers only through the app-scoped
runtime. |
| Registry-shaped tests preserved old internal boundaries. |
Runtime-shaped tests assert provider config, construction, cache,
discovery, validation, and import boundaries. |

<!-- greptile_comment -->

<details open><summary><h3>Greptile Summary</h3></summary>

This PR moves provider lifecycle ownership from the old registry module
into an app-scoped runtime package. The main changes are:

- Split provider config, factory wiring, instance cache, model cache,
discovery, validation, and cleanup into `providers.runtime` modules.
- Updated API and admin routes to resolve providers and model metadata
through `app.state.provider_runtime`.
- Removed legacy process-global provider helpers and the deleted
`providers.registry` module.
- Updated docs, smoke metadata, import-boundary checks, and tests for
the new runtime ownership model.
- Bumped the package version and lockfile metadata for the production
refactor.
</details>

<h3>Confidence Score: 5/5</h3>

The provider runtime refactor appears merge-safe with no identified
blocking issues.

The changes consistently move provider ownership to app-scoped runtime
modules and update API, admin, docs, smoke metadata, import-boundary
checks, and tests around that architecture.

<details><summary><h3><a href="https://www.greptile.com/trex"><img
alt="T-Rex"
src="https://greptile-static-assets.s3.amazonaws.com/trex/trex_green.svg"
height="20" align="absmiddle"></a> T-Rex Logs</h3></summary>

**What T-Rex did**
- Ran a baseline and head comparison of provider registry and runtime
states, verifying the after-state shows head
state\_has\_provider\_registry=False and
state\_has\_provider\_runtime=True, that GET /v1/models and admin
endpoints respond with 200, and that provider\_resolver\_called via
runtime, with assertions passing.
- Verified that the four focused provider-runtime contract tests passed
in both the before and after refactor runs, including runtime split
checks, with exit code 0.
- Identified environmental blockers that prevented the smoke-runtime
workflow from running, including uv unavailability, missing pytest for
/usr/local/bin/python, and Python 3.11 being used despite pyproject.toml
requiring \>=3.14.

<a
href="https://app.greptile.com/trex/runs/12528505/artifacts"><picture><source
media="(prefers-color-scheme: dark)"
srcset="https://greptile-static-assets.s3.amazonaws.com/badges/ViewAllArtifactsDark.svg?v=1"><source
media="(prefers-color-scheme: light)"
srcset="https://greptile-static-assets.s3.amazonaws.com/badges/ViewAllArtifacts.svg?v=1"><img
alt="View all artifacts"
src="https://greptile-static-assets.s3.amazonaws.com/badges/ViewAllArtifacts.svg?v=1"
height="32"></picture></a>

_{<a href="https://www.greptile.com/trex"><img alt="T-Rex"
src="https://greptile-static-assets.s3.amazonaws.com/trex/trex_green.svg"
height="14" align="absmiddle"></a> Ran code and verified through
T-Rex}
</details>

_{Reviews (1): Last reviewed commit: ["Refactor provider runtime
ownership"](https://github.com/alishahryar1/free-claude-code/commit/01d589488185c1f85112f1a49c47f04512846161)
| [Re-trigger
Greptile](https://app.greptile.com/api/retrigger?id=40312173)}

<!-- /greptile_comment -->

ARCHITECTURE.md

@@ -35,7 +35,7 @@ flowchart LR
     ProxyAPI --> Handlers[API Product Handlers]
     Handlers --> Router[ModelRouter]
     Handlers --> Executor[ProviderExecutionService]
-    Executor --> Providers[ProviderRegistry]
+    Executor --> Providers[ProviderRuntime]
     Providers --> OpenAIChat[OpenAI Chat Providers]
     Providers --> NativeAnthropic[Anthropic Messages Providers]
 ```
@@ -156,7 +156,7 @@ reported without noisy Starlette tracebacks.
 [api/runtime.py](api/runtime.py) owns process-lifetime resources through
 `AppRuntime`:
 
-- creates and publishes an app-scoped `ProviderRegistry`;
+- creates and publishes an app-scoped `ProviderRuntime`;
 - validates configured models best-effort without blocking first-run admin access;
 - starts provider model-list refresh;
 - starts optional Discord or Telegram messaging when configured;
@@ -198,7 +198,7 @@ Model routing configuration is tiered:
 and writes managed env updates. [api/admin_routes.py](api/admin_routes.py)
 exposes local-only admin endpoints that load, validate, apply, and test config.
 After an apply, settings are cache-cleared. Depending on the changed fields, the
-server either replaces the app provider registry or asks the supervised server to
+server either replaces the app provider runtime or asks the supervised server to
 restart.
 
 Admin routes call `require_loopback_admin()`, which rejects non-loopback clients
@@ -241,7 +241,7 @@ sequenceDiagram
     participant Handler as ProductHandler
     participant Router as ModelRouter
     participant Exec as ProviderExecution
-    participant Registry as ProviderRegistry
+    participant Runtime as ProviderRuntime
     participant Provider
 
     Client->>Route: POST /v1/messages
@@ -250,8 +250,8 @@ sequenceDiagram
     Handler->>Router: resolve model and thinking
     Handler->>Handler: server tools or optimizations
     Handler->>Exec: stream routed request
-    Exec->>Registry: resolve provider
-    Registry->>Provider: cached or new provider
+    Exec->>Runtime: resolve provider
+    Runtime->>Provider: cached or new provider
     Exec->>Provider: preflight_stream
     Exec->>Provider: stream_response
     Provider-->>Client: Anthropic SSE events
@@ -287,7 +287,7 @@ overrides or the global setting.
 - no-thinking variants when appropriate;
 - built-in Claude model IDs for compatibility with Claude clients.
 
-Provider model discovery is app-scoped through `ProviderRegistry`, which caches
+Provider model discovery is app-scoped through `ProviderRuntime`, which caches
 model IDs and optional thinking capability metadata for the model-list route and
 admin status.
 
@@ -304,10 +304,12 @@ Provider metadata is neutral and centralized in
 `ProviderDescriptor` declares provider ID, transport type, capabilities,
 credential env var, default base URL, settings attribute names, and proxy support.
 
-[providers/registry.py](providers/registry.py) owns provider factories and the
-runtime registry. It validates that descriptors, factories, and supported IDs are
-in sync, builds shared `ProviderConfig`, checks required credentials, creates
-providers lazily, caches them, refreshes model lists, and cleans up transports.
+[providers/runtime/](providers/runtime/) owns the app-scoped provider runtime.
+It validates that descriptors, factories, and supported IDs are in sync, builds
+shared `ProviderConfig`, checks required credentials, creates providers lazily,
+caches them, refreshes model lists, validates configured models, and cleans up
+transports. The package splits factory wiring, config building, provider instance
+cache, model metadata cache, discovery, and validation into separate modules.
 
 [providers/base.py](providers/base.py) defines:
 
@@ -342,7 +344,7 @@ where supported, and returning Anthropic SSE strings to the service layer.
    the setting should be editable in the Admin UI.
 4. Implement the provider under [providers/](providers/) using the appropriate
    shared transport family.
-5. Add a factory in [providers/registry.py](providers/registry.py).
+5. Add a factory in [providers/runtime/factory.py](providers/runtime/factory.py).
 6. Add deterministic tests under [tests/providers/](tests/providers/) and any
    relevant contract tests.
 7. Add smoke coverage or smoke config in [smoke/](smoke/) when the provider can
@@ -697,7 +699,7 @@ Update this file when a change adds or meaningfully changes:
 - a public route or wire protocol;
 - startup, shutdown, or resource ownership;
 - configuration precedence or managed config behavior;
-- provider registry, catalog, or transport architecture;
+- provider runtime, catalog, or transport architecture;
 - model routing or thinking behavior;
 - CLI adapter behavior;
 - messaging platform behavior;

README.md

@@ -577,7 +577,7 @@ free-claude-code/
 ├── api/                   # FastAPI routes, service layer, routing, optimizations
 ├── core/                  # Shared Anthropic protocol helpers, SSE, OpenAI Responses
 │   └── openai_responses/  # Responses ↔ Anthropic conversion and SSE mapping
-├── providers/             # Provider transports, registry, rate limiting
+├── providers/             # Provider runtime, transports, rate limiting
 ├── messaging/             # Discord/Telegram runtimes, outbound ports, voice
 ├── cli/                   # Package entry points and client CLI process management
 ├── config/                # Settings, provider catalog, logging
@@ -639,7 +639,7 @@ CI also enforces a ban on `# type: ignore` / `# ty: ignore` suppressions; `scrip
 - Add OpenAI-compatible providers by extending `OpenAIChatTransport`.
 - Add Anthropic Messages providers by extending `AnthropicMessagesTransport`.
 - Extend OpenAI Responses conversion in `core/openai_responses/` when Codex adds new request or stream shapes.
-- Register provider metadata in `config.provider_catalog` and factory wiring in `providers.registry`.
+- Register provider metadata in `config.provider_catalog` and factory wiring in `providers.runtime`.
 - Add messaging platforms by wiring runtime, outbound, and inbound-normalizer ports in `messaging/platforms/`.
 
 ## Contributing

api/admin_routes.py

@@ -15,7 +15,7 @@ from pydantic import BaseModel, Field
 
 from config.settings import Settings
 from config.settings import get_settings as get_cached_settings
-from providers.registry import ProviderRegistry
+from providers.runtime import ProviderRuntime
 
 from .admin_config import (
     FIELD_BY_KEY,
@@ -126,10 +126,10 @@ async def apply_admin_config(
         request.app.state.admin_pending_fields = []
         return result
 
-    old_registry = getattr(request.app.state, "provider_registry", None)
-    if isinstance(old_registry, ProviderRegistry):
-        await old_registry.cleanup()
-    request.app.state.provider_registry = ProviderRegistry()
+    old_runtime = getattr(request.app.state, "provider_runtime", None)
+    if isinstance(old_runtime, ProviderRuntime):
+        await old_runtime.cleanup()
+    request.app.state.provider_runtime = ProviderRuntime(get_cached_settings())
     request.app.state.admin_pending_fields = result["pending_fields"]
     return result
 
@@ -138,12 +138,12 @@ async def apply_admin_config(
 async def admin_status(request: Request):
     require_loopback_admin(request)
     settings = get_cached_settings()
-    registry = getattr(request.app.state, "provider_registry", None)
+    runtime = getattr(request.app.state, "provider_runtime", None)
     cached_models: dict[str, list[str]] = {}
-    if isinstance(registry, ProviderRegistry):
+    if isinstance(runtime, ProviderRuntime):
         cached_models = {
             provider_id: sorted(model_ids)
-            for provider_id, model_ids in registry.cached_model_ids().items()
+            for provider_id, model_ids in runtime.cached_model_ids().items()
         }
     return {
         "status": "running",
@@ -173,12 +173,9 @@ async def local_provider_status(request: Request):
 async def test_provider(provider_id: str, request: Request):
     require_loopback_admin(request)
     settings = get_cached_settings()
-    registry = getattr(request.app.state, "provider_registry", None)
-    if not isinstance(registry, ProviderRegistry):
-        registry = ProviderRegistry()
-        request.app.state.provider_registry = registry
+    runtime = _provider_runtime_for_admin(request, settings)
     try:
-        provider = registry.get(provider_id, settings)
+        provider = runtime.resolve_provider(provider_id)
         infos = await provider.list_model_infos()
     except Exception as exc:
         return {
@@ -186,7 +183,7 @@ async def test_provider(provider_id: str, request: Request):
             "ok": False,
             "error_type": type(exc).__name__,
         }
-    registry.cache_model_infos(provider_id, infos)
+    runtime.cache_model_infos(provider_id, infos)
     return {
         "provider_id": provider_id,
         "ok": True,
@@ -198,19 +195,27 @@ async def test_provider(provider_id: str, request: Request):
 async def refresh_models(request: Request):
     require_loopback_admin(request)
     settings = get_cached_settings()
-    registry = getattr(request.app.state, "provider_registry", None)
-    if not isinstance(registry, ProviderRegistry):
-        registry = ProviderRegistry()
-        request.app.state.provider_registry = registry
-    await registry.refresh_model_list_cache(settings)
+    runtime = _provider_runtime_for_admin(request, settings)
+    await runtime.refresh_model_list_cache()
     return {
         "cached_models": {
             provider_id: sorted(model_ids)
-            for provider_id, model_ids in registry.cached_model_ids().items()
+            for provider_id, model_ids in runtime.cached_model_ids().items()
         }
     }
 
 
+def _provider_runtime_for_admin(
+    request: Request, settings: Settings
+) -> ProviderRuntime:
+    runtime = getattr(request.app.state, "provider_runtime", None)
+    if isinstance(runtime, ProviderRuntime):
+        return runtime
+    runtime = ProviderRuntime(settings)
+    request.app.state.provider_runtime = runtime
+    return runtime
+
+
 def _filtered_values(values: dict[str, Any]) -> dict[str, Any]:
     return {key: value for key, value in values.items() if key in FIELD_BY_KEY}
 

api/dependencies.py

@@ -6,6 +6,7 @@ from fastapi import Depends, HTTPException, Request
 from loguru import logger
 from starlette.applications import Starlette
 
+from config.provider_catalog import PROVIDER_CATALOG
 from config.settings import Settings
 from config.settings import get_settings as _get_settings
 from core.anthropic import get_user_facing_error_message
@@ -15,12 +16,7 @@ from providers.exceptions import (
     ServiceUnavailableError,
     UnknownProviderTypeError,
 )
-from providers.registry import PROVIDER_DESCRIPTORS, ProviderRegistry
-
-# Process-level cache: only for :func:`get_provider_for_type` / :func:`get_provider`
-# when there is no ``Request``/``app`` (unit tests, scripts). HTTP handlers must pass
-# ``app`` to :func:`resolve_provider` so the app-scoped registry is used.
-_providers: dict[str, BaseProvider] = {}
+from providers.runtime import ProviderRuntime
 
 
 def get_settings() -> Settings:
@@ -28,39 +24,33 @@ def get_settings() -> Settings:
     return _get_settings()
 
 
+def get_provider_runtime(app: Starlette) -> ProviderRuntime:
+    """Return the app-scoped provider runtime installed by ``AppRuntime``."""
+    runtime = getattr(app.state, "provider_runtime", None)
+    if isinstance(runtime, ProviderRuntime):
+        return runtime
+    raise ServiceUnavailableError(
+        "Provider runtime is not configured. Ensure AppRuntime startup ran "
+        "or assign app.state.provider_runtime for test apps."
+    )
+
+
+def maybe_provider_runtime(app: Starlette) -> ProviderRuntime | None:
+    """Return the app-scoped provider runtime when it is installed."""
+    runtime = getattr(app.state, "provider_runtime", None)
+    return runtime if isinstance(runtime, ProviderRuntime) else None
+
+
 def resolve_provider(
     provider_type: str,
     *,
-    app: Starlette | None,
-    settings: Settings,
-) -> BaseProvider:
-    """Resolve a provider using the app-scoped registry when ``app`` is set.
-
-    When ``app`` is not ``None``, the app-owned :attr:`app.state.provider_registry`
-    must exist (installed by :class:`~api.runtime.AppRuntime` during startup).
-    Callers that construct a bare ``FastAPI`` without lifespan must set
-    ``app.state.provider_registry`` explicitly.
-
-    When ``app`` is ``None`` (no HTTP context), uses the process-level
-    :data:`_providers` cache only.
-    """
-    if app is not None:
-        reg = getattr(app.state, "provider_registry", None)
-        if reg is None:
-            raise ServiceUnavailableError(
-                "Provider registry is not configured. Ensure AppRuntime startup ran "
-                "or assign app.state.provider_registry for test apps."
-            )
-        return _resolve_with_registry(reg, provider_type, settings)
-    return _resolve_with_registry(ProviderRegistry(_providers), provider_type, settings)
-
-
-def _resolve_with_registry(
-    registry: ProviderRegistry, provider_type: str, settings: Settings
+    app: Starlette,
 ) -> BaseProvider:
-    should_log_init = not registry.is_cached(provider_type)
+    """Resolve a provider through the app-scoped provider runtime."""
+    runtime = get_provider_runtime(app)
+    should_log_init = not runtime.is_cached(provider_type)
     try:
-        provider = registry.get(provider_type, settings)
+        provider = runtime.resolve_provider(provider_type)
     except AuthenticationError as e:
         # Provider :class:`~providers.exceptions.AuthenticationError` messages are
         # curated configuration hints (env var names, docs links), not upstream noise.
@@ -70,7 +60,7 @@ def _resolve_with_registry(
         logger.error(
             "Unknown provider_type: '{}'. Supported: {}",
             provider_type,
-            ", ".join(f"'{key}'" for key in PROVIDER_DESCRIPTORS),
+            ", ".join(f"'{key}'" for key in PROVIDER_CATALOG),
         )
         raise
     if should_log_init:
@@ -78,16 +68,6 @@ def _resolve_with_registry(
     return provider
 
 
-def get_provider_for_type(provider_type: str) -> BaseProvider:
-    """Get or create a provider in the process-level cache (no ``app``/Request).
-
-    HTTP route handlers should call :func:`resolve_provider` with the active
-    :attr:`request.app` (via :class:`~api.runtime.AppRuntime`) instead of this
-    process-wide cache.
-    """
-    return resolve_provider(provider_type, app=None, settings=get_settings())
-
-
 def require_api_key(
     request: Request, settings: Settings = Depends(get_settings)
 ) -> None:
@@ -124,21 +104,3 @@ def require_api_key(
         token.encode("utf-8"), anthropic_auth_token.encode("utf-8")
     ):
         raise HTTPException(status_code=401, detail="Invalid API key")
-
-
-def get_provider() -> BaseProvider:
-    """Get or create the default provider (``MODEL`` / ``provider_type``).
-
-    Process-cache helper for scripts, unit tests, and non-FastAPI callers. HTTP
-    handlers must use :func:`resolve_provider` with :attr:`request.app` so the
-    app-scoped :class:`~providers.registry.ProviderRegistry` is used.
-    """
-    return get_provider_for_type(get_settings().provider_type)
-
-
-async def cleanup_provider():
-    """Cleanup all provider resources."""
-    global _providers
-    await ProviderRegistry(_providers).cleanup()
-    _providers = {}
-    logger.debug("Provider cleanup completed")

api/model_catalog.py

@@ -3,7 +3,7 @@
 from __future__ import annotations
 
 from config.settings import Settings
-from providers.registry import ProviderRegistry
+from providers.runtime import ProviderRuntime
 
 from .gateway_model_ids import gateway_model_id, no_thinking_gateway_model_id
 from .models.responses import ModelResponse, ModelsListResponse
@@ -51,7 +51,7 @@ SUPPORTED_CLAUDE_MODELS = [
 
 
 def build_models_list_response(
-    settings: Settings, provider_registry: ProviderRegistry | None
+    settings: Settings, provider_runtime: ProviderRuntime | None
 ) -> ModelsListResponse:
     """Return configured, cached, and compatibility model ids."""
     models: list[ModelResponse] = []
@@ -59,8 +59,8 @@ def build_models_list_response(
 
     for ref in settings.configured_chat_model_refs():
         supports_thinking = None
-        if provider_registry is not None:
-            supports_thinking = provider_registry.cached_model_supports_thinking(
+        if provider_runtime is not None:
+            supports_thinking = provider_runtime.cached_model_supports_thinking(
                 ref.provider_id, ref.model_id
             )
         _append_provider_model_variants(
@@ -70,8 +70,8 @@ def build_models_list_response(
             supports_thinking=supports_thinking,
         )
 
-    if provider_registry is not None:
-        for model_info in provider_registry.cached_prefixed_model_infos():
+    if provider_runtime is not None:
+        for model_info in provider_runtime.cached_prefixed_model_infos():
             _append_provider_model_variants(
                 models,
                 seen,

api/routes.py

@@ -6,7 +6,6 @@ from loguru import logger
 from config.settings import Settings
 from core.anthropic import get_token_count
 from core.trace import trace_event
-from providers.registry import ProviderRegistry
 
 from . import dependencies
 from .dependencies import get_settings, require_api_key
@@ -21,7 +20,7 @@ router = APIRouter()
 
 def _provider_getter(request: Request, settings: Settings):
     return lambda provider_type: dependencies.resolve_provider(
-        provider_type, app=request.app, settings=settings
+        provider_type, app=request.app
     )
 
 
@@ -149,9 +148,8 @@ async def list_models(
 ):
     """List the model ids this proxy advertises to Claude-compatible clients."""
     trace_event(stage="ingress", event="api.models.list", source="api")
-    registry = getattr(request.app.state, "provider_registry", None)
-    provider_registry = registry if isinstance(registry, ProviderRegistry) else None
-    return build_models_list_response(settings, provider_registry)
+    provider_runtime = dependencies.maybe_provider_runtime(request.app)
+    return build_models_list_response(settings, provider_runtime)
 
 
 @router.post("/stop")

api/runtime.py

@@ -14,7 +14,7 @@ from loguru import logger
 from api.admin_urls import local_admin_url
 from config.settings import Settings, get_settings
 from providers.exceptions import ServiceUnavailableError
-from providers.registry import ProviderRegistry
+from providers.runtime import ProviderRuntime
 
 if TYPE_CHECKING:
     from cli.managed import ManagedClaudeSessionManager
@@ -87,7 +87,7 @@ class AppRuntime:
 
     app: FastAPI
     settings: Settings
-    _provider_registry: ProviderRegistry | None = field(default=None, init=False)
+    _provider_runtime: ProviderRuntime | None = field(default=None, init=False)
     messaging_runtime: MessagingRuntime | None = None
     messaging_workflow: MessagingWorkflow | None = None
     cli_manager: ManagedClaudeSessionManager | None = None
@@ -103,12 +103,12 @@ class AppRuntime:
     async def startup(self) -> None:
         logger.info("Starting Claude Code Proxy...")
         admin_url = local_admin_url(self.settings)
-        self._provider_registry = ProviderRegistry()
-        self.app.state.provider_registry = self._provider_registry
+        self._provider_runtime = ProviderRuntime(self.settings)
+        self.app.state.provider_runtime = self._provider_runtime
         try:
             warn_if_process_auth_token(self.settings)
             await self._validate_configured_models_best_effort()
-            self._provider_registry.start_model_list_refresh(self.settings)
+            self._provider_runtime.start_model_list_refresh()
             await self._start_messaging_if_configured()
             self._publish_state()
             logging.getLogger("uvicorn.error").info(
@@ -117,18 +117,18 @@ class AppRuntime:
         except Exception as exc:
             log_startup_failure(self.settings, exc)
             await best_effort(
-                "provider_registry.cleanup",
-                self._provider_registry.cleanup(),
+                "provider_runtime.cleanup",
+                self._provider_runtime.cleanup(),
                 log_verbose_errors=self.settings.log_api_error_tracebacks,
             )
             raise
 
     async def _validate_configured_models_best_effort(self) -> None:
         """Warm validation status without blocking first-run/admin access."""
-        if self._provider_registry is None:
+        if self._provider_runtime is None:
             return
         try:
-            await self._provider_registry.validate_configured_models(self.settings)
+            await self._provider_runtime.validate_configured_models()
         except ServiceUnavailableError as exc:
             self.app.state.startup_validation_error = exc.message
             logger.warning(
@@ -165,10 +165,10 @@ class AppRuntime:
                 self.cli_manager.stop_all(),
                 log_verbose_errors=verbose,
             )
-        if self._provider_registry is not None:
+        if self._provider_runtime is not None:
             await best_effort(
-                "provider_registry.cleanup",
-                self._provider_registry.cleanup(),
+                "provider_runtime.cleanup",
+                self._provider_runtime.cleanup(),
                 log_verbose_errors=verbose,
             )
         await self._shutdown_limiter()

config/provider_catalog.py

@@ -1,6 +1,6 @@
 """Neutral provider catalog: IDs, credentials, defaults, proxy and capability metadata.
 
-Adapter factories live in :mod:`providers.registry`; this module stays free of
+Adapter factories live in :mod:`providers.runtime.factory`; this module stays free of
 provider implementation imports (see contract tests).
 """
 

providers/registry.py

@@ -1,527 +0,0 @@
-"""Provider descriptors, factory, and runtime registry."""
-
-from __future__ import annotations
-
-import asyncio
-from collections import defaultdict
-from collections.abc import Callable, Iterable, MutableMapping
-from contextlib import suppress
-
-import httpx
-from loguru import logger
-
-from config.provider_catalog import (
-    PROVIDER_CATALOG,
-    SUPPORTED_PROVIDER_IDS,
-    ProviderDescriptor,
-)
-from config.settings import ConfiguredChatModelRef, Settings
-from providers.base import BaseProvider, ProviderConfig
-from providers.exceptions import (
-    AuthenticationError,
-    ModelListResponseError,
-    ProviderError,
-    ServiceUnavailableError,
-    UnknownProviderTypeError,
-)
-from providers.model_listing import ProviderModelInfo, model_infos_from_ids
-
-ProviderFactory = Callable[[ProviderConfig, Settings], BaseProvider]
-
-# Backwards-compatible name for the catalog (single source: ``config.provider_catalog``).
-PROVIDER_DESCRIPTORS: dict[str, ProviderDescriptor] = PROVIDER_CATALOG
-
-
-def _create_nvidia_nim(config: ProviderConfig, settings: Settings) -> BaseProvider:
-    from providers.nvidia_nim import NvidiaNimProvider
-
-    return NvidiaNimProvider(config, nim_settings=settings.nim)
-
-
-def _create_open_router(config: ProviderConfig, _settings: Settings) -> BaseProvider:
-    from providers.open_router import OpenRouterProvider
-
-    return OpenRouterProvider(config)
-
-
-def _create_mistral(config: ProviderConfig, _settings: Settings) -> BaseProvider:
-    from providers.mistral import MistralProvider
-
-    return MistralProvider(config)
-
-
-def _create_mistral_codestral(
-    config: ProviderConfig, _settings: Settings
-) -> BaseProvider:
-    from providers.codestral import CodestralProvider
-
-    return CodestralProvider(config)
-
-
-def _create_deepseek(config: ProviderConfig, _settings: Settings) -> BaseProvider:
-    from providers.deepseek import DeepSeekProvider
-
-    return DeepSeekProvider(config)
-
-
-def _create_lmstudio(config: ProviderConfig, _settings: Settings) -> BaseProvider:
-    from providers.lmstudio import LMStudioProvider
-
-    return LMStudioProvider(config)
-
-
-def _create_llamacpp(config: ProviderConfig, _settings: Settings) -> BaseProvider:
-    from providers.llamacpp import LlamaCppProvider
-
-    return LlamaCppProvider(config)
-
-
-def _create_ollama(config: ProviderConfig, _settings: Settings) -> BaseProvider:
-    from providers.ollama import OllamaProvider
-
-    return OllamaProvider(config)
-
-
-def _create_kimi(config: ProviderConfig, _settings: Settings) -> BaseProvider:
-    from providers.kimi import KimiProvider
-
-    return KimiProvider(config)
-
-
-def _create_wafer(config: ProviderConfig, _settings: Settings) -> BaseProvider:
-    from providers.wafer import WaferProvider
-
-    return WaferProvider(config)
-
-
-def _create_opencode(config: ProviderConfig, _settings: Settings) -> BaseProvider:
-    from providers.opencode import OpenCodeProvider
-
-    return OpenCodeProvider(config)
-
-
-def _create_opencode_go(config: ProviderConfig, _settings: Settings) -> BaseProvider:
-    from providers.opencode import OpenCodeProvider
-
-    return OpenCodeProvider(config, provider_name="OPENCODE_GO")
-
-
-def _create_zai(config: ProviderConfig, _settings: Settings) -> BaseProvider:
-    from providers.zai import ZaiProvider
-
-    return ZaiProvider(config)
-
-
-def _create_fireworks(config: ProviderConfig, _settings: Settings) -> BaseProvider:
-    from providers.fireworks import FireworksProvider
-
-    return FireworksProvider(config)
-
-
-def _create_gemini(config: ProviderConfig, _settings: Settings) -> BaseProvider:
-    from providers.gemini import GeminiProvider
-
-    return GeminiProvider(config)
-
-
-def _create_groq(config: ProviderConfig, _settings: Settings) -> BaseProvider:
-    from providers.groq import GroqProvider
-
-    return GroqProvider(config)
-
-
-def _create_cerebras(config: ProviderConfig, _settings: Settings) -> BaseProvider:
-    from providers.cerebras import CerebrasProvider
-
-    return CerebrasProvider(config)
-
-
-PROVIDER_FACTORIES: dict[str, ProviderFactory] = {
-    "nvidia_nim": _create_nvidia_nim,
-    "open_router": _create_open_router,
-    "gemini": _create_gemini,
-    "deepseek": _create_deepseek,
-    "mistral": _create_mistral,
-    "mistral_codestral": _create_mistral_codestral,
-    "opencode": _create_opencode,
-    "opencode_go": _create_opencode_go,
-    "wafer": _create_wafer,
-    "kimi": _create_kimi,
-    "cerebras": _create_cerebras,
-    "groq": _create_groq,
-    "fireworks": _create_fireworks,
-    "zai": _create_zai,
-    "lmstudio": _create_lmstudio,
-    "llamacpp": _create_llamacpp,
-    "ollama": _create_ollama,
-}
-
-if set(PROVIDER_DESCRIPTORS) != set(SUPPORTED_PROVIDER_IDS) or set(
-    PROVIDER_FACTORIES
-) != set(SUPPORTED_PROVIDER_IDS):
-    raise AssertionError(
-        "PROVIDER_DESCRIPTORS, PROVIDER_FACTORIES, and SUPPORTED_PROVIDER_IDS are out of sync: "
-        f"descriptors={set(PROVIDER_DESCRIPTORS)!r} factories={set(PROVIDER_FACTORIES)!r} "
-        f"ids={set(SUPPORTED_PROVIDER_IDS)!r}"
-    )
-
-
-def _string_attr(settings: Settings, attr_name: str | None, default: str = "") -> str:
-    if attr_name is None:
-        return default
-    value = getattr(settings, attr_name, default)
-    return value if isinstance(value, str) else default
-
-
-def _credential_for(descriptor: ProviderDescriptor, settings: Settings) -> str:
-    if descriptor.static_credential is not None:
-        return descriptor.static_credential
-    if descriptor.credential_attr:
-        return _string_attr(settings, descriptor.credential_attr)
-    return ""
-
-
-def _require_credential(descriptor: ProviderDescriptor, credential: str) -> None:
-    if descriptor.credential_env is None:
-        return
-    if credential and credential.strip():
-        return
-    message = f"{descriptor.credential_env} is not set. Add it to your .env file."
-    if descriptor.credential_url:
-        message = f"{message} Get a key at {descriptor.credential_url}"
-    raise AuthenticationError(message)
-
-
-def build_provider_config(
-    descriptor: ProviderDescriptor, settings: Settings
-) -> ProviderConfig:
-    credential = _credential_for(descriptor, settings)
-    _require_credential(descriptor, credential)
-    base_url = _string_attr(
-        settings, descriptor.base_url_attr, descriptor.default_base_url or ""
-    )
-    proxy = _string_attr(settings, descriptor.proxy_attr)
-    return ProviderConfig(
-        api_key=credential,
-        base_url=base_url or descriptor.default_base_url,
-        rate_limit=settings.provider_rate_limit,
-        rate_window=settings.provider_rate_window,
-        max_concurrency=settings.provider_max_concurrency,
-        http_read_timeout=settings.http_read_timeout,
-        http_write_timeout=settings.http_write_timeout,
-        http_connect_timeout=settings.http_connect_timeout,
-        enable_thinking=settings.enable_model_thinking,
-        proxy=proxy,
-        log_raw_sse_events=settings.log_raw_sse_events,
-        log_api_error_tracebacks=settings.log_api_error_tracebacks,
-    )
-
-
-def create_provider(provider_id: str, settings: Settings) -> BaseProvider:
-    descriptor = PROVIDER_DESCRIPTORS.get(provider_id)
-    if descriptor is None:
-        supported = "', '".join(PROVIDER_DESCRIPTORS)
-        raise UnknownProviderTypeError(
-            f"Unknown provider_type: '{provider_id}'. Supported: '{supported}'"
-        )
-
-    config = build_provider_config(descriptor, settings)
-    factory = PROVIDER_FACTORIES.get(provider_id)
-    if factory is None:
-        raise AssertionError(f"Unhandled provider descriptor: {provider_id}")
-    return factory(config, settings)
-
-
-def _format_provider_query_failures(
-    refs: list[ConfiguredChatModelRef],
-    exc: BaseException,
-    settings: Settings,
-) -> list[str]:
-    reason = _provider_query_failure_reason(exc, settings)
-    return [_format_model_validation_failure(ref, reason) for ref in refs]
-
-
-def _format_missing_model_failure(ref: ConfiguredChatModelRef) -> str:
-    return _format_model_validation_failure(ref, "missing model")
-
-
-def _format_model_validation_failure(ref: ConfiguredChatModelRef, problem: str) -> str:
-    return (
-        f"sources={','.join(ref.sources)} provider={ref.provider_id} "
-        f"model={ref.model_id} problem={problem}"
-    )
-
-
-def _provider_query_failure_reason(
-    exc: BaseException,
-    settings: Settings,
-) -> str:
-    if isinstance(exc, ModelListResponseError):
-        return f"malformed model-list response: {exc.message}"
-    if isinstance(exc, httpx.HTTPStatusError):
-        return f"query failure: HTTP {exc.response.status_code}"
-    if isinstance(exc, AuthenticationError):
-        return f"query failure: {exc.message}"
-    if isinstance(exc, ProviderError) and settings.log_api_error_tracebacks:
-        return f"query failure: {exc.message}"
-    return f"query failure: {type(exc).__name__}"
-
-
-def _referenced_provider_ids(settings: Settings) -> frozenset[str]:
-    return frozenset(ref.provider_id for ref in settings.configured_chat_model_refs())
-
-
-def _model_list_provider_ids_for_settings(settings: Settings) -> tuple[str, ...]:
-    """Return providers worth discovering for this process configuration."""
-    referenced_provider_ids = _referenced_provider_ids(settings)
-    provider_ids: list[str] = []
-    for provider_id, descriptor in PROVIDER_DESCRIPTORS.items():
-        if descriptor.static_credential is not None:
-            if provider_id in referenced_provider_ids:
-                provider_ids.append(provider_id)
-            continue
-        if (
-            descriptor.credential_env is not None
-            and _credential_for(descriptor, settings).strip()
-        ):
-            provider_ids.append(provider_id)
-    return tuple(provider_ids)
-
-
-def _log_model_discovery_failure(
-    provider_id: str, exc: BaseException, settings: Settings
-) -> None:
-    logger.warning(
-        "Provider model discovery skipped: provider={} reason={}",
-        provider_id,
-        _provider_query_failure_reason(exc, settings),
-    )
-
-
-class ProviderRegistry:
-    """Cache and clean up provider instances by provider id."""
-
-    def __init__(self, providers: MutableMapping[str, BaseProvider] | None = None):
-        self._providers = providers if providers is not None else {}
-        self._model_ids_by_provider: dict[str, frozenset[str]] = {}
-        self._model_infos_by_provider: dict[str, dict[str, ProviderModelInfo]] = {}
-        self._model_list_refresh_task: asyncio.Task[None] | None = None
-
-    def is_cached(self, provider_id: str) -> bool:
-        """Return whether a provider for this id is already in the cache."""
-        return provider_id in self._providers
-
-    def get(self, provider_id: str, settings: Settings) -> BaseProvider:
-        if provider_id not in self._providers:
-            self._providers[provider_id] = create_provider(provider_id, settings)
-        return self._providers[provider_id]
-
-    def cache_model_ids(self, provider_id: str, model_ids: Iterable[str]) -> None:
-        """Store a provider model-list result for later instant API responses."""
-        self.cache_model_infos(provider_id, model_infos_from_ids(model_ids))
-
-    def cache_model_infos(
-        self, provider_id: str, model_infos: Iterable[ProviderModelInfo]
-    ) -> None:
-        """Store provider model metadata for later instant API responses."""
-        clean_infos = {
-            info.model_id: info for info in model_infos if info.model_id.strip()
-        }
-        self._model_infos_by_provider[provider_id] = clean_infos
-        self._model_ids_by_provider[provider_id] = frozenset(clean_infos)
-
-    def cached_model_ids(self) -> dict[str, frozenset[str]]:
-        """Return a copy of cached raw provider model ids."""
-        return dict(self._model_ids_by_provider)
-
-    def cached_model_supports_thinking(
-        self, provider_id: str, model_id: str
-    ) -> bool | None:
-        """Return cached thinking support when a provider exposes it."""
-        info = self._model_infos_by_provider.get(provider_id, {}).get(model_id)
-        if info is None:
-            return None
-        return info.supports_thinking
-
-    def cached_prefixed_model_refs(self) -> tuple[str, ...]:
-        """Return cached provider models in user-selectable ``provider/model`` form."""
-        return tuple(info.model_id for info in self.cached_prefixed_model_infos())
-
-    def cached_prefixed_model_infos(self) -> tuple[ProviderModelInfo, ...]:
-        """Return cached provider models with user-selectable prefixed ids."""
-        infos: list[ProviderModelInfo] = []
-        for provider_id in SUPPORTED_PROVIDER_IDS:
-            provider_infos = self._model_infos_by_provider.get(provider_id, {})
-            infos.extend(
-                ProviderModelInfo(
-                    model_id=f"{provider_id}/{info.model_id}",
-                    supports_thinking=info.supports_thinking,
-                )
-                for info in sorted(
-                    provider_infos.values(), key=lambda item: item.model_id
-                )
-            )
-        return tuple(infos)
-
-    async def refresh_model_list_cache(
-        self, settings: Settings, *, only_missing: bool = False
-    ) -> None:
-        """Best-effort refresh of model lists for providers usable in this process."""
-        provider_ids = _model_list_provider_ids_for_settings(settings)
-        if only_missing:
-            provider_ids = tuple(
-                provider_id
-                for provider_id in provider_ids
-                if provider_id not in self._model_ids_by_provider
-            )
-        await self._refresh_model_ids(settings, provider_ids)
-
-    def start_model_list_refresh(self, settings: Settings) -> None:
-        """Start a non-blocking cache warmup for missing eligible provider lists."""
-        if (
-            self._model_list_refresh_task is not None
-            and not self._model_list_refresh_task.done()
-        ):
-            return
-
-        provider_ids = tuple(
-            provider_id
-            for provider_id in _model_list_provider_ids_for_settings(settings)
-            if provider_id not in self._model_ids_by_provider
-        )
-        if not provider_ids:
-            logger.info(
-                "Provider model discovery cache already warm: providers={}",
-                len(self._model_ids_by_provider),
-            )
-            return
-
-        self._model_list_refresh_task = asyncio.create_task(
-            self._run_model_list_refresh(settings, provider_ids)
-        )
-
-    async def _run_model_list_refresh(
-        self, settings: Settings, provider_ids: tuple[str, ...]
-    ) -> None:
-        try:
-            await self._refresh_model_ids(settings, provider_ids)
-        except asyncio.CancelledError:
-            raise
-        except Exception as exc:
-            logger.warning(
-                "Provider model discovery task failed: exc_type={}",
-                type(exc).__name__,
-            )
-
-    async def _refresh_model_ids(
-        self, settings: Settings, provider_ids: tuple[str, ...]
-    ) -> None:
-        tasks: dict[str, asyncio.Task[frozenset[ProviderModelInfo]]] = {}
-        for provider_id in provider_ids:
-            try:
-                provider = self.get(provider_id, settings)
-            except Exception as exc:
-                _log_model_discovery_failure(provider_id, exc, settings)
-                continue
-            tasks[provider_id] = asyncio.create_task(provider.list_model_infos())
-
-        if not tasks:
-            return
-
-        results = await asyncio.gather(*tasks.values(), return_exceptions=True)
-        for (provider_id, _task), result in zip(tasks.items(), results, strict=True):
-            if isinstance(result, BaseException):
-                if isinstance(result, asyncio.CancelledError):
-                    raise result
-                _log_model_discovery_failure(provider_id, result, settings)
-                continue
-            self.cache_model_infos(provider_id, result)
-            logger.info(
-                "Provider model discovery cached: provider={} models={}",
-                provider_id,
-                len(result),
-            )
-
-    async def validate_configured_models(self, settings: Settings) -> None:
-        """Fail fast unless every configured chat model exists upstream."""
-        refs = settings.configured_chat_model_refs()
-        refs_by_provider: dict[str, list[ConfiguredChatModelRef]] = defaultdict(list)
-        for ref in refs:
-            refs_by_provider[ref.provider_id].append(ref)
-
-        failures: list[str] = []
-        tasks: dict[str, asyncio.Task[frozenset[ProviderModelInfo]]] = {}
-        for provider_id, provider_refs in refs_by_provider.items():
-            try:
-                provider = self.get(provider_id, settings)
-            except Exception as exc:
-                failures.extend(
-                    _format_provider_query_failures(provider_refs, exc, settings)
-                )
-                continue
-            tasks[provider_id] = asyncio.create_task(provider.list_model_infos())
-
-        if tasks:
-            results = await asyncio.gather(*tasks.values(), return_exceptions=True)
-            for (provider_id, _task), result in zip(
-                tasks.items(), results, strict=True
-            ):
-                provider_refs = refs_by_provider[provider_id]
-                if isinstance(result, BaseException):
-                    if isinstance(result, asyncio.CancelledError):
-                        raise result
-                    failures.extend(
-                        _format_provider_query_failures(provider_refs, result, settings)
-                    )
-                    continue
-                self.cache_model_infos(provider_id, result)
-                model_ids = self._model_ids_by_provider[provider_id]
-                failures.extend(
-                    _format_missing_model_failure(ref)
-                    for ref in provider_refs
-                    if ref.model_id not in model_ids
-                )
-
-        if failures:
-            message = "Configured model validation failed:\n" + "\n".join(
-                f"- {failure}" for failure in failures
-            )
-            raise ServiceUnavailableError(message)
-
-        logger.info(
-            "Configured provider models validated: models={} providers={}",
-            len(refs),
-            len(refs_by_provider),
-        )
-
-    async def cleanup(self) -> None:
-        """Call ``cleanup`` on every cached provider, then clear the cache.
-
-        Attempts all providers even if one fails. A single failure is re-raised
-        as-is; multiple failures are wrapped in :exc:`ExceptionGroup`.
-        """
-        if (
-            self._model_list_refresh_task is not None
-            and not self._model_list_refresh_task.done()
-        ):
-            self._model_list_refresh_task.cancel()
-            with suppress(asyncio.CancelledError):
-                await self._model_list_refresh_task
-
-        items = list(self._providers.items())
-        errors: list[Exception] = []
-        try:
-            for _pid, provider in items:
-                try:
-                    await provider.cleanup()
-                except Exception as e:
-                    errors.append(e)
-        finally:
-            self._providers.clear()
-            self._model_ids_by_provider.clear()
-            self._model_infos_by_provider.clear()
-        if len(errors) == 1:
-            raise errors[0]
-        if len(errors) > 1:
-            msg = "One or more provider cleanups failed"
-            raise ExceptionGroup(msg, errors)

providers/runtime/__init__.py

@@ -0,0 +1,12 @@
+"""App-scoped provider runtime facade."""
+
+from .config import build_provider_config
+from .factory import PROVIDER_FACTORIES, create_provider
+from .runtime import ProviderRuntime
+
+__all__ = [
+    "PROVIDER_FACTORIES",
+    "ProviderRuntime",
+    "build_provider_config",
+    "create_provider",
+]

providers/runtime/cache.py

@@ -0,0 +1,54 @@
+"""Provider instance cache and cleanup."""
+
+from __future__ import annotations
+
+from collections.abc import Callable, MutableMapping
+
+from config.settings import Settings
+from providers.base import BaseProvider
+
+from .factory import create_provider
+
+ProviderCreator = Callable[[str, Settings], BaseProvider]
+
+
+class ProviderCache:
+    """Cache provider instances for one settings snapshot."""
+
+    def __init__(
+        self,
+        settings: Settings,
+        providers: MutableMapping[str, BaseProvider] | None = None,
+        *,
+        creator: ProviderCreator = create_provider,
+    ) -> None:
+        self._settings = settings
+        self._providers = providers if providers is not None else {}
+        self._creator = creator
+
+    def is_cached(self, provider_id: str) -> bool:
+        """Return whether a provider for this id is already cached."""
+        return provider_id in self._providers
+
+    def get(self, provider_id: str) -> BaseProvider:
+        """Return an existing provider or create it lazily."""
+        if provider_id not in self._providers:
+            self._providers[provider_id] = self._creator(provider_id, self._settings)
+        return self._providers[provider_id]
+
+    async def cleanup(self) -> None:
+        """Clean up every cached provider, then clear the cache."""
+        items = list(self._providers.items())
+        errors: list[Exception] = []
+        try:
+            for _provider_id, provider in items:
+                try:
+                    await provider.cleanup()
+                except Exception as exc:
+                    errors.append(exc)
+        finally:
+            self._providers.clear()
+        if len(errors) == 1:
+            raise errors[0]
+        if len(errors) > 1:
+            raise ExceptionGroup("One or more provider cleanups failed", errors)

providers/runtime/config.py

@@ -0,0 +1,65 @@
+"""Provider configuration construction from neutral catalog metadata."""
+
+from __future__ import annotations
+
+from config.provider_catalog import ProviderDescriptor
+from config.settings import Settings
+from providers.base import ProviderConfig
+from providers.exceptions import AuthenticationError
+
+
+def string_setting(settings: Settings, attr_name: str | None, default: str = "") -> str:
+    """Return a string-valued settings attribute, ignoring non-string mocks."""
+    if attr_name is None:
+        return default
+    value = getattr(settings, attr_name, default)
+    return value if isinstance(value, str) else default
+
+
+def provider_credential(descriptor: ProviderDescriptor, settings: Settings) -> str:
+    """Return the configured credential for a provider descriptor."""
+    if descriptor.static_credential is not None:
+        return descriptor.static_credential
+    if descriptor.credential_attr:
+        return string_setting(settings, descriptor.credential_attr)
+    return ""
+
+
+def require_provider_credential(
+    descriptor: ProviderDescriptor, credential: str
+) -> None:
+    """Raise a user-facing configuration error when a required key is missing."""
+    if descriptor.credential_env is None:
+        return
+    if credential and credential.strip():
+        return
+    message = f"{descriptor.credential_env} is not set. Add it to your .env file."
+    if descriptor.credential_url:
+        message = f"{message} Get a key at {descriptor.credential_url}"
+    raise AuthenticationError(message)
+
+
+def build_provider_config(
+    descriptor: ProviderDescriptor, settings: Settings
+) -> ProviderConfig:
+    """Build shared provider configuration for one provider descriptor."""
+    credential = provider_credential(descriptor, settings)
+    require_provider_credential(descriptor, credential)
+    base_url = string_setting(
+        settings, descriptor.base_url_attr, descriptor.default_base_url or ""
+    )
+    proxy = string_setting(settings, descriptor.proxy_attr)
+    return ProviderConfig(
+        api_key=credential,
+        base_url=base_url or descriptor.default_base_url,
+        rate_limit=settings.provider_rate_limit,
+        rate_window=settings.provider_rate_window,
+        max_concurrency=settings.provider_max_concurrency,
+        http_read_timeout=settings.http_read_timeout,
+        http_write_timeout=settings.http_write_timeout,
+        http_connect_timeout=settings.http_connect_timeout,
+        enable_thinking=settings.enable_model_thinking,
+        proxy=proxy,
+        log_raw_sse_events=settings.log_raw_sse_events,
+        log_api_error_tracebacks=settings.log_api_error_tracebacks,
+    )

providers/runtime/discovery.py

@@ -0,0 +1,140 @@
+"""Provider model-list discovery and background refresh."""
+
+from __future__ import annotations
+
+import asyncio
+from collections.abc import Callable
+from contextlib import suppress
+
+from loguru import logger
+
+from config.provider_catalog import PROVIDER_CATALOG
+from config.settings import Settings
+from providers.base import BaseProvider
+from providers.model_listing import ProviderModelInfo
+
+from .config import provider_credential
+from .model_cache import ProviderModelCache
+from .validation import provider_query_failure_reason
+
+ProviderResolver = Callable[[str], BaseProvider]
+
+
+def referenced_provider_ids(settings: Settings) -> frozenset[str]:
+    """Return provider ids referenced by configured chat model refs."""
+    return frozenset(ref.provider_id for ref in settings.configured_chat_model_refs())
+
+
+def model_list_provider_ids_for_settings(settings: Settings) -> tuple[str, ...]:
+    """Return providers worth discovering for this process configuration."""
+    referenced_ids = referenced_provider_ids(settings)
+    provider_ids: list[str] = []
+    for provider_id, descriptor in PROVIDER_CATALOG.items():
+        if descriptor.static_credential is not None:
+            if provider_id in referenced_ids:
+                provider_ids.append(provider_id)
+            continue
+        if (
+            descriptor.credential_env is not None
+            and provider_credential(descriptor, settings).strip()
+        ):
+            provider_ids.append(provider_id)
+    return tuple(provider_ids)
+
+
+class ProviderModelDiscovery:
+    """Refresh provider model-list metadata for one provider runtime."""
+
+    def __init__(
+        self,
+        settings: Settings,
+        provider_resolver: ProviderResolver,
+        model_cache: ProviderModelCache,
+    ) -> None:
+        self._settings = settings
+        self._provider_resolver = provider_resolver
+        self._model_cache = model_cache
+        self._refresh_task: asyncio.Task[None] | None = None
+
+    async def refresh_model_list_cache(self, *, only_missing: bool = False) -> None:
+        """Best-effort refresh of model lists for usable providers."""
+        provider_ids = model_list_provider_ids_for_settings(self._settings)
+        if only_missing:
+            provider_ids = tuple(
+                provider_id
+                for provider_id in provider_ids
+                if not self._model_cache.has_provider(provider_id)
+            )
+        await self._refresh_model_infos(provider_ids)
+
+    def start_model_list_refresh(self) -> None:
+        """Start a non-blocking cache warmup for missing eligible provider lists."""
+        if self._refresh_task is not None and not self._refresh_task.done():
+            return
+
+        provider_ids = tuple(
+            provider_id
+            for provider_id in model_list_provider_ids_for_settings(self._settings)
+            if not self._model_cache.has_provider(provider_id)
+        )
+        if not provider_ids:
+            logger.info(
+                "Provider model discovery cache already warm: providers={}",
+                len(self._model_cache.cached_model_ids()),
+            )
+            return
+
+        self._refresh_task = asyncio.create_task(self._run_refresh(provider_ids))
+
+    async def cleanup(self) -> None:
+        """Cancel any background model-list refresh."""
+        if self._refresh_task is None or self._refresh_task.done():
+            return
+        self._refresh_task.cancel()
+        with suppress(asyncio.CancelledError):
+            await self._refresh_task
+
+    async def _run_refresh(self, provider_ids: tuple[str, ...]) -> None:
+        try:
+            await self._refresh_model_infos(provider_ids)
+        except asyncio.CancelledError:
+            raise
+        except Exception as exc:
+            logger.warning(
+                "Provider model discovery task failed: exc_type={}",
+                type(exc).__name__,
+            )
+
+    async def _refresh_model_infos(self, provider_ids: tuple[str, ...]) -> None:
+        tasks: dict[str, asyncio.Task[frozenset[ProviderModelInfo]]] = {}
+        for provider_id in provider_ids:
+            try:
+                provider = self._provider_resolver(provider_id)
+            except Exception as exc:
+                self._log_discovery_failure(provider_id, exc)
+                continue
+            tasks[provider_id] = asyncio.create_task(provider.list_model_infos())
+
+        if not tasks:
+            return
+
+        results = await asyncio.gather(*tasks.values(), return_exceptions=True)
+        for (provider_id, _task), result in zip(tasks.items(), results, strict=True):
+            if isinstance(result, BaseException):
+                if isinstance(result, asyncio.CancelledError):
+                    raise result
+                self._log_discovery_failure(provider_id, result)
+                continue
+            self._model_cache.cache_model_infos(provider_id, result)
+            logger.info(
+                "Provider model discovery cached: provider={} models={}",
+                provider_id,
+                len(result),
+            )
+
+    def _log_discovery_failure(self, provider_id: str, exc: BaseException) -> None:
+        logger.warning(
+            "Provider model discovery skipped: provider={} reason={}",
+            provider_id,
+            provider_query_failure_reason(exc, self._settings),
+        )

providers/runtime/factory.py

@@ -0,0 +1,166 @@
+"""Provider factory wiring and lazy adapter construction."""
+
+from __future__ import annotations
+
+from collections.abc import Callable
+
+from config.provider_catalog import (
+    PROVIDER_CATALOG,
+    SUPPORTED_PROVIDER_IDS,
+)
+from config.settings import Settings
+from providers.base import BaseProvider, ProviderConfig
+from providers.exceptions import UnknownProviderTypeError
+
+from .config import build_provider_config
+
+ProviderFactory = Callable[[ProviderConfig, Settings], BaseProvider]
+
+
+def _create_nvidia_nim(config: ProviderConfig, settings: Settings) -> BaseProvider:
+    from providers.nvidia_nim import NvidiaNimProvider
+
+    return NvidiaNimProvider(config, nim_settings=settings.nim)
+
+
+def _create_open_router(config: ProviderConfig, _settings: Settings) -> BaseProvider:
+    from providers.open_router import OpenRouterProvider
+
+    return OpenRouterProvider(config)
+
+
+def _create_mistral(config: ProviderConfig, _settings: Settings) -> BaseProvider:
+    from providers.mistral import MistralProvider
+
+    return MistralProvider(config)
+
+
+def _create_mistral_codestral(
+    config: ProviderConfig, _settings: Settings
+) -> BaseProvider:
+    from providers.codestral import CodestralProvider
+
+    return CodestralProvider(config)
+
+
+def _create_deepseek(config: ProviderConfig, _settings: Settings) -> BaseProvider:
+    from providers.deepseek import DeepSeekProvider
+
+    return DeepSeekProvider(config)
+
+
+def _create_lmstudio(config: ProviderConfig, _settings: Settings) -> BaseProvider:
+    from providers.lmstudio import LMStudioProvider
+
+    return LMStudioProvider(config)
+
+
+def _create_llamacpp(config: ProviderConfig, _settings: Settings) -> BaseProvider:
+    from providers.llamacpp import LlamaCppProvider
+
+    return LlamaCppProvider(config)
+
+
+def _create_ollama(config: ProviderConfig, _settings: Settings) -> BaseProvider:
+    from providers.ollama import OllamaProvider
+
+    return OllamaProvider(config)
+
+
+def _create_kimi(config: ProviderConfig, _settings: Settings) -> BaseProvider:
+    from providers.kimi import KimiProvider
+
+    return KimiProvider(config)
+
+
+def _create_wafer(config: ProviderConfig, _settings: Settings) -> BaseProvider:
+    from providers.wafer import WaferProvider
+
+    return WaferProvider(config)
+
+
+def _create_opencode(config: ProviderConfig, _settings: Settings) -> BaseProvider:
+    from providers.opencode import OpenCodeProvider
+
+    return OpenCodeProvider(config)
+
+
+def _create_opencode_go(config: ProviderConfig, _settings: Settings) -> BaseProvider:
+    from providers.opencode import OpenCodeProvider
+
+    return OpenCodeProvider(config, provider_name="OPENCODE_GO")
+
+
+def _create_zai(config: ProviderConfig, _settings: Settings) -> BaseProvider:
+    from providers.zai import ZaiProvider
+
+    return ZaiProvider(config)
+
+
+def _create_fireworks(config: ProviderConfig, _settings: Settings) -> BaseProvider:
+    from providers.fireworks import FireworksProvider
+
+    return FireworksProvider(config)
+
+
+def _create_gemini(config: ProviderConfig, _settings: Settings) -> BaseProvider:
+    from providers.gemini import GeminiProvider
+
+    return GeminiProvider(config)
+
+
+def _create_groq(config: ProviderConfig, _settings: Settings) -> BaseProvider:
+    from providers.groq import GroqProvider
+
+    return GroqProvider(config)
+
+
+def _create_cerebras(config: ProviderConfig, _settings: Settings) -> BaseProvider:
+    from providers.cerebras import CerebrasProvider
+
+    return CerebrasProvider(config)
+
+
+PROVIDER_FACTORIES: dict[str, ProviderFactory] = {
+    "nvidia_nim": _create_nvidia_nim,
+    "open_router": _create_open_router,
+    "gemini": _create_gemini,
+    "deepseek": _create_deepseek,
+    "mistral": _create_mistral,
+    "mistral_codestral": _create_mistral_codestral,
+    "opencode": _create_opencode,
+    "opencode_go": _create_opencode_go,
+    "wafer": _create_wafer,
+    "kimi": _create_kimi,
+    "cerebras": _create_cerebras,
+    "groq": _create_groq,
+    "fireworks": _create_fireworks,
+    "zai": _create_zai,
+    "lmstudio": _create_lmstudio,
+    "llamacpp": _create_llamacpp,
+    "ollama": _create_ollama,
+}
+
+if set(PROVIDER_CATALOG) != set(SUPPORTED_PROVIDER_IDS) or set(
+    PROVIDER_FACTORIES
+) != set(SUPPORTED_PROVIDER_IDS):
+    raise AssertionError(
+        "PROVIDER_CATALOG, PROVIDER_FACTORIES, and SUPPORTED_PROVIDER_IDS are out of sync: "
+        f"catalog={set(PROVIDER_CATALOG)!r} factories={set(PROVIDER_FACTORIES)!r} "
+        f"ids={set(SUPPORTED_PROVIDER_IDS)!r}"
+    )
+
+
+def create_provider(provider_id: str, settings: Settings) -> BaseProvider:
+    """Create a provider instance for a supported provider id."""
+    descriptor = PROVIDER_CATALOG.get(provider_id)
+    if descriptor is None:
+        supported = "', '".join(PROVIDER_CATALOG)
+        raise UnknownProviderTypeError(
+            f"Unknown provider_type: '{provider_id}'. Supported: '{supported}'"
+        )
+
+    factory = PROVIDER_FACTORIES.get(provider_id)
+    if factory is None:
+        raise AssertionError(f"Unhandled provider descriptor: {provider_id}")
+    return factory(build_provider_config(descriptor, settings), settings)

providers/runtime/model_cache.py

@@ -0,0 +1,72 @@
+"""Provider model-list metadata cache."""
+
+from __future__ import annotations
+
+from collections.abc import Iterable
+
+from config.provider_catalog import SUPPORTED_PROVIDER_IDS
+from providers.model_listing import ProviderModelInfo, model_infos_from_ids
+
+
+class ProviderModelCache:
+    """Store provider model metadata for instant model-list responses."""
+
+    def __init__(self) -> None:
+        self._model_infos_by_provider: dict[str, dict[str, ProviderModelInfo]] = {}
+
+    def cache_model_ids(self, provider_id: str, model_ids: Iterable[str]) -> None:
+        """Store raw provider model ids with unknown capability metadata."""
+        self.cache_model_infos(provider_id, model_infos_from_ids(model_ids))
+
+    def cache_model_infos(
+        self, provider_id: str, model_infos: Iterable[ProviderModelInfo]
+    ) -> None:
+        """Store provider model metadata by raw provider model id."""
+        clean_infos = {
+            info.model_id: info for info in model_infos if info.model_id.strip()
+        }
+        self._model_infos_by_provider[provider_id] = clean_infos
+
+    def cached_model_ids(self) -> dict[str, frozenset[str]]:
+        """Return cached raw provider model ids by provider."""
+        return {
+            provider_id: frozenset(infos)
+            for provider_id, infos in self._model_infos_by_provider.items()
+        }
+
+    def has_provider(self, provider_id: str) -> bool:
+        """Return whether this provider has any cached model-list result."""
+        return provider_id in self._model_infos_by_provider
+
+    def cached_model_supports_thinking(
+        self, provider_id: str, model_id: str
+    ) -> bool | None:
+        """Return cached thinking support when a provider exposes it."""
+        info = self._model_infos_by_provider.get(provider_id, {}).get(model_id)
+        if info is None:
+            return None
+        return info.supports_thinking
+
+    def cached_prefixed_model_refs(self) -> tuple[str, ...]:
+        """Return cached provider models in user-selectable ``provider/model`` form."""
+        return tuple(info.model_id for info in self.cached_prefixed_model_infos())
+
+    def cached_prefixed_model_infos(self) -> tuple[ProviderModelInfo, ...]:
+        """Return cached provider models with user-selectable prefixed ids."""
+        infos: list[ProviderModelInfo] = []
+        for provider_id in SUPPORTED_PROVIDER_IDS:
+            provider_infos = self._model_infos_by_provider.get(provider_id, {})
+            infos.extend(
+                ProviderModelInfo(
+                    model_id=f"{provider_id}/{info.model_id}",
+                    supports_thinking=info.supports_thinking,
+                )
+                for info in sorted(
+                    provider_infos.values(), key=lambda item: item.model_id
+                )
+            )
+        return tuple(infos)
+
+    def clear(self) -> None:
+        """Clear all cached model metadata."""
+        self._model_infos_by_provider.clear()

providers/runtime/runtime.py

@@ -0,0 +1,93 @@
+"""App-scoped provider runtime orchestration."""
+
+from __future__ import annotations
+
+from collections.abc import Iterable, MutableMapping
+
+from config.settings import Settings
+from providers.base import BaseProvider
+from providers.model_listing import ProviderModelInfo
+
+from .cache import ProviderCache
+from .discovery import ProviderModelDiscovery
+from .model_cache import ProviderModelCache
+from .validation import ConfiguredModelValidator
+
+
+class ProviderRuntime:
+    """Own provider instances, model discovery, validation, and cleanup."""
+
+    def __init__(
+        self,
+        settings: Settings,
+        providers: MutableMapping[str, BaseProvider] | None = None,
+    ) -> None:
+        self.settings = settings
+        self._provider_cache = ProviderCache(settings, providers)
+        self._model_cache = ProviderModelCache()
+        self._discovery = ProviderModelDiscovery(
+            settings,
+            self.resolve_provider,
+            self._model_cache,
+        )
+        self._validator = ConfiguredModelValidator(
+            settings,
+            self.resolve_provider,
+            self._model_cache,
+        )
+
+    def is_cached(self, provider_id: str) -> bool:
+        """Return whether a provider for this id is already cached."""
+        return self._provider_cache.is_cached(provider_id)
+
+    def resolve_provider(self, provider_id: str) -> BaseProvider:
+        """Return an existing provider or create it lazily."""
+        return self._provider_cache.get(provider_id)
+
+    def cache_model_ids(self, provider_id: str, model_ids: Iterable[str]) -> None:
+        """Store raw provider model ids for later instant API responses."""
+        self._model_cache.cache_model_ids(provider_id, model_ids)
+
+    def cache_model_infos(
+        self, provider_id: str, model_infos: Iterable[ProviderModelInfo]
+    ) -> None:
+        """Store provider model metadata for later instant API responses."""
+        self._model_cache.cache_model_infos(provider_id, model_infos)
+
+    def cached_model_ids(self) -> dict[str, frozenset[str]]:
+        """Return cached raw provider model ids by provider."""
+        return self._model_cache.cached_model_ids()
+
+    def cached_model_supports_thinking(
+        self, provider_id: str, model_id: str
+    ) -> bool | None:
+        """Return cached thinking support when a provider exposes it."""
+        return self._model_cache.cached_model_supports_thinking(provider_id, model_id)
+
+    def cached_prefixed_model_refs(self) -> tuple[str, ...]:
+        """Return cached provider models in user-selectable ``provider/model`` form."""
+        return self._model_cache.cached_prefixed_model_refs()
+
+    def cached_prefixed_model_infos(self) -> tuple[ProviderModelInfo, ...]:
+        """Return cached provider models with user-selectable prefixed ids."""
+        return self._model_cache.cached_prefixed_model_infos()
+
+    async def refresh_model_list_cache(self, *, only_missing: bool = False) -> None:
+        """Best-effort refresh of model lists for usable providers."""
+        await self._discovery.refresh_model_list_cache(only_missing=only_missing)
+
+    def start_model_list_refresh(self) -> None:
+        """Start a non-blocking cache warmup for missing eligible provider lists."""
+        self._discovery.start_model_list_refresh()
+
+    async def validate_configured_models(self) -> None:
+        """Fail unless every configured chat model exists upstream."""
+        await self._validator.validate_configured_models()
+
+    async def cleanup(self) -> None:
+        """Cancel discovery, clean provider instances, and clear model metadata."""
+        try:
+            await self._discovery.cleanup()
+            await self._provider_cache.cleanup()
+        finally:
+            self._model_cache.clear()

providers/runtime/validation.py

@@ -0,0 +1,123 @@
+"""Configured provider model validation."""
+
+from __future__ import annotations
+
+import asyncio
+from collections import defaultdict
+from collections.abc import Callable
+
+import httpx
+from loguru import logger
+
+from config.settings import ConfiguredChatModelRef, Settings
+from providers.base import BaseProvider
+from providers.exceptions import (
+    AuthenticationError,
+    ModelListResponseError,
+    ProviderError,
+    ServiceUnavailableError,
+)
+from providers.model_listing import ProviderModelInfo
+
+from .model_cache import ProviderModelCache
+
+ProviderResolver = Callable[[str], BaseProvider]
+
+
+def provider_query_failure_reason(exc: BaseException, settings: Settings) -> str:
+    """Return a concise model-list query failure reason for user-facing logs."""
+    if isinstance(exc, ModelListResponseError):
+        return f"malformed model-list response: {exc.message}"
+    if isinstance(exc, httpx.HTTPStatusError):
+        return f"query failure: HTTP {exc.response.status_code}"
+    if isinstance(exc, AuthenticationError):
+        return f"query failure: {exc.message}"
+    if isinstance(exc, ProviderError) and settings.log_api_error_tracebacks:
+        return f"query failure: {exc.message}"
+    return f"query failure: {type(exc).__name__}"
+
+
+class ConfiguredModelValidator:
+    """Validate configured provider/model refs against upstream model lists."""
+
+    def __init__(
+        self,
+        settings: Settings,
+        provider_resolver: ProviderResolver,
+        model_cache: ProviderModelCache,
+    ) -> None:
+        self._settings = settings
+        self._provider_resolver = provider_resolver
+        self._model_cache = model_cache
+
+    async def validate_configured_models(self) -> None:
+        """Fail unless every configured chat model exists upstream."""
+        refs = self._settings.configured_chat_model_refs()
+        refs_by_provider: dict[str, list[ConfiguredChatModelRef]] = defaultdict(list)
+        for ref in refs:
+            refs_by_provider[ref.provider_id].append(ref)
+
+        failures: list[str] = []
+        tasks: dict[str, asyncio.Task[frozenset[ProviderModelInfo]]] = {}
+        for provider_id, provider_refs in refs_by_provider.items():
+            try:
+                provider = self._provider_resolver(provider_id)
+            except Exception as exc:
+                failures.extend(
+                    self._format_provider_query_failures(provider_refs, exc)
+                )
+                continue
+            tasks[provider_id] = asyncio.create_task(provider.list_model_infos())
+
+        if tasks:
+            results = await asyncio.gather(*tasks.values(), return_exceptions=True)
+            for (provider_id, _task), result in zip(
+                tasks.items(), results, strict=True
+            ):
+                provider_refs = refs_by_provider[provider_id]
+                if isinstance(result, BaseException):
+                    if isinstance(result, asyncio.CancelledError):
+                        raise result
+                    failures.extend(
+                        self._format_provider_query_failures(provider_refs, result)
+                    )
+                    continue
+                self._model_cache.cache_model_infos(provider_id, result)
+                model_ids = self._model_cache.cached_model_ids()[provider_id]
+                failures.extend(
+                    self._format_missing_model_failure(ref)
+                    for ref in provider_refs
+                    if ref.model_id not in model_ids
+                )
+
+        if failures:
+            message = "Configured model validation failed:\n" + "\n".join(
+                f"- {failure}" for failure in failures
+            )
+            raise ServiceUnavailableError(message)
+
+        logger.info(
+            "Configured provider models validated: models={} providers={}",
+            len(refs),
+            len(refs_by_provider),
+        )
+
+    def _format_provider_query_failures(
+        self,
+        refs: list[ConfiguredChatModelRef],
+        exc: BaseException,
+    ) -> list[str]:
+        reason = provider_query_failure_reason(exc, self._settings)
+        return [self._format_model_validation_failure(ref, reason) for ref in refs]
+
+    def _format_missing_model_failure(self, ref: ConfiguredChatModelRef) -> str:
+        return self._format_model_validation_failure(ref, "missing model")
+
+    @staticmethod
+    def _format_model_validation_failure(
+        ref: ConfiguredChatModelRef, problem: str
+    ) -> str:
+        return (
+            f"sources={','.join(ref.sources)} provider={ref.provider_id} "
+            f"model={ref.model_id} problem={problem}"
+        )

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "free-claude-code"
-version = "2.3.16"
+version = "2.3.17"
 description = "Middleware between Claude Code CLI (Anthropic API) and NVIDIA NIM"
 readme = "README.md"
 requires-python = ">=3.14.0"

smoke/README.md

@@ -58,7 +58,7 @@ Default targets do not send real bot messages or load voice backends:
 | `cli` | `fcc-init`, server entrypoint, Claude CLI adaptive thinking, session cleanup | Claude CLI binary and provider only for real CLI |
 | `clients` | VS Code and JetBrains protocol payloads | configured provider |
 | `config` | env precedence, removed-env migration, proxy/timeouts | none |
-| `extensibility` | provider registry and platform factory construction | none |
+| `extensibility` | provider runtime and platform factory construction | none |
 | `messaging` | fake Discord/Telegram full flow, commands, trees, persistence, voice cancel | none |
 | `providers` | multi-turn text, adaptive thinking history, tools, disconnect, errors | configured providers, optional `FCC_SMOKE_MODEL_*` |
 | `tools` | forced tool_use and tool_result continuation | tool-capable configured provider |

smoke/capabilities.py

@@ -112,13 +112,13 @@ CAPABILITY_CONTRACTS: tuple[CapabilityContract, ...] = (
     ),
     CapabilityContract(
         "provider_routing",
-        "provider_registry",
+        "provider_runtime",
         "provider_matrix",
-        "providers.registry.ProviderRegistry",
+        "providers.runtime.ProviderRuntime",
         "provider id and Settings",
         "configured BaseProvider instance",
         "503 for missing credentials; invalid_request_error for unknown provider",
-        ("tests/api/test_dependencies.py", "tests/providers/test_registry.py"),
+        ("tests/api/test_dependencies.py", "tests/providers/test_provider_runtime.py"),
         (
             "test_configured_provider_models_stream_successfully",
             "test_provider_matrix_presence_e2e",
@@ -150,17 +150,17 @@ CAPABILITY_CONTRACTS: tuple[CapabilityContract, ...] = (
         "provider_routing",
         "provider_runtime_config",
         "provider_proxy_timeout_config",
-        "providers.registry.ProviderRegistry",
+        "providers.runtime.ProviderRuntime",
         "provider proxy, timeout, and rate-limit settings",
         "provider client and scoped limiter config",
         "provider construction failure",
-        ("tests/api/test_dependencies.py", "tests/providers/test_registry.py"),
+        ("tests/api/test_dependencies.py", "tests/providers/test_provider_runtime.py"),
     ),
     CapabilityContract(
         "provider_routing",
         "zero_cost_backends",
         "zero_cost_provider_access",
-        "providers.registry.ProviderRegistry",
+        "providers.runtime.ProviderRuntime",
         "configured free/local provider",
         "streaming response from selected backend",
         "missing env or upstream unavailable skip in smoke",
@@ -454,13 +454,13 @@ CAPABILITY_CONTRACTS: tuple[CapabilityContract, ...] = (
         "extensibility",
         "provider_platform_abcs",
         "extensible_provider_platform_abcs",
-        "providers.registry and messaging.platforms.factory",
+        "providers.runtime and messaging.platforms.factory",
         "new provider/platform implementations",
         "registered BaseProvider or messaging component bundle",
         "unknown platform returns None; unknown provider errors",
         (
             "tests/contracts/test_feature_manifest.py",
-            "tests/providers/test_registry.py",
+            "tests/providers/test_provider_runtime.py",
         ),
     ),
 )

smoke/features.py

@@ -235,10 +235,10 @@ FEATURE_INVENTORY: tuple[FeatureCoverage, ...] = (
         "readme",
         (
             "tests/contracts/test_feature_manifest.py",
-            "tests/providers/test_registry.py",
+            "tests/providers/test_provider_runtime.py",
         ),
         (),
-        ("test_provider_registry_e2e", "test_platform_factory_e2e"),
+        ("test_provider_runtime_config_e2e", "test_platform_factory_e2e"),
         ("extensibility",),
         (),
         "always runnable with isolated settings",
@@ -341,7 +341,7 @@ FEATURE_INVENTORY: tuple[FeatureCoverage, ...] = (
         "provider_proxy_timeout_config",
         "Provider proxies and HTTP timeout settings reach provider config",
         "public_surface",
-        ("tests/api/test_dependencies.py", "tests/providers/test_registry.py"),
+        ("tests/api/test_dependencies.py", "tests/providers/test_provider_runtime.py"),
         (),
         ("test_proxy_timeout_config_e2e",),
         ("config",),

smoke/product/test_config_extensibility_product_live.py

@@ -5,9 +5,10 @@ import subprocess
 
 import pytest
 
+from config.provider_catalog import PROVIDER_CATALOG
 from config.settings import Settings
 from messaging.platforms.factory import create_messaging_components
-from providers.registry import PROVIDER_DESCRIPTORS, build_provider_config
+from providers.runtime import build_provider_config
 from smoke.lib.child_process import cmd_free_claude_code_serve, cmd_python_c
 from smoke.lib.config import SmokeConfig
 from smoke.lib.e2e import SmokeServerDriver
@@ -112,8 +113,9 @@ def test_proxy_timeout_config_e2e(smoke_config: SmokeConfig, tmp_path) -> None:
     env["FCC_ENV_FILE"] = str(env_file)
     script = (
         "from config.settings import Settings; "
-        "from providers.registry import PROVIDER_DESCRIPTORS, build_provider_config; "
-        "s=Settings(); c=build_provider_config(PROVIDER_DESCRIPTORS['open_router'], s); "
+        "from config.provider_catalog import PROVIDER_CATALOG; "
+        "from providers.runtime import build_provider_config; "
+        "s=Settings(); c=build_provider_config(PROVIDER_CATALOG['open_router'], s); "
         "print(c.proxy); print(c.http_read_timeout); "
         "print(c.http_connect_timeout); print(c.http_write_timeout)"
     )
@@ -136,9 +138,9 @@ def test_proxy_timeout_config_e2e(smoke_config: SmokeConfig, tmp_path) -> None:
 
 
 @pytest.mark.smoke_target("extensibility")
-def test_provider_registry_e2e() -> None:
+def test_provider_runtime_config_e2e() -> None:
     settings_kwargs: dict[str, str] = {}
-    for descriptor in PROVIDER_DESCRIPTORS.values():
+    for descriptor in PROVIDER_CATALOG.values():
         if descriptor.credential_attr is not None:
             settings_kwargs[_settings_init_key(descriptor.credential_attr)] = (
                 f"{descriptor.provider_id}-key"
@@ -148,7 +150,7 @@ def test_provider_registry_e2e() -> None:
                 descriptor.default_base_url
             )
     settings = Settings.model_validate(settings_kwargs)
-    for descriptor in PROVIDER_DESCRIPTORS.values():
+    for descriptor in PROVIDER_CATALOG.values():
         config = build_provider_config(descriptor, settings)
         assert config.base_url
         assert config.api_key

tests/api/test_api.py

@@ -31,10 +31,10 @@ def client():
     with (
         patch("api.dependencies.resolve_provider", return_value=mock_provider),
         patch(
-            "providers.registry.ProviderRegistry.validate_configured_models",
+            "providers.runtime.ProviderRuntime.validate_configured_models",
             new_callable=AsyncMock,
         ),
-        patch("providers.registry.ProviderRegistry.start_model_list_refresh"),
+        patch("providers.runtime.ProviderRuntime.start_model_list_refresh"),
         TestClient(app) as test_client,
     ):
         yield test_client

tests/api/test_app_lifespan_and_errors.py

@@ -10,7 +10,7 @@ from fastapi.testclient import TestClient
 
 from config.settings import Settings
 from providers.exceptions import ServiceUnavailableError
-from providers.registry import ProviderRegistry
+from providers.runtime import ProviderRuntime
 
 _RUNTIME_EXTRAS = {
     "voice_note_enabled": True,
@@ -110,9 +110,9 @@ async def test_runtime_startup_logs_admin_url_without_printed_server_banner(tmp_
             api_runtime_mod.logging, "getLogger", return_value=uvicorn_logger
         ) as get_logger,
         patch.object(api_runtime_mod.logger, "info") as app_info,
-        patch.object(ProviderRegistry, "validate_configured_models", new=AsyncMock()),
-        patch.object(ProviderRegistry, "start_model_list_refresh"),
-        patch.object(ProviderRegistry, "cleanup", new=AsyncMock()),
+        patch.object(ProviderRuntime, "validate_configured_models", new=AsyncMock()),
+        patch.object(ProviderRuntime, "start_model_list_refresh"),
+        patch.object(ProviderRuntime, "cleanup", new=AsyncMock()),
         patch(
             "messaging.platforms.factory.create_messaging_components",
             return_value=None,
@@ -155,7 +155,7 @@ def test_create_app_provider_error_handler_returns_anthropic_format():
     )
     with (
         patch.object(api_app_mod, "get_settings", return_value=settings),
-        patch.object(ProviderRegistry, "cleanup", new=AsyncMock()),
+        patch.object(ProviderRuntime, "cleanup", new=AsyncMock()),
     ):
         with TestClient(app) as client:
             resp = client.get("/raise_provider")
@@ -192,7 +192,7 @@ def test_create_app_provider_error_default_logs_exclude_provider_message():
     )
     with (
         patch.object(api_app_mod, "get_settings", return_value=settings),
-        patch.object(ProviderRegistry, "cleanup", new=AsyncMock()),
+        patch.object(ProviderRuntime, "cleanup", new=AsyncMock()),
         patch.object(api_app_mod.logger, "error") as log_err,
     ):
         with TestClient(app) as client:
@@ -228,7 +228,7 @@ def test_create_app_general_exception_handler_returns_500():
     )
     with (
         patch.object(api_app_mod, "get_settings", return_value=settings),
-        patch.object(ProviderRegistry, "cleanup", new=AsyncMock()),
+        patch.object(ProviderRuntime, "cleanup", new=AsyncMock()),
     ):
         with TestClient(app, raise_server_exceptions=False) as client:
             resp = client.get("/raise_general")
@@ -265,7 +265,7 @@ def test_create_app_general_exception_default_logs_exclude_exception_message():
     )
     with (
         patch.object(api_app_mod, "get_settings", return_value=settings),
-        patch.object(ProviderRegistry, "cleanup", new=AsyncMock()),
+        patch.object(ProviderRuntime, "cleanup", new=AsyncMock()),
         patch.object(api_app_mod.logger, "error") as log_err,
     ):
         with TestClient(app, raise_server_exceptions=False) as client:
@@ -325,10 +325,10 @@ def test_app_lifespan_sets_state_and_cleans_up(tmp_path, messaging_enabled):
 
     api_app_mod = importlib.import_module("api.app")
 
-    registry_cleanup = AsyncMock()
+    runtime_cleanup = AsyncMock()
     with (
         patch.object(api_app_mod, "get_settings", return_value=settings),
-        patch.object(ProviderRegistry, "cleanup", new=registry_cleanup),
+        patch.object(ProviderRuntime, "cleanup", new=runtime_cleanup),
         patch(
             "messaging.platforms.factory.create_messaging_components",
             return_value=fake_components if messaging_enabled else None,
@@ -360,7 +360,7 @@ def test_app_lifespan_sets_state_and_cleans_up(tmp_path, messaging_enabled):
         cli_manager.stop_all.assert_not_awaited()
         assert getattr(app.state, "messaging_runtime", "missing") is None
 
-    registry_cleanup.assert_awaited_once()
+    runtime_cleanup.assert_awaited_once()
 
 
 def test_app_lifespan_cleanup_continues_if_platform_stop_raises(tmp_path):
@@ -396,10 +396,10 @@ def test_app_lifespan_cleanup_continues_if_platform_stop_raises(tmp_path):
     cli_manager.stop_all = AsyncMock()
 
     api_app_mod = importlib.import_module("api.app")
-    registry_cleanup = AsyncMock()
+    runtime_cleanup = AsyncMock()
     with (
         patch.object(api_app_mod, "get_settings", return_value=settings),
-        patch.object(ProviderRegistry, "cleanup", new=registry_cleanup),
+        patch.object(ProviderRuntime, "cleanup", new=runtime_cleanup),
         patch(
             "messaging.platforms.factory.create_messaging_components",
             return_value=fake_components,
@@ -412,7 +412,7 @@ def test_app_lifespan_cleanup_continues_if_platform_stop_raises(tmp_path):
 
     fake_platform.stop.assert_awaited_once()
     cli_manager.stop_all.assert_awaited_once()
-    registry_cleanup.assert_awaited_once()
+    runtime_cleanup.assert_awaited_once()
 
 
 @pytest.mark.asyncio
@@ -439,8 +439,8 @@ async def test_runtime_startup_validation_failure_does_not_block_server(tmp_path
     validation = AsyncMock(side_effect=ServiceUnavailableError("bad model"))
     cleanup = AsyncMock()
     with (
-        patch.object(ProviderRegistry, "validate_configured_models", new=validation),
-        patch.object(ProviderRegistry, "cleanup", new=cleanup),
+        patch.object(ProviderRuntime, "validate_configured_models", new=validation),
+        patch.object(ProviderRuntime, "cleanup", new=cleanup),
         patch.object(api_runtime_mod.logger, "warning") as log_warning,
         patch(
             "messaging.platforms.factory.create_messaging_components",
@@ -450,7 +450,7 @@ async def test_runtime_startup_validation_failure_does_not_block_server(tmp_path
         await runtime.startup()
         await runtime.shutdown()
 
-    validation.assert_awaited_once_with(settings)
+    validation.assert_awaited_once_with()
     cleanup.assert_awaited_once()
     create_components.assert_called_once()
     logged = " ".join(
@@ -494,8 +494,8 @@ async def test_graceful_asgi_lifespan_model_validation_failure_starts(tmp_path):
     cleanup = AsyncMock()
     with (
         patch.object(api_app_mod, "get_settings", return_value=settings),
-        patch.object(ProviderRegistry, "validate_configured_models", new=validation),
-        patch.object(ProviderRegistry, "cleanup", new=cleanup),
+        patch.object(ProviderRuntime, "validate_configured_models", new=validation),
+        patch.object(ProviderRuntime, "cleanup", new=cleanup),
     ):
         await app({"type": "lifespan"}, receive, send)
 
@@ -524,10 +524,10 @@ def test_app_lifespan_messaging_import_error_no_crash(tmp_path, caplog):
     )
 
     api_app_mod = importlib.import_module("api.app")
-    registry_cleanup = AsyncMock()
+    runtime_cleanup = AsyncMock()
     with (
         patch.object(api_app_mod, "get_settings", return_value=settings),
-        patch.object(ProviderRegistry, "cleanup", new=registry_cleanup),
+        patch.object(ProviderRuntime, "cleanup", new=runtime_cleanup),
         patch(
             "messaging.platforms.factory.create_messaging_components",
             side_effect=ImportError("discord not installed"),
@@ -537,7 +537,7 @@ def test_app_lifespan_messaging_import_error_no_crash(tmp_path, caplog):
         pass
 
     assert getattr(app.state, "messaging_runtime", None) is None
-    registry_cleanup.assert_awaited_once()
+    runtime_cleanup.assert_awaited_once()
 
 
 def test_app_lifespan_platform_start_exception_cleanup_still_runs(tmp_path):
@@ -574,10 +574,10 @@ def test_app_lifespan_platform_start_exception_cleanup_still_runs(tmp_path):
     cli_manager.stop_all = AsyncMock()
 
     api_app_mod = importlib.import_module("api.app")
-    registry_cleanup = AsyncMock()
+    runtime_cleanup = AsyncMock()
     with (
         patch.object(api_app_mod, "get_settings", return_value=settings),
-        patch.object(ProviderRegistry, "cleanup", new=registry_cleanup),
+        patch.object(ProviderRuntime, "cleanup", new=runtime_cleanup),
         patch(
             "messaging.platforms.factory.create_messaging_components",
             return_value=fake_components,
@@ -588,7 +588,7 @@ def test_app_lifespan_platform_start_exception_cleanup_still_runs(tmp_path):
     ):
         pass
 
-    registry_cleanup.assert_awaited_once()
+    runtime_cleanup.assert_awaited_once()
 
 
 def test_app_lifespan_flush_pending_save_exception_warning_only(tmp_path):
@@ -626,10 +626,10 @@ def test_app_lifespan_flush_pending_save_exception_warning_only(tmp_path):
     cli_manager.stop_all = AsyncMock()
 
     api_app_mod = importlib.import_module("api.app")
-    registry_cleanup = AsyncMock()
+    runtime_cleanup = AsyncMock()
     with (
         patch.object(api_app_mod, "get_settings", return_value=settings),
-        patch.object(ProviderRegistry, "cleanup", new=registry_cleanup),
+        patch.object(ProviderRuntime, "cleanup", new=runtime_cleanup),
         patch(
             "messaging.platforms.factory.create_messaging_components",
             return_value=fake_components,
@@ -641,7 +641,7 @@ def test_app_lifespan_flush_pending_save_exception_warning_only(tmp_path):
         pass
 
     session_store.flush_pending_save.assert_called_once()
-    registry_cleanup.assert_awaited_once()
+    runtime_cleanup.assert_awaited_once()
 
 
 def test_create_app_writes_server_log_under_fcc_home(monkeypatch, tmp_path):

tests/api/test_dependencies.py

@@ -1,6 +1,6 @@
 from types import SimpleNamespace
 from typing import cast
-from unittest.mock import AsyncMock, MagicMock, patch
+from unittest.mock import MagicMock, patch
 
 import pytest
 from fastapi import HTTPException
@@ -8,37 +8,24 @@ from starlette.applications import Starlette
 from starlette.datastructures import State
 
 from api.dependencies import (
-    cleanup_provider,
-    get_provider,
-    get_provider_for_type,
+    get_provider_runtime,
     get_settings,
+    maybe_provider_runtime,
+    require_api_key,
     resolve_provider,
 )
 from config.nim import NimSettings
-from providers.cerebras import CerebrasProvider
-from providers.codestral import CodestralProvider
-from providers.deepseek import DeepSeekProvider
-from providers.exceptions import ServiceUnavailableError, UnknownProviderTypeError
-from providers.gemini import GeminiProvider
-from providers.groq import GroqProvider
-from providers.lmstudio import LMStudioProvider
-from providers.mistral import MistralProvider
+from providers.exceptions import ServiceUnavailableError
 from providers.nvidia_nim import NvidiaNimProvider
-from providers.ollama import OllamaProvider
-from providers.open_router import OpenRouterProvider
-from providers.registry import ProviderRegistry
-from providers.wafer import WaferProvider
+from providers.runtime import ProviderRuntime
 
 
 def _make_mock_settings(**overrides):
-    """Create a mock settings object with all required fields for get_provider()."""
+    """Create a mock settings object with provider runtime fields."""
     mock = MagicMock()
     mock.model = "nvidia_nim/meta/llama3"
     mock.provider_type = "nvidia_nim"
     mock.nvidia_nim_api_key = "test_key"
-    mock.provider_rate_limit = 40
-    mock.provider_rate_window = 60
-    mock.provider_max_concurrency = 5
     mock.open_router_api_key = "test_openrouter_key"
     mock.mistral_api_key = "test_mistral_key"
     mock.codestral_api_key = "test_codestral_key"
@@ -56,6 +43,7 @@ def _make_mock_settings(**overrides):
     mock.lmstudio_proxy = ""
     mock.llamacpp_proxy = ""
     mock.kimi_proxy = ""
+    mock.kimi_api_key = "test_kimi_key"
     mock.wafer_proxy = ""
     mock.opencode_proxy = ""
     mock.opencode_go_proxy = ""
@@ -68,624 +56,147 @@ def _make_mock_settings(**overrides):
     mock.groq_proxy = ""
     mock.cerebras_api_key = ""
     mock.cerebras_proxy = ""
-    mock.nim = NimSettings()
+    mock.provider_rate_limit = 40
+    mock.provider_rate_window = 60
+    mock.provider_max_concurrency = 5
     mock.http_read_timeout = 300.0
     mock.http_write_timeout = 10.0
     mock.http_connect_timeout = 10.0
     mock.enable_model_thinking = True
+    mock.log_raw_sse_events = False
+    mock.log_api_error_tracebacks = False
+    mock.configured_chat_model_refs.return_value = ()
+    mock.nim = NimSettings()
     for key, value in overrides.items():
         setattr(mock, key, value)
     return mock
 
 
-@pytest.fixture(autouse=True)
-def reset_provider():
-    """Reset the global _providers registry between tests."""
-    import api.dependencies
-
-    saved = api.dependencies._providers
-    api.dependencies._providers = {}
-    yield
-    api.dependencies._providers = saved
+def _app_with_runtime(settings=None):
+    app = SimpleNamespace(state=State())
+    app.state.provider_runtime = ProviderRuntime(settings or _make_mock_settings())
+    return cast(Starlette, app)
 
 
-@pytest.mark.asyncio
-async def test_get_provider_singleton():
-    with patch("api.dependencies.get_settings") as mock_settings:
-        mock_settings.return_value = _make_mock_settings()
+def _request(headers=None, token: str = ""):
+    return SimpleNamespace(
+        headers=headers or {},
+    ), SimpleNamespace(anthropic_auth_token=token)
 
-        p1 = get_provider()
-        p2 = get_provider()
 
-        assert isinstance(p1, NvidiaNimProvider)
-        assert p1 is p2
-
-
-@pytest.mark.asyncio
-async def test_get_settings():
+def test_get_settings():
     settings = get_settings()
     assert settings is not None
-    # Verify it calls the internal _get_settings
     with patch("api.dependencies._get_settings") as mock_get:
         get_settings()
         mock_get.assert_called_once()
 
 
-@pytest.mark.asyncio
-async def test_cleanup_provider():
-    with patch("api.dependencies.get_settings") as mock_settings:
-        mock_settings.return_value = _make_mock_settings()
-
-        provider = get_provider()
-        assert isinstance(provider, NvidiaNimProvider)
-        provider._client = AsyncMock()
-
-        await cleanup_provider()
-
-        provider._client.close.assert_called_once()
-
-
-@pytest.mark.asyncio
-async def test_cleanup_provider_no_client():
-    with patch("api.dependencies.get_settings") as mock_settings:
-        mock_settings.return_value = _make_mock_settings()
-
-        provider = get_provider()
-        if hasattr(provider, "_client"):
-            del provider._client
-
-        await cleanup_provider()
-        # Should not raise
-
-
-@pytest.mark.asyncio
-async def test_get_provider_open_router():
-    """Test that provider_type=open_router returns OpenRouterProvider."""
-    with patch("api.dependencies.get_settings") as mock_settings:
-        mock_settings.return_value = _make_mock_settings(provider_type="open_router")
-
-        provider = get_provider()
-
-        assert isinstance(provider, OpenRouterProvider)
-        assert provider._base_url == "https://openrouter.ai/api/v1"
-        assert provider._api_key == "test_openrouter_key"
-
-
-@pytest.mark.asyncio
-async def test_get_provider_lmstudio():
-    """Test that provider_type=lmstudio returns LMStudioProvider."""
-    with patch("api.dependencies.get_settings") as mock_settings:
-        mock_settings.return_value = _make_mock_settings(provider_type="lmstudio")
-
-        provider = get_provider()
-
-        assert isinstance(provider, LMStudioProvider)
-        assert provider._base_url == "http://localhost:1234/v1"
-
-
-@pytest.mark.asyncio
-async def test_get_provider_ollama():
-    """Test that provider_type=ollama returns OllamaProvider without an API key."""
-    with patch("api.dependencies.get_settings") as mock_settings:
-        mock_settings.return_value = _make_mock_settings(provider_type="ollama")
-
-        provider = get_provider()
-
-        assert isinstance(provider, OllamaProvider)
-        assert provider._base_url == "http://localhost:11434"
-        assert provider._api_key == "ollama"
-
-
-@pytest.mark.asyncio
-async def test_get_provider_deepseek():
-    """Test that provider_type=deepseek returns DeepSeekProvider."""
-    with patch("api.dependencies.get_settings") as mock_settings:
-        mock_settings.return_value = _make_mock_settings(provider_type="deepseek")
-
-        provider = get_provider()
-
-        assert isinstance(provider, DeepSeekProvider)
-        assert provider._base_url == "https://api.deepseek.com/anthropic"
-        assert provider._api_key == "test_deepseek_key"
-        assert provider._config.enable_thinking is True
-
+def test_get_provider_runtime_returns_app_scoped_runtime() -> None:
+    app = _app_with_runtime()
 
-@pytest.mark.asyncio
-async def test_get_provider_deepseek_uses_fixed_base_url():
-    """DeepSeek provider always uses the fixed provider base URL."""
-    with patch("api.dependencies.get_settings") as mock_settings:
-        mock_settings.return_value = _make_mock_settings(
-            provider_type="deepseek",
-        )
+    assert isinstance(get_provider_runtime(app), ProviderRuntime)
+    assert maybe_provider_runtime(app) is get_provider_runtime(app)
 
-        provider = get_provider()
 
-        assert isinstance(provider, DeepSeekProvider)
-        assert provider._base_url == "https://api.deepseek.com/anthropic"
+def test_get_provider_runtime_missing_runtime_raises_service_unavailable() -> None:
+    app = cast(Starlette, SimpleNamespace(state=State()))
 
+    assert maybe_provider_runtime(app) is None
+    with pytest.raises(
+        ServiceUnavailableError, match="Provider runtime is not configured"
+    ):
+        get_provider_runtime(app)
 
-@pytest.mark.asyncio
-async def test_get_provider_deepseek_passes_enable_model_thinking():
-    """DeepSeek provider receives the fallback thinking toggle."""
-    with patch("api.dependencies.get_settings") as mock_settings:
-        mock_settings.return_value = _make_mock_settings(
-            provider_type="deepseek",
-            enable_model_thinking=False,
-        )
-
-        provider = get_provider()
-
-        assert isinstance(provider, DeepSeekProvider)
-        assert provider._config.enable_thinking is False
-
-
-@pytest.mark.asyncio
-async def test_get_provider_mistral():
-    """Test that provider_type=mistral returns MistralProvider."""
-    with patch("api.dependencies.get_settings") as mock_settings:
-        mock_settings.return_value = _make_mock_settings(provider_type="mistral")
-
-        provider = get_provider()
-
-        assert isinstance(provider, MistralProvider)
-        assert provider._base_url == "https://api.mistral.ai/v1"
-        assert provider._api_key == "test_mistral_key"
-
-
-@pytest.mark.asyncio
-async def test_get_provider_mistral_codestral():
-    """provider_type=mistral_codestral returns CodestralProvider."""
-    with patch("api.dependencies.get_settings") as mock_settings:
-        mock_settings.return_value = _make_mock_settings(
-            provider_type="mistral_codestral",
-        )
-
-        provider = get_provider()
-
-        assert isinstance(provider, CodestralProvider)
-        assert provider._base_url == "https://codestral.mistral.ai/v1"
-        assert provider._api_key == "test_codestral_key"
-
-
-@pytest.mark.asyncio
-async def test_get_provider_gemini():
-    """Test that provider_type=gemini returns GeminiProvider."""
-    with patch("api.dependencies.get_settings") as mock_settings:
-        mock_settings.return_value = _make_mock_settings(
-            provider_type="gemini",
-            gemini_api_key="secret",
-        )
-
-        provider = get_provider()
-
-        assert isinstance(provider, GeminiProvider)
-        assert provider._base_url == (
-            "https://generativelanguage.googleapis.com/v1beta/openai"
-        )
-        assert provider._api_key == "secret"
-
-
-@pytest.mark.asyncio
-async def test_get_provider_gemini_missing_api_key():
-    """Gemini with empty API key raises HTTPException 503."""
-    with patch("api.dependencies.get_settings") as mock_settings:
-        mock_settings.return_value = _make_mock_settings(
-            provider_type="gemini",
-            gemini_api_key="",
-        )
-
-        with pytest.raises(HTTPException) as exc_info:
-            get_provider()
-
-        assert exc_info.value.status_code == 503
-        assert "GEMINI_API_KEY" in exc_info.value.detail
-        assert "aistudio.google.com" in exc_info.value.detail
-
-
-@pytest.mark.asyncio
-async def test_get_provider_groq():
-    """Test that provider_type=groq returns GroqProvider."""
-    with patch("api.dependencies.get_settings") as mock_settings:
-        mock_settings.return_value = _make_mock_settings(
-            provider_type="groq",
-            groq_api_key="secret",
-        )
-
-        provider = get_provider()
-
-        assert isinstance(provider, GroqProvider)
-        assert provider._base_url == "https://api.groq.com/openai/v1"
-        assert provider._api_key == "secret"
-
-
-@pytest.mark.asyncio
-async def test_get_provider_groq_missing_api_key():
-    """Groq with empty API key raises HTTPException 503."""
-    with patch("api.dependencies.get_settings") as mock_settings:
-        mock_settings.return_value = _make_mock_settings(
-            provider_type="groq",
-            groq_api_key="",
-        )
-
-        with pytest.raises(HTTPException) as exc_info:
-            get_provider()
-
-        assert exc_info.value.status_code == 503
-        assert "GROQ_API_KEY" in exc_info.value.detail
-        assert "console.groq.com" in exc_info.value.detail
-
-
-@pytest.mark.asyncio
-async def test_get_provider_cerebras():
-    """Test that provider_type=cerebras returns CerebrasProvider."""
-    with patch("api.dependencies.get_settings") as mock_settings:
-        mock_settings.return_value = _make_mock_settings(
-            provider_type="cerebras",
-            cerebras_api_key="secret",
-        )
-
-        provider = get_provider()
-
-        assert isinstance(provider, CerebrasProvider)
-        assert provider._base_url == "https://api.cerebras.ai/v1"
-        assert provider._api_key == "secret"
-
-
-@pytest.mark.asyncio
-async def test_get_provider_cerebras_missing_api_key():
-    """Cerebras with empty API key raises HTTPException 503."""
-    with patch("api.dependencies.get_settings") as mock_settings:
-        mock_settings.return_value = _make_mock_settings(
-            provider_type="cerebras",
-            cerebras_api_key="",
-        )
-
-        with pytest.raises(HTTPException) as exc_info:
-            get_provider()
 
-        assert exc_info.value.status_code == 503
-        assert "CEREBRAS_API_KEY" in exc_info.value.detail
-        assert "cloud.cerebras.ai" in exc_info.value.detail
+def test_resolve_provider_per_app_uses_separate_runtimes() -> None:
+    app1 = _app_with_runtime()
+    app2 = _app_with_runtime()
 
+    with patch("providers.transports.openai_chat.transport.AsyncOpenAI"):
+        p1 = resolve_provider("nvidia_nim", app=app1)
+        p2 = resolve_provider("nvidia_nim", app=app2)
 
-@pytest.mark.asyncio
-async def test_get_provider_wafer():
-    """Test that provider_type=wafer returns WaferProvider."""
-    with patch("api.dependencies.get_settings") as mock_settings:
-        mock_settings.return_value = _make_mock_settings(provider_type="wafer")
-
-        provider = get_provider()
+    assert isinstance(p1, NvidiaNimProvider)
+    assert isinstance(p2, NvidiaNimProvider)
+    assert p1 is not p2
 
-        assert isinstance(provider, WaferProvider)
-        assert provider._base_url == "https://pass.wafer.ai/v1"
-        assert provider._api_key == "test_wafer_key"
 
+def test_resolve_provider_missing_key_raises_503() -> None:
+    app = _app_with_runtime(_make_mock_settings(open_router_api_key=""))
 
-@pytest.mark.asyncio
-async def test_get_provider_lmstudio_uses_lm_studio_base_url():
-    """LM Studio provider uses lm_studio_base_url from settings."""
-    with patch("api.dependencies.get_settings") as mock_settings:
-        mock_settings.return_value = _make_mock_settings(
-            provider_type="lmstudio",
-            lm_studio_base_url="http://custom:9999/v1",
-        )
+    with pytest.raises(HTTPException) as exc_info:
+        resolve_provider("open_router", app=app)
 
-        provider = get_provider()
+    assert exc_info.value.status_code == 503
+    assert "OPENROUTER_API_KEY" in exc_info.value.detail
+    assert "openrouter.ai" in exc_info.value.detail
 
-        assert isinstance(provider, LMStudioProvider)
-        assert provider._base_url == "http://custom:9999/v1"
 
+def test_resolve_provider_missing_runtime_raises_service_unavailable() -> None:
+    app = cast(Starlette, SimpleNamespace(state=State()))
 
-@pytest.mark.asyncio
-async def test_get_provider_passes_http_timeouts_from_settings():
-    """Provider receives http timeouts from settings when creating client."""
-    with (
-        patch("api.dependencies.get_settings") as mock_settings,
-        patch("providers.transports.openai_chat.transport.AsyncOpenAI") as mock_openai,
-    ):
-        mock_settings.return_value = _make_mock_settings(
-            http_read_timeout=600.0,
-            http_write_timeout=20.0,
-            http_connect_timeout=5.0,
-        )
-        provider = get_provider()
-        assert isinstance(provider, NvidiaNimProvider)
-        call_kwargs = mock_openai.call_args[1]
-        timeout = call_kwargs["timeout"]
-        assert timeout.read == 600.0
-        assert timeout.write == 20.0
-        assert timeout.connect == 5.0
-
-
-@pytest.mark.asyncio
-async def test_get_provider_passes_proxy_from_settings():
-    """Provider receives configured proxy and builds a proxied HTTP client."""
-    with (
-        patch("api.dependencies.get_settings") as mock_settings,
-        patch(
-            "providers.transports.openai_chat.transport.httpx.AsyncClient"
-        ) as mock_http_client,
-        patch("providers.transports.openai_chat.transport.AsyncOpenAI") as mock_openai,
+    with pytest.raises(
+        ServiceUnavailableError, match="Provider runtime is not configured"
     ):
-        mock_settings.return_value = _make_mock_settings(
-            nvidia_nim_proxy="http://proxy.example:8080"
-        )
+        resolve_provider("nvidia_nim", app=app)
 
-        provider = get_provider()
 
-        assert isinstance(provider, NvidiaNimProvider)
-        mock_http_client.assert_called_once()
-        assert mock_http_client.call_args.kwargs["proxy"] == "http://proxy.example:8080"
-        assert (
-            mock_openai.call_args.kwargs["http_client"] is mock_http_client.return_value
-        )
+def test_resolve_provider_unrelated_value_error_is_not_unknown_provider_log() -> None:
+    import api.dependencies as deps
 
+    app = _app_with_runtime()
+    runtime = get_provider_runtime(app)
 
-@pytest.mark.asyncio
-async def test_get_provider_ignores_non_string_proxy_value():
-    """Mock settings without proxy attrs should not fail provider construction."""
     with (
-        patch("api.dependencies.get_settings") as mock_settings,
-        patch("providers.transports.openai_chat.transport.AsyncOpenAI") as mock_openai,
+        patch.object(
+            runtime,
+            "resolve_provider",
+            side_effect=ValueError("unrelated config"),
+        ),
+        patch.object(deps.logger, "error") as log_err,
+        pytest.raises(ValueError, match="unrelated config"),
     ):
-        mock_settings.return_value = _make_mock_settings(
-            nvidia_nim_proxy=MagicMock(name="proxy")
-        )
-
-        provider = get_provider()
-
-        assert isinstance(provider, NvidiaNimProvider)
-        assert mock_openai.call_args.kwargs["http_client"] is None
-
-
-@pytest.mark.asyncio
-async def test_get_provider_nvidia_nim_missing_api_key():
-    """NVIDIA NIM with empty API key raises HTTPException 503."""
-    with patch("api.dependencies.get_settings") as mock_settings:
-        mock_settings.return_value = _make_mock_settings(nvidia_nim_api_key="")
-
-        with pytest.raises(HTTPException) as exc_info:
-            get_provider()
-
-        assert exc_info.value.status_code == 503
-        assert "NVIDIA_NIM_API_KEY" in exc_info.value.detail
-        assert "build.nvidia.com" in exc_info.value.detail
-
-
-@pytest.mark.asyncio
-async def test_get_provider_nvidia_nim_whitespace_only_api_key():
-    """NVIDIA NIM with whitespace-only API key raises HTTPException 503."""
-    with patch("api.dependencies.get_settings") as mock_settings:
-        mock_settings.return_value = _make_mock_settings(nvidia_nim_api_key="   ")
-
-        with pytest.raises(HTTPException) as exc_info:
-            get_provider()
-
-        assert exc_info.value.status_code == 503
-        assert "NVIDIA_NIM_API_KEY" in exc_info.value.detail
-
-
-@pytest.mark.asyncio
-async def test_get_provider_open_router_missing_api_key():
-    """OpenRouter with empty API key raises HTTPException 503."""
-    with patch("api.dependencies.get_settings") as mock_settings:
-        mock_settings.return_value = _make_mock_settings(
-            provider_type="open_router",
-            open_router_api_key="",
-        )
-
-        with pytest.raises(HTTPException) as exc_info:
-            get_provider()
-
-        assert exc_info.value.status_code == 503
-        assert "OPENROUTER_API_KEY" in exc_info.value.detail
-        assert "openrouter.ai" in exc_info.value.detail
-
-
-@pytest.mark.asyncio
-async def test_get_provider_mistral_missing_api_key():
-    """Mistral with empty API key raises HTTPException 503."""
-    with patch("api.dependencies.get_settings") as mock_settings:
-        mock_settings.return_value = _make_mock_settings(
-            provider_type="mistral",
-            mistral_api_key="",
-        )
-
-        with pytest.raises(HTTPException) as exc_info:
-            get_provider()
-
-        assert exc_info.value.status_code == 503
-        assert "MISTRAL_API_KEY" in exc_info.value.detail
-        assert "console.mistral.ai" in exc_info.value.detail
-
-
-@pytest.mark.asyncio
-async def test_get_provider_mistral_codestral_missing_api_key():
-    """Mistral Codestral with empty API key raises HTTPException 503."""
-    with patch("api.dependencies.get_settings") as mock_settings:
-        mock_settings.return_value = _make_mock_settings(
-            provider_type="mistral_codestral",
-            codestral_api_key="",
-        )
-
-        with pytest.raises(HTTPException) as exc_info:
-            get_provider()
-
-        assert exc_info.value.status_code == 503
-        assert "CODESTRAL_API_KEY" in exc_info.value.detail
-        assert "console.mistral.ai" in exc_info.value.detail
-
-
-@pytest.mark.asyncio
-async def test_get_provider_deepseek_missing_api_key():
-    """DeepSeek with empty API key raises HTTPException 503."""
-    with patch("api.dependencies.get_settings") as mock_settings:
-        mock_settings.return_value = _make_mock_settings(
-            provider_type="deepseek",
-            deepseek_api_key="",
-        )
-
-        with pytest.raises(HTTPException) as exc_info:
-            get_provider()
-
-        assert exc_info.value.status_code == 503
-        assert "DEEPSEEK_API_KEY" in exc_info.value.detail
-        assert "platform.deepseek.com" in exc_info.value.detail
-
-
-@pytest.mark.asyncio
-async def test_get_provider_wafer_missing_api_key():
-    """Wafer with empty API key raises HTTPException 503."""
-    with patch("api.dependencies.get_settings") as mock_settings:
-        mock_settings.return_value = _make_mock_settings(
-            provider_type="wafer",
-            wafer_api_key="",
-        )
-
-        with pytest.raises(HTTPException) as exc_info:
-            get_provider()
-
-        assert exc_info.value.status_code == 503
-        assert "WAFER_API_KEY" in exc_info.value.detail
-        assert "wafer.ai" in exc_info.value.detail
-
-
-@pytest.mark.asyncio
-async def test_get_provider_unknown_type():
-    """Unknown ``provider_type`` raises :exc:`~providers.exceptions.UnknownProviderTypeError`."""
-    with patch("api.dependencies.get_settings") as mock_settings:
-        mock_settings.return_value = _make_mock_settings(provider_type="unknown")
-
-        with pytest.raises(UnknownProviderTypeError, match="Unknown provider_type"):
-            get_provider()
-
-
-@pytest.mark.asyncio
-async def test_cleanup_provider_close_raises():
-    """cleanup_provider handles close() raising an exception."""
-    with patch("api.dependencies.get_settings") as mock_settings:
-        mock_settings.return_value = _make_mock_settings()
-
-        provider = get_provider()
-        assert isinstance(provider, NvidiaNimProvider)
-        provider._client = AsyncMock()
-        provider._client.close = AsyncMock(side_effect=RuntimeError("cleanup failed"))
-
-        # Should propagate the error
-        with pytest.raises(RuntimeError, match="cleanup failed"):
-            await cleanup_provider()
-
-
-# --- Provider Registry Tests ---
-
-
-@pytest.mark.asyncio
-async def test_get_provider_for_type_caches():
-    """get_provider_for_type returns cached provider on second call."""
-    with patch("api.dependencies.get_settings") as mock_settings:
-        mock_settings.return_value = _make_mock_settings()
-
-        p1 = get_provider_for_type("nvidia_nim")
-        p2 = get_provider_for_type("nvidia_nim")
-
-        assert p1 is p2
-        assert isinstance(p1, NvidiaNimProvider)
-
-
-@pytest.mark.asyncio
-async def test_get_provider_for_type_different_types():
-    """get_provider_for_type creates separate providers per type."""
-    with patch("api.dependencies.get_settings") as mock_settings:
-        mock_settings.return_value = _make_mock_settings()
-
-        nim = get_provider_for_type("nvidia_nim")
-        lmstudio = get_provider_for_type("lmstudio")
-
-        assert isinstance(nim, NvidiaNimProvider)
-        assert isinstance(lmstudio, LMStudioProvider)
-        assert nim is not lmstudio
-
+        deps.resolve_provider("nvidia_nim", app=app)
+    log_err.assert_not_called()
 
-@pytest.mark.asyncio
-async def test_get_provider_for_type_missing_key_raises_503():
-    """get_provider_for_type raises HTTPException 503 for missing API key."""
-    with patch("api.dependencies.get_settings") as mock_settings:
-        mock_settings.return_value = _make_mock_settings(open_router_api_key="")
 
-        with pytest.raises(HTTPException) as exc_info:
-            get_provider_for_type("open_router")
+def test_require_api_key_allows_when_no_token_configured():
+    request, settings = _request(headers={}, token="")
 
-        assert exc_info.value.status_code == 503
-        assert "OPENROUTER_API_KEY" in exc_info.value.detail
+    require_api_key(request, settings)
 
 
-@pytest.mark.asyncio
-async def test_cleanup_provider_cleans_all():
-    """cleanup_provider cleans up all providers in the registry."""
-    with patch("api.dependencies.get_settings") as mock_settings:
-        mock_settings.return_value = _make_mock_settings()
+def test_require_api_key_rejects_missing_token():
+    request, settings = _request(headers={}, token="secret")
 
-        nim = get_provider_for_type("nvidia_nim")
-        lmstudio = get_provider_for_type("lmstudio")
+    with pytest.raises(HTTPException) as exc_info:
+        require_api_key(request, settings)
 
-        assert isinstance(nim, NvidiaNimProvider)
-        assert isinstance(lmstudio, LMStudioProvider)
+    assert exc_info.value.status_code == 401
+    assert exc_info.value.detail == "Missing API key"
 
-        nim._client = AsyncMock()
-        lmstudio._client = AsyncMock()
 
-        await cleanup_provider()
+def test_require_api_key_accepts_x_api_key():
+    request, settings = _request(headers={"x-api-key": "secret"}, token="secret")
 
-        nim._client.close.assert_called_once()
-        lmstudio._client.aclose.assert_called_once()
+    require_api_key(request, settings)
 
 
-def test_resolve_provider_per_app_uses_separate_registries() -> None:
-    """With app set, each app gets its own provider cache (not process _providers)."""
-    with patch("api.dependencies.get_settings") as mock_settings:
-        mock_settings.return_value = _make_mock_settings()
-        settings = _make_mock_settings()
-        app1 = SimpleNamespace(state=State())
-        app2 = SimpleNamespace(state=State())
-        app1.state.provider_registry = ProviderRegistry()
-        app2.state.provider_registry = ProviderRegistry()
-        p1 = resolve_provider(
-            "nvidia_nim", app=cast(Starlette, app1), settings=settings
-        )
-        p2 = resolve_provider(
-            "nvidia_nim", app=cast(Starlette, app2), settings=settings
-        )
-        assert isinstance(p1, NvidiaNimProvider)
-        assert isinstance(p2, NvidiaNimProvider)
-        assert p1 is not p2
+def test_require_api_key_accepts_bearer_token_and_strips_model_suffix():
+    request, settings = _request(
+        headers={"authorization": "Bearer secret:claude-sonnet"},
+        token="secret",
+    )
 
+    require_api_key(request, settings)
 
-def test_resolve_provider_missing_registry_raises_service_unavailable() -> None:
-    """HTTP apps must install app.state.provider_registry (e.g. via AppRuntime)."""
-    with patch("api.dependencies.get_settings") as mock_settings:
-        mock_settings.return_value = _make_mock_settings()
-        settings = _make_mock_settings()
-        app = SimpleNamespace(state=State())
-        assert getattr(app.state, "provider_registry", None) is None
-        with pytest.raises(
-            ServiceUnavailableError, match="Provider registry is not configured"
-        ):
-            resolve_provider("nvidia_nim", app=cast(Starlette, app), settings=settings)
 
+def test_require_api_key_rejects_invalid_token():
+    request, settings = _request(headers={"x-api-key": "wrong"}, token="secret")
 
-def test_resolve_provider_unrelated_value_error_is_not_unknown_provider_log() -> None:
-    """Only :exc:`~providers.exceptions.UnknownProviderTypeError` logs unknown provider."""
-    import api.dependencies as deps
+    with pytest.raises(HTTPException) as exc_info:
+        require_api_key(request, settings)
 
-    with (
-        patch.object(deps, "get_settings", return_value=_make_mock_settings()),
-        patch.object(
-            ProviderRegistry,
-            "get",
-            side_effect=ValueError("unrelated config"),
-        ),
-        patch.object(deps.logger, "error") as log_err,
-        pytest.raises(ValueError, match="unrelated config"),
-    ):
-        deps.resolve_provider("nvidia_nim", app=None, settings=_make_mock_settings())
-    log_err.assert_not_called()
+    assert exc_info.value.status_code == 401
+    assert exc_info.value.detail == "Invalid API key"

tests/api/test_model_listing.py

@@ -4,7 +4,7 @@ from api.app import create_app
 from api.dependencies import get_settings
 from config.settings import Settings
 from providers.model_listing import ProviderModelInfo
-from providers.registry import ProviderRegistry
+from providers.runtime import ProviderRuntime
 
 
 def _settings(
@@ -25,10 +25,10 @@ def _settings(
 def test_models_list_includes_configured_refs_cached_provider_models_and_aliases():
     app = create_app(lifespan_enabled=False)
     settings = _settings()
-    registry = ProviderRegistry()
-    registry.cache_model_ids("deepseek", {"deepseek-chat"})
-    registry.cache_model_ids("open_router", {"meta/llama-3.3", "anthropic/claude-opus"})
-    app.state.provider_registry = registry
+    runtime = ProviderRuntime(settings)
+    runtime.cache_model_ids("deepseek", {"deepseek-chat"})
+    runtime.cache_model_ids("open_router", {"meta/llama-3.3", "anthropic/claude-opus"})
+    app.state.provider_runtime = runtime
     app.dependency_overrides[get_settings] = lambda: settings
 
     try:
@@ -72,16 +72,16 @@ def test_models_list_includes_configured_refs_cached_provider_models_and_aliases
 def test_models_list_uses_openrouter_thinking_metadata_for_cached_models():
     app = create_app(lifespan_enabled=False)
     settings = _settings(model_opus=None)
-    registry = ProviderRegistry()
-    registry.cache_model_ids("deepseek", {"deepseek-chat"})
-    registry.cache_model_infos(
+    runtime = ProviderRuntime(settings)
+    runtime.cache_model_ids("deepseek", {"deepseek-chat"})
+    runtime.cache_model_infos(
         "open_router",
         {
             ProviderModelInfo("reasoning-model", supports_thinking=True),
             ProviderModelInfo("plain-model", supports_thinking=False),
         },
     )
-    app.state.provider_registry = registry
+    app.state.provider_runtime = runtime
     app.dependency_overrides[get_settings] = lambda: settings
 
     try:
@@ -104,12 +104,12 @@ def test_models_list_uses_cached_metadata_for_configured_openrouter_refs():
         model_opus=None,
         model_haiku=None,
     )
-    registry = ProviderRegistry()
-    registry.cache_model_infos(
+    runtime = ProviderRuntime(settings)
+    runtime.cache_model_infos(
         "open_router",
         {ProviderModelInfo("plain-model", supports_thinking=False)},
     )
-    app.state.provider_registry = registry
+    app.state.provider_runtime = runtime
     app.dependency_overrides[get_settings] = lambda: settings
 
     try:
@@ -130,9 +130,9 @@ def test_models_list_includes_cached_wafer_models():
         model_opus=None,
         model_haiku=None,
     )
-    registry = ProviderRegistry()
-    registry.cache_model_ids("wafer", {"DeepSeek-V4-Pro", "MiniMax-M2.7"})
-    app.state.provider_registry = registry
+    runtime = ProviderRuntime(settings)
+    runtime.cache_model_ids("wafer", {"DeepSeek-V4-Pro", "MiniMax-M2.7"})
+    app.state.provider_runtime = runtime
     app.dependency_overrides[get_settings] = lambda: settings
 
     try:
@@ -148,7 +148,7 @@ def test_models_list_includes_cached_wafer_models():
     assert "claude-3-freecc-no-thinking/wafer/MiniMax-M2.7" in ids
 
 
-def test_models_list_works_without_provider_registry():
+def test_models_list_works_without_provider_runtime():
     app = create_app(lifespan_enabled=False)
     settings = _settings()
     app.dependency_overrides[get_settings] = lambda: settings

tests/contracts/test_import_boundaries.py

@@ -11,7 +11,7 @@ _API_ALLOWED_PROVIDER_MODULES = frozenset(
         "providers",
         "providers.base",
         "providers.exceptions",
-        "providers.registry",
+        "providers.runtime",
     }
 )
 
@@ -56,13 +56,27 @@ def test_core_does_not_import_product_packages() -> None:
 
 def test_provider_catalog_is_single_source_for_supported_ids() -> None:
     from config.provider_catalog import PROVIDER_CATALOG, SUPPORTED_PROVIDER_IDS
-    from providers.registry import PROVIDER_DESCRIPTORS, PROVIDER_FACTORIES
+    from providers.runtime import PROVIDER_FACTORIES
 
     assert tuple(PROVIDER_CATALOG.keys()) == SUPPORTED_PROVIDER_IDS
-    assert PROVIDER_DESCRIPTORS is PROVIDER_CATALOG
     assert set(SUPPORTED_PROVIDER_IDS) == set(PROVIDER_FACTORIES)
 
 
+def test_provider_runtime_replaces_old_registry_module() -> None:
+    repo_root = Path(__file__).resolve().parents[2]
+
+    assert not (repo_root / "providers" / "registry.py").exists()
+    assert (repo_root / "providers" / "runtime" / "runtime.py").exists()
+    assert (repo_root / "providers" / "runtime" / "factory.py").exists()
+    assert (repo_root / "providers" / "runtime" / "discovery.py").exists()
+
+    offenders = _imports_matching(
+        [repo_root / "api", repo_root / "tests", repo_root / "smoke"],
+        forbidden_prefixes=("providers.registry",),
+    )
+    assert offenders == []
+
+
 def test_config_does_not_import_non_config_packages() -> None:
     """Settings and env handling must not depend on transport or protocol layers."""
     repo_root = Path(__file__).resolve().parents[2]

tests/providers/test_model_validation.py

@@ -19,7 +19,7 @@ from providers.model_listing import ProviderModelInfo
 from providers.nvidia_nim import NvidiaNimProvider
 from providers.ollama import OllamaProvider
 from providers.open_router import OpenRouterProvider
-from providers.registry import ProviderRegistry
+from providers.runtime import ProviderRuntime
 from providers.wafer import WaferProvider
 
 
@@ -353,32 +353,34 @@ class FakeProvider(BaseProvider):
 
 
 @pytest.mark.asyncio
-async def test_registry_validation_succeeds_for_all_configured_models() -> None:
-    registry = ProviderRegistry(
+async def test_runtime_validation_succeeds_for_all_configured_models() -> None:
+    settings = _settings(model_opus="open_router/anthropic/claude-opus")
+    runtime = ProviderRuntime(
+        settings,
         {
             "nvidia_nim": FakeProvider(frozenset({"nim-model"})),
             "open_router": FakeProvider(frozenset({"anthropic/claude-opus"})),
-        }
+        },
     )
-    settings = _settings(model_opus="open_router/anthropic/claude-opus")
 
-    await registry.validate_configured_models(settings)
+    await runtime.validate_configured_models()
 
-    assert registry.cached_model_ids() == {
+    assert runtime.cached_model_ids() == {
         "nvidia_nim": frozenset({"nim-model"}),
         "open_router": frozenset({"anthropic/claude-opus"}),
     }
 
 
 @pytest.mark.asyncio
-async def test_registry_validation_reports_missing_model_with_sources() -> None:
-    registry = ProviderRegistry(
-        {"nvidia_nim": FakeProvider(frozenset({"different-model"}))}
-    )
+async def test_runtime_validation_reports_missing_model_with_sources() -> None:
     settings = _settings(model_sonnet="nvidia_nim/nim-model")
+    runtime = ProviderRuntime(
+        settings,
+        {"nvidia_nim": FakeProvider(frozenset({"different-model"}))},
+    )
 
     with pytest.raises(ServiceUnavailableError) as exc_info:
-        await registry.validate_configured_models(settings)
+        await runtime.validate_configured_models()
 
     message = exc_info.value.message
     assert "sources=MODEL,MODEL_SONNET" in message
@@ -388,19 +390,20 @@ async def test_registry_validation_reports_missing_model_with_sources() -> None:
 
 
 @pytest.mark.asyncio
-async def test_registry_validation_aggregates_multiple_failures() -> None:
-    registry = ProviderRegistry(
+async def test_runtime_validation_aggregates_multiple_failures() -> None:
+    settings = _settings(model_opus="open_router/anthropic/claude-opus")
+    runtime = ProviderRuntime(
+        settings,
         {
             "nvidia_nim": FakeProvider(frozenset({"different-model"})),
             "open_router": FakeProvider(
                 error=ModelListResponseError("bad model-list shape")
             ),
-        }
+        },
     )
-    settings = _settings(model_opus="open_router/anthropic/claude-opus")
 
     with pytest.raises(ServiceUnavailableError) as exc_info:
-        await registry.validate_configured_models(settings)
+        await runtime.validate_configured_models()
 
     message = exc_info.value.message
     assert "sources=MODEL provider=nvidia_nim model=nim-model" in message
@@ -412,10 +415,12 @@ async def test_registry_validation_aggregates_multiple_failures() -> None:
 
 
 @pytest.mark.asyncio
-async def test_registry_validation_queries_providers_concurrently() -> None:
+async def test_runtime_validation_queries_providers_concurrently() -> None:
     nim_started = asyncio.Event()
     router_started = asyncio.Event()
-    registry = ProviderRegistry(
+    settings = _settings(model_opus="open_router/anthropic/claude-opus")
+    runtime = ProviderRuntime(
+        settings,
         {
             "nvidia_nim": FakeProvider(
                 frozenset({"nim-model"}),
@@ -427,56 +432,57 @@ async def test_registry_validation_queries_providers_concurrently() -> None:
                 started=router_started,
                 peer_started=nim_started,
             ),
-        }
+        },
     )
-    settings = _settings(model_opus="open_router/anthropic/claude-opus")
 
-    await asyncio.wait_for(registry.validate_configured_models(settings), timeout=1.0)
+    await asyncio.wait_for(runtime.validate_configured_models(), timeout=1.0)
 
 
 @pytest.mark.asyncio
-async def test_registry_refresh_model_list_cache_uses_configured_remote_keys_and_referenced_local() -> (
+async def test_runtime_refresh_model_list_cache_uses_configured_remote_keys_and_referenced_local() -> (
     None
 ):
-    registry = ProviderRegistry(
+    settings = _settings(
+        model="lmstudio/local-qwen",
+        open_router_api_key="open-router-key",
+    )
+    runtime = ProviderRuntime(
+        settings,
         {
             "open_router": FakeProvider(frozenset({"anthropic/claude-sonnet"})),
             "lmstudio": FakeProvider(frozenset({"local-qwen"})),
             "ollama": FakeProvider(frozenset({"llama3.1"})),
-        }
-    )
-    settings = _settings(
-        model="lmstudio/local-qwen",
-        open_router_api_key="open-router-key",
+        },
     )
 
-    await registry.refresh_model_list_cache(settings)
+    await runtime.refresh_model_list_cache()
 
-    assert registry.cached_model_ids() == {
+    assert runtime.cached_model_ids() == {
         "open_router": frozenset({"anthropic/claude-sonnet"}),
         "lmstudio": frozenset({"local-qwen"}),
     }
 
 
 @pytest.mark.asyncio
-async def test_registry_refresh_model_list_cache_keeps_prior_cache_on_failure() -> None:
-    registry = ProviderRegistry(
-        {"nvidia_nim": FakeProvider(error=RuntimeError("upstream down"))}
-    )
-    registry.cache_model_ids("nvidia_nim", {"cached-model"})
+async def test_runtime_refresh_model_list_cache_keeps_prior_cache_on_failure() -> None:
     settings = _settings(
         model="nvidia_nim/cached-model",
         nvidia_nim_api_key="nim-key",
     )
+    runtime = ProviderRuntime(
+        settings,
+        {"nvidia_nim": FakeProvider(error=RuntimeError("upstream down"))},
+    )
+    runtime.cache_model_ids("nvidia_nim", {"cached-model"})
 
-    await registry.refresh_model_list_cache(settings)
+    await runtime.refresh_model_list_cache()
 
-    assert registry.cached_model_ids() == {"nvidia_nim": frozenset({"cached-model"})}
+    assert runtime.cached_model_ids() == {"nvidia_nim": frozenset({"cached-model"})}
 
 
-def test_registry_metadata_cache_exposes_ids_and_prefixed_infos() -> None:
-    registry = ProviderRegistry()
-    registry.cache_model_infos(
+def test_runtime_metadata_cache_exposes_ids_and_prefixed_infos() -> None:
+    runtime = ProviderRuntime(_settings())
+    runtime.cache_model_infos(
         "open_router",
         {
             ProviderModelInfo("reasoning-model", supports_thinking=True),
@@ -484,39 +490,36 @@ def test_registry_metadata_cache_exposes_ids_and_prefixed_infos() -> None:
         },
     )
 
-    assert registry.cached_model_ids() == {
+    assert runtime.cached_model_ids() == {
         "open_router": frozenset({"reasoning-model", "plain-model"})
     }
     assert (
-        registry.cached_model_supports_thinking("open_router", "reasoning-model")
-        is True
-    )
-    assert (
-        registry.cached_model_supports_thinking("open_router", "plain-model") is False
+        runtime.cached_model_supports_thinking("open_router", "reasoning-model") is True
     )
-    assert registry.cached_prefixed_model_infos() == (
+    assert runtime.cached_model_supports_thinking("open_router", "plain-model") is False
+    assert runtime.cached_prefixed_model_infos() == (
         ProviderModelInfo("open_router/plain-model", supports_thinking=False),
         ProviderModelInfo("open_router/reasoning-model", supports_thinking=True),
     )
 
 
-def test_registry_legacy_model_id_cache_keeps_unknown_thinking_support() -> None:
-    registry = ProviderRegistry()
-    registry.cache_model_ids("open_router", {"plain-model"})
+def test_runtime_model_id_cache_keeps_unknown_thinking_support() -> None:
+    runtime = ProviderRuntime(_settings())
+    runtime.cache_model_ids("open_router", {"plain-model"})
 
-    assert registry.cached_model_ids() == {"open_router": frozenset({"plain-model"})}
-    assert registry.cached_model_supports_thinking("open_router", "plain-model") is None
-    assert registry.cached_prefixed_model_infos() == (
+    assert runtime.cached_model_ids() == {"open_router": frozenset({"plain-model"})}
+    assert runtime.cached_model_supports_thinking("open_router", "plain-model") is None
+    assert runtime.cached_prefixed_model_infos() == (
         ProviderModelInfo("open_router/plain-model", supports_thinking=None),
     )
 
 
-def test_registry_cached_prefixed_model_refs_are_deterministic() -> None:
-    registry = ProviderRegistry()
-    registry.cache_model_ids("deepseek", {"deepseek-chat"})
-    registry.cache_model_ids("open_router", {"z-model", "a-model"})
+def test_runtime_cached_prefixed_model_refs_are_deterministic() -> None:
+    runtime = ProviderRuntime(_settings())
+    runtime.cache_model_ids("deepseek", {"deepseek-chat"})
+    runtime.cache_model_ids("open_router", {"z-model", "a-model"})
 
-    assert registry.cached_prefixed_model_refs() == (
+    assert runtime.cached_prefixed_model_refs() == (
         "open_router/a-model",
         "open_router/z-model",
         "deepseek/deepseek-chat",

tests/providers/{test_registry.py → test_provider_runtime.py}

@@ -22,12 +22,7 @@ from providers.nvidia_nim import NvidiaNimProvider
 from providers.ollama import OllamaProvider
 from providers.open_router import OpenRouterProvider
 from providers.opencode import OpenCodeProvider
-from providers.registry import (
-    PROVIDER_DESCRIPTORS,
-    ProviderRegistry,
-    build_provider_config,
-    create_provider,
-)
+from providers.runtime import ProviderRuntime, build_provider_config, create_provider
 from providers.wafer import WaferProvider
 from providers.zai import ZaiProvider
 
@@ -74,17 +69,19 @@ def _make_settings(**overrides):
     mock.http_write_timeout = 10.0
     mock.http_connect_timeout = 10.0
     mock.enable_model_thinking = True
+    mock.log_raw_sse_events = False
+    mock.log_api_error_tracebacks = False
     mock.nim = NimSettings()
     for key, value in overrides.items():
         setattr(mock, key, value)
     return mock
 
 
-def test_importing_registry_does_not_eager_load_other_adapters() -> None:
-    """Registry metadata must not import every provider adapter up front."""
+def test_importing_runtime_does_not_eager_load_other_adapters() -> None:
+    """Runtime metadata must not import every provider adapter up front."""
     code = (
         "import sys\n"
-        "import providers.registry\n"
+        "import providers.runtime\n"
         "assert 'providers.open_router' not in sys.modules\n"
     )
     proc = subprocess.run(
@@ -96,16 +93,16 @@ def test_importing_registry_does_not_eager_load_other_adapters() -> None:
     assert proc.returncode == 0, proc.stderr or proc.stdout
 
 
-def test_descriptors_cover_advertised_provider_ids():
-    assert set(PROVIDER_DESCRIPTORS) == set(SUPPORTED_PROVIDER_IDS)
-    for descriptor in PROVIDER_DESCRIPTORS.values():
+def test_provider_catalog_covers_advertised_provider_ids():
+    assert set(PROVIDER_CATALOG) == set(SUPPORTED_PROVIDER_IDS)
+    for descriptor in PROVIDER_CATALOG.values():
         assert descriptor.provider_id
         assert descriptor.transport_type in {"openai_chat", "anthropic_messages"}
         assert descriptor.capabilities
 
 
 def test_ollama_descriptor_uses_native_anthropic_transport():
-    descriptor = PROVIDER_DESCRIPTORS["ollama"]
+    descriptor = PROVIDER_CATALOG["ollama"]
 
     assert descriptor.transport_type == "anthropic_messages"
     assert descriptor.default_base_url == "http://localhost:11434"
@@ -113,14 +110,14 @@ def test_ollama_descriptor_uses_native_anthropic_transport():
 
 
 def test_zai_descriptor_uses_fixed_cloud_base_url():
-    descriptor = PROVIDER_DESCRIPTORS["zai"]
+    descriptor = PROVIDER_CATALOG["zai"]
 
     assert descriptor.default_base_url == ZAI_DEFAULT_BASE
     assert descriptor.base_url_attr is None
 
 
 def test_zai_provider_config_ignores_stale_base_url_setting():
-    descriptor = PROVIDER_DESCRIPTORS["zai"]
+    descriptor = PROVIDER_CATALOG["zai"]
 
     config = build_provider_config(
         descriptor,
@@ -198,13 +195,12 @@ def test_create_provider_instantiates_each_builtin():
             assert isinstance(create_provider(provider_id, settings), provider_cls)
 
 
-def test_provider_registry_caches_by_provider_id():
-    registry = ProviderRegistry()
-    settings = _make_settings()
+def test_provider_runtime_caches_by_provider_id():
+    runtime = ProviderRuntime(_make_settings())
 
     with patch("providers.transports.openai_chat.transport.AsyncOpenAI"):
-        first = registry.get("nvidia_nim", settings)
-        second = registry.get("nvidia_nim", settings)
+        first = runtime.resolve_provider("nvidia_nim")
+        second = runtime.resolve_provider("nvidia_nim")
 
     assert first is second
 
@@ -215,32 +211,34 @@ def test_unknown_provider_raises_unknown_provider_type_error():
 
 
 @pytest.mark.asyncio
-async def test_provider_registry_cleanup_runs_all_even_if_one_fails() -> None:
+async def test_provider_runtime_cleanup_runs_all_even_if_one_fails() -> None:
     """Every provider gets cleanup; cache is cleared even when one raises."""
-    reg = ProviderRegistry()
     p1 = MagicMock()
     p1.cleanup = AsyncMock(side_effect=RuntimeError("first"))
     p2 = MagicMock()
     p2.cleanup = AsyncMock()
-    reg._providers["a"] = p1
-    reg._providers["b"] = p2
+    runtime = ProviderRuntime(_make_settings(), {"a": p1, "b": p2})
+
     with pytest.raises(RuntimeError, match="first"):
-        await reg.cleanup()
+        await runtime.cleanup()
+
     p1.cleanup.assert_awaited_once()
     p2.cleanup.assert_awaited_once()
-    assert reg._providers == {}
+    assert not runtime.is_cached("a")
+    assert not runtime.is_cached("b")
 
 
 @pytest.mark.asyncio
-async def test_provider_registry_cleanup_exceptiongroup_on_multiple_failures() -> None:
-    reg = ProviderRegistry()
+async def test_provider_runtime_cleanup_exceptiongroup_on_multiple_failures() -> None:
     p1 = MagicMock()
     p1.cleanup = AsyncMock(side_effect=RuntimeError("a"))
     p2 = MagicMock()
     p2.cleanup = AsyncMock(side_effect=RuntimeError("b"))
-    reg._providers["x"] = p1
-    reg._providers["y"] = p2
+    runtime = ProviderRuntime(_make_settings(), {"x": p1, "y": p2})
+
     with pytest.raises(ExceptionGroup) as exc_info:
-        await reg.cleanup()
+        await runtime.cleanup()
+
     assert len(exc_info.value.exceptions) == 2
-    assert reg._providers == {}
+    assert not runtime.is_cached("x")
+    assert not runtime.is_cached("y")

uv.lock

@@ -561,7 +561,7 @@ wheels = [
 
 [[package]]
 name = "free-claude-code"
-version = "2.3.16"
+version = "2.3.17"
 source = { editable = "." }
 dependencies = [
     { name = "aiohttp" },