import json from nisse_dev.canonical import ( AssistantText, Compaction, GateOpened, StoredEvent, ToolCall, ToolResult, UserMessage, ) from nisse_dev.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"]] assert contents == ["kept", "[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 (A3) # 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"]