nisse/nisse-core/src/nisse_core/render.py

"""L7 renderer: provider request bodies are built from the canonical log at
call time, never stored provider-shaped.

First adapter: the OpenAI-compatible chat format LiteLLM's normalized route
speaks — the executor role's path (L9). The Anthropic passthrough adapter
arrives with delegations (L9). Token budgeting and prompt-cache breakpoints
are render-time concerns and are never stored (L7); neither exists yet in
this skeleton.
"""

import json
from typing import Any

from nisse_core.canonical import (
    AssistantText,
    Compaction,
    DelegationCompleted,
    DelegationStarted,
    Fact,
    GateOpened,
    PlanSubmitted,
    StoredEvent,
    ToolCall,
    ToolResult,
    UserMessage,
)


def effective_events(stored: list[StoredEvent]) -> list[StoredEvent]:
    """Apply compactions: events whose ids fall in a compaction's replaced
    range are dropped; the compaction event itself stays as their summary."""
    replaced: list[tuple[int, int]] = [
        (e.event.replaces[0], e.event.replaces[1])
        for e in stored
        if isinstance(e.event, Compaction)
    ]

    def is_replaced(event_id: int) -> bool:
        return any(first <= event_id <= last for first, last in replaced)

    return [
        e for e in stored if isinstance(e.event, Compaction) or not is_replaced(e.id)
    ]


class OpenAIChatRenderer:
    """Mappings for orchestrator-side events (gates, plans, delegations,
    compaction summaries) are provisional; they get tuned when the executor
    loop (step 5) and delegations (step 7) consume them for real."""

    def render(
        self,
        events: list[StoredEvent],
        tools: list[dict[str, Any]] | None,
        model: str,
        system: str | None = None,
    ) -> dict[str, Any]:
        messages: list[dict[str, Any]] = []
        if system is not None:
            messages.append({"role": "system", "content": system})
        kept = effective_events(events)
        answered = {e.event.call_id for e in kept if isinstance(e.event, ToolResult)}
        for stored in kept:
            _append_merged(messages, self._message_for(stored.event))
            # Providers reject a tool call with no result following it. One
            # can legitimately be missing: a crash between the call and its
            # result (L8), or a turn-ending gate tool. Render a placeholder so
            # the transcript stays replayable rather than poisoned.
            if isinstance(stored.event, ToolCall) and (
                stored.event.call_id not in answered
            ):
                messages.append(
                    {
                        "role": "tool",
                        "tool_call_id": stored.event.call_id,
                        "content": "[no result recorded]",
                    }
                )
        body: dict[str, Any] = {"model": model, "messages": messages}
        if tools:
            body["tools"] = tools
        return body

    def _message_for(self, event: object) -> dict[str, Any]:
        match event:
            case UserMessage(text=text):
                return {"role": "user", "content": text}
            case AssistantText(text=text):
                return {"role": "assistant", "content": text}
            case ToolCall(call_id=call_id, tool=tool, arguments=arguments):
                return {
                    "role": "assistant",
                    "content": None,
                    "tool_calls": [
                        {
                            "id": call_id,
                            "type": "function",
                            "function": {
                                "name": tool,
                                "arguments": json.dumps(arguments),
                            },
                        }
                    ],
                }
            case ToolResult(call_id=call_id, result=result):
                return {
                    "role": "tool",
                    "tool_call_id": call_id,
                    "content": result,
                }
            case PlanSubmitted(plan=plan):
                return {"role": "assistant", "content": plan}
            case GateOpened(gate=gate, reason=reason):
                return {"role": "user", "content": f"[gate {gate} opened: {reason}]"}
            case DelegationStarted(brief=brief, model=model):
                return {
                    "role": "assistant",
                    "content": f"[delegation started ({model}): {brief}]",
                }
            case DelegationCompleted(handback=handback):
                return {
                    "role": "assistant",
                    "content": f"[delegation handback]\n{handback}",
                }
            case Fact(text=text):
                return {"role": "user", "content": f"[orchestrator] {text}"}
            case Compaction(summary=summary):
                return {
                    "role": "user",
                    "content": f"[summary of earlier conversation]\n{summary}",
                }
            case _:
                raise ValueError(f"unrenderable event: {event!r}")


def _append_merged(messages: list[dict[str, Any]], message: dict[str, Any]) -> None:
    """Append, folding into the previous message when both are the same user
    or assistant role.

    One model turn is often several events (some text, then one or more tool
    calls), and the OpenAI format carries a turn as one assistant message
    with a tool_calls list. Strict chat templates enforce that: Devstral's
    (Mistral) rejected a turn rendered as a text message then a tool-call
    message with a 500, failing runs whenever the model spoke before acting."""
    previous = messages[-1] if messages else None
    role = message["role"]
    if (
        previous is None
        or previous["role"] != role
        or role not in ("user", "assistant")
    ):
        messages.append(message)
        return
    texts = [t for t in (previous.get("content"), message.get("content")) if t]
    previous["content"] = "\n\n".join(texts) if texts else None
    calls = previous.get("tool_calls", []) + message.get("tool_calls", [])
    if calls:
        previous["tool_calls"] = calls