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)
    # The summary takes the place of what it replaces.
    assert [e.id for e in remaining] == [4, 3]
    assert effective_events(remaining) == remaining  # applying again: no change
    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 == ["[summary of earlier conversation]\nthe old stuff\n\nkept"]


def test_a_later_compaction_folds_in_an_earlier_one() -> None:
    events = stored([UserMessage(text=f"e{i}", source="m") for i in range(1, 7)])
    events.append(
        StoredEvent(id=7, ts="t", event=Compaction("first", "q", replaces=[2, 3]))
    )
    events.append(
        StoredEvent(id=8, ts="t", event=Compaction("second", "q", replaces=[2, 5]))
    )
    remaining = effective_events(events)
    assert [e.id for e in remaining] == [1, 8, 6]


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


def test_each_event_maps_to_the_message_render_puts_it_in() -> None:
    # A turn's text and tool call fold into one message; a call with no
    # result gets a placeholder that came from no event; the system prompt
    # is message 0.
    events = stored(
        [
            UserMessage(text="go", source="m:1"),
            AssistantText(text="Looking.", model="m"),
            ToolCall(call_id="c1", tool="read_file", arguments={"path": "a"}),
            ToolResult(call_id="c1", result="text", success=True),
            ToolCall(call_id="c2", tool="finish", arguments={}),
            UserMessage(text="and then?", source="m:2"),
        ]
    )
    renderer = OpenAIChatRenderer()
    body = renderer.render(events, tools=None, model="m", system="sys")
    indexes = renderer.message_indexes(events, system="sys")
    assert indexes == [1, 2, 2, 3, 4, 6]
    assert body["messages"][5]["content"] == "[no result recorded]"
    for event, index in zip(events, indexes, strict=True):
        message = body["messages"][index]
        if isinstance(event.event, ToolCall):
            ids = [c["id"] for c in message["tool_calls"]]
            assert event.event.call_id in ids
        elif isinstance(event.event, (UserMessage, AssistantText)):
            assert event.event.text in message["content"]
    assert renderer.message_indexes(events) == [i - 1 for i in indexes]


def test_events_a_compaction_replaced_map_to_no_message() -> None:
    events = stored(
        [
            UserMessage(text="one", source="m:1"),
            AssistantText(text="two", model="m"),
            UserMessage(text="three", source="m:3"),
            Compaction(summary="one and two", model="m", replaces=[1, 2]),
        ]
    )
    # The summary takes the place of what it replaced, and the user message
    # after it folds into the same user message.
    indexes = OpenAIChatRenderer().message_indexes(events)
    assert indexes == [None, None, 0, 0]