"""ProviderRegistry - the one place a provider is declared (R03-T02). Replaces the three-way split between ``providers/factory.py::_REGISTRY``, ``config.py::DEFAULT_CONFIG["providers"]`` and ``config.py::PROVIDER_LABELS`` with a single catalogue of :class:`ProviderDescriptor` objects plus the implementation class each one maps to. Adding a provider is now one entry in :data:`BUILT_IN_PROVIDERS` (declarative facts) and one line in :data:`_IMPLEMENTATIONS` (which class speaks that protocol) - see ``docs/governance/contributor-recipes.md`` (R10-T04). Migration note (strangler fig, ADR-001 section 4): this registry does not re-implement any provider. It builds the SAME classes ``providers/factory.py`` builds, so both entry points stay behaviourally identical while call sites move over one at a time. """ from __future__ import annotations from typing import Any, Dict, Iterable, List, Mapping, Optional from cowork_local.domain.models.provider_descriptor import ( ProviderCapability, ProviderDescriptor, ) from cowork_local.providers.base import Provider, ProviderError _CAP = ProviderCapability # Every provider the app ships with, described once. # # The capability sets are deliberately conservative: a capability listed here is # one the adapter genuinely implements today. Claiming VISION for a provider # whose chat() cannot translate an image block would route an image turn into a # guaranteed failure, so an unimplemented capability must stay off the list. BUILT_IN_PROVIDERS: tuple = ( ProviderDescriptor( id="openai_compat", label="OpenAI-compatible (Internal Gateway)", protocol="openai_compat", default_model="gpt-4o-mini", capabilities=frozenset({_CAP.STREAMING, _CAP.TOOLS, _CAP.VISION, _CAP.REASONING, _CAP.MODEL_LISTING}), notes="Any endpoint speaking the OpenAI Chat Completions protocol.", ), ProviderDescriptor( id="anthropic", label="Anthropic Claude", protocol="anthropic", default_model="claude-sonnet-4-6", capabilities=frozenset({_CAP.STREAMING, _CAP.TOOLS, _CAP.VISION, _CAP.MODEL_LISTING}), ), ProviderDescriptor( id="ollama", label="Ollama (local models)", protocol="openai_compat", default_model="llama3.1", capabilities=frozenset({_CAP.STREAMING, _CAP.TOOLS, _CAP.REASONING, _CAP.MODEL_LISTING}), # Ollama ignores the key, but the OpenAI client layer requires a value, # so the default config ships a placeholder rather than an empty string. requires_api_key=False, local=True, notes="Runs on this machine - no data leaves the device, no token cost.", ), ProviderDescriptor( id="github_copilot", label="GitHub Copilot", protocol="openai_compat", default_model="gpt-4o", capabilities=frozenset({_CAP.STREAMING, _CAP.TOOLS, _CAP.MODEL_LISTING}), notes="Paste a Copilot token as the API key.", ), ProviderDescriptor( id="codex", label="OpenAI (Codex / GPT)", protocol="openai_compat", default_model="gpt-4o-mini", capabilities=frozenset({_CAP.STREAMING, _CAP.TOOLS, _CAP.VISION, _CAP.REASONING, _CAP.MODEL_LISTING}), ), ) def _implementations() -> Dict[str, type]: """Protocol -> adapter class. Imported lazily inside the function because ``providers/anthropic.py`` and ``providers/openai_compat.py`` pull in ``requests`` at import time; keeping that out of module import means a test that only inspects descriptors pays no import cost at all. """ from cowork_local.providers.anthropic import AnthropicProvider from cowork_local.providers.openai_compat import OpenAICompatProvider return { "openai_compat": OpenAICompatProvider, "anthropic": AnthropicProvider, } class ProviderRegistry: """Catalogue of known providers + the factory that instantiates them. Intentionally holds no config and no app context: it is a pure lookup table plus a build step, so it can be constructed in a test with a custom descriptor list and no application running. """ def __init__(self, descriptors: Optional[Iterable[ProviderDescriptor]] = None) -> None: # Dict preserves declaration order (Python 3.7+), which is the order # Settings lists providers in - so the catalogue order is data, not luck. self._by_id: Dict[str, ProviderDescriptor] = { d.id: d for d in (descriptors if descriptors is not None else BUILT_IN_PROVIDERS) } # -- catalogue queries ------------------------------------------------ # def ids(self) -> List[str]: """Known provider ids, in declaration order.""" return list(self._by_id) def all(self) -> List[ProviderDescriptor]: """Every descriptor, in declaration order.""" return list(self._by_id.values()) def get(self, provider_id: str) -> Optional[ProviderDescriptor]: """The descriptor for ``provider_id``, or None when unknown. Returns None rather than raising because the caller is often reacting to a config file that may name a provider from a newer version; the UI should be able to skip it, not crash. """ return self._by_id.get(provider_id) def require(self, provider_id: str) -> ProviderDescriptor: """Like :meth:`get` but raises :class:`ProviderError` when unknown. Same error type ``providers/factory.py::build_provider`` already raises, so callers that migrate to the registry keep their existing except clause. """ descriptor = self._by_id.get(provider_id) if descriptor is None: known = ", ".join(self._by_id) or "(none)" raise ProviderError(f"Unsupported provider: {provider_id} (known: {known})") return descriptor def labels(self) -> Dict[str, str]: """``{id: label}`` - the drop-in replacement for ``config.PROVIDER_LABELS``.""" return {d.id: d.label for d in self._by_id.values()} def supporting(self, capability: ProviderCapability) -> List[ProviderDescriptor]: """Every descriptor advertising ``capability`` - used to answer "which providers could serve this turn?" before any of them is built.""" return [d for d in self._by_id.values() if d.supports(capability)] def configured(self, providers_conf: Mapping[str, Mapping[str, Any]] ) -> List[ProviderDescriptor]: """Descriptors whose config section is complete enough to actually call. ``providers_conf`` is ``AppConfig.data["providers"]``. Passing the raw mapping (not the AppConfig object) keeps this layer independent of the config implementation, which EPIC R02 is rewriting in parallel. """ return [d for d in self._by_id.values() if d.is_configured(providers_conf.get(d.id, {}) or {})] # -- construction ----------------------------------------------------- # def build(self, provider_id: str, conf: Mapping[str, Any], model: str = "") -> Provider: """Instantiate the adapter for ``provider_id``. ``model`` overrides the configured model for this instance only - that is how the routing layer runs one turn on a different model without mutating the user's saved settings. """ descriptor = self.require(provider_id) impl = _implementations().get(descriptor.protocol) if impl is None: # pragma: no cover - only reachable via a bad descriptor raise ProviderError( f"Provider '{provider_id}' declares unknown protocol " f"'{descriptor.protocol}'." ) # Copy before mutating: conf is the caller's live config dict, and # writing the routed model into it would silently change the user's # saved default for every later turn. resolved = dict(conf or {}) resolved["model"] = descriptor.resolve_model(conf, model) instance = impl(resolved) # The adapter class is shared by several ids (three of them are # OpenAI-compatible), so its class-level `name` cannot identify which # provider this is. Stamping the instance keeps usage records, audit # entries and routing candidate keys attributed to the right provider. instance.name = descriptor.id return instance def describe(self, provider_id: str, conf: Optional[Mapping[str, Any]] = None) -> str: """One-line description used in logs and error messages.""" return self.require(provider_id).describe(conf) # Shared default instance. Callers that need the built-in catalogue use this # instead of constructing a registry each time; tests build their own with an # explicit descriptor list. default_registry = ProviderRegistry() __all__ = ["ProviderRegistry", "BUILT_IN_PROVIDERS", "default_registry"]