diff --git a/agent_core/providers/anthropic.py b/agent_core/providers/anthropic.py index 2b8ac6e..43e010b 100644 --- a/agent_core/providers/anthropic.py +++ b/agent_core/providers/anthropic.py @@ -136,6 +136,9 @@ def _build_kwargs( elif self.default_timeout is not None: kwargs["timeout"] = self.default_timeout _add_prompt_cache(kwargs, transient_tail=transient_tail) + # After the cache breakpoint is placed, so it stays on the last + # persistent block rather than moving onto the per-call text. + kwargs["messages"] = _fold_transient_tail(kwargs["messages"], transient_tail) return kwargs async def chat( @@ -412,6 +415,36 @@ def only_results(msg: dict[str, Any]) -> bool: return out +def _fold_transient_tail(msgs: list[dict[str, Any]], transient_tail: int) -> list[dict[str, Any]]: + """Append trailing per-call user text to the user message before it. + + The runtime addendum (``system_addendum_per_call_role="user"``) arrives as + one more user message after the tool results. Anthropic accepts the pair, + but translating gateways do not reliably: llm-hub in front of deepseek-flash + answered six of 195 requests in one Forge run (2026-10-01) with + "An assistant message with 'tool_calls' must be followed by tool messages + responding to each 'tool_call_id'", and the loop then gave up on the + episode's history. One user message holding the results and then the text + is the shape every translator maps cleanly. + """ + if transient_tail <= 0 or len(msgs) <= transient_tail: + return msgs + head, tail = msgs[:-transient_tail], msgs[-transient_tail:] + target = head[-1] + if target.get("role") != "user" or any(m.get("role") != "user" for m in tail): + return msgs + blocks = target.get("content") + blocks = [{"type": "text", "text": blocks}] if isinstance(blocks, str) else list(blocks or []) + for m in tail: + extra = m.get("content") + if isinstance(extra, str): + if extra: + blocks.append({"type": "text", "text": extra}) + elif isinstance(extra, list): + blocks.extend(extra) + return [*head[:-1], {**target, "content": blocks}] + + def _split_system(messages: list[Message]) -> tuple[str, list[Message]]: """Pull out the (single) leading system message; Anthropic takes it as a top-level kwarg, not as a message.""" diff --git a/agent_core/runtime/loop/_call.py b/agent_core/runtime/loop/_call.py index ee145f7..4dfacdc 100644 --- a/agent_core/runtime/loop/_call.py +++ b/agent_core/runtime/loop/_call.py @@ -170,6 +170,17 @@ def _stream_recovery_budget_too_small( and remaining_s < attempt_budget_s / 2 ) +def _runaway_retry_messages(messages: list[Message], guidance: str) -> list[Message]: + """Append per-call guidance without persisting or caching the reminder. + + Every recovery branch uses the same projection so providers can fold the + reminder and any runtime addendum into the preceding persistent user turn. + """ + reminder = user_msg(guidance) + reminder["transient"] = True + return [*messages, reminder] + + async def call_llm( llm: Any, messages: list[Message], @@ -760,7 +771,7 @@ async def _finish_attempt( expanded_attempted=runaway_expanded, ) ) - messages_active = [*messages, user_msg(recovery_guidance)] + messages_active = _runaway_retry_messages(messages, recovery_guidance) await _finish_attempt( outcome=ATTEMPT_DISCARDED, reason="reasoning_runaway", @@ -882,7 +893,7 @@ async def _finish_attempt( expanded_attempted=runaway_expanded, ) ) - messages_active = [*messages, user_msg(recovery_guidance)] + messages_active = _runaway_retry_messages(messages, recovery_guidance) await _finish_attempt( outcome=ATTEMPT_DISCARDED, reason="reasoning_runaway_early", @@ -1027,7 +1038,7 @@ async def _finish_attempt( retry_thinking, recovery_guidance, _ = _runaway_retry_policy( runaway_retries, next_cap, ) - messages_active = [*messages, user_msg(recovery_guidance)] + messages_active = _runaway_retry_messages(messages, recovery_guidance) await _finish_attempt( outcome=ATTEMPT_DISCARDED, reason="context_length", diff --git a/changes/anthropic-fold-transient-addendum.fix.md b/changes/anthropic-fold-transient-addendum.fix.md new file mode 100644 index 0000000..f6bdde7 --- /dev/null +++ b/changes/anthropic-fold-transient-addendum.fix.md @@ -0,0 +1 @@ +Anthropic requests fold the trailing per-call runtime addendum into the preceding user message instead of sending a second consecutive user message; translating gateways (llm-hub before deepseek-flash) intermittently rejected the split form. The prompt-cache breakpoint stays on the last persistent block. diff --git a/changes/runaway-reminder-transient.fix.md b/changes/runaway-reminder-transient.fix.md new file mode 100644 index 0000000..f772f79 --- /dev/null +++ b/changes/runaway-reminder-transient.fix.md @@ -0,0 +1 @@ +The runaway-retry reminder is now marked per-call (`transient`) in every recovery path, including early-stopped streams and retries after expanded-context overflow, like the runtime addendum it follows. Unmarked, it ended the Anthropic fold of that addendum and a retried request went out as three consecutive user messages. Cache breakpoints stay on persistent content and reminders never accumulate in history. diff --git a/tests/test_provider_native_clients.py b/tests/test_provider_native_clients.py index 6d1726d..da097fc 100644 --- a/tests/test_provider_native_clients.py +++ b/tests/test_provider_native_clients.py @@ -1169,8 +1169,12 @@ def test_anthropic_rolling_breakpoint_skips_transient_addendum(monkeypatch): call2 = {**call, "id": "t2"} history += [assistant_msg("", tool_calls=[call2]), tool_msg("t2", "out2")] second = build([*history, _transient("[Runtime environment metadata]")]) - # Everything up to last turn's breakpoint is byte-identical this turn. - assert _strip_cache(first["messages"][:3]) == _strip_cache(second["messages"][:3]) + # Everything up to and including last turn's breakpoint block is + # byte-identical this turn. The addendum is folded into message 2 after + # that block, which the cached prefix does not reach. + first_prefix = _strip_cache(first["messages"][:3]) + first_prefix[2] = {**first_prefix[2], "content": first_prefix[2]["content"][:-1]} + assert first_prefix == _strip_cache(second["messages"][:3]) assert _tail_breakpoints(second) == [4] @@ -1204,9 +1208,9 @@ def test_anthropic_parallel_tool_results_share_one_user_message(monkeypatch): tools=None, temperature=None, max_tokens=None, extra_headers=None, timeout=None, ) msgs = kwargs["messages"] - assert [m["role"] for m in msgs] == ["user", "assistant", "user", "user"] - assert [b["tool_use_id"] for b in msgs[2]["content"]] == ["a", "b"] - assert msgs[3]["content"] == "[env]" + assert [m["role"] for m in msgs] == ["user", "assistant", "user"] + assert [b.get("tool_use_id") for b in msgs[2]["content"]] == ["a", "b", None] + assert msgs[2]["content"][-1] == {"type": "text", "text": "[env]"} def test_anthropic_cache_breakpoint_after_merged_tool_results(monkeypatch): @@ -1219,5 +1223,35 @@ def test_anthropic_cache_breakpoint_after_merged_tool_results(monkeypatch): tool_msg("ra", "a"), tool_msg("rb", "b"), _transient("[env]")], tools=None, temperature=None, max_tokens=None, extra_headers=None, timeout=None, ) - assert _tail_breakpoints(kwargs) == [2] - assert "cache_control" in kwargs["messages"][2]["content"][-1] + # The breakpoint stays on the last tool_result, not on the folded-in + # per-call text that the next request will not repeat. + content = kwargs["messages"][2]["content"] + assert "cache_control" in content[-2] and "cache_control" not in content[-1] + + +def test_anthropic_transient_addendum_after_a_plain_user_turn_is_folded(monkeypatch): + monkeypatch.setenv("ANTHROPIC_PROMPT_CACHE", "0") + c = ac.AnthropicClient("claude-x", api_key="x") + kwargs = c._build_kwargs([system_msg("s"), user_msg("q"), _transient("[env]")], + tools=None, temperature=None, max_tokens=None, + extra_headers=None, timeout=None) + assert kwargs["messages"] == [{"role": "user", "content": [ + {"type": "text", "text": "q"}, {"type": "text", "text": "[env]"}]}] + + +def test_anthropic_folds_every_trailing_per_call_message(monkeypatch): + """A runaway retry adds its reminder after the runtime addendum: both are + per-call and both join the tool results, breakpoint on the last result.""" + monkeypatch.delenv("ANTHROPIC_PROMPT_CACHE", raising=False) + c = ac.AnthropicClient("claude-x", api_key="x") + call = {"id": "a", "type": "function", "function": {"name": "bash", "arguments": "{}"}} + kwargs = c._build_kwargs( + [system_msg("s"), user_msg("q"), assistant_msg("", tool_calls=[call]), + tool_msg("ra", "a"), _transient("[env]"), _transient("[system reminder] retry")], + tools=None, temperature=None, max_tokens=None, extra_headers=None, timeout=None, + ) + msgs = kwargs["messages"] + assert [m["role"] for m in msgs] == ["user", "assistant", "user"] + content = msgs[2]["content"] + assert [b["type"] for b in content] == ["tool_result", "text", "text"] + assert "cache_control" in content[0] and all("cache_control" not in b for b in content[1:]) diff --git a/tests/test_runaway_retry_guidance.py b/tests/test_runaway_retry_guidance.py index 2c5e522..5a428ad 100644 --- a/tests/test_runaway_retry_guidance.py +++ b/tests/test_runaway_retry_guidance.py @@ -111,3 +111,17 @@ async def on_delta(*_args: Any, **_kwargs: Any) -> None: reminder = llm.requests[1][-1]["content"] assert "previous attempt stopped" in reminder assert "full private-reasoning budget" not in reminder + + +def test_retry_reminder_is_per_call_so_providers_can_fold_it(monkeypatch: pytest.MonkeyPatch) -> None: + """Like the runtime addendum it follows, the reminder is never stored in + history; unmarked, it ended the Anthropic fold of the addendum and the + retry went out as three consecutive user messages.""" + monkeypatch.setattr(_call, "_RUNAWAY_EXPAND_ENABLED", False) + monkeypatch.setattr(_call, "_RUNAWAY_BACKOFF_S", 0.0) + llm = _RunawayThenAnswer() + addendum = {"role": "user", "content": "[env]", "transient": True} + asyncio.run(_call.call_llm(llm, [{"role": "user", "content": "q"}, addendum], + timeout=30, max_retries=3, turn=1)) + retry = llm.requests[1] + assert retry[-2] is addendum and retry[-1].get("transient") is True diff --git a/tests/test_runaway_retry_transient.py b/tests/test_runaway_retry_transient.py new file mode 100644 index 0000000..6266ece --- /dev/null +++ b/tests/test_runaway_retry_transient.py @@ -0,0 +1,160 @@ +"""Every runaway recovery path must preserve the Anthropic request boundary.""" + +from __future__ import annotations + +from copy import deepcopy +from types import SimpleNamespace +from typing import Any +from unittest.mock import AsyncMock + +import pytest + +from agent_core.messages import Message, assistant_msg, for_wire, system_msg, tool_msg, user_msg +from agent_core.providers.anthropic import AnthropicClient +from agent_core.runtime.loop import _call +from agent_core.runtime.loop._bind import bind_max_tokens + + +@pytest.mark.parametrize("retry_path", [ + "completed_chat", "completed_stream", "early_stream", "overflow_chat", "overflow_stream", +]) +@pytest.mark.parametrize("history_kind", ["plain_user", "parallel_tools"]) +@pytest.mark.parametrize("with_addendum", [False, True]) +@pytest.mark.parametrize("cache_enabled", [False, True]) +@pytest.mark.asyncio +async def test_all_runaway_retries_fold_without_caching_or_persisting_reminders( + monkeypatch: pytest.MonkeyPatch, + retry_path: str, + history_kind: str, + with_addendum: bool, + cache_enabled: bool, +) -> None: + monkeypatch.setenv("ANTHROPIC_PROMPT_CACHE", "1" if cache_enabled else "0") + monkeypatch.setattr(_call, "_RUNAWAY_MAX_RETRIES", 3) + monkeypatch.setattr(_call, "_RUNAWAY_EXPAND_ENABLED", True) + monkeypatch.setattr(_call, "_RUNAWAY_BACKOFF_S", 0.0) + client = AnthropicClient("claude-test", api_key="test-key") + messages = [system_msg("system"), user_msg("question")] + if history_kind == "parallel_tools": + messages += [ + assistant_msg("", tool_calls=[{ + "id": call_id, "type": "function", + "function": {"name": "bash", "arguments": "{}"}, + } for call_id in ("a", "b")]), + tool_msg("result a", "a"), tool_msg("result b", "b"), + ] + if with_addendum: + messages.append({**user_msg("[env]"), "transient": True}) + original = deepcopy(messages) + projections: list[list[Message]] = [] + build_kwargs = client._build_kwargs + + def capture_projection(request: list[Message], **kwargs: Any) -> dict[str, Any]: + projections.append(deepcopy(request)) + return build_kwargs(request, **kwargs) + + monkeypatch.setattr(client, "_build_kwargs", capture_projection) + wire_requests: list[dict[str, Any]] = [] + + async def create(**kwargs: Any) -> Any: + wire_requests.append(deepcopy(kwargs)) + attempt = len(wire_requests) + if retry_path.startswith("overflow") and attempt == 2: + raise RuntimeError("maximum context length exceeded") + done = attempt == 4 + if not kwargs.get("stream"): + return SimpleNamespace( + content=[SimpleNamespace(type="text", text="done")] if done else [ + SimpleNamespace(type="thinking", thinking="x" * 800, signature="sig"), + ], + stop_reason="end_turn" if done else "max_tokens", + model="claude-test", id=f"response-{attempt}", + usage=SimpleNamespace(input_tokens=10, output_tokens=2 if done else 2048), + ) + + async def events() -> Any: + block_type = "text" if done else "thinking" + yield SimpleNamespace( + type="content_block_start", index=0, + content_block=SimpleNamespace(type=block_type), + ) + yield SimpleNamespace( + type="content_block_delta", index=0, + delta=SimpleNamespace( + type="text_delta" if done else "thinking_delta", + text="done" if done else "", thinking="" if done else "x" * 800, + ), + ) + yield SimpleNamespace( + type="message_delta", + delta=SimpleNamespace(stop_reason="end_turn" if done else "max_tokens"), + usage=SimpleNamespace(output_tokens=2 if done else 2048), + ) + + return events() + + client._client = SimpleNamespace(messages=SimpleNamespace(create=AsyncMock(side_effect=create))) + attempts: list[dict[str, Any]] = [] + + async def on_attempt(event: dict[str, Any]) -> None: + attempts.append(event) + + async def on_delta(*_args: Any, **_kwargs: Any) -> None: + pass + + response = await _call.call_llm( + bind_max_tokens(client, 2048), messages, timeout=30, max_retries=4, turn=1, + on_delta=on_delta if retry_path.endswith("stream") else None, + reasoning_only_max_tokens=100 if retry_path == "early_stream" else None, + on_attempt=on_attempt, + ) + assert response is not None and response.content == "done" + assert len(wire_requests) == len(projections) == 4 + # Exercise all three retry phases, including the context-overflow downgrade. + started = [event for event in attempts if event["phase"] == "started"] + assert [event["thinking_mode"] for event in started] == [ + "profile_default", "expanded", "reduced", "disabled", + ] + finished = [event for event in attempts if event["phase"] == "finished"] + if retry_path == "early_stream": + expected_reasons = ["reasoning_runaway_early"] * 3 + [""] + elif retry_path.startswith("overflow"): + expected_reasons = ["reasoning_runaway", "context_length", "reasoning_runaway", ""] + else: + expected_reasons = ["reasoning_runaway"] * 3 + [""] + assert [event["reason"] for event in finished] == expected_reasons + assert projections[0] == original + assert messages == original + for projection, request in zip(projections[1:], wire_requests[1:], strict=True): + # Each retry replaces the reminder; neither the addendum nor reminders + # accumulate in history or leak their bookkeeping onto the wire. + assert projection[:-1] == original + reminder = projection[-1] + assert reminder["role"] == "user" and reminder["transient"] is True + assert str(reminder["content"]).startswith("[system reminder]") + assert "transient" not in for_wire([reminder])[0] + wire = request["messages"] + assert all("transient" not in message for message in wire) + assert [message["role"] for message in wire] == ( + ["user", "assistant", "user"] if history_kind == "parallel_tools" else ["user"] + ) + content = wire[-1]["content"] + persistent = [ + {"type": "tool_result", "tool_use_id": "a", "content": "result a"}, + {"type": "tool_result", "tool_use_id": "b", "content": "result b"}, + ] if history_kind == "parallel_tools" else [{"type": "text", "text": "question"}] + if cache_enabled: + persistent[-1]["cache_control"] = {"type": "ephemeral"} + assert content == [ + *persistent, + *([{"type": "text", "text": "[env]"}] if with_addendum else []), + {"type": "text", "text": reminder["content"]}, + ] + assert [i for i, block in enumerate(content) if "cache_control" in block] == ( + [len(persistent) - 1] if cache_enabled else [] + ) + # Rebuilding the original history for another logical call drops every reminder. + next_call = build_kwargs( + messages, tools=None, temperature=None, max_tokens=None, extra_headers=None, timeout=None, + ) + assert next_call["messages"] == wire_requests[0]["messages"]