From d6825175cacf23ddd0e00d8becae4e9fb5b7c74b Mon Sep 17 00:00:00 2001 From: Hans Date: Thu, 1 Oct 2026 17:59:02 +0800 Subject: [PATCH 1/3] fix(anthropic): fold the per-call runtime addendum into the preceding user message The runtime addendum (system_addendum_per_call_role="user") was sent as one more user message after the tool results. Anthropic accepts that, but translating gateways do not reliably: in Forge runs r-20261001-150020-9895 and r-20261001-160244-46f2 (deepseek-flash behind llm-hub, Anthropic dialect), all six tool-pairing 400s came from apodexharness requests shaped [assistant tool_use] [user tool_result] [user addendum]; claude-code, which sends one user message per turn, had none. Each 400 made the loop drop the episode's history and answer from a recovery context. Append the addendum as a text block after the tool results instead, and do it after the cache breakpoint is placed so the breakpoint stays on the last persistent block. The prefix through that block is unchanged turn to turn, so caching behaves as before. Not reproduced after the fact: replaying the three failing requests 10 times each, as recorded and folded, returned 200 every time, so the offending backend is intermittent. The folded shape is the one every translator handles. Co-Authored-By: Claude Opus 5.5 (1M context) --- agent_core/providers/anthropic.py | 33 +++++++++++++++++++ .../anthropic-fold-transient-addendum.fix.md | 1 + tests/test_provider_native_clients.py | 30 +++++++++++++---- 3 files changed, 57 insertions(+), 7 deletions(-) create mode 100644 changes/anthropic-fold-transient-addendum.fix.md 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/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/tests/test_provider_native_clients.py b/tests/test_provider_native_clients.py index 6d1726d..ab62ccd 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,17 @@ 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]"}]}] From 550374a7e29df625f3d1c6dbda30c28628132e79 Mon Sep 17 00:00:00 2001 From: Hans Date: Thu, 1 Oct 2026 19:36:25 +0800 Subject: [PATCH 2/3] fix(loop): mark the runaway-retry reminder per-call so providers fold it MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A reasoning-runaway retry appended its reminder as a plain user message after the transient runtime addendum. The addendum was then no longer the request's tail, so the Anthropic adapter folded nothing and the retry went out as [tool_result] [addendum] [reminder] — three consecutive user messages, the shape #54 exists to avoid. Seen in Forge run r-20261001-180502-801e: 8 of 222 apodexharness requests, all retries. The reminder is never stored in history, so mark it transient: the adapter folds both into the tool-result message and the cache breakpoint stays on the last tool_result. Co-Authored-By: Claude Opus 5.5 (1M context) --- agent_core/runtime/loop/_call.py | 10 +++++++++- changes/runaway-reminder-transient.fix.md | 1 + tests/test_provider_native_clients.py | 18 ++++++++++++++++++ tests/test_runaway_retry_guidance.py | 14 ++++++++++++++ 4 files changed, 42 insertions(+), 1 deletion(-) create mode 100644 changes/runaway-reminder-transient.fix.md diff --git a/agent_core/runtime/loop/_call.py b/agent_core/runtime/loop/_call.py index ee145f7..5803988 100644 --- a/agent_core/runtime/loop/_call.py +++ b/agent_core/runtime/loop/_call.py @@ -760,7 +760,15 @@ async def _finish_attempt( expanded_attempted=runaway_expanded, ) ) - messages_active = [*messages, user_msg(recovery_guidance)] + # Per-call, like the runtime addendum: the loop never stores + # it in history. Marked so providers treat it that way — + # the Anthropic adapter folds it into the preceding user + # message and keeps the cache breakpoint off it. Unmarked, + # the addendum ahead of it was no longer the tail, so the + # request went out as three consecutive user messages. + reminder = user_msg(recovery_guidance) + reminder["transient"] = True + messages_active = [*messages, reminder] await _finish_attempt( outcome=ATTEMPT_DISCARDED, reason="reasoning_runaway", diff --git a/changes/runaway-reminder-transient.fix.md b/changes/runaway-reminder-transient.fix.md new file mode 100644 index 0000000..fc79553 --- /dev/null +++ b/changes/runaway-reminder-transient.fix.md @@ -0,0 +1 @@ +The runaway-retry reminder is now marked per-call (`transient`), 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. diff --git a/tests/test_provider_native_clients.py b/tests/test_provider_native_clients.py index ab62ccd..da097fc 100644 --- a/tests/test_provider_native_clients.py +++ b/tests/test_provider_native_clients.py @@ -1237,3 +1237,21 @@ def test_anthropic_transient_addendum_after_a_plain_user_turn_is_folded(monkeypa 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 From 0afdbb79554a285aff031e36ad8b136698559469 Mon Sep 17 00:00:00 2001 From: zhanghanduo Date: Thu, 1 Oct 2026 20:54:05 +0800 Subject: [PATCH 3/3] fix(runtime): mark all runaway retry reminders transient --- agent_core/runtime/loop/_call.py | 25 ++-- changes/runaway-reminder-transient.fix.md | 2 +- tests/test_runaway_retry_transient.py | 160 ++++++++++++++++++++++ 3 files changed, 175 insertions(+), 12 deletions(-) create mode 100644 tests/test_runaway_retry_transient.py diff --git a/agent_core/runtime/loop/_call.py b/agent_core/runtime/loop/_call.py index 5803988..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,15 +771,7 @@ async def _finish_attempt( expanded_attempted=runaway_expanded, ) ) - # Per-call, like the runtime addendum: the loop never stores - # it in history. Marked so providers treat it that way — - # the Anthropic adapter folds it into the preceding user - # message and keeps the cache breakpoint off it. Unmarked, - # the addendum ahead of it was no longer the tail, so the - # request went out as three consecutive user messages. - reminder = user_msg(recovery_guidance) - reminder["transient"] = True - messages_active = [*messages, reminder] + messages_active = _runaway_retry_messages(messages, recovery_guidance) await _finish_attempt( outcome=ATTEMPT_DISCARDED, reason="reasoning_runaway", @@ -890,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", @@ -1035,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/runaway-reminder-transient.fix.md b/changes/runaway-reminder-transient.fix.md index fc79553..f772f79 100644 --- a/changes/runaway-reminder-transient.fix.md +++ b/changes/runaway-reminder-transient.fix.md @@ -1 +1 @@ -The runaway-retry reminder is now marked per-call (`transient`), 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. +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_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"]