nisse/tests/integration/test_model_capability.py

"""Capability matrix across models (spec Part H, A15).

Part H says the thresholds that depend on local capability — A12's failure
count, A14's consult budget, A15's manifest size — must be tuned "at
build-order step 5 on the real hardware". This is the measurement that makes
that possible, rather than reasoning about model strength from reputation.

What it measures is the thing that actually breaks: whether a model can hold
the tool-call contract our loops depend on. llama3.2 could not — it returned
`{` as an entire plan — and no amount of prompt work fixed it.

Opt in by naming the models to measure, for one run:

    uv run pytest -m integration tests/integration/test_model_capability.py \
        --capability-models=local-fast,local-coder

Skipped otherwise, because it spends real tokens across every model listed.
Which models to measure is a question for a particular run, so it is an
argument, not configuration.
"""

import pytest

from nisse_dev.canonical import (
    AssistantText,
    StoredEvent,
    ToolCall,
    ToolResult,
    UserMessage,
)
from nisse_dev.catalog import ModelCatalog, select_client
from nisse_dev.config import AgentConfig
from nisse_dev.executor import FINISH_TOOL
from nisse_dev.expert import parse_answer
from nisse_dev.llm import AnthropicPassthroughClient, LiteLLMClient, LLMError
from nisse_dev.planner import PLANNER_TOOLS
from nisse_dev.tools import EXECUTOR_TOOLS

pytestmark = pytest.mark.integration


def pytest_generate_tests(metafunc: pytest.Metafunc) -> None:
    if "model" not in metafunc.fixturenames:
        return
    raw = metafunc.config.getoption("--capability-models")
    models = [m.strip() for m in raw.split(",") if m.strip()]
    if models:
        metafunc.parametrize("model", models)
    else:
        metafunc.parametrize(
            "model",
            [
                pytest.param(
                    None,
                    marks=pytest.mark.skip(
                        reason="no models to measure; pass --capability-models"
                    ),
                )
            ],
        )


@pytest.fixture
def routes(
    litellm: LiteLLMClient, passthrough: AnthropicPassthroughClient
) -> tuple[LiteLLMClient, AnthropicPassthroughClient]:
    return litellm, passthrough


@pytest.fixture(scope="module")
def catalog(live_config: AgentConfig) -> 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(live_config.llm_base_url, live_config.llm_token)


def assert_answered(client: object, model: str, catalog: ModelCatalog) -> None:
    """A proxy fallback answers under another model's name. Crediting its
    answer to this model is how a local model once 'held the tool contract
    20/20' while never answering at all."""
    reported = getattr(client, "last_model", None)
    assert catalog.facts(model).answered(reported), (
        f"{reported} answered instead of {model}: a fallback, not a measurement"
    )


def user(text: str) -> list[StoredEvent]:
    return [StoredEvent(id=1, ts="t", event=UserMessage(text=text, source="probe"))]


def test_model_answers_at_all(model: str, catalog: ModelCatalog, routes) -> None:
    normalized, passthrough = routes
    client = select_client(model, normalized, passthrough, catalog)
    produced = client.complete(
        user("Reply with exactly one word: pong"), None, model, system="Be terse."
    )
    assert_answered(client, model, catalog)
    texts = [e.text for e in produced if isinstance(e, AssistantText)]
    assert texts, f"{model} produced no text: {produced!r}"


def test_model_holds_the_planner_contract(
    model: str, catalog: ModelCatalog, routes
) -> 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 = routes
    client = select_client(model, normalized, passthrough, catalog)
    produced = client.complete(
        user("Draft an architecture plan for a command-line slug generator."),
        PLANNER_TOOLS,
        model,
        system=(
            "You must answer by calling exactly one tool: submit_plan or "
            "ask_human. Plain prose is not accepted."
        ),
    )
    assert_answered(client, model, catalog)
    calls = [e for e in produced if isinstance(e, ToolCall)]
    assert calls, f"{model} answered in prose instead of calling a tool"
    assert calls[0].tool in {"submit_plan", "ask_human"}
    if calls[0].tool == "submit_plan":
        content = calls[0].arguments.get("content", "")
        # llama3.2 emitted "{" here — structurally a tool call, substantively
        # nothing. Length is a crude proxy, but it separates the two.
        assert len(str(content)) > 80, (
            f"{model} submitted a degenerate plan: {content!r}"
        )


def test_model_round_trips_a_tool_result(
    model: str, catalog: ModelCatalog, routes
) -> None:
    """The executor loop feeds results back; a model that cannot consume its
    own tool result cannot run a milestone."""
    normalized, passthrough = routes
    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)]
    assert calls, f"{model} made no tool call"

    next_id = 2
    for event in produced:
        events.append(StoredEvent(id=next_id, ts="t", event=event))
        next_id += 1
    events.append(
        StoredEvent(
            id=next_id,
            ts="t",
            event=ToolResult(
                call_id=calls[0].call_id, result="app.py\ntests/", success=True
            ),
        )
    )
    assert_answered(client, model, catalog)
    follow_up = client.complete(events, EXECUTOR_TOOLS, model, system="Use a tool.")
    assert follow_up, f"{model} produced nothing after its tool result"


def test_model_holds_the_prompt_level_json_contract(
    model: str, catalog: ModelCatalog, routes
) -> None:
    """A14 contracts structured output at the prompt level because vendors
    differ on JSON enforcement. This measures whether that holds per model."""
    normalized, passthrough = routes
    client = select_client(model, normalized, passthrough, catalog)
    produced = client.complete(
        user("Is SQLite sufficient for a single-process agent's task state?"),
        None,
        model,
        system=(
            "Reply with a single JSON object and nothing else: "
            '{"answer": "...", "confidence": "high"|"medium"|"low", '
            '"caveats": "..."}'
        ),
    )
    assert_answered(client, model, catalog)
    text = "\n".join(e.text for e in produced if isinstance(e, AssistantText))
    assert parse_answer(text) is not None, (
        f"{model} did not produce parseable JSON: {text[:200]!r}"
    )


def test_route_selection_matches_the_model(
    model: str, catalog: ModelCatalog, routes
) -> None:
    """A non-Anthropic model must not be sent Anthropic-shaped bodies. This
    was hardcoded until gpt-6-astra made it observable, and name-sniffed until
    semantic aliases made the name stop meaning anything."""
    normalized, passthrough = routes
    facts = catalog.facts(model)
    chosen = select_client(model, normalized, passthrough, catalog)
    if facts.provider == "anthropic":
        assert chosen is passthrough
    else:
        assert chosen is normalized
    # And the chosen route must actually work for this model.
    try:
        chosen.complete(user("say pong"), None, model, system="Be terse.")
    except LLMError as exc:
        pytest.fail(f"{model} failed on its selected route: {exc}")
    assert_answered(chosen, model, catalog)


FINISH_TOOLS = [*EXECUTOR_TOOLS, FINISH_TOOL]
OFFERED = {tool["function"]["name"] for tool in FINISH_TOOLS}


def test_model_calls_only_the_tools_it_was_offered(
    model: str, catalog: ModelCatalog, routes
) -> None:
    """Measured on a server that did not enforce tool_choice: a model asked
    to chat called "chat", "say_hello" and "respond", none of them offered.
    The executor counts those as failures; this measures how often."""
    normalized, passthrough = routes
    client = select_client(model, normalized, passthrough, catalog)
    produced = client.complete(
        user("Hi! How are you today? Just chat with me for a moment."),
        FINISH_TOOLS,
        model,
        system="You are an agent. Every reply must be a tool call.",
    )
    assert_answered(client, model, catalog)
    called = {e.tool for e in produced if isinstance(e, ToolCall)}
    assert called <= OFFERED, (
        f"{model} called tools it was never offered: {called - OFFERED}"
    )


def test_supports_tool_choice_is_declared_truthfully(
    model: str, catalog: ModelCatalog, routes
) -> None:
    """The agent asks for tool_choice only of models LiteLLM declares
    supports_tool_choice for, so the declaration must be true. Asked of a
    model that can't honour it, "required" can be worse than not asking:
    qwen3-14b can't stop, and writes until the output ceiling."""
    facts = catalog.facts(model)
    if facts.supports_tool_choice is not True:
        pytest.skip(
            f"{model} is not declared to support tool_choice, so is never asked"
        )
    normalized, passthrough = routes
    client = select_client(model, normalized, passthrough, catalog)
    produced = client.complete(
        user("Hi! How are you today? Just chat with me for a moment."),
        FINISH_TOOLS,
        model,
        system="You are a helpful assistant.",
        require_tool_call=True,
    )
    assert_answered(client, model, catalog)
    calls = [e for e in produced if isinstance(e, ToolCall)]
    assert calls, (
        f"{model} is declared supports_tool_choice but answered a required call "
        "in prose; set it to false in LiteLLM"
    )
    assert {c.tool for c in calls} <= OFFERED