cowork-local/domain/models/provider_descriptor.py

"""Provider catalog metadata — the domain-layer description of ONE LLM provider.

Before R03 the answer to "which providers exist, what do they cost, what can
they do?" was spread over three places: the class table in
``providers/factory.py``, the hand-maintained pricing table in
``core/routing/metadata.py`` and a handful of ``if provider == "anthropic"``
branches in the UI. :class:`ProviderDescriptor` is the single declarative
record those call sites now read from.

Layer rules (see ``docs/architecture/ADR-001-layered-architecture.md``): this
module is 100% pure Python — no PySide6, no ``requests``, no filesystem, and no
import of the concrete ``providers/*`` adapters. It only *describes* a provider;
constructing one is the infrastructure layer's job
(``infrastructure/providers/provider_registry.py``).
"""

from __future__ import annotations

from dataclasses import dataclass, field, replace
from enum import Enum
from typing import Any, Dict, Optional, Tuple


class AuthKind(str, Enum):
    """How a provider authenticates, so Settings/onboarding can ask for the
    right thing instead of hard-coding per-provider form fields.

    Inherits ``str`` so a descriptor round-trips through JSON unchanged (the
    value is written as a plain string), matching how the routing models in
    ``core/routing/models.py`` already serialize their enums.
    """

    NONE = "none"          # local runtimes (Ollama) — nothing to supply
    API_KEY = "api_key"    # bearer/x-api-key style secret
    OAUTH_TOKEN = "oauth"  # token minted by an external login flow (Copilot)


class WireProtocol(str, Enum):
    """The on-the-wire dialect a provider speaks.

    Several *distinct* providers share one protocol (Ollama, Codex, GitHub
    Copilot and generic gateways are all OpenAI Chat Completions), which is
    exactly why protocol is a separate field from the provider id: the registry
    picks the adapter class from the protocol, while everything user-facing
    keys off the id.
    """

    OPENAI_COMPAT = "openai_compat"
    ANTHROPIC = "anthropic"


@dataclass(frozen=True)
class ProviderDescriptor:
    """Immutable metadata for one provider the app can route work to.

    Frozen because descriptors are shared process-wide by the registry, the
    routing service and (eventually) the Settings screen; making them read-only
    removes any chance one caller mutates the catalog another caller is
    iterating. Use :meth:`with_models` to derive an updated copy instead.

    Unknown pricing/context values stay ``None`` rather than being guessed —
    the routing scorer needs to distinguish "free" from "we don't know", the
    same contract ``core/routing/models.py::ModelMetadata`` already follows.
    """

    provider_id: str                              # config key, e.g. "anthropic"
    display_name: str                             # human label for Settings/UI
    wire_protocol: WireProtocol                   # which adapter class implements it
    auth_kind: AuthKind = AuthKind.API_KEY
    default_model: str = ""                       # used when no model is selected
    models: Tuple[str, ...] = ()                  # known model ids (may be empty)
    max_context: Optional[int] = None             # tokens; None = unknown
    cost_per_1k_input: Optional[float] = None     # USD per 1K input tokens
    cost_per_1k_output: Optional[float] = None    # USD per 1K output tokens
    supports_vision: bool = False
    supports_tools: bool = True
    supports_streaming: bool = True
    requires_base_url: bool = False               # gateway endpoints must be configured
    # Extra ids that should resolve to this descriptor (renames/aliases kept for
    # backwards compatibility with configs written by older app versions).
    aliases: Tuple[str, ...] = ()
    # Free-form extension point so a team can attach provider-specific hints
    # without another schema migration.
    extras: Dict[str, Any] = field(default_factory=dict)

    def __post_init__(self) -> None:
        """Reject descriptors that could never be looked up.

        Raising here (rather than at registration time) means a malformed
        descriptor cannot exist at all, so every consumer downstream may assume
        ``provider_id`` is a usable dict key.
        """
        if not self.provider_id:
            raise ValueError("ProviderDescriptor.provider_id must not be empty")
        if not isinstance(self.wire_protocol, WireProtocol):
            raise TypeError("ProviderDescriptor.wire_protocol must be a WireProtocol")

    # -- identity ------------------------------------------------------- #
    @property
    def identifiers(self) -> Tuple[str, ...]:
        """Every id this descriptor answers to (canonical id first)."""
        return (self.provider_id, *self.aliases)

    def matches(self, provider_id: str) -> bool:
        """Case-insensitive id/alias match — config files and CLI flags are
        typed by humans, so lookup must not be case sensitive."""
        needle = (provider_id or "").strip().lower()
        return any(needle == known.lower() for known in self.identifiers)

    # -- capability queries --------------------------------------------- #
    def knows_model(self, model_id: str) -> bool:
        """Whether ``model_id`` is in this provider's declared catalog.

        A miss is NOT proof the model is unusable: gateways expose models we
        cannot enumerate offline, so callers treat this as a hint (used to
        resolve a bare model id back to its provider) and never as a gate that
        blocks a request.
        """
        needle = (model_id or "").strip().lower()
        return any(needle == known.strip().lower() for known in self.models)

    def has_capability(self, capability: str) -> bool:
        """Capability check by name, mirroring the vocabulary the routing
        selector already filters on (``"vision"``, ``"tools"``, ``"streaming"``)
        so a descriptor can be fed straight into ``rank_models``."""
        return capability in self.capabilities

    @property
    def capabilities(self) -> frozenset:
        """Capability set in the same vocabulary as
        ``core/routing/models.py::ModelMetadata.capabilities``."""
        caps = set()
        if self.supports_vision:
            caps.add("vision")
        if self.supports_tools:
            caps.add("tools")
        if self.supports_streaming:
            caps.add("streaming")
        return frozenset(caps)

    @property
    def avg_cost_per_1k(self) -> Optional[float]:
        """Blended input/output price, or ``None`` when either side is unknown.

        Uses the same 1:3 input:output weighting as
        ``ModelMetadata.avg_cost_per_1k`` so a descriptor and an assessment
        never disagree about what a model costs.
        """
        ci, co = self.cost_per_1k_input, self.cost_per_1k_output
        if ci is None or co is None:
            return None
        return (ci + 3.0 * co) / 4.0

    def resolve_model(self, requested: str = "") -> str:
        """The model id to actually call: the caller's choice when they made
        one, otherwise this provider's default. Centralised here because every
        surface (chat, Co4E, AI-Edit) previously re-implemented the same
        ``model or config_default`` fallback inline."""
        return (requested or "").strip() or self.default_model

    # -- derivation / serialization ------------------------------------- #
    def with_models(self, models, *, default_model: str = "") -> "ProviderDescriptor":
        """A copy carrying a freshly discovered model list.

        Providers can enumerate their models at runtime (``list_models()``);
        because the descriptor is frozen, discovery produces a NEW descriptor
        that the registry swaps in atomically instead of mutating one that other
        threads may be reading.
        """
        ordered = tuple(dict.fromkeys(m for m in models if m))  # de-dup, keep order
        chosen = default_model or self.default_model
        # Keep the default pointing at something real: fall back to the first
        # discovered model when the configured default vanished from the catalog.
        if ordered and chosen not in ordered:
            chosen = ordered[0]
        return replace(self, models=ordered, default_model=chosen)

    def to_dict(self) -> Dict[str, Any]:
        """JSON-friendly view for config persistence and the Settings UI."""
        return {
            "provider_id": self.provider_id,
            "display_name": self.display_name,
            "wire_protocol": self.wire_protocol.value,
            "auth_kind": self.auth_kind.value,
            "default_model": self.default_model,
            "models": list(self.models),
            "max_context": self.max_context,
            "cost_per_1k_input": self.cost_per_1k_input,
            "cost_per_1k_output": self.cost_per_1k_output,
            "capabilities": sorted(self.capabilities),
            "requires_base_url": self.requires_base_url,
            "aliases": list(self.aliases),
        }


__all__ = ["AuthKind", "WireProtocol", "ProviderDescriptor"]