nisse/nisse-core/tests/test_render_anthropic.py

"""The Anthropic Messages shape differs structurally from the OpenAI one;
these tests pin the differences that a normalising middlebox would blur."""

import httpx
import pytest

from nisse_core.log.canonical import (
    AssistantText,
    Compaction,
    StoredEvent,
    ToolCall,
    ToolResult,
    UserMessage,
)
from nisse_core.models.llm import AnthropicPassthroughClient, LLMError
from nisse_core.models.render_anthropic import (
    AnthropicMessagesRenderer,
    to_anthropic_tools,
)

TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "run_command",
            "description": "Run a shell command.",
            "parameters": {
                "type": "object",
                "properties": {"command": {"type": "string"}},
                "required": ["command"],
            },
        },
    }
]


def stored(events: list) -> list[StoredEvent]:
    return [StoredEvent(id=i + 1, ts="t", event=e) for i, e in enumerate(events)]


def render(events: list, **kw) -> dict:
    return AnthropicMessagesRenderer().render(
        stored(events), kw.pop("tools", None), "claude-haiku-4-5", **kw
    )


def test_system_is_a_top_level_field_not_a_message() -> None:
    body = render([UserMessage(text="hi", source="m")], system="you are nisse")
    assert body["system"] == "you are nisse"
    assert all(m["role"] != "system" for m in body["messages"])


def test_max_tokens_is_always_present() -> None:
    # The Messages API rejects a request without it.
    assert render([UserMessage(text="hi", source="m")])["max_tokens"] > 0


def test_tool_call_is_a_content_block_not_an_array() -> None:
    body = render(
        [ToolCall(call_id="call-1", tool="run_command", arguments={"command": "ls"})]
    )
    block = body["messages"][0]["content"][0]
    assert block["type"] == "tool_use"
    assert block["id"] == "call-1"
    assert block["input"] == {"command": "ls"}
    assert "tool_calls" not in body["messages"][0]


def test_tool_result_is_a_user_block_directly_after_the_call() -> None:
    body = render(
        [
            ToolCall(call_id="call-1", tool="run_command", arguments={}),
            ToolResult(call_id="call-1", result="exit code: 0", success=True),
        ]
    )
    assistant, user = body["messages"]
    assert assistant["role"] == "assistant"
    assert assistant["content"][0]["type"] == "tool_use"
    assert user["role"] == "user"  # not a "tool" role, as OpenAI would have
    assert user["content"][0] == {
        "type": "tool_result",
        "tool_use_id": "call-1",
        "content": "exit code: 0",
    }


def test_adjacent_same_role_events_are_merged() -> None:
    body = render(
        [
            UserMessage(text="one", source="m"),
            UserMessage(text="two", source="m"),
            AssistantText(text="reply", model="m"),
        ]
    )
    assert [m["role"] for m in body["messages"]] == ["user", "assistant"]
    assert [b["text"] for b in body["messages"][0]["content"]] == ["one", "two"]


def test_dangling_tool_call_gets_a_placeholder_result() -> None:
    body = render([ToolCall(call_id="call-x", tool="submit_plan", arguments={})])
    assert [m["role"] for m in body["messages"]] == ["assistant", "user"]
    assert body["messages"][1]["content"][0]["tool_use_id"] == "call-x"


def test_tools_are_converted_to_input_schema() -> None:
    [tool] = to_anthropic_tools(TOOLS)
    assert tool["name"] == "run_command"
    assert tool["input_schema"]["properties"] == {"command": {"type": "string"}}
    assert "parameters" not in tool
    assert "function" not in tool


def test_compaction_still_replaces_its_range() -> None:
    events = stored(
        [
            UserMessage(text="old", source="m"),
            UserMessage(text="kept", source="m"),
        ]
    )
    events.append(
        StoredEvent(
            id=3,
            ts="t",
            event=Compaction(summary="summary", model="m", replaces=[1, 1]),
        )
    )
    body = AnthropicMessagesRenderer().render(events, None, "claude-haiku-4-5")
    texts = [b["text"] for m in body["messages"] for b in m["content"]]
    assert "old" not in texts
    assert any("summary" in t for t in texts)


# --- passthrough client --------------------------------------------------


def client(handler) -> AnthropicPassthroughClient:
    return AnthropicPassthroughClient(
        "https://litellm.test", "tok", transport=httpx.MockTransport(handler)
    )


def test_client_posts_to_the_messages_route_with_version_header() -> None:
    seen: dict = {}

    def handler(request: httpx.Request) -> httpx.Response:
        seen["url"] = str(request.url)
        seen["version"] = request.headers.get("anthropic-version")
        return httpx.Response(
            200,
            json={
                "model": "claude-haiku-4-5",
                "content": [{"type": "text", "text": "hi"}],
            },
        )

    events = stored([UserMessage(text="hello", source="m")])
    [event] = client(handler).complete(events, None, "claude-haiku-4-5")
    assert seen["url"] == "https://litellm.test/v1/messages"
    assert seen["version"] == "2023-06-01"
    assert event == AssistantText(text="hi", model="claude-haiku-4-5")


def test_client_maps_tool_use_to_local_call_ids() -> None:
    response = {
        "model": "claude-haiku-4-5",
        "content": [
            {"type": "text", "text": "running it"},
            {
                "type": "tool_use",
                "id": "toolu_vendor_xyz",
                "name": "run_command",
                "input": {"command": "pytest"},
            },
        ],
    }
    events = stored([UserMessage(text="go", source="m")])
    text, call = client(lambda r: httpx.Response(200, json=response)).complete(
        events, TOOLS, "claude-haiku-4-5"
    )
    assert isinstance(text, AssistantText)
    assert isinstance(call, ToolCall)
    assert call.arguments == {"command": "pytest"}
    assert call.call_id != "toolu_vendor_xyz"  # provider id never a join key
    assert call.call_id.startswith("call-")


def test_client_errors_are_llm_errors() -> None:
    events = stored([UserMessage(text="go", source="m")])
    with pytest.raises(LLMError, match="400"):
        client(lambda r: httpx.Response(400, text="bad")).complete(
            events, None, "claude-haiku-4-5"
        )
    with pytest.raises(LLMError, match="malformed"):
        client(lambda r: httpx.Response(200, json={"oops": 1})).complete(
            events, None, "claude-haiku-4-5"
        )