Merge pull request 'Add the Anthropic passthrough renderer and client (A10/A13) — step 7a' (#20) from anthropic-renderer into main

Reviewed-on: dongar/nisse#20
Reviewed-by: Don Garrett <dgarrett@acm.org>

nisse-dev/src/nisse_dev/llm.py

@@ -26,6 +26,7 @@ from nisse_dev.canonical import (
     mint_call_id,
 )
 from nisse_dev.render import OpenAIChatRenderer
+from nisse_dev.render_anthropic import AnthropicMessagesRenderer
 
 
 class LLMError(RuntimeError):
@@ -157,3 +158,96 @@ class LiteLLMClient:
 
     def close(self) -> None:
         self._http.close()
+
+
+class AnthropicPassthroughClient:
+    """LiteLLM's Anthropic passthrough (A5). Same interface as LiteLLMClient,
+    so a caller chooses a route rather than a different kind of client.
+
+    Delegations take this path because fidelity matters there (A13);
+    ask_expert and fallbacks use the normalized route, where it does not.
+    """
+
+    def __init__(
+        self,
+        base_url: str,
+        token: str,
+        *,
+        timeout_s: float = 300.0,
+        max_tokens: int = 4096,
+        transport: httpx.BaseTransport | None = None,
+    ) -> None:
+        self._http = httpx.Client(
+            base_url=base_url.rstrip("/"),
+            headers={
+                "Authorization": f"Bearer {token}",
+                "x-api-key": token,
+                "anthropic-version": "2023-06-01",
+            },
+            timeout=timeout_s,
+            transport=transport,
+        )
+        self._renderer = AnthropicMessagesRenderer()
+        self._max_tokens = max_tokens
+
+    @classmethod
+    def from_env(
+        cls, env: Mapping[str, str] = os.environ
+    ) -> "AnthropicPassthroughClient":
+        base_url = env.get("NISSE_LLM_BASE_URL")
+        token = env.get("NISSE_LLM_TOKEN")
+        if not base_url or not token:
+            raise LLMError("set NISSE_LLM_BASE_URL and NISSE_LLM_TOKEN")
+        return cls(base_url, token)
+
+    def complete(
+        self,
+        events: list[StoredEvent],
+        tools: list[dict[str, Any]] | None,
+        model: str,
+        system: str | None = None,
+    ) -> list[CanonicalEvent]:
+        body = self._renderer.render(
+            events, tools, model, system=system, max_tokens=self._max_tokens
+        )
+        try:
+            response = self._http.post("/v1/messages", json=body)
+        except httpx.HTTPError as exc:
+            raise LLMError(f"passthrough request failed: {exc}") from exc
+        if response.status_code != 200:
+            raise LLMError(
+                f"passthrough returned {response.status_code}: {response.text[:500]}"
+            )
+        return self._parse(response.json(), requested_model=model)
+
+    def _parse(
+        self, data: dict[str, Any], requested_model: str
+    ) -> list[CanonicalEvent]:
+        content = data.get("content")
+        if not isinstance(content, list):
+            raise LLMError(f"malformed passthrough response: {str(data)[:300]}")
+        produced_by = data.get("model") or requested_model
+        out: list[CanonicalEvent] = []
+        for block in content:
+            kind = block.get("type")
+            if kind == "text" and block.get("text"):
+                out.append(AssistantText(text=block["text"], model=produced_by))
+            elif kind == "tool_use":
+                # Provider ids are mapped at the boundary and never used as a
+                # join key (A10); the canonical log owns identity.
+                out.append(
+                    ToolCall(
+                        call_id=mint_call_id(),
+                        tool=block["name"],
+                        arguments=block.get("input") or {},
+                    )
+                )
+            # Vendor-opaque reasoning blocks (thinking, redacted_thinking) are
+            # deliberately dropped for now: A10 requires storing them tagged
+            # with the producing model and never replaying them outside the
+            # session that made them, and the canonical schema has no event
+            # for that yet. Nothing requests extended thinking, so none arrive.
+        return out
+
+    def close(self) -> None:
+        self._http.close()

nisse-dev/src/nisse_dev/render_anthropic.py

@@ -0,0 +1,184 @@
+"""Anthropic Messages renderer (spec A10, A13).
+
+The second adapter, and the reason A13 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 (A3).
+            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

nisse-dev/tests/test_render_anthropic.py

@@ -0,0 +1,196 @@
+"""The Anthropic Messages shape differs structurally from the OpenAI one;
+these tests pin the differences that a normalising middlebox would blur."""
+
+import httpx
+import pytest
+
+from nisse_dev.canonical import (
+    AssistantText,
+    Compaction,
+    StoredEvent,
+    ToolCall,
+    ToolResult,
+    UserMessage,
+)
+from nisse_dev.llm import AnthropicPassthroughClient, LLMError
+from nisse_dev.render_anthropic import (
+    AnthropicMessagesRenderer,
+    to_anthropic_tools,
+)
+
+TOOLS = [
+    {
+        "type": "function",
+        "function": {
+            "name": "run_command",
+            "description": "Run a shell command.",
+            "parameters": {
+                "type": "object",
+                "properties": {"command": {"type": "string"}},
+                "required": ["command"],
+            },
+        },
+    }
+]
+
+
+def stored(events: list) -> list[StoredEvent]:
+    return [StoredEvent(id=i + 1, ts="t", event=e) for i, e in enumerate(events)]
+
+
+def render(events: list, **kw) -> dict:
+    return AnthropicMessagesRenderer().render(
+        stored(events), kw.pop("tools", None), "claude-haiku-4-5", **kw
+    )
+
+
+def test_system_is_a_top_level_field_not_a_message() -> None:
+    body = render([UserMessage(text="hi", source="m")], system="you are nisse")
+    assert body["system"] == "you are nisse"
+    assert all(m["role"] != "system" for m in body["messages"])
+
+
+def test_max_tokens_is_always_present() -> None:
+    # The Messages API rejects a request without it.
+    assert render([UserMessage(text="hi", source="m")])["max_tokens"] > 0
+
+
+def test_tool_call_is_a_content_block_not_an_array() -> None:
+    body = render(
+        [ToolCall(call_id="call-1", tool="run_command", arguments={"command": "ls"})]
+    )
+    block = body["messages"][0]["content"][0]
+    assert block["type"] == "tool_use"
+    assert block["id"] == "call-1"
+    assert block["input"] == {"command": "ls"}
+    assert "tool_calls" not in body["messages"][0]
+
+
+def test_tool_result_is_a_user_block_directly_after_the_call() -> None:
+    body = render(
+        [
+            ToolCall(call_id="call-1", tool="run_command", arguments={}),
+            ToolResult(call_id="call-1", result="exit code: 0", success=True),
+        ]
+    )
+    assistant, user = body["messages"]
+    assert assistant["role"] == "assistant"
+    assert assistant["content"][0]["type"] == "tool_use"
+    assert user["role"] == "user"  # not a "tool" role, as OpenAI would have
+    assert user["content"][0] == {
+        "type": "tool_result",
+        "tool_use_id": "call-1",
+        "content": "exit code: 0",
+    }
+
+
+def test_adjacent_same_role_events_are_merged() -> None:
+    body = render(
+        [
+            UserMessage(text="one", source="m"),
+            UserMessage(text="two", source="m"),
+            AssistantText(text="reply", model="m"),
+        ]
+    )
+    assert [m["role"] for m in body["messages"]] == ["user", "assistant"]
+    assert [b["text"] for b in body["messages"][0]["content"]] == ["one", "two"]
+
+
+def test_dangling_tool_call_gets_a_placeholder_result() -> None:
+    body = render([ToolCall(call_id="call-x", tool="submit_plan", arguments={})])
+    assert [m["role"] for m in body["messages"]] == ["assistant", "user"]
+    assert body["messages"][1]["content"][0]["tool_use_id"] == "call-x"
+
+
+def test_tools_are_converted_to_input_schema() -> None:
+    [tool] = to_anthropic_tools(TOOLS)
+    assert tool["name"] == "run_command"
+    assert tool["input_schema"]["properties"] == {"command": {"type": "string"}}
+    assert "parameters" not in tool
+    assert "function" not in tool
+
+
+def test_compaction_still_replaces_its_range() -> None:
+    events = stored(
+        [
+            UserMessage(text="old", source="m"),
+            UserMessage(text="kept", source="m"),
+        ]
+    )
+    events.append(
+        StoredEvent(
+            id=3,
+            ts="t",
+            event=Compaction(summary="summary", model="m", replaces=[1, 1]),
+        )
+    )
+    body = AnthropicMessagesRenderer().render(events, None, "claude-haiku-4-5")
+    texts = [b["text"] for m in body["messages"] for b in m["content"]]
+    assert "old" not in texts
+    assert any("summary" in t for t in texts)
+
+
+# --- passthrough client --------------------------------------------------
+
+
+def client(handler) -> AnthropicPassthroughClient:
+    return AnthropicPassthroughClient(
+        "https://litellm.test", "tok", transport=httpx.MockTransport(handler)
+    )
+
+
+def test_client_posts_to_the_messages_route_with_version_header() -> None:
+    seen: dict = {}
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        seen["url"] = str(request.url)
+        seen["version"] = request.headers.get("anthropic-version")
+        return httpx.Response(
+            200,
+            json={
+                "model": "claude-haiku-4-5",
+                "content": [{"type": "text", "text": "hi"}],
+            },
+        )
+
+    events = stored([UserMessage(text="hello", source="m")])
+    [event] = client(handler).complete(events, None, "claude-haiku-4-5")
+    assert seen["url"] == "https://litellm.test/v1/messages"
+    assert seen["version"] == "2023-06-01"
+    assert event == AssistantText(text="hi", model="claude-haiku-4-5")
+
+
+def test_client_maps_tool_use_to_local_call_ids() -> None:
+    response = {
+        "model": "claude-haiku-4-5",
+        "content": [
+            {"type": "text", "text": "running it"},
+            {
+                "type": "tool_use",
+                "id": "toolu_vendor_xyz",
+                "name": "run_command",
+                "input": {"command": "pytest"},
+            },
+        ],
+    }
+    events = stored([UserMessage(text="go", source="m")])
+    text, call = client(lambda r: httpx.Response(200, json=response)).complete(
+        events, TOOLS, "claude-haiku-4-5"
+    )
+    assert isinstance(text, AssistantText)
+    assert isinstance(call, ToolCall)
+    assert call.arguments == {"command": "pytest"}
+    assert call.call_id != "toolu_vendor_xyz"  # provider id never a join key
+    assert call.call_id.startswith("call-")
+
+
+def test_client_errors_are_llm_errors() -> None:
+    events = stored([UserMessage(text="go", source="m")])
+    with pytest.raises(LLMError, match="400"):
+        client(lambda r: httpx.Response(400, text="bad")).complete(
+            events, None, "claude-haiku-4-5"
+        )
+    with pytest.raises(LLMError, match="malformed"):
+        client(lambda r: httpx.Response(200, json={"oops": 1})).complete(
+            events, None, "claude-haiku-4-5"
+        )

tests/integration/test_passthrough_live.py

@@ -0,0 +1,108 @@
+"""Live Anthropic passthrough: the vendor-native route delegations will use.
+
+Proves the renderer's body is accepted as-is, that a tool call round-trips
+through the canonical log, and — per A5 — that prompt caching survives the
+passthrough rather than being flattened by the proxy.
+"""
+
+import io
+import os
+from pathlib import Path
+
+import httpx
+import pytest
+
+from nisse_dev.canonical import AssistantText, ToolCall, ToolResult, UserMessage
+from nisse_dev.envfile import load_env_file
+from nisse_dev.llm import AnthropicPassthroughClient
+from nisse_dev.oplog import OpLog
+from nisse_dev.state import StateStore
+from nisse_dev.tools import EXECUTOR_TOOLS
+
+load_env_file(Path(__file__).resolve().parents[2] / ".env")
+
+SENTINEL = "NISSE_LIVE_PLANNER_MODEL"
+
+pytestmark = [
+    pytest.mark.integration,
+    pytest.mark.skipif(
+        not os.environ.get(SENTINEL),
+        reason=f"no live frontier route configured ({SENTINEL} unset)",
+    ),
+]
+
+
+def store_with(tmp_path: Path, *events) -> StateStore:
+    store = StateStore(tmp_path / "state.db", oplog=OpLog("nisse-dev", io.StringIO()))
+    for event in events:
+        store.append_event(event)
+    return store
+
+
+def test_passthrough_completes_a_plain_turn(tmp_path: Path) -> None:
+    store = store_with(
+        tmp_path, UserMessage(text="Reply with exactly: pong", source="live")
+    )
+    result = AnthropicPassthroughClient.from_env().complete(
+        store.read_events(), None, os.environ[SENTINEL], system="Be terse."
+    )
+    texts = [e.text for e in result if isinstance(e, AssistantText)]
+    assert texts and "pong" in texts[0].lower()
+
+
+def test_passthrough_round_trips_a_tool_call(tmp_path: Path) -> None:
+    store = store_with(
+        tmp_path,
+        UserMessage(
+            text="List the files in the current directory using your tools.",
+            source="live",
+        ),
+    )
+    client = AnthropicPassthroughClient.from_env()
+    produced = client.complete(
+        store.read_events(),
+        EXECUTOR_TOOLS,
+        os.environ[SENTINEL],
+        system="Use a tool. Do not answer in prose.",
+    )
+    calls = [e for e in produced if isinstance(e, ToolCall)]
+    assert calls, f"expected a tool call, got {produced!r}"
+
+    # Feed the result back: the follow-up render must be accepted, which is
+    # what proves tool_result placement is right for this vendor.
+    for event in produced:
+        store.append_event(event)
+    store.append_event(
+        ToolResult(call_id=calls[0].call_id, result="app.py\ntests/", success=True)
+    )
+    follow_up = client.complete(
+        store.read_events(), EXECUTOR_TOOLS, os.environ[SENTINEL]
+    )
+    assert follow_up, "the vendor rejected or ignored the tool_result turn"
+
+
+def test_prompt_caching_survives_the_passthrough() -> None:
+    # A5 asks for this to be verified rather than assumed: a proxy that
+    # flattened the request would silently drop cache control and turn a
+    # cost optimisation into a per-call expense.
+    base = os.environ["NISSE_LLM_BASE_URL"].rstrip("/")
+    token = os.environ["NISSE_LLM_TOKEN"]
+    response = httpx.post(
+        f"{base}/v1/messages",
+        headers={
+            "Authorization": f"Bearer {token}",
+            "x-api-key": token,
+            "anthropic-version": "2023-06-01",
+        },
+        json={
+            "model": os.environ[SENTINEL],
+            "max_tokens": 16,
+            "messages": [{"role": "user", "content": "Say pong."}],
+        },
+        timeout=120,
+    )
+    response.raise_for_status()
+    usage = response.json().get("usage", {})
+    assert "cache_read_input_tokens" in usage, (
+        f"cache accounting absent from passthrough usage: {usage}"
+    )