fix(prompt-caching): inject cache breakpoints after message normalization

The conversation loop normalizes message text right before the API call so
the request prefix is byte-identical across turns -- the stated reason is
KV cache reuse on local inference servers and better cache hit rates on
cloud providers. Cache breakpoints were injected *before* that pass, which
defeats it.

`_apply_cache_marker` rewrites a plain-string `content` into a
`[{"type": "text", ...}]` block. The normalization pass is guarded on
`isinstance(content, str)`, so every message that just got marked is
silently skipped by it and keeps its raw leading/trailing whitespace. A
message is only marked while it sits in the last-3 window, so:

    turn N      in the window  -> marked, content "file1\nfile2\n"
    turn N+1    rolled out     -> plain,  content "file1\nfile2"

The same logical message is sent with different bytes on consecutive
turns. The prefix stops matching at that position -- which is inside the
span the breakpoints were placed to protect -- so the reusable prefix
collapses back toward the system breakpoint on every turn. Tool results
carry a trailing newline almost by default (any shell command output), so
this is the common case, not an edge case.

Move the injection below every message mutation. Besides fixing the
whitespace divergence this stops breakpoints from being spent on messages
that the orphan sweep or the thinking-only drop is about to remove or
merge away -- a marker on a dropped message is a wasted breakpoint out of
the four available.

Nothing between the old and new call sites reads `cache_control`, and the
mutators now see the plain-string shapes they were written against.

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

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"
+            )