nisse/nisse-dev/src/nisse_dev/render_anthropic.py

"""Anthropic Messages renderer (architecture L7, L9).

The second adapter, and the reason L9 puts normalisation in the renderer
rather than on the wire: the two vendors disagree *structurally*, not
cosmetically.

| | OpenAI-compatible | Anthropic Messages |
|---|---|---|
| system prompt | a message with role `system` | a top-level `system` field |
| tool call | `tool_calls` array on the message | a `tool_use` content block |
| tool result | a message with role `tool` | a `tool_result` block in a **user** message |
| tool schema | `function.parameters` | `input_schema` |

A middlebox that normalised between these would be lossy exactly where a
long-running delegation lives, which is why delegations take the passthrough
route and this renderer builds the vendor-native body.
"""

from typing import Any

from nisse_dev.canonical import (
    AssistantText,
    Compaction,
    DelegationCompleted,
    DelegationStarted,
    GateOpened,
    PlanSubmitted,
    StoredEvent,
    ToolCall,
    ToolResult,
    UserMessage,
)
from nisse_dev.render import effective_events

DEFAULT_MAX_TOKENS = 4096
NO_RESULT_PLACEHOLDER = "[no result recorded]"


def to_anthropic_tools(tools: list[dict[str, Any]] | None) -> list[dict[str, Any]]:
    """Convert the OpenAI-shaped manifests the tool modules already define,
    so a tool is declared once and rendered per vendor."""
    converted: list[dict[str, Any]] = []
    for tool in tools or []:
        fn = tool.get("function", tool)
        converted.append(
            {
                "name": fn["name"],
                "description": fn.get("description", ""),
                "input_schema": fn.get(
                    "parameters", {"type": "object", "properties": {}}
                ),
            }
        )
    return converted


class AnthropicMessagesRenderer:
    def render(
        self,
        events: list[StoredEvent],
        tools: list[dict[str, Any]] | None,
        model: str,
        system: str | None = None,
        *,
        max_tokens: int = DEFAULT_MAX_TOKENS,
    ) -> dict[str, Any]:
        kept = effective_events(events)
        answered = {e.event.call_id for e in kept if isinstance(e.event, ToolResult)}
        blocks: list[tuple[str, dict[str, Any]]] = []
        for stored in kept:
            blocks.extend(self._blocks_for(stored.event))
            # A dangling tool_use is rejected outright by this API — more
            # strictly than the OpenAI shape — so a crash between a call and
            # its result must not poison the transcript (L8).
            if isinstance(stored.event, ToolCall) and (
                stored.event.call_id not in answered
            ):
                blocks.append(
                    (
                        "user",
                        {
                            "type": "tool_result",
                            "tool_use_id": stored.event.call_id,
                            "content": NO_RESULT_PLACEHOLDER,
                        },
                    )
                )

        body: dict[str, Any] = {
            "model": model,
            "max_tokens": max_tokens,
            "messages": _coalesce(blocks),
        }
        if system is not None:
            body["system"] = system
        if tools:
            body["tools"] = to_anthropic_tools(tools)
        return body

    def _blocks_for(self, event: object) -> list[tuple[str, dict[str, Any]]]:
        match event:
            case UserMessage(text=text):
                return [("user", {"type": "text", "text": text})]
            case AssistantText(text=text):
                return [("assistant", {"type": "text", "text": text})]
            case ToolCall(call_id=call_id, tool=tool, arguments=arguments):
                return [
                    (
                        "assistant",
                        {
                            "type": "tool_use",
                            "id": call_id,
                            "name": tool,
                            "input": arguments,
                        },
                    )
                ]
            case ToolResult(call_id=call_id, result=result):
                return [
                    (
                        "user",
                        {
                            "type": "tool_result",
                            "tool_use_id": call_id,
                            "content": result,
                        },
                    )
                ]
            case PlanSubmitted(plan=plan):
                return [("assistant", {"type": "text", "text": plan})]
            case GateOpened(gate=gate, reason=reason):
                return [
                    (
                        "user",
                        {"type": "text", "text": f"[gate {gate} opened: {reason}]"},
                    )
                ]
            case DelegationStarted(brief=brief, model=model):
                return [
                    (
                        "assistant",
                        {
                            "type": "text",
                            "text": f"[delegation started ({model}): {brief}]",
                        },
                    )
                ]
            case DelegationCompleted(handback=handback):
                return [
                    (
                        "assistant",
                        {"type": "text", "text": f"[delegation handback]\n{handback}"},
                    )
                ]
            case Compaction(summary=summary):
                return [
                    (
                        "user",
                        {
                            "type": "text",
                            "text": f"[summary of earlier conversation]\n{summary}",
                        },
                    )
                ]
            case _:
                raise ValueError(f"unrenderable event: {event!r}")


def _coalesce(
    blocks: list[tuple[str, dict[str, Any]]],
) -> list[dict[str, Any]]:
    """Merge adjacent blocks of the same role into one message.

    The Messages API expects alternating roles, and a tool_result must sit in
    the user message directly after the assistant's tool_use — which happens
    naturally once same-role runs are merged.
    """
    messages: list[dict[str, Any]] = []
    for role, block in blocks:
        if messages and messages[-1]["role"] == role:
            messages[-1]["content"].append(block)
        else:
            messages.append({"role": role, "content": [block]})
    return messages