nisse/nisse-dev/src/nisse_dev/catalog.py

"""Runtime model discovery (spec A5, A11, A13).

LiteLLM is the VM's single LLM destination, which makes it the authority on
what any given model *is*. Asking it at startup keeps vendor knowledge in one
place instead of duplicating it as name patterns in agent code -- and that
stops being a tidiness argument the moment a role alias no longer names its
vendor. An alias like ``remote-architecture`` is deliberately free to point at
Anthropic today and something else tomorrow; nothing in the name can tell us
which request shape the model behind it expects.

Discovery is per key, not global: ``/model/info`` returns only the models the
presented token is allowed to use. So the same check that catches a typo'd
alias also catches a role pointed at a model this key cannot reach.
"""

from collections.abc import Mapping
from dataclasses import dataclass
from typing import Any

import httpx

from nisse_dev.llm import LLMClient, LLMError

# LiteLLM's provider name for models we have a vendor-native adapter for.
ANTHROPIC_PROVIDER = "anthropic"


class CatalogUnavailable(LLMError):
    """The catalog could not be fetched: the proxy is down, unreachable, or
    refused the token. An outage, not a mistake in the configuration."""


class UnknownModel(LLMError):
    """A model was named that this key's catalog does not contain: a typo, a
    stale alias, or a key without access. Always a misconfiguration."""


@dataclass(frozen=True)
class ModelFacts:
    alias: str  # what we ask for, e.g. "remote-architecture"
    backing: str  # what LiteLLM calls, e.g. "anthropic/claude-opus-5"
    provider: str  # "anthropic", "openai", "ollama", ...
    supports_tools: bool | None  # None when LiteLLM does not say

    @property
    def native_anthropic(self) -> bool:
        return self.provider == ANTHROPIC_PROVIDER


class ModelCatalog:
    """What one LiteLLM key can see, as of the moment it was asked."""

    def __init__(self, facts: Mapping[str, ModelFacts]) -> None:
        self._facts = dict(facts)

    @classmethod
    def from_payload(cls, payload: Mapping[str, Any]) -> "ModelCatalog":
        facts: dict[str, ModelFacts] = {}
        for entry in payload.get("data") or []:
            alias = entry.get("model_name")
            if not alias:
                continue
            params = entry.get("litellm_params") or {}
            info = entry.get("model_info") or {}
            backing = str(params.get("model") or "")
            # litellm_provider is the direct answer; the backing model's
            # "vendor/model" prefix says the same thing and is present even
            # when the provider field is not.
            provider = info.get("litellm_provider") or backing.partition("/")[0]
            facts[alias] = ModelFacts(
                alias=alias,
                backing=backing,
                provider=str(provider),
                supports_tools=info.get("supports_function_calling"),
            )
        return cls(facts)

    @classmethod
    def discover(
        cls,
        base_url: str,
        token: str,
        *,
        timeout_s: float = 30.0,
        transport: httpx.BaseTransport | None = None,
    ) -> "ModelCatalog":
        try:
            with httpx.Client(
                base_url=base_url.rstrip("/"),
                headers={"Authorization": f"Bearer {token}"},
                timeout=timeout_s,
                transport=transport,
            ) as http:
                response = http.get("/model/info")
        except httpx.HTTPError as exc:
            raise CatalogUnavailable(
                f"could not reach the LiteLLM proxy at {base_url} to discover "
                f"models: {exc}"
            ) from exc
        if response.status_code != 200:
            raise CatalogUnavailable(
                f"the LiteLLM proxy refused a model listing with "
                f"{response.status_code}: {response.text[:300]}"
            )
        return cls.from_payload(response.json())

    def aliases(self) -> list[str]:
        return sorted(self._facts)

    def facts(self, alias: str) -> ModelFacts:
        try:
            return self._facts[alias]
        except KeyError:
            raise UnknownModel(
                f"{alias!r} is not a model this LiteLLM key can see; "
                f"available: {', '.join(self.aliases()) or '(none)'}"
            ) from None

    def validate(self, roles: Mapping[str, str]) -> None:
        """Check a whole role mapping at once, before anything uses it.

        Reported together rather than one per boot: an operator fixing a
        config wants the whole list, not to rediscover the next typo on the
        next start (the same reason ``AgentConfig`` reports every missing
        setting at once).
        """
        unknown = {
            role: alias for role, alias in roles.items() if alias not in self._facts
        }
        if unknown:
            named = ", ".join(
                f"{role}={alias!r}" for role, alias in sorted(unknown.items())
            )
            raise UnknownModel(
                f"role(s) point at models this LiteLLM key cannot see: {named}; "
                f"available: {', '.join(self.aliases()) or '(none)'}"
            )

    def pinned(self, roles: Mapping[str, str]) -> dict[str, dict[str, Any]]:
        """The resolved mapping to record at task start (A11, A16).

        Recording only the alias would let an alias that silently changes
        models muddle the eval history behind it: "remote-architecture: 80%
        good" averaged over however many models it pointed at is not a fact
        about anything.
        """
        return {
            role: {
                "alias": alias,
                "backing": self.facts(alias).backing,
                "provider": self.facts(alias).provider,
            }
            for role, alias in roles.items()
        }


def select_client(
    model: str,
    normalized: LLMClient,
    passthrough: LLMClient,
    catalog: ModelCatalog,
) -> LLMClient:
    """Choose the route from what the model *is*, not from what it is called.

    A13 spells out the rule: use the vendor-native adapter where one exists,
    and otherwise provisionally accept LiteLLM's normalization for that
    vendor -- validating it against the A16 eval set before trusting it with
    delegations. There is deliberately no name-based fallback here: a second
    heuristic could disagree with the proxy, and an unknown model raises
    rather than quietly taking the normalized route.
    """
    return passthrough if catalog.facts(model).native_anthropic else normalized