nisse/nisse-core/tests/test_render.py

import json
from itertools import pairwise

from nisse_core.log.canonical import (
    AssistantText,
    Compaction,
    GateOpened,
    StoredEvent,
    ToolCall,
    ToolResult,
    UserMessage,
)
from nisse_core.models.render import OpenAIChatRenderer, effective_events


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


def test_renders_mixed_transcript_in_order() -> None:
    events = stored(
        [
            UserMessage(text="run the tests", source="m:1"),
            ToolCall(call_id="call-9", tool="run_tests", arguments={"path": "x"}),
            ToolResult(call_id="call-9", result="1 passed", success=True),
            AssistantText(text="all green", model="local/qwen"),
        ]
    )
    body = OpenAIChatRenderer().render(events, tools=None, model="local/qwen")
    assert body["model"] == "local/qwen"
    roles = [m["role"] for m in body["messages"]]
    assert roles == ["user", "assistant", "tool", "assistant"]
    call = body["messages"][1]["tool_calls"][0]
    assert call["id"] == "call-9"
    assert call["function"]["name"] == "run_tests"
    assert json.loads(call["function"]["arguments"]) == {"path": "x"}
    assert body["messages"][2]["tool_call_id"] == "call-9"
    assert "tools" not in body


def test_system_prompt_and_tools_pass_through() -> None:
    tools = [{"type": "function", "function": {"name": "run_tests"}}]
    body = OpenAIChatRenderer().render(
        stored([UserMessage(text="hi", source="m:1")]),
        tools=tools,
        model="local/qwen",
        system="you are nisse-dev",
    )
    assert body["messages"][0] == {"role": "system", "content": "you are nisse-dev"}
    assert body["tools"] == tools


def test_compaction_replaces_its_range_only() -> None:
    events = stored(
        [
            UserMessage(text="old-1", source="m:1"),
            AssistantText(text="old-2", model="q"),
            UserMessage(text="kept", source="m:3"),
        ]
    )
    events.append(
        StoredEvent(
            id=4,
            ts="t",
            event=Compaction(summary="the old stuff", model="q", replaces=[1, 2]),
        )
    )
    remaining = effective_events(events)
    assert [e.id for e in remaining] == [3, 4]
    body = OpenAIChatRenderer().render(events, tools=None, model="q")
    contents = [m["content"] for m in body["messages"]]
    # Both render as user messages, so they share one: roles must alternate.
    assert contents == ["kept\n\n[summary of earlier conversation]\nthe old stuff"]


def test_tool_call_without_a_result_gets_a_placeholder() -> None:
    # Providers reject a dangling tool_use; this happens after a crash (L8)
    # or with turn-ending gate tools like submit_plan.
    events = stored(
        [
            UserMessage(text="go", source="m:1"),
            ToolCall(call_id="call-x", tool="submit_plan", arguments={}),
        ]
    )
    body = OpenAIChatRenderer().render(events, tools=None, model="q")
    assert [m["role"] for m in body["messages"]] == ["user", "assistant", "tool"]
    assert body["messages"][2]["tool_call_id"] == "call-x"
    assert body["messages"][2]["content"] == "[no result recorded]"


def test_answered_tool_call_gets_no_placeholder() -> None:
    events = stored(
        [
            ToolCall(call_id="call-x", tool="run_command", arguments={}),
            ToolResult(call_id="call-x", result="ok", success=True),
        ]
    )
    body = OpenAIChatRenderer().render(events, tools=None, model="q")
    assert [m["role"] for m in body["messages"]] == ["assistant", "tool"]
    assert body["messages"][1]["content"] == "ok"


def test_gate_event_renders_as_context() -> None:
    events = stored([GateOpened(gate="arch", reason="plan submitted (rev-1)")])
    body = OpenAIChatRenderer().render(events, tools=None, model="q")
    assert body["messages"][0]["role"] == "user"
    assert "arch" in body["messages"][0]["content"]


def _roles(body: dict) -> list[str]:
    return [m["role"] for m in body["messages"]]


def test_one_turn_is_one_assistant_message() -> None:
    # Text and two parallel tool calls from one model turn: the OpenAI format
    # carries them as a single assistant message with a tool_calls list.
    events = stored(
        [
            UserMessage(text="go", source="m:1"),
            AssistantText(text="Looking around first.", model="q"),
            ToolCall(call_id="a", tool="list_dir", arguments={"path": "."}),
            ToolCall(call_id="b", tool="read_file", arguments={"path": "README.md"}),
            ToolResult(call_id="a", result="README.md", success=True),
            ToolResult(call_id="b", result="hello", success=True),
        ]
    )
    body = OpenAIChatRenderer().render(events, tools=None, model="q")
    assert _roles(body) == ["user", "assistant", "tool", "tool"]
    turn = body["messages"][1]
    assert turn["content"] == "Looking around first."
    assert [c["id"] for c in turn["tool_calls"]] == ["a", "b"]


def test_roles_never_repeat_for_strict_templates() -> None:
    """Mistral's chat template raises on two user or two assistant messages
    in a row; every mix of events must render without one."""
    events = stored(
        [
            UserMessage(text="one", source="m:1"),
            UserMessage(text="two", source="m:2"),
            AssistantText(text="thinking", model="q"),
            ToolCall(call_id="a", tool="list_dir", arguments={}),
            ToolResult(call_id="a", result="x", success=True),
            AssistantText(text="more", model="q"),
            AssistantText(text="and more", model="q"),
            ToolCall(call_id="b", tool="finish", arguments={}),
        ]
    )
    roles = _roles(OpenAIChatRenderer().render(events, tools=None, model="q"))
    for first, second in pairwise(roles):
        assert not (first == second and first in ("user", "assistant")), roles