fix(anthropic): fold the per-call runtime addendum into the preceding user message

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."""

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",

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.

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.

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:])

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

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"]