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

"""Runtime model discovery (architecture L9).

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."""


class UnknownLimits(LLMError):
    """A model in use has no declared context window. LiteLLM knows the
    limits of commercial models; for a local one they are whatever its server
    allocates, and have to be declared in LiteLLM's ``model_info``. Guessing
    is not safe: a server that silently truncates a prompt (Ollama does)
    drops its start, which is where the system prompt and the plan are."""


@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
    # The whole context window, prompt plus response, as LiteLLM declares it
    # (model_info.max_input_tokens); None when it does not say.
    max_input_tokens: int | None = None
    # The most one response may use (model_info.max_output_tokens).
    max_output_tokens: int | None = None
    # Whether its server enforces tool_choice (model_info.supports_tool_choice).
    # Only a declared True counts: a model that says nothing is not asked.
    supports_tool_choice: bool | None = None

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

    def answered(self, reported: str | None) -> bool:
        """Whether a response reporting this model name came from this model.

        A proxy fallback answers under another model's name. LiteLLM reports
        the alias; a provider may report its own name for the backing model,
        sometimes with a date suffix ("claude-haiku-4-5-20251001").
        """
        if not reported:
            return True  # nothing reported, nothing contradicted
        own_name = self.backing.partition("/")[2] or self.backing
        return reported in (self.alias, self.backing) or reported.startswith(own_name)


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"),
                max_input_tokens=_positive_int(info.get("max_input_tokens")),
                max_output_tokens=_positive_int(info.get("max_output_tokens")),
                supports_tool_choice=info.get("supports_tool_choice"),
            )
        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 require_limits(self, roles: Mapping[str, str]) -> None:
        """Refuse a role mapping that uses a model with no declared context
        window, naming every such model at once."""
        missing = {
            role: alias
            for role, alias in roles.items()
            if self.facts(alias).max_input_tokens is None
        }
        if missing:
            named = ", ".join(
                f"{role}={alias!r}" for role, alias in sorted(missing.items())
            )
            raise UnknownLimits(
                f"no context window is declared for: {named}. Set "
                "model_info.max_input_tokens (and max_output_tokens) for these "
                "in LiteLLM; docs/Tuning-Guide.md says how to find the values"
            )

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

        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 _positive_int(value: Any) -> int | None:
    if isinstance(value, bool) or not isinstance(value, int | float):
        return None
    return int(value) if value > 0 else None


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.

    L9 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 L9 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