diff --git a/agent/conversation_loop.py b/agent/conversation_loop.py index 51147f209843..daa6cac807aa 100644 --- a/agent/conversation_loop.py +++ b/agent/conversation_loop.py @@ -1365,25 +1365,6 @@ def run_conversation( logger=request_logger, ) - # Apply Anthropic prompt caching for Claude models on native - # Anthropic, OpenRouter, and third-party Anthropic-compatible - # gateways. Auto-detected: if ``_use_prompt_caching`` is set, inject - # cache_control breakpoints for the static system prefix, full system - # prompt, and last two messages (or the legacy system-and-3 layout - # when no static prefix is available). - if agent._use_prompt_caching: - _static_system_prefix = getattr(agent, "_cached_system_prompt_static", None) - api_messages = apply_anthropic_cache_control( - api_messages, - cache_ttl=agent._cache_ttl, - native_anthropic=agent._use_native_cache_layout, - static_system_prefix=( - _static_system_prefix - if isinstance(_static_system_prefix, str) - else None - ), - ) - # Safety net: strip orphaned tool results / add stubs for missing # results before sending to the API. Runs unconditionally — not # gated on context_compressor — so orphans from session loading or @@ -1442,6 +1423,39 @@ def run_conversation( # the OpenAI SDK. Sanitizing here prevents the 3-retry cycle. _sanitize_messages_surrogates(api_messages) + # Apply Anthropic prompt caching for Claude models on native + # Anthropic, OpenRouter, and third-party Anthropic-compatible + # gateways. Auto-detected: if ``_use_prompt_caching`` is set, inject + # cache_control breakpoints for the static system prefix, full system + # prompt, and last two messages (or the legacy system-and-3 layout + # when no static prefix is available). + # + # Runs LAST, after every message mutation above. Marking earlier + # defeats the prefix stability the mutations exist to create: + # ``_apply_cache_marker`` rewrites ``content`` from a plain string + # into a ``[{"type": "text", ...}]`` block, so the marked messages + # no longer match the ``isinstance(content, str)`` test in the + # whitespace-normalization pass and silently keep their raw + # leading/trailing whitespace. A tool result ending in "\n" is + # therefore sent unstripped while it sits in the last-3 window and + # stripped once it rolls out of it — the same message, different + # bytes on consecutive turns, which breaks the prefix match at + # exactly the point the breakpoints were meant to protect. Marking + # last also keeps breakpoints off messages that the orphan sweep or + # the thinking-only drop is about to remove or merge away. + if agent._use_prompt_caching: + _static_system_prefix = getattr(agent, "_cached_system_prompt_static", None) + api_messages = apply_anthropic_cache_control( + api_messages, + cache_ttl=agent._cache_ttl, + native_anthropic=agent._use_native_cache_layout, + static_system_prefix=( + _static_system_prefix + if isinstance(_static_system_prefix, str) + else None + ), + ) + # Build a persistent-MoA request before measuring compression pressure. # MoA reference output is injected into the aggregator prompt, but it # is deliberately ephemeral and therefore absent from ``messages``. diff --git a/tests/agent/test_prompt_caching.py b/tests/agent/test_prompt_caching.py index b0b94244fbe8..ba21e4c91951 100644 --- a/tests/agent/test_prompt_caching.py +++ b/tests/agent/test_prompt_caching.py @@ -271,3 +271,59 @@ class TestApplyAnthropicCacheControl: assert isinstance(result[1]["content"], list) assert result[1]["content"][0]["cache_control"] == {"type": "ephemeral"} assert "cache_control" not in result[1] + + +class TestNormalizationOrdering: + """The conversation loop normalizes message text for prefix stability and + injects cache breakpoints. Marking must happen AFTER normalization. + + ``_apply_cache_marker`` rewrites a plain-string ``content`` into a + ``[{"type": "text", ...}]`` block. The loop's whitespace pass is guarded + on ``isinstance(content, str)``, so anything marked first is skipped by + it — and a message is only marked while it sits in the last-3 window. + The same message would then be sent raw on one turn and stripped on the + next, breaking the prefix match the breakpoints exist to protect. + """ + + def test_marking_a_string_hides_it_from_string_normalization(self): + """The mechanism: marking changes content out of ``str`` shape.""" + msgs = [{"role": "user", "content": "hello \n"}] + marked = apply_anthropic_cache_control(msgs, native_anthropic=False) + assert not isinstance(marked[0]["content"], str) + # Raw whitespace survives, now unreachable by an isinstance(str) pass. + assert marked[0]["content"][0]["text"] == "hello \n" + + def test_normalized_then_marked_matches_the_unmarked_wire_text(self): + """Normalize-then-mark keeps a message byte-identical across the + turn where it rolls out of the cache window.""" + raw = "file1\nfile2\n" # trailing newline: every shell tool result + + # Turn N+1, message has left the window: plain string, normalized. + out_of_window = raw.strip() + + # Turn N, message is in the window: normalized first, then marked. + marked = apply_anthropic_cache_control( + [{"role": "tool", "content": raw.strip(), "tool_call_id": "t1"}], + native_anthropic=False, + ) + in_window = marked[0]["content"][0]["text"] + + assert in_window == out_of_window + + def test_cache_marking_runs_after_every_message_mutation(self): + """Ordering invariant, locked against regression.""" + import inspect + + from agent import conversation_loop + + src = inspect.getsource(conversation_loop) + mark = src.index("apply_anthropic_cache_control(\n") + for earlier in ( + 'am["content"].strip()', # whitespace normalization + "_sanitize_api_messages(api_messages)", # orphan sweep + "_drop_thinking_only_and_merge_users(", # drop / merge + "_sanitize_messages_surrogates(api_messages)", + ): + assert src.index(earlier) < mark, ( + f"{earlier!r} must run before cache breakpoints are injected" + )