Add the Anthropic passthrough renderer and client (A10/A13) — 7a
CI / checks (pull_request) Successful in 43s
CI / checks (push) Successful in 44s

The second renderer adapter, and the concrete case for A13's decision
to normalise in the renderer rather than on the wire. The two vendors
disagree structurally, not cosmetically: the system prompt is a
top-level field rather than a message, a tool call is a tool_use
content block rather than an array, a tool result is a block inside a
user message rather than a "tool" role, and tool schemas use
input_schema rather than parameters. A middlebox flattening those
would be lossy exactly where a long-running delegation lives.

- render_anthropic.py builds vendor-native /v1/messages bodies from
  the canonical log, coalescing adjacent same-role blocks so a
  tool_result lands directly after its tool_use
- Tool manifests are declared once in OpenAI shape and converted per
  vendor, so adding a tool does not mean editing two places
- AnthropicPassthroughClient implements the same LLMClient protocol,
  so callers pick a route rather than a different kind of client, and
  maps provider tool ids to canonical ones at the boundary
- Dangling tool calls get a placeholder result here too; this API
  rejects them more strictly than the OpenAI shape
- Vendor-opaque reasoning blocks are dropped for now with the reason
  recorded: A10 wants them stored and never replayed across sessions,
  and the canonical schema has no event for that yet

Live tests prove the body is accepted as sent, that a tool_result
turn round-trips, and — as A5 asks — that prompt cache accounting
survives the passthrough rather than being flattened by the proxy.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
dongarandClaude Opus 5 committed 2026-09-07 23:13:22 +00:00
1 parent 1f292793c1
commit f4a94281ef
4 files changed
+582

No files matched your search

+108
View File
@@ -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}"
)