diff --git a/nisse-dev/src/nisse_dev/llm.py b/nisse-dev/src/nisse_dev/llm.py index e6aa61e..269eece 100644 --- a/nisse-dev/src/nisse_dev/llm.py +++ b/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() diff --git a/nisse-dev/src/nisse_dev/render_anthropic.py b/nisse-dev/src/nisse_dev/render_anthropic.py new file mode 100644 index 0000000..88c1f6e --- /dev/null +++ b/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 diff --git a/nisse-dev/tests/test_render_anthropic.py b/nisse-dev/tests/test_render_anthropic.py new file mode 100644 index 0000000..f516d59 --- /dev/null +++ b/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" + ) diff --git a/tests/integration/test_passthrough_live.py b/tests/integration/test_passthrough_live.py new file mode 100644 index 0000000..f5c9191 --- /dev/null +++ b/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}" + )