tests/backends/providers/test_anthropic_wire.py
Rama main · réplica de lectura del repositorio original.
"""Wire rendering of :class:`AnthropicProvider`: host-injected turns,
output_config / thinking, images, empty text blocks and prompt-cache
breakpoints."""
from __future__ import annotations
import base64
import json
from typing import Any
import pytest
import respx
from noeta.core.errors import (
Fatal,
)
from noeta.core.types import (
Image,
Message,
SystemPart,
Text,
Thinking,
ToolResult,
ToolUse,
user,
)
from tests.backends.providers._anthropic_helpers import (
_PNG,
_TEXT_ONLY_MODEL,
_UNKNOWN_MODEL,
_VISION_MODEL,
EPHEMERAL,
MESSAGES_ENDPOINT,
_call,
_injected,
_provider,
_req,
_resp,
_respond,
_result,
_send,
_stream,
_tool,
register_text_only_model,
)
@pytest.fixture(autouse=True)
def _text_only_catalog_row(monkeypatch: pytest.MonkeyPatch) -> None:
register_text_only_model(monkeypatch)
# ---------------------------------------------------------------------------
# origin rendering — the vendor tag syntax lives only in the adapter
# ---------------------------------------------------------------------------
async def _wire_messages(messages: list[Message], **req_kw: Any) -> list[dict[str, Any]]:
body, _ = await _send(_req(messages=messages, **req_kw))
return list(body["messages"])
@respx.mock
async def test_origin_system_wrapped_and_merged_into_prev_user_turn() -> None:
wire = await _wire_messages([user("real human words"), _injected("host says hi", "system")])
assert wire == [
{
"role": "user",
"content": [
# Claude Code joins two user messages with a newline.
{"type": "text", "text": "real human words\n"},
{"type": "text", "text": "<system-reminder>\nhost says hi\n</system-reminder>"},
],
}
]
@respx.mock
async def test_origin_memory_before_user_merges_into_next_user_turn() -> None:
wire = await _wire_messages([_injected("recalled note", "memory"), user("the actual ask")])
assert wire == [
{
"role": "user",
"content": [
{"type": "text",
"text": "<system-reminder>\nrecalled note\n</system-reminder>\n"},
{"type": "text", "text": "the actual ask"},
],
}
]
@respx.mock
async def test_origin_system_with_no_adjacent_user_turn_stands_alone() -> None:
assistant = Message(role="assistant", content=[Text(text="working on it")])
wire = await _wire_messages([user("q"), assistant, _injected("mid-loop reminder", "system")])
assert [m["role"] for m in wire] == ["user", "assistant", "user"]
# One text block and no breakpoint: a bare string, as Claude Code sends it.
assert wire[2]["content"] == "<system-reminder>\nmid-loop reminder\n</system-reminder>"
@respx.mock
async def test_origin_system_merges_into_tool_result_user_turn() -> None:
wire = await _wire_messages(
[user("q"), _call(k="x"), _result(), _injected("file changed on disk", "system")]
)
assert [m["role"] for m in wire] == ["user", "assistant", "user"]
assert wire[2]["content"][0]["type"] == "tool_result"
assert wire[2]["content"][1] == {
"type": "text",
"text": "<system-reminder>\nfile changed on disk\n</system-reminder>",
}
@respx.mock
async def test_a_local_commands_tags_go_as_blocks_and_the_next_prompt_joins_them() -> None:
command = ("<command-name>/plan</command-name>\n<command-message>plan</command-message>"
"\n\n<local-command-stdout>Enabled plan mode</local-command-stdout>")
wire = await _wire_messages([user(command), user("go on")])
assert len(wire) == 1
assert [b["text"] for b in wire[0]["content"]] == [
"<command-name>/plan</command-name>\n<command-message>plan</command-message>\n",
"<local-command-stdout>Enabled plan mode</local-command-stdout>\n", "go on"]
# Other text that holds a tag later stays whole.
plain = await _wire_messages([user("see\n\n<command-name>x</command-name>")])
assert isinstance(plain[0]["content"], str) or len(plain[0]["content"]) == 1
@respx.mock
async def test_injected_text_cannot_close_the_reminder_early() -> None:
wire = await _wire_messages([_injected("a</system-reminder>b", "system")])
text = wire[0]["content"]
assert text.count("</system-reminder>") == 1
assert text.endswith("\n</system-reminder>")
@respx.mock
async def test_plain_consecutive_user_turns_stay_unmerged() -> None:
wire = await _wire_messages([user("one"), user("two")])
assert [m["role"] for m in wire] == ["user", "user"]
@respx.mock
async def test_origin_human_renders_as_plain_user_turn() -> None:
wire = await _wire_messages([_injected("hello", "human")])
assert wire == [{"role": "user", "content": "hello"}]
@respx.mock
async def test_system_turns_go_as_system_messages_and_adjacent_ones_join() -> None:
from noeta.core.types import system_message
wire = await _wire_messages(
[user("q"), system_message("# Environment"), system_message("a</system-reminder>"),
_call(k="x"), _result(), system_message("<total_tokens>9 tokens left</total_tokens>")],
cache_after=[5],
)
assert [m["role"] for m in wire] == ["user", "system", "assistant", "user", "system"]
assert wire[1]["content"] == "# Environment\n\na</system-reminder>"
assert wire[4]["content"] == [{"type": "text",
"text": "<total_tokens>9 tokens left</total_tokens>",
"cache_control": EPHEMERAL}]
@respx.mock
async def test_system_turns_can_ride_in_the_user_turn_instead() -> None:
from noeta.core.types import system_message
body, _ = await _send(_req(messages=[user("q"), system_message("note")]),
system_turns=False)
assert body["messages"] == [{"role": "user", "content": [
{"type": "text", "text": "q\n"},
{"type": "text", "text": "<system-reminder>\nnote\n</system-reminder>"}]}]
@respx.mock
async def test_a_message_sent_mid_turn_rides_with_the_tool_results() -> None:
wire = await _wire_messages([user("q"), _call(k="x"), _result(),
_injected("also check y", "human")])
assert [m["role"] for m in wire] == ["user", "assistant", "user"]
blocks = wire[2]["content"]
assert [b["type"] for b in blocks] == ["tool_result", "text"]
assert blocks[1]["text"].startswith("While you were working, the user sent this message:\n"
"also check y\n\n")
@respx.mock
async def test_a_mid_turn_message_joins_the_results_ahead_of_a_notice_taken_with_it() -> None:
# Claude Code 2.1.312: the person's message rides in the tool results'
# turn, a background job's end notice taken in the same step is the system
# turn after it (with the step's notes).
from noeta.core.types import system_message
wire = await _wire_messages([user("q"), _call(k="x"), _result(), system_message("job ended"),
_injected("also check y", "human"),
system_message("<total_tokens>9 tokens left</total_tokens>")])
assert [m["role"] for m in wire] == ["user", "assistant", "user", "system"]
assert [b["type"] for b in wire[2]["content"]] == ["tool_result", "text"]
assert wire[2]["content"][1]["text"].startswith("While you were working")
assert wire[3]["content"] == "job ended\n\n<total_tokens>9 tokens left</total_tokens>"
@respx.mock
async def test_the_empty_reply_nudge_joins_the_user_turn_before_the_steps_notes() -> None:
# Claude Code 2.1.312: the nudge rides in the tool results' turn, and the
# step's system note still ends the request (and carries the cache mark).
from noeta.core.types import system_message
from noeta.toolkit.empty_reply import EMPTY_REPLY_NUDGE
wire = await _wire_messages(
[user("q"), _call(k="x"), _result(),
system_message("<total_tokens>9 tokens left</total_tokens>"),
user(EMPTY_REPLY_NUDGE, origin="inline")],
cache_after=[4])
assert [m["role"] for m in wire] == ["user", "assistant", "user", "system"]
assert [b["type"] for b in wire[2]["content"]] == ["tool_result", "text"]
assert wire[2]["content"][1] == {"type": "text", "text": EMPTY_REPLY_NUDGE}
assert wire[3]["content"][-1]["cache_control"] == EPHEMERAL
@respx.mock
async def test_the_empty_reply_nudge_joins_the_prompt_before_the_environment() -> None:
from noeta.core.types import system_message
from noeta.toolkit.empty_reply import EMPTY_REPLY_NUDGE
wire = await _wire_messages([user("q"), system_message("# Environment"),
user(EMPTY_REPLY_NUDGE, origin="inline")])
assert [m["role"] for m in wire] == ["user", "system"]
# Claude Code 2.1.312 sends ``q\n`` there (run p-opus-first, request 2).
assert [b["text"] for b in wire[0]["content"]] == ["q\n", EMPTY_REPLY_NUDGE]
@respx.mock
async def test_api_key_request_settings_follow_claude_code() -> None:
import uuid
from noeta.backends.providers.anthropic import DEFAULT_BETAS
route = respx.post(MESSAGES_ENDPOINT).mock(return_value=_stream(_resp()))
root = uuid.uuid4().hex
await _provider(device_id="d" * 64).complete(_req(model=_VISION_MODEL, root_id=root))
sent = json.loads(route.calls.last.request.content)
# claude-opus-4-8 has no per-turn effort: Claude Code leaves that beta out.
assert route.calls.last.request.headers["anthropic-beta"] == ",".join(
b for b in DEFAULT_BETAS if b != "per-turn-control-2026-07-01")
assert sent["thinking"] == {"type": "adaptive", "display": "updates"}
assert sent["context_management"] == {
"edits": [{"type": "clear_thinking_20251015", "keep": "all"}]}
# Claude Code's catalog: Opus 4.8's default effort is high.
assert sent["output_config"] == {"effort": "high"}
# An opaque machine id, not Claude Code's JSON (no claim to be Claude Code).
assert sent["metadata"]["user_id"] == "d" * 64
assert "claude-code-20250219" not in route.calls.last.request.headers["anthropic-beta"]
@respx.mock
@pytest.mark.parametrize(("model", "max_tokens", "effort"), [
("claude-opus-5-5", 128_000, "medium"), ("claude-sonnet-5-5", 128_000, "medium"),
("claude-haiku-5-5", 128_000, "medium"), ("claude-fable-5-1", 64_000, "high"),
("claude-opus-5", 64_000, "high"), ("claude-sonnet-5", 64_000, "high"),
("claude-opus-4-8", 64_000, "high"), ("claude-sonnet-4-6", 32_000, "high"),
("claude-haiku-4-5", 32_000, None)])
async def test_output_cap_and_effort_default_to_claude_codes_catalog(
model: str, max_tokens: int, effort: str | None) -> None:
route = respx.post(MESSAGES_ENDPOINT).mock(return_value=_stream(_resp()))
await _provider(default_max_tokens=None).complete(
_req(model=model, max_tokens=None, thinking=True))
sent = json.loads(route.calls.last.request.content)
assert sent["max_tokens"] == max_tokens
assert sent.get("output_config", {}).get("effort") == effort
@respx.mock
@pytest.mark.parametrize(("model", "sent"), [
("claude-opus-5-5", True), ("claude-opus-5-5[1m]", True),
("us.anthropic.claude-sonnet-5-5-v1:0", True), ("claude-haiku-5-5", True),
("claude-fable-5-1", True), ("claude-opus-5", False), ("claude-sonnet-4-6", False),
("gateway/other-model", False), ("gpt-5.4", False)])
async def test_the_per_turn_beta_goes_only_with_models_that_have_per_turn_effort(
model: str, sent: bool) -> None:
from noeta.backends.providers.anthropic import DEFAULT_BETAS, PER_TURN_BETA
route = respx.post(MESSAGES_ENDPOINT).mock(return_value=_stream(_resp()))
await _provider().complete(_req(model=model, max_tokens=100))
betas = route.calls.last.request.headers["anthropic-beta"].split(",")
assert (PER_TURN_BETA in betas) is sent
assert [b for b in betas if b != PER_TURN_BETA] == [
b for b in DEFAULT_BETAS if b != PER_TURN_BETA]
@respx.mock
async def test_a_beta_set_the_host_chose_is_sent_as_given() -> None:
from noeta.backends.providers.anthropic import PER_TURN_BETA
route = respx.post(MESSAGES_ENDPOINT).mock(return_value=_stream(_resp()))
await _provider(betas=("one", PER_TURN_BETA)).complete(_req(model="gpt-5.4", max_tokens=9))
assert route.calls.last.request.headers["anthropic-beta"] == f"one,{PER_TURN_BETA}"
@respx.mock
async def test_betas_a_request_adds_take_claude_codes_order() -> None:
from noeta.backends.providers.anthropic import DEFAULT_BETAS
route = respx.post(MESSAGES_ENDPOINT).mock(return_value=_stream(_resp()))
request = _req(model="claude-opus-5-5", max_tokens=9,
betas=["dangerous-tool-use-2026-09-03", "afk-mode-2026-01-31", "zz-own"])
await _provider().complete(request)
# As recorded from Claude Code 2.1.312 in auto mode.
assert route.calls.last.request.headers["anthropic-beta"] == ",".join([
*DEFAULT_BETAS[:-1], "dangerous-tool-use-2026-09-03",
"thinking-display-updates-2026-08-18", "afk-mode-2026-01-31", "zz-own"])
# ---------------------------------------------------------------------------
# output_schema / effort / thinking
# ---------------------------------------------------------------------------
@respx.mock
async def test_output_schema_wired_to_output_config_json_schema() -> None:
schema = {"type": "object", "properties": {"n": {"type": "number"}}}
body, _ = await _send(_req(output_schema=schema))
assert body["output_config"] == {"format": {"type": "json_schema", "schema": schema}}
@respx.mock
async def test_effort_wired_to_output_config_effort() -> None:
body, _ = await _send(_req(effort="high")) # a model outside the catalog
assert body["output_config"] == {"effort": "high"}
@respx.mock
async def test_effort_on_a_registered_model_without_efforts_is_sent_when_set(
monkeypatch: pytest.MonkeyPatch) -> None:
"""A row a host registered with no ``efforts`` (a ``config.toml``
provider's ``models``): not a reasoning model, yet a task's effort goes
out; with none set, none does."""
register_text_only_model(monkeypatch)
body, _ = await _send(_req(model=_TEXT_ONLY_MODEL, effort="high"))
assert body["output_config"] == {"effort": "high"}
body, _ = await _send(_req(model=_TEXT_ONLY_MODEL))
assert "output_config" not in body
@respx.mock
@pytest.mark.parametrize("effort", ["xhigh", "max"])
async def test_top_effort_levels_pass_through_verbatim(effort: str) -> None:
body, _ = await _send(_req(effort=effort))
assert body["output_config"] == {"effort": effort}
@respx.mock
async def test_output_schema_and_effort_merge_inside_output_config() -> None:
schema = {"type": "string"}
body, _ = await _send(_req(output_schema=schema, effort="xhigh"))
assert body["output_config"] == {
"format": {"type": "json_schema", "schema": schema},
"effort": "xhigh",
}
@respx.mock
async def test_thinking_on_a_catalogued_reasoning_model_is_adaptive() -> None:
body, _ = await _send(_req(model=_VISION_MODEL))
assert body["thinking"] == {"type": "adaptive", "display": "updates"}
@respx.mock
async def test_thinking_off_is_explicitly_disabled() -> None:
body, _ = await _send(_req(thinking=False))
assert body["thinking"] == {"type": "disabled"}
@respx.mock
async def test_thinking_on_an_unknown_model_leaves_vendor_default() -> None:
body, _ = await _send(_req(model=_UNKNOWN_MODEL))
assert "thinking" not in body
assert "output_config" not in body
@respx.mock
async def test_a_claude_model_without_adaptive_thinking_gets_claude_codes_budget() -> None:
# Claude Code: the output ceiling less one, under max_tokens, at least 1,024.
body, _ = await _send(_req(model="claude-haiku-4-5", max_tokens=32_000))
assert body["thinking"] == {"type": "enabled", "budget_tokens": 31_999}
assert "output_config" not in body and "context_management" not in body
# No room for the API's least budget: no thinking (a 400 otherwise).
for cap in (500, 1024):
body, _ = await _send(_req(model="claude-haiku-4-5", max_tokens=cap))
assert "thinking" not in body
body, _ = await _send(_req(model="claude-haiku-4-5", max_tokens=4096))
assert body["thinking"] == {"type": "enabled", "budget_tokens": 4095}
@respx.mock
async def test_budget_thinking_waits_out_a_tool_loop_begun_without_it() -> None:
# A turn whose assistant message has a tool call and no thinking block
# (begun before the upgrade) continues without thinking; the API
# refuses budget thinking there.
loop = [Message(role="user", content=[Text(text="go")]),
Message(role="assistant", content=[ToolUse(id="t1", name="Read", input={})]),
Message(role="user", content=[ToolResult(call_id="t1", content=[Text(text="x")])])]
body, _ = await _send(_req(model="claude-haiku-4-5", max_tokens=32_000, messages=loop))
assert "thinking" not in body
# ---------------------------------------------------------------------------
# Images
# ---------------------------------------------------------------------------
def _image() -> Image:
return Image(media_type="image/png", data=_PNG)
_WIRE_IMAGE = {
"type": "image",
"source": {
"type": "base64",
"media_type": "image/png",
"data": base64.b64encode(_PNG).decode("ascii"),
},
}
def _image_result() -> Message:
return Message(
role="user",
content=[ToolResult(call_id="c1", content=[Text(text="read ok"), _image()])],
)
@respx.mock
async def test_tool_result_with_images_vision_model_emits_content_array() -> None:
body, _ = await _send(_req(model=_VISION_MODEL, messages=[user("q"), _call(), _image_result()]))
assert body["messages"][2]["content"][0]["content"] == [
{"type": "text", "text": "read ok"},
_WIRE_IMAGE,
]
@respx.mock
async def test_a_block_list_result_goes_as_a_list_even_as_plain_text() -> None:
plain = Message(role="user", content=[ToolResult(
call_id="c1", content=[Text(text="a"), Text(text="b")], block_list=True)])
body, _ = await _send(_req(model=_VISION_MODEL, messages=[user("q"), _call(), plain]))
assert body["messages"][2]["content"][0]["content"] == [
{"type": "text", "text": "a"}, {"type": "text", "text": "b"}]
@respx.mock
async def test_tool_result_with_images_non_vision_model_degrades_to_string() -> None:
body, _ = await _send(
_req(model=_TEXT_ONLY_MODEL, messages=[user("q"), _call(), _image_result()])
)
content = body["messages"][2]["content"][0]["content"]
assert isinstance(content, str)
assert content.startswith("read ok")
assert "image omitted" in content
@respx.mock
async def test_tool_result_images_inlined_on_unknown_model() -> None:
body, _ = await _send(_req(model=_UNKNOWN_MODEL, messages=[user("q"), _call(), _image_result()]))
assert body["messages"][2]["content"][0]["content"][1] == _WIRE_IMAGE
@respx.mock
async def test_image_in_user_message_vision_model_translates() -> None:
msg = Message(role="user", content=[Text(text="look"), _image()])
body, _ = await _send(_req(model=_VISION_MODEL, messages=[msg]))
assert body["messages"][0]["content"] == [{"type": "text", "text": "look"}, _WIRE_IMAGE]
@respx.mock
async def test_image_in_assistant_message_vision_model_translates() -> None:
assistant = Message(role="assistant", content=[Text(text="here"), _image()])
body, _ = await _send(_req(model=_VISION_MODEL, messages=[user("q"), assistant, user("n")]))
assert body["messages"][1]["content"] == [{"type": "text", "text": "here"}, _WIRE_IMAGE]
@respx.mock
@pytest.mark.parametrize("role", ["user", "assistant"])
async def test_image_bound_for_non_vision_model_guards_fatal_before_sending(role: str) -> None:
route = respx.post(MESSAGES_ENDPOINT).mock(return_value=_stream(_resp()))
msg = Message(role=role, content=[_image()]) # type: ignore[arg-type]
messages = [msg] if role == "user" else [user("q"), msg]
with pytest.raises(Fatal, match="supports_vision=False"):
await _provider().complete(_req(model=_TEXT_ONLY_MODEL, messages=messages))
assert not route.called
@respx.mock
async def test_image_on_unknown_model_passes_the_guard() -> None:
msg = Message(role="user", content=[_image()])
body, _ = await _send(_req(model=_UNKNOWN_MODEL, messages=[msg]))
assert body["messages"][0]["content"] == [_WIRE_IMAGE]
# ---------------------------------------------------------------------------
# Empty text blocks
# ---------------------------------------------------------------------------
@respx.mock
async def test_response_with_an_empty_text_block_drops_it() -> None:
response = await _respond(
_resp(
content=[
{"type": "text", "text": ""},
{"type": "text", "text": " \n"},
{"type": "tool_use", "id": "t", "name": "e", "input": {}},
],
stop_reason="tool_use",
)
)
assert response.message.content == [ToolUse(id="t", name="e", input={})]
@respx.mock
async def test_response_of_only_an_empty_text_block_is_empty() -> None:
response = await _respond(_resp(content=[{"type": "text", "text": ""}]))
assert response.message.content == []
@respx.mock
async def test_recorded_empty_text_block_is_not_resent() -> None:
"""History recorded with an empty text block must not go out: the API
rejects it ("text content blocks must be non-empty")."""
assistant = Message(
role="assistant", content=[Text(text=""), Text(text=" "), ToolUse(id="t", name="e", input={})]
)
body, _ = await _send(_req(messages=[user("q"), assistant, _result("t")]))
assert body["messages"][1]["content"] == [
{"type": "tool_use", "id": "t", "name": "e", "input": {}}
]
# ---------------------------------------------------------------------------
# Prompt caching — cache_control breakpoints
# ---------------------------------------------------------------------------
def _stamped(body: dict[str, Any]) -> list[tuple[int, int]]:
return [
(mi, bi)
for mi, m in enumerate(body["messages"])
for bi, block in enumerate(m["content"])
if "cache_control" in block
]
@respx.mock
async def test_no_breakpoints_unless_asked() -> None:
body, _ = await _send(_req(system=[SystemPart(text="s")], messages=[user("a"), user("b")]))
assert json.dumps(body).count('"cache_control"') == 0
@respx.mock
async def test_cache_control_stamped_on_cached_system_and_cache_after() -> None:
body, _ = await _send(
_req(
system=[SystemPart(text="you are helpful", cache=True), SystemPart(text="volatile")],
tools=[_tool("a"), _tool("z")],
messages=[user("first"), user("last message")],
cache_after=[1],
)
)
assert body["system"] == [
{"type": "text", "text": "you are helpful", "cache_control": EPHEMERAL},
{"type": "text", "text": "volatile"},
]
assert all("cache_control" not in t for t in body["tools"])
assert _stamped(body) == [(1, 0)]
assert json.dumps(body).count('"cache_control"') == 2
@respx.mock
async def test_cache_after_stamps_the_last_block_of_that_message() -> None:
"""The composer anchors on the last recorded message; a trailing
host-injected reminder merged into the same wire turn stays outside."""
wire = await _wire_messages(
[user("q"), _call(k="x"), _result(), _injected("todo: keep going", "system"),
_injected("recalled pointer", "memory")],
cache_after=[0, 2],
)
assert [m["role"] for m in wire] == ["user", "assistant", "user"]
blocks = wire[2]["content"]
assert [b["type"] for b in blocks] == ["tool_result", "text", "text"]
assert blocks[0]["cache_control"] == EPHEMERAL
assert all("cache_control" not in b for b in blocks[1:])
assert wire[0]["content"][-1]["cache_control"] == EPHEMERAL
assert json.dumps(wire).count('"cache_control"') == 2
@respx.mock
async def test_cache_prefix_is_stable_when_only_the_reminder_tail_changes() -> None:
"""Two steps resending the same history under a different reminder tail
produce byte-identical wire bytes through the breakpoint."""
history = [user("q"), _call(k="x"), _result()]
one = await _wire_messages([*history, _injected("todo: 2 left", "system")], cache_after=[2])
two = await _wire_messages([*history, _injected("todo: 1 left", "system")], cache_after=[2])
def through_breakpoint(wire: list[dict[str, Any]]) -> str:
out: list[Any] = []
for message in wire:
blocks = []
for block in message["content"]:
blocks.append(block)
if "cache_control" in block:
return json.dumps([*out, {**message, "content": blocks}])
out.append(message)
raise AssertionError("no breakpoint in messages")
assert through_breakpoint(one) == through_breakpoint(two)
assert one[-1]["content"][-1] != two[-1]["content"][-1]
@respx.mock
async def test_cache_after_on_a_message_with_no_wire_content_walks_back() -> None:
"""An assistant turn whose only block is foreign thinking renders empty;
its breakpoint lands on the previous message's last block."""
assistant = Message(role="assistant", content=[Thinking(text="x", family="openai_compat")])
body, _ = await _send(_req(messages=[user("q"), assistant], cache_after=[1]))
assert _stamped(body) == [(0, 0)]
@respx.mock
async def test_out_of_range_cache_after_is_ignored() -> None:
body, _ = await _send(_req(cache_after=[5, -1]))
assert json.dumps(body).count('"cache_control"') == 0
@respx.mock
async def test_breakpoints_capped_at_four_keeping_messages_then_system() -> None:
"""Anthropic allows four breakpoints. Message breakpoints (newest first)
win, then cached system parts (last first); tools get none."""
body, _ = await _send(
_req(
system=[SystemPart(text="a", cache=True), SystemPart(text="b", cache=True)],
tools=[_tool()],
messages=[user("1"), user("2"), user("3")],
cache_after=[0, 2],
)
)
assert _stamped(body) == [(0, 0), (2, 0)]
assert [("cache_control" in s) for s in body["system"]] == [True, True]
assert "cache_control" not in body["tools"][0]
assert json.dumps(body).count('"cache_control"') == 4
@respx.mock
async def test_many_message_breakpoints_keep_the_newest_four() -> None:
body, _ = await _send(
_req(
system=[SystemPart(text="s", cache=True)],
messages=[user(str(i)) for i in range(6)],
cache_after=[0, 1, 2, 3, 4, 5],
)
)
assert _stamped(body) == [(2, 0), (3, 0), (4, 0), (5, 0)]
assert "cache_control" not in body["system"][0]
@respx.mock
async def test_parallel_step_previous_breakpoint_reads_back_the_prior_entry() -> None:
"""A step of twelve parallel calls appends ~25 blocks — past Anthropic's
lookback. Two message breakpoints (the last recorded block and the
previous step's) re-read the previous request's prefix."""
first = _call("r0", "read", p="a")
first_result = _result("r0", "body")
fan = Message(
role="assistant",
content=[Text(text="Reading in parallel.")]
+ [ToolUse(id=f"p{k}", name="grep", input={"k": k}) for k in range(12)],
)
fan_results = Message(
role="user",
content=[ToolResult(call_id=f"p{k}", content=[Text(text=f"hit {k}")]) for k in range(12)],
)
history = [user("q"), first, first_result, fan, fan_results]
request = _req(
system=[SystemPart(text="sys", cache=True)],
tools=[_tool()],
messages=[*history, _injected("todo: 1 left", "system")],
cache_after=[2, 4],
)
body, _ = await _send(request)
assert _stamped(body) == [(2, 0), (4, 11)]
assert "cache_control" not in body["messages"][4]["content"][-1]
assert json.dumps(body).count('"cache_control"') == 3
prev, _ = await _send(
_req(
system=[SystemPart(text="sys", cache=True)],
tools=[_tool()],
messages=[*history[:3], _injected("todo: 2 left", "system")],
cache_after=[2],
)
)
assert _stamped(prev)[-1] == (2, 0)
msgs, prev_msgs = body["messages"], prev["messages"]
assert json.dumps([*prev_msgs[:2], {**prev_msgs[2], "content": prev_msgs[2]["content"][:1]}]) == (
json.dumps([*msgs[:2], {**msgs[2], "content": msgs[2]["content"][:1]}])
)
@respx.mock
async def test_cache_control_never_mutates_the_request() -> None:
request = _req(system=[SystemPart(text="sys", cache=True)], tools=[_tool()], cache_after=[0])
before = request
snapshot = json.dumps(
[m.text for m in request.messages] + [p.text for p in request.system]
)
body, _ = await _send(request)
assert json.dumps(body).count('"cache_control"') == 2
assert request == before
assert snapshot == json.dumps(
[m.text for m in request.messages] + [p.text for p in request.system]
)
# A second build from the same request is identical (no stamps leak back).
provider = _provider()
assert provider.build_body(request) == provider.build_body(request)
@respx.mock
async def test_side_request_knobs_reach_the_wire() -> None:
"""Stop sequences, per-block breakpoints, extra betas and vendor fields
(what the auto-mode classifier and the first auto-mode request use);
``thinking=None`` sends no thinking parameter."""
route = respx.post(MESSAGES_ENDPOINT).mock(return_value=_stream(_resp()))
request = _req(
messages=[user("ctx"), Message(role="user", content=[Text(text="a"), Text(text="b"),
Text(text="c")])],
stop_sequences=["</block>"], cache_blocks=[(0, 0), (1, 1)], thinking=None,
betas=["afk-mode-2026-01-31"], vendor_fields={"safeguards": [{"type": "x"}]})
await _provider(betas=("one",)).complete(request)
sent = route.calls.last.request
body = json.loads(sent.content)
assert body["stop_sequences"] == ["</block>"]
assert "thinking" not in body
assert body["safeguards"] == [{"type": "x"}]
assert sent.headers["anthropic-beta"] == "one,afk-mode-2026-01-31"
assert body["messages"][0]["content"][0]["cache_control"] == EPHEMERAL
stamped = [b.get("cache_control") for b in body["messages"][1]["content"]]
assert stamped == [None, EPHEMERAL, None]
@respx.mock
async def test_the_opening_context_is_one_message_to_claude_code() -> None:
# Claude Code sends the opening context (instructions, workspace context,
# attribution) as blocks of one message: no newline between them; the
# prompt joined after them adds one, as a later reminder does.
wire = await _wire_messages([
_injected("instructions", "system"), _injected("context", "system"),
user("hi"), _injected("later note", "system"), user("again")])
assert [b["text"] for b in wire[0]["content"]] == [
"<system-reminder>\ninstructions\n</system-reminder>",
"<system-reminder>\ncontext\n</system-reminder>\n", "hi\n",
"<system-reminder>\nlater note\n</system-reminder>\n", "again"]
def test_tool_references_are_held_to_the_tools_the_request_offers() -> None:
"""A reference by an MCP tool's old name goes as the tool's current
name; one to a tool the request does not offer is left out, and a
result left with none says so in text (the API refuses a reference to
a tool the request does not define)."""
from noeta.backends.providers._anthropic_wire import REFERENCES_REMOVED
from noeta.backends.providers.anthropic import AnthropicProvider
from noeta.core.types import ModelRequest, ToolReference, ToolSpec
def refs(*names: str) -> ToolResult:
return ToolResult(call_id="c", content=[ToolReference(tool_name=n) for n in names])
def spec(name: str) -> ToolSpec:
return ToolSpec(name=name, description="d", input_schema={"type": "object"})
history = [user("q"),
Message(role="assistant", content=[ToolUse(id="c", name="ToolSearch",
input={"query": "gh"})]),
Message(role="user", content=[
refs("mcp__github__delete_repo", "WebFetch", "mcp__GitHub__list")])]
offered = [spec("ToolSearch"), spec("mcp__GitHub__delete_repo"), spec("mcp__GitHub__list")]
body = AnthropicProvider(api_key="k").build_body(
ModelRequest(model="m", messages=history, tools=offered, max_tokens=64))
content = body["messages"][2]["content"][0]["content"]
assert content == [{"type": "tool_reference", "tool_name": "mcp__GitHub__delete_repo"},
{"type": "tool_reference", "tool_name": "mcp__GitHub__list"}]
# A request offering none of them (the compaction's fallback call keeps
# only Read when tool search is off).
bare = AnthropicProvider(api_key="k").build_body(
ModelRequest(model="m", messages=history, tools=[spec("Read")], max_tokens=64))
assert bare["messages"][2]["content"][0]["content"] == [
{"type": "text", "text": REFERENCES_REMOVED}]
def test_the_compaction_fallback_call_sends_no_reference_to_a_tool_it_drops() -> None:
"""The fallback summarize call keeps Read, ToolSearch and the MCP
tools; a loaded built-in deferred tool the history references
(WebFetch) is not among them, so its reference does not go."""
from noeta.backends.providers.anthropic import AnthropicProvider
from noeta.core.types import ModelRequest, ToolReference, ToolSpec
from noeta.plugins.compaction.summary import fallback_request
def spec(name: str) -> ToolSpec:
return ToolSpec(name=name, description="d", input_schema={"type": "object"})
history = [user("q"),
Message(role="assistant", content=[ToolUse(id="c", name="ToolSearch",
input={"query": "web"})]),
Message(role="user", content=[ToolResult(call_id="c", content=[
ToolReference(tool_name="WebFetch"),
ToolReference(tool_name="mcp__db__query")])]),
user("summarize")]
main = ModelRequest(model="m", messages=history, max_tokens=64, tools=[
spec("Read"), spec("ToolSearch"), spec("WebFetch"), spec("mcp__db__query"),
ToolSpec(name="DeferredToolPlaceholder", description="d", input_schema={},
defer_loading=True)])
body = AnthropicProvider(api_key="k").build_body(fallback_request(main))
offered = {t["name"] for t in body["tools"]}
sent = [b["tool_name"] for m in body["messages"] if isinstance(m["content"], list)
for blk in m["content"] if blk.get("type") == "tool_result"
and isinstance(blk.get("content"), list)
for b in blk["content"] if b.get("type") == "tool_reference"]
assert sent == ["mcp__db__query"] and set(sent) <= offered