Discover a model's vendor from LiteLLM instead of its name #35

Merged
dongar merged 1 commits from alias-discovery into main 2026-09-23 01:49:36 +00:00
10 changed files with 574 additions and 76 deletions

No files matched your search

+37 -6
View File
@@ -12,6 +12,7 @@ from collections.abc import Callable
from dataclasses import dataclass
from nisse_dev.approvals import PlanBoard
from nisse_dev.catalog import ModelCatalog, select_client
from nisse_dev.config import AgentConfig
from nisse_dev.control import Orchestrator
from nisse_dev.delegation import Delegation
@@ -19,11 +20,7 @@ from nisse_dev.executor import LLMExecutor
from nisse_dev.expert import ExpertDesk
from nisse_dev.git import GitRepo, authenticated_url
from nisse_dev.gitea import Comment, GiteaClient
from nisse_dev.llm import (
AnthropicPassthroughClient,
LiteLLMClient,
select_client,
)
from nisse_dev.llm import AnthropicPassthroughClient, LiteLLMClient
from nisse_dev.matrix import MatrixTransport
from nisse_dev.oplog import OpLog
from nisse_dev.planner import LLMPlanningAgent
@@ -32,6 +29,35 @@ from nisse_dev.publish import MilestonePublisher
from nisse_dev.state import StateStore
from nisse_dev.tools import Workspace
TOOL_DRIVING_ROLES = ("planner", "executor", "delegate")
def discover_models(config: AgentConfig, oplog: OpLog) -> ModelCatalog:
"""Resolve every configured role against LiteLLM before anything runs.
Refusing to boot is the whole point: a role pointed at a model this key
cannot see is a misconfiguration, and the cheapest place to learn that is
here rather than ten minutes into a milestone. An unreachable proxy is
reported separately -- it is an outage, not a mistake, and telling them
apart is what lets an operator know whether to edit config or wait.
"""
catalog = ModelCatalog.discover(config.llm_base_url, config.llm_token)
catalog.validate(config.roles.pinned())
for role in TOOL_DRIVING_ROLES:
facts = catalog.facts(config.roles.resolve(role))
if facts.supports_tools is False:
# A warning, not a refusal: models LiteLLM declares tool-less are
# observed driving tool loops anyway, and A15 expects weak local
# models to be accommodated rather than excluded.
oplog.emit(
"model_tools_unsupported",
cause="startup",
role=role,
model=facts.alias,
backing=facts.backing,
)
return catalog
@dataclass
class Agent:
@@ -91,6 +117,9 @@ def build_agent(
) -> Agent:
oplog = OpLog("nisse-dev")
store = StateStore(config.state_db, oplog=oplog)
# Before the clone, before the token checks: a typo'd role name should
# cost a second, not a repository.
catalog = discover_models(config, oplog)
gitea = GiteaClient(config.gitea_url, config.gitea_token)
# Whoami rather than another config variable: it names the identity every
@@ -113,11 +142,12 @@ def build_agent(
expert = ExpertDesk(normalized, store, config.roles)
delegate_model = config.roles.resolve("delegate")
delegation = Delegation(
select_client(delegate_model, normalized, passthrough),
select_client(delegate_model, normalized, passthrough, catalog),
store,
workspace,
git_repo,
model=delegate_model,
backing_model=catalog.facts(delegate_model).backing,
)
executor = LLMExecutor(
normalized,
@@ -162,6 +192,7 @@ def build_agent(
board=PlanBoard(gitea, config.gitea_repo),
delegate=delegation,
poller_factory=poller_factory,
catalog=catalog,
)
return Agent(
config=config,
+5 -1
View File
@@ -56,8 +56,12 @@ class GateOpened:
class DelegationStarted:
brief: str
role: str
model: str
model: str # the alias asked for, e.g. "remote-architecture"
caps: dict[str, Any] # turns / tokens / wall-clock limits
# What the alias resolved to at task start. An alias is free to change
# models between tasks, so eval history (A16) keyed on the alias alone
# would average results across whatever it happened to point at.
backing_model: str = ""
@dataclass(frozen=True)
+172
View File
@@ -0,0 +1,172 @@
"""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.from_env`` batches its
missing variables).
"""
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
+17 -1
View File
@@ -143,6 +143,10 @@ def _fresh_data() -> dict[str, Any]:
"milestone": 1,
"task_id": None,
"roles": None, # A11 mapping, pinned at task start
# What each pinned alias resolved to, discovered from LiteLLM. An
# alias is free to point somewhere else tomorrow, so the transcript
# has to say what actually answered (A11, A16).
"models": None,
"pr": None, # the open PR for the current milestone, once published
# Pointer to the Gitea plan record (Part H): the state store holds
# only this, so gate state can be re-derived from Gitea after a
@@ -165,11 +169,13 @@ class Orchestrator:
board: "PlanBoard | None" = None,
delegate: Any = None,
poller_factory: Any = None,
catalog: Any = None,
) -> None:
self._store = store
self._transport = transport
self._agent = agent
self._roles = roles
self._catalog = catalog
self._executor = executor
self._publisher = publisher
self._operators = set(operators)
@@ -287,7 +293,17 @@ class Orchestrator:
# Pin the role mapping now (A11): a mid-task config change must not
# split one task across two planners.
pinned = self._roles.pinned() if self._roles is not None else None
self._data = {**_fresh_data(), "task_id": task_id, "roles": pinned}
resolved = (
self._catalog.pinned(pinned)
if self._catalog is not None and pinned is not None
else None
)
self._data = {
**_fresh_data(),
"task_id": task_id,
"roles": pinned,
"models": resolved,
}
self._project = project
self._store.oplog.task_id = task_id
if self._board is not None:
+3
View File
@@ -165,6 +165,7 @@ class Delegation:
git_repo: GitRepo,
*,
model: str,
backing_model: str = "",
caps: Caps | None = None,
now: Callable[[], float] = time.monotonic,
) -> None:
@@ -173,6 +174,7 @@ class Delegation:
self._workspace = workspace
self._git = git_repo
self._model = model
self._backing_model = backing_model
self._caps = caps or Caps()
self._now = now
@@ -210,6 +212,7 @@ class Delegation:
role=role,
model=self._model,
caps=self._caps.as_payload(),
backing_model=self._backing_model,
)
)
result = self._loop(brief, branch, started_at)
-22
View File
@@ -371,28 +371,6 @@ def _post_with_rate_limit_retry(
# Model families we have a native renderer adapter for. A13: delegations take
# the vendor's passthrough route because a normalising middlebox is lossy
# exactly where a long-running delegation lives.
PASSTHROUGH_PREFIXES = ("claude-", "anthropic/")
def has_native_adapter(model: str) -> bool:
return model.startswith(PASSTHROUGH_PREFIXES)
def select_client(
model: str, normalized: LLMClient, passthrough: LLMClient
) -> LLMClient:
"""Choose the route from the model, not from where the caller sits.
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. Hard-wiring the passthrough would mean pointing a role at a
non-Anthropic model silently sent it Anthropic-shaped request bodies,
which is the opposite of the vendor fluidity A5 requires.
"""
return passthrough if has_native_adapter(model) else normalized
def _raise_for_upstream(response: httpx.Response) -> None:
"""Classify a non-200 as upstream trouble where it plainly is.
+111 -1
View File
@@ -11,9 +11,28 @@ import httpx
import pytest
from test_config import COMPLETE
from nisse_dev.agent import build_agent, prepare_workspace
from nisse_dev.agent import build_agent, discover_models, prepare_workspace
from nisse_dev.catalog import ModelCatalog, UnknownModel
from nisse_dev.config import AgentConfig
from nisse_dev.git import GitRepo
from nisse_dev.oplog import OpLog
# What the test roles resolve to, as LiteLLM would report them.
TEST_CATALOG = ModelCatalog.from_payload(
{
"data": [
{
"model_name": alias,
"litellm_params": {"model": f"anthropic/{alias}"},
"model_info": {
"litellm_provider": "anthropic",
"supports_function_calling": True,
},
}
for alias in ("claude-opus-5", "claude-haiku-4-5")
]
}
)
def config_for(tmp_path: Path, **overrides: str) -> AgentConfig:
@@ -94,6 +113,10 @@ def build_offline(tmp_path: Path, origin: Path, monkeypatch: Any) -> Any:
monkeypatch.setattr(
"nisse_dev.agent.GiteaClient.whoami", lambda self: "nisse-dev-testing"
)
monkeypatch.setattr(
"nisse_dev.agent.ModelCatalog.discover",
classmethod(lambda cls, *a, **kw: TEST_CATALOG),
)
monkeypatch.setattr("nisse_dev.agent.MatrixTransport", lambda cfg: FakeTransport())
monkeypatch.setattr(
"nisse_dev.agent.authenticated_url",
@@ -177,3 +200,90 @@ def test_httpx_clients_are_not_shared_between_routes(
assert isinstance(normalized, httpx.Client)
assert normalized is not passthrough # different base URLs and headers
agent.stop()
# --- startup model discovery (A5/A11) ------------------------------------
def test_a_typod_role_refuses_to_boot(tmp_path: Path, monkeypatch) -> None:
# Misconfiguration should cost a second at startup, not be discovered ten
# minutes into a milestone.
config = config_for(tmp_path, NISSE_ROLE_PLANNER="clause-opus-5")
monkeypatch.setattr(
"nisse_dev.agent.ModelCatalog.discover",
classmethod(lambda cls, *a, **kw: TEST_CATALOG),
)
with pytest.raises(UnknownModel, match="planner='clause-opus-5'"):
discover_models(config, OpLog("test"))
def test_a_role_this_key_cannot_reach_refuses_to_boot(
tmp_path: Path, monkeypatch
) -> None:
# /model/info is scoped per key, so the same check catches a role pointed
# at a real model that this token has no access to.
config = config_for(tmp_path, NISSE_ROLE_EXECUTOR="local-fast")
monkeypatch.setattr(
"nisse_dev.agent.ModelCatalog.discover",
classmethod(lambda cls, *a, **kw: TEST_CATALOG),
)
with pytest.raises(UnknownModel, match="executor='local-fast'"):
discover_models(config, OpLog("test"))
def test_a_tool_less_model_warns_but_still_boots(tmp_path: Path, monkeypatch) -> None:
# A15 expects weak local models to be accommodated, and models LiteLLM
# declares tool-less are observed driving tool loops anyway.
import io
tool_less = ModelCatalog.from_payload(
{
"data": [
{
"model_name": alias,
"litellm_params": {"model": "openai/qwen3-8b-awq"},
"model_info": {
"litellm_provider": "openai",
"supports_function_calling": False,
},
}
for alias in ("claude-opus-5", "claude-haiku-4-5")
]
}
)
monkeypatch.setattr(
"nisse_dev.agent.ModelCatalog.discover",
classmethod(lambda cls, *a, **kw: tool_less),
)
stream = io.StringIO()
discover_models(config_for(tmp_path), OpLog("test", stream))
logged = stream.getvalue()
assert "model_tools_unsupported" in logged
assert '"role": "executor"' in logged
assert '"role": "consultant"' not in logged # ask_expert drives no tools
def test_the_resolved_model_is_pinned_at_task_start(
tmp_path: Path, origin: Path, monkeypatch
) -> None:
agent = build_offline(tmp_path, origin, monkeypatch)
agent.orchestrator._begin_project("greet")
pinned = agent.orchestrator._data["models"]
assert pinned["planner"] == {
"alias": "claude-opus-5",
"backing": "anthropic/claude-opus-5",
"provider": "anthropic",
}
agent.stop()
def test_delegation_records_what_the_alias_resolved_to(
tmp_path: Path, origin: Path, monkeypatch
) -> None:
# A16 eval history keyed on an alias alone would average across whatever
# that alias pointed at over time.
agent = build_offline(tmp_path, origin, monkeypatch)
assert agent.orchestrator._delegate._backing_model == "anthropic/claude-opus-5"
agent.stop()
+202
View File
@@ -0,0 +1,202 @@
"""Runtime model discovery (spec A5/A11/A13).
The point of these tests is that nothing here reads the model's *name*. A
semantic alias like "remote-architecture" is free to point anywhere, so the
route has to come from what LiteLLM says the model is.
"""
import httpx
import pytest
from nisse_dev.catalog import (
CatalogUnavailable,
ModelCatalog,
UnknownModel,
select_client,
)
def entry(alias: str, backing: str, provider: str | None, tools: bool | None = None):
"""One /model/info row, shaped as the live proxy returns it."""
info: dict = {"supports_function_calling": tools}
if provider is not None:
info["litellm_provider"] = provider
return {
"model_name": alias,
"litellm_params": {"model": backing, "api_base": "http://proxy.test/v1"},
"model_info": info,
}
def catalog(*entries) -> ModelCatalog:
return ModelCatalog.from_payload({"data": list(entries)})
# --- parsing -------------------------------------------------------------
def test_facts_are_read_from_the_payload() -> None:
cat = catalog(
entry("claude-haiku-4-5", "anthropic/claude-haiku-4-5", "anthropic", True)
)
facts = cat.facts("claude-haiku-4-5")
assert facts.backing == "anthropic/claude-haiku-4-5"
assert facts.provider == "anthropic"
assert facts.supports_tools is True
assert facts.native_anthropic
def test_provider_falls_back_to_the_backing_prefix() -> None:
# litellm_provider is absent on some deployments; the backing model's
# own "vendor/model" prefix says the same thing.
cat = catalog(entry("mystery", "anthropic/claude-opus-5", None))
assert cat.facts("mystery").provider == "anthropic"
def test_unnamed_entries_are_skipped() -> None:
cat = ModelCatalog.from_payload({"data": [{"litellm_params": {"model": "x"}}]})
assert cat.aliases() == []
def test_missing_tool_support_is_unknown_not_false() -> None:
# LiteLLM reports null for models it has no capability data on. That is
# not the same as declaring the model tool-less, and the startup warning
# depends on the difference.
cat = catalog(entry("local-fast", "openai/qwen3-8b-awq", "openai", None))
assert cat.facts("local-fast").supports_tools is None
# --- route selection by provider, never by name --------------------------
def test_a_semantic_alias_routes_by_its_provider() -> None:
# The whole feature: nothing in "remote-architecture" says Anthropic.
cat = catalog(entry("remote-architecture", "anthropic/claude-opus-5", "anthropic"))
normalized, passthrough = object(), object()
chosen = select_client("remote-architecture", normalized, passthrough, cat)
assert chosen is passthrough
def test_the_same_alias_reroutes_when_it_is_repointed() -> None:
# A13: provisionally accept LiteLLM's normalization for a vendor with no
# native adapter, rather than sending it Anthropic-shaped bodies.
cat = catalog(entry("remote-architecture", "openai/gpt-6-astra", "openai"))
normalized, passthrough = object(), object()
assert (
select_client("remote-architecture", normalized, passthrough, cat) is normalized
)
def test_a_name_that_merely_mentions_a_vendor_is_not_that_vendor() -> None:
cat = catalog(entry("claude-clone-9", "openai/some-clone", "openai"))
normalized, passthrough = object(), object()
assert select_client("claude-clone-9", normalized, passthrough, cat) is normalized
def test_local_models_take_the_normalized_route() -> None:
cat = catalog(
entry("local-coder", "openai/qwen3-14b", "openai"),
entry("local-fast", "openai/qwen3-8b-awq", "openai"),
)
normalized, passthrough = object(), object()
for alias in ("local-coder", "local-fast"):
assert select_client(alias, normalized, passthrough, cat) is normalized
# --- misconfiguration is loud --------------------------------------------
def test_an_unknown_model_raises_rather_than_guessing_a_route() -> None:
cat = catalog(entry("local-fast", "openai/qwen3-8b-awq", "openai"))
with pytest.raises(UnknownModel, match="local-fsat"):
select_client("local-fsat", object(), object(), cat)
def test_the_error_lists_what_is_available() -> None:
cat = catalog(
entry("local-fast", "openai/qwen3-8b-awq", "openai"),
entry("claude-opus-5", "anthropic/claude-opus-5", "anthropic"),
)
with pytest.raises(UnknownModel, match="claude-opus-5, local-fast"):
cat.facts("typo")
def test_validate_names_every_bad_role_at_once() -> None:
# One boot should surface the whole list, not make the operator
# rediscover the next typo on the next start.
cat = catalog(entry("local-fast", "openai/qwen3-8b-awq", "openai"))
with pytest.raises(UnknownModel) as caught:
cat.validate(
{
"executor": "local-fast",
"planner": "clause-opus-5",
"delegate": "remote-architecture",
"consultant": "local-fast",
}
)
message = str(caught.value)
assert "planner='clause-opus-5'" in message
assert "delegate='remote-architecture'" in message
assert "executor" not in message # the good ones are not noise
def test_validate_passes_a_whole_good_mapping() -> None:
cat = catalog(
entry("local-fast", "openai/qwen3-8b-awq", "openai"),
entry("claude-opus-5", "anthropic/claude-opus-5", "anthropic"),
)
cat.validate({"executor": "local-fast", "planner": "claude-opus-5"})
# --- an outage is not a misconfiguration ---------------------------------
def test_an_unreachable_proxy_is_reported_as_an_outage() -> None:
def handler(request: httpx.Request) -> httpx.Response:
raise httpx.ConnectError("no route to host")
with pytest.raises(CatalogUnavailable, match="could not reach"):
ModelCatalog.discover(
"http://litellm.test", "sk-test", transport=httpx.MockTransport(handler)
)
def test_a_refused_listing_is_reported_as_an_outage() -> None:
cat = httpx.MockTransport(lambda request: httpx.Response(401, text="bad key"))
with pytest.raises(CatalogUnavailable, match="401"):
ModelCatalog.discover("http://litellm.test", "sk-test", transport=cat)
def test_discovery_asks_the_proxy_with_the_token() -> None:
seen: dict = {}
def handler(request: httpx.Request) -> httpx.Response:
seen["url"] = str(request.url)
seen["auth"] = request.headers.get("Authorization")
return httpx.Response(
200, json={"data": [entry("local-fast", "openai/qwen3-8b-awq", "openai")]}
)
cat = ModelCatalog.discover(
"http://litellm.test/", "sk-test", transport=httpx.MockTransport(handler)
)
assert seen["url"] == "http://litellm.test/model/info"
assert seen["auth"] == "Bearer sk-test"
assert cat.aliases() == ["local-fast"]
# --- what gets pinned ----------------------------------------------------
def test_pinning_records_what_the_alias_resolved_to() -> None:
# An alias may point somewhere else tomorrow; eval history (A16) keyed on
# the alias alone would average across whatever it pointed at.
cat = catalog(entry("remote-architecture", "anthropic/claude-opus-5", "anthropic"))
pinned = cat.pinned({"planner": "remote-architecture"})
assert pinned == {
"planner": {
"alias": "remote-architecture",
"backing": "anthropic/claude-opus-5",
"provider": "anthropic",
}
}
-27
View File
@@ -21,9 +21,7 @@ from nisse_dev.llm import (
RoleMap,
TruncatedResponse,
UpstreamUnavailable,
has_native_adapter,
max_output_tokens,
select_client,
)
EVENTS = [StoredEvent(id=1, ts="t", event=UserMessage(text="hi", source="m:1"))]
@@ -218,31 +216,6 @@ def test_load_env_file_missing_is_noop(tmp_path: Path) -> None:
assert env == {}
# --- route selection (A5 vendor fluidity) --------------------------------
def test_anthropic_models_take_the_native_route() -> None:
normalized, passthrough = object(), object()
for model in ("claude-haiku-4-5", "claude-opus-5", "anthropic/claude-sonnet-5"):
assert select_client(model, normalized, passthrough) is passthrough
def test_other_vendors_take_the_normalized_route() -> None:
# A13: provisionally accept LiteLLM's normalization for a vendor whose
# renderer adapter does not exist yet, rather than sending it request
# bodies shaped for someone else.
normalized, passthrough = object(), object()
for model in ("gpt-6-astra", "local-coder", "local-fast", "llama3.2"):
assert select_client(model, normalized, passthrough) is normalized
def test_native_adapter_detection_is_explicit() -> None:
assert has_native_adapter("claude-opus-5")
assert not has_native_adapter("gpt-6-astra")
# A model merely mentioning the vendor is not the vendor.
assert not has_native_adapter("my-claude-clone")
# --- truncation (silent until measured against Opus) ---------------------
+27 -18
View File
@@ -25,14 +25,10 @@ from nisse_dev.canonical import (
ToolResult,
UserMessage,
)
from nisse_dev.catalog import ModelCatalog, select_client
from nisse_dev.envfile import load_env_file
from nisse_dev.expert import parse_answer
from nisse_dev.llm import (
AnthropicPassthroughClient,
LiteLLMClient,
LLMError,
select_client,
)
from nisse_dev.llm import AnthropicPassthroughClient, LiteLLMClient, LLMError
from nisse_dev.planner import PLANNER_TOOLS
from nisse_dev.tools import EXECUTOR_TOOLS
@@ -53,14 +49,23 @@ def clients() -> tuple[LiteLLMClient, AnthropicPassthroughClient]:
return LiteLLMClient.from_env(), AnthropicPassthroughClient.from_env()
@pytest.fixture(scope="module")
def catalog() -> ModelCatalog:
"""The live catalog this key can see. /model/info is scoped per key, so
the probe list and the key have to agree."""
return ModelCatalog.discover(
os.environ["NISSE_LLM_BASE_URL"], os.environ["NISSE_LLM_TOKEN"]
)
def user(text: str) -> list[StoredEvent]:
return [StoredEvent(id=1, ts="t", event=UserMessage(text=text, source="probe"))]
@pytest.mark.parametrize("model", MODELS)
def test_model_answers_at_all(model: str) -> None:
def test_model_answers_at_all(model: str, catalog: ModelCatalog) -> None:
normalized, passthrough = clients()
client = select_client(model, normalized, passthrough)
client = select_client(model, normalized, passthrough, catalog)
produced = client.complete(
user("Reply with exactly one word: pong"), None, model, system="Be terse."
)
@@ -69,11 +74,11 @@ def test_model_answers_at_all(model: str) -> None:
@pytest.mark.parametrize("model", MODELS)
def test_model_holds_the_planner_contract(model: str) -> None:
def test_model_holds_the_planner_contract(model: str, catalog: ModelCatalog) -> None:
"""A4's gating is structural at the tool layer: a model that answers in
prose cannot be gated, because there is no plan to approve."""
normalized, passthrough = clients()
client = select_client(model, normalized, passthrough)
client = select_client(model, normalized, passthrough, catalog)
produced = client.complete(
user("Draft an architecture plan for a command-line slug generator."),
PLANNER_TOOLS,
@@ -96,11 +101,11 @@ def test_model_holds_the_planner_contract(model: str) -> None:
@pytest.mark.parametrize("model", MODELS)
def test_model_round_trips_a_tool_result(model: str) -> None:
def test_model_round_trips_a_tool_result(model: str, catalog: ModelCatalog) -> None:
"""The executor loop feeds results back; a model that cannot consume its
own tool result cannot run a milestone."""
normalized, passthrough = clients()
client = select_client(model, normalized, passthrough)
client = select_client(model, normalized, passthrough, catalog)
events = user("List the files in the current directory using your tools.")
produced = client.complete(events, EXECUTOR_TOOLS, model, system="Use a tool.")
calls = [e for e in produced if isinstance(e, ToolCall)]
@@ -124,11 +129,13 @@ def test_model_round_trips_a_tool_result(model: str) -> None:
@pytest.mark.parametrize("model", MODELS)
def test_model_holds_the_prompt_level_json_contract(model: str) -> None:
def test_model_holds_the_prompt_level_json_contract(
model: str, catalog: ModelCatalog
) -> None:
"""A14 contracts structured output at the prompt level because vendors
differ on JSON enforcement. This measures whether that holds per model."""
normalized, passthrough = clients()
client = select_client(model, normalized, passthrough)
client = select_client(model, normalized, passthrough, catalog)
produced = client.complete(
user("Is SQLite sufficient for a single-process agent's task state?"),
None,
@@ -146,12 +153,14 @@ def test_model_holds_the_prompt_level_json_contract(model: str) -> None:
@pytest.mark.parametrize("model", MODELS)
def test_route_selection_matches_the_model(model: str) -> None:
def test_route_selection_matches_the_model(model: str, catalog: ModelCatalog) -> None:
"""A non-Anthropic model must not be sent Anthropic-shaped bodies. This
was hardcoded until gpt-6-astra made it observable."""
was hardcoded until gpt-6-astra made it observable, and name-sniffed until
semantic aliases made the name stop meaning anything."""
normalized, passthrough = clients()
chosen = select_client(model, normalized, passthrough)
if model.startswith(("claude-", "anthropic/")):
facts = catalog.facts(model)
chosen = select_client(model, normalized, passthrough, catalog)
if facts.provider == "anthropic":
assert chosen is passthrough
else:
assert chosen is normalized