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 with NISSE_CAPABILITY_MODELS, comma separated. Skipped otherwise,
because it spends real tokens across several models.
"""

import os
from pathlib import Path

import pytest

from nisse_dev.canonical import (
    AssistantText,
    StoredEvent,
    ToolCall,
    ToolResult,
    UserMessage,
)
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.planner import PLANNER_TOOLS
from nisse_dev.tools import EXECUTOR_TOOLS

load_env_file(Path(__file__).resolve().parents[2] / ".env")

SENTINEL = "NISSE_CAPABILITY_MODELS"
MODELS = [m.strip() for m in os.environ.get(SENTINEL, "").split(",") if m.strip()]

pytestmark = [
    pytest.mark.integration,
    pytest.mark.skipif(
        not MODELS, reason=f"set {SENTINEL} to a comma-separated model list"
    ),
]


def clients() -> tuple[LiteLLMClient, AnthropicPassthroughClient]:
    return LiteLLMClient.from_env(), AnthropicPassthroughClient.from_env()


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:
    normalized, passthrough = clients()
    client = select_client(model, normalized, passthrough)
    produced = client.complete(
        user("Reply with exactly one word: pong"), None, model, system="Be terse."
    )
    texts = [e.text for e in produced if isinstance(e, AssistantText)]
    assert texts, f"{model} produced no text: {produced!r}"


@pytest.mark.parametrize("model", MODELS)
def test_model_holds_the_planner_contract(model: str) -> 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)
    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."
        ),
    )
    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}"
        )


@pytest.mark.parametrize("model", MODELS)
def test_model_round_trips_a_tool_result(model: str) -> 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)
    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
            ),
        )
    )
    follow_up = client.complete(events, EXECUTOR_TOOLS, model, system="Use a tool.")
    assert follow_up, f"{model} produced nothing after its tool result"


@pytest.mark.parametrize("model", MODELS)
def test_model_holds_the_prompt_level_json_contract(model: str) -> 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)
    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": "..."}'
        ),
    )
    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}"
    )


@pytest.mark.parametrize("model", MODELS)
def test_route_selection_matches_the_model(model: str) -> None:
    """A non-Anthropic model must not be sent Anthropic-shaped bodies. This
    was hardcoded until gpt-6-astra made it observable."""
    normalized, passthrough = clients()
    chosen = select_client(model, normalized, passthrough)
    if model.startswith(("claude-", "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}")