Discover a model's vendor from LiteLLM instead of its name
CI / checks (push) Skipped
CI / checks (pull_request) Successful in 47s

Route selection sniffed model names: anything starting "claude-" or
"anthropic/" took the vendor-native passthrough, everything else took the
normalized route. That works only while an alias names its vendor. A
semantic role alias like "remote-architecture" is meant to point at
Anthropic today and somewhere else tomorrow, and the name says nothing
either way -- so it would have taken the normalized route regardless of
what was behind it, silently losing the fidelity A13 wants for delegations.

LiteLLM already knows the answer. /model/info reports each alias's backing
model, provider, and declared tool support, so the agent asks at startup and
routes on the provider it discovers. That keeps vendor knowledge where A5
already put it -- in the proxy -- rather than duplicating it as name patterns
here.

There is deliberately no name-based fallback: PASSTHROUGH_PREFIXES and
has_native_adapter are gone, and select_client now requires a catalog. A
second heuristic could only ever disagree with the proxy. An unknown model
raises instead of guessing a route.

Misconfiguration is therefore loud and early. build_agent validates every
role against the catalog before it clones anything, naming all the bad ones
at once. Discovery is per key -- /model/info returns only what the presented
token may use -- so the same check catches a role pointed at a real model
this key cannot reach. An unreachable proxy is reported separately, since an
outage and a typo call for different responses from the operator.

Two things now get recorded rather than inferred: the resolved model is
pinned alongside the role mapping at task start, for the reason A11 already
pins roles, and DelegationStarted carries backing_model so A16 eval history
is not averaged across whatever an alias happened to point at.

A model LiteLLM declares tool-less warns at startup rather than refusing;
A15 expects weak local models to be accommodated, and they are observed
driving tool loops anyway.

Verified live: all four aliases the production key can see route correctly
through discovery, and each chosen route works.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
dongarandClaude Opus 5 committed 2026-09-23 01:47:58 +00:00
1 parent 231b406393
commit e38adf40ca
10 files changed
+574 -76

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