fix(agent): make the send-path copy structural — close the write-through class

The api_messages build used a shallow msg.copy(), decoupling only
top-level fields. Every nested container (tool_calls entries and their
function dicts, multimodal content-part lists, reasoning_details) stayed
aliased to the persisted history, so ANY in-place transform on the send
copy silently rewrote the stored transcript.

Probed every send-path transform against that aliasing shape on main:

  content strip loop                       safe (top-level reassign)
  _canonicalize_api_tool_calls (repair)    LEAKED  <- #80616's fix
  _sanitize_messages_surrogates            LEAKED  (multimodal parts,
                                                    tc ids/args, reasoning)
  _sanitize_messages_non_ascii             LEAKED  (multimodal parts)
  _sanitize_api_messages                   safe
  _drop_thinking_only_and_merge_users      safe

The retry loop already believed the copies were independent - it
sanitizes messages AND api_messages separately (~L3555) - so the
aliasing was accidental everywhere.

Fix at the chokepoint: _clone_message_for_send clones every container
(dict/list) recursively while sharing immutable leaves, so every
downstream in-place transform - current and future - is safe by
construction. Cost is container-count, not string-bytes: 100KB argument
strings and base64 payloads are shared (measured ~0.5ms vs ~0.1ms per
1500-message build; noise next to one json round-trip). Same clone
applied to the prefill-message insert (same class, same pipeline).

The class-wide invariant test runs the full send-path transform
pipeline over an adversarial fixture (malformed args, surrogates,
non-ASCII, multimodal parts, reasoning fields) and asserts the history
stays byte-identical; an AST contract pins the build-site wiring so the
shallow copy can't quietly return. Both mutation-verified: reverting
the clone to shallow fails 4 isolation tests, unwiring the build site
fails the AST contract.

0xGr1mm's branch fix (previous commit) remains as defense in depth at
the exact site the #80498 incident hit; his regression tests and the
class-wide invariant give layered coverage.

agent/conversation_loop.py

@@ -795,6 +795,50 @@ def _canonicalize_tool_call_arguments(arg_str: str) -> str:
     return canonical
 
 
+def _clone_message_for_send(msg):
+    """Structural clone of a history message for the per-call API copy.
+
+    The send path builds ``api_messages`` from the persisted history and
+    then rewrites the copies in place (canonicalization/repair of tool-call
+    arguments, surrogate and non-ASCII sanitization, content strips, cache
+    decoration). A shallow ``msg.copy()`` only decouples TOP-LEVEL fields:
+    nested containers — ``tool_calls`` entries and their ``function`` dicts,
+    multimodal ``content`` part lists, ``reasoning_details`` — remain the
+    SAME objects the persisted history holds, so any in-place write there
+    silently rewrites the stored transcript (#80498: an unrepairable
+    ``write_file`` argument string was replaced with ``{}`` in the persisted
+    turn, destroying the streamed file content).
+
+    Cloning every container (dict/list) recursively while SHARING immutable
+    leaves (strings, numbers, None) makes every downstream in-place
+    transform safe by construction — current and future — at container-count
+    cost, not string-byte cost: big argument strings and base64 image
+    payloads are shared, never copied. Measured: ~1-5ms per 2000-message
+    pathological build (20% multimodal, 30% tool calls) vs ~0.4ms for the
+    shallow copy; compression keeps real request histories far smaller, and
+    the build runs once per API call — noise next to the call itself.
+    copy.deepcopy would be equally correct (CPython deepcopy also shares
+    immutable str) but ~4x slower again and needs its memo machinery;
+    history messages are JSON-shaped and acyclic (depth < 10 in practice;
+    a >~1000-deep pathological nest would hit the recursion limit, exactly
+    as deepcopy would), so cycle handling isn't needed here. Tuples are
+    shared as leaves: JSON-derived message content never contains tuples,
+    so a mutable container smuggled inside one is not a reachable shape on
+    this path.
+    """
+    if isinstance(msg, dict):
+        return {
+            k: _clone_message_for_send(v) if isinstance(v, (dict, list)) else v
+            for k, v in msg.items()
+        }
+    if isinstance(msg, list):
+        return [
+            _clone_message_for_send(v) if isinstance(v, (dict, list)) else v
+            for v in msg
+        ]
+    return msg
+
+
 def _canonicalize_api_tool_calls(api_messages) -> None:
     """Canonicalize tool-call argument JSON on the send-path message copy.
 
@@ -821,16 +865,18 @@ def _canonicalize_api_tool_calls(api_messages) -> None:
                         ),
                     }}
                 except Exception:
-                    # Copy-on-write here too. ``api_messages`` holds shallow
-                    # per-message copies (``msg.copy()`` at the send-path
-                    # build), so the tool_call dicts are the SAME objects as
-                    # the persisted history's — assigning into
-                    # ``tc["function"]`` rewrites the stored turn. On the
-                    # unrepairable path the repair returns "{}", so that
-                    # in-place write replaced the model's real arguments with
-                    # an empty object in the transcript: a stream that died
-                    # mid ``write_file`` lost the file content it had already
-                    # streamed, with only a WARNING to show for it (#80498).
+                    # Copy-on-write here too. The send-path build now hands
+                    # this pass structurally-cloned messages (see
+                    # _clone_message_for_send), but this branch keeps its own
+                    # copy as defense in depth: some callers (tests, future
+                    # call sites) pass shallow copies, and assigning into a
+                    # shared ``tc["function"]`` would rewrite the stored
+                    # turn. On the unrepairable path the repair returns "{}",
+                    # so a write-through here replaced the model's real
+                    # arguments with an empty object in the transcript: a
+                    # stream that died mid ``write_file`` lost the file
+                    # content it had already streamed, with only a WARNING
+                    # to show for it (#80498).
                     tc = {**tc, "function": {
                         **tc["function"],
                         "arguments": _repair_tool_call_arguments(
@@ -1157,10 +1203,13 @@ def _apply_context_engine_selection(
     # and it may be replaced wholesale via the return value — never mutated in
     # place either. ``conversation_messages`` / ``incoming_message`` are
     # read-only context; copying enforces the request-only contract rather than
-    # merely documenting it.
-    _conv_copy = [dict(m) if isinstance(m, dict) else m for m in conversation_messages] \
+    # merely documenting it. Structural clones, not dict(m): a shallow copy
+    # would leave nested containers (tool_calls, content parts) aliased to
+    # the persisted history, so an engine writing into them would rewrite
+    # the transcript (#80498 aliasing class).
+    _conv_copy = [_clone_message_for_send(m) for m in conversation_messages] \
         if conversation_messages is not None else None
-    _incoming_copy = dict(incoming_message) if isinstance(incoming_message, dict) else incoming_message
+    _incoming_copy = _clone_message_for_send(incoming_message) if isinstance(incoming_message, dict) else incoming_message
     try:
         selected = engine.select_context(
             api_messages,
@@ -1231,7 +1280,10 @@ def _notify_context_engine_turn_complete(
 
     try:
         hook(
-            [dict(m) if isinstance(m, dict) else m for m in messages],
+            # Structural clones: on_turn_complete receives the PERSISTED
+            # history; a shallow dict(m) would let a hook write through
+            # nested containers into the transcript (#80498 aliasing class).
+            [_clone_message_for_send(m) for m in messages],
             usage=usage,
             **meta,
         )
@@ -1599,7 +1651,12 @@ def run_conversation(
 
         api_messages = []
         for idx, msg in enumerate(messages):
-            api_msg = msg.copy()
+            # Structural clone, NOT msg.copy(): every in-place transform
+            # below (canonicalize/repair, surrogate + non-ASCII sanitizers,
+            # cache decoration) must be unable to reach the persisted
+            # history through shared nested containers. See
+            # _clone_message_for_send.
+            api_msg = _clone_message_for_send(msg)
 
             # api_content is the persistence sidecar carrying the exact bytes
             # sent to the API for this message when they differ from the clean
@@ -1781,7 +1838,11 @@ def run_conversation(
         if agent.prefill_messages:
             sys_offset = 1 if (api_messages and api_messages[0].get("role") == "system") else 0
             for idx, pfm in enumerate(agent.prefill_messages):
-                api_messages.insert(sys_offset + idx, pfm.copy())
+                # Structural clone: the sanitizers below run over
+                # api_messages in place, and a shallow copy would let them
+                # write through into agent.prefill_messages' nested
+                # containers (same aliasing class as the history build).
+                api_messages.insert(sys_offset + idx, _clone_message_for_send(pfm))
 
         # Per-turn context selection hook (additive, no-op by default).
         # Lets a context engine select/replace which context enters the

agent/prompt_caching.py

@@ -179,9 +179,10 @@ def strip_anthropic_cache_control(
     multi-part text (merged user turns, imported transcripts) and parts
     carrying extra keys (``citations`` etc.) keep their structure; only
     per-part markers are removed. Marker removal is copy-on-write on the
-    part dicts: content parts may alias the persistent conversation history
-    (the per-call copy is shallow), and stripping must never rewrite the
-    stored transcript.
+    part dicts: content parts can alias caller-held message lists (the main
+    send path now hands structurally-cloned copies via
+    _clone_message_for_send, but other callers may pass shallow copies),
+    and stripping must never rewrite the stored transcript.
 
     Mutates the top-level message dicts of ``api_messages`` in place and
     returns the same list.

tests/agent/test_send_path_history_isolation.py

@@ -0,0 +1,209 @@
+"""Send-path transforms must never write through into persisted history.
+
+The send path builds ``api_messages`` from the persisted conversation
+history and then rewrites the copies IN PLACE (tool-call argument
+canonicalization/repair, surrogate sanitization, non-ASCII sanitization,
+content strips). The build previously used a shallow ``msg.copy()``, which
+decouples only top-level fields: nested containers (tool_calls entries and
+their function dicts, multimodal content-part lists, reasoning_details)
+stayed aliased to the history's objects, so those in-place transforms
+silently rewrote the stored transcript. Incident #80498: an unrepairable
+``write_file`` argument string was replaced with ``{}`` in the persisted
+turn, destroying the streamed file content.
+
+``_clone_message_for_send`` closes the whole class at the chokepoint: it
+clones every container while sharing immutable leaves. These tests pin the
+invariant CLASS-WIDE — every send-path in-place transform runs over an
+adversarial fixture and the history must remain byte-identical — so any
+future transform added to the pipeline inherits the guarantee (or fails
+here loudly).
+"""
+
+import copy
+import json
+
+import agent.conversation_loop as cl
+from agent.message_sanitization import (
+    _sanitize_messages_non_ascii,
+    _sanitize_messages_surrogates,
+)
+
+TRUNCATED_ARGS = '{"content": "# chapter draft\\nline one'  # unrepairable
+VALID_ARGS = json.dumps({"path": "a.txt", "text": "héllo"})
+LONE_SURROGATE = "hello \ud83d world"
+
+
+def _adversarial_history():
+    """Every nested container + every dirty-leaf shape the transforms touch."""
+    return [
+        {
+            "role": "user",
+            "content": [
+                {"type": "text", "text": "look at this " + LONE_SURROGATE},
+                {
+                    "type": "image_url",
+                    "image_url": {"url": "data:image/png;base64,AAAA"},
+                },
+            ],
+        },
+        {
+            "role": "assistant",
+            "content": "ok " + LONE_SURROGATE,
+            "reasoning_content": "thinking … " + LONE_SURROGATE,
+            "reasoning_details": [
+                {"type": "reasoning.text", "text": "chaîne " + LONE_SURROGATE}
+            ],
+            "tool_calls": [
+                {
+                    "id": "c1" + LONE_SURROGATE,
+                    "type": "function",
+                    "function": {"name": "write_file", "arguments": TRUNCATED_ARGS},
+                },
+                {
+                    "id": "c2",
+                    "type": "function",
+                    "function": {"name": "read_file", "arguments": VALID_ARGS},
+                },
+            ],
+        },
+        {"role": "tool", "tool_call_id": "c1", "content": "done"},
+    ]
+
+
+def _api_copy(history):
+    """Exactly the send path's build shape."""
+    return [cl._clone_message_for_send(m) for m in history]
+
+
+# Every send-path transform that rewrites api_messages in place. Add new
+# transforms here when the pipeline grows — the invariant is class-wide.
+def _run_full_pipeline(api_messages):
+    for am in api_messages:
+        if isinstance(am.get("content"), str):
+            am["content"] = am["content"].strip()
+    cl._canonicalize_api_tool_calls(api_messages)
+    _sanitize_messages_surrogates(api_messages)
+    _sanitize_messages_non_ascii(api_messages)
+
+
+class TestSendPathNeverMutatesHistory:
+    def test_full_pipeline_leaves_history_byte_identical(self):
+        history = _adversarial_history()
+        before = copy.deepcopy(history)
+
+        api_messages = _api_copy(history)
+        _run_full_pipeline(api_messages)
+
+        assert history == before, (
+            "a send-path transform wrote through the api copy into the "
+            "persisted history — the clone is no longer structural"
+        )
+
+    def test_each_transform_in_isolation(self):
+        transforms = {
+            "canonicalize/repair": cl._canonicalize_api_tool_calls,
+            "surrogate sanitizer": _sanitize_messages_surrogates,
+            "non-ascii sanitizer": _sanitize_messages_non_ascii,
+        }
+        for name, fn in transforms.items():
+            history = _adversarial_history()
+            before = copy.deepcopy(history)
+            fn(_api_copy(history))
+            assert history == before, f"{name} mutated persisted history"
+
+    def test_clone_decouples_every_nested_container(self):
+        history = _adversarial_history()
+        api = _api_copy(history)
+
+        # Write into every nested container of the copy.
+        api[0]["content"][0]["text"] = "MUTATED"
+        api[0]["content"][1]["image_url"]["url"] = "MUTATED"
+        api[1]["tool_calls"][0]["function"]["arguments"] = "{}"
+        api[1]["tool_calls"][1]["id"] = "MUTATED"
+        api[1]["reasoning_details"][0]["text"] = "MUTATED"
+
+        assert history[0]["content"][0]["text"].startswith("look at this")
+        assert history[0]["content"][1]["image_url"]["url"].startswith("data:")
+        assert (
+            history[1]["tool_calls"][0]["function"]["arguments"]
+            == TRUNCATED_ARGS
+        )
+        assert history[1]["tool_calls"][1]["id"] == "c2"
+        assert history[1]["reasoning_details"][0]["text"].startswith("chaîne")
+
+    def test_clone_shares_immutable_leaves(self):
+        """Cost model: containers copied, strings shared (not duplicated)."""
+        history = _adversarial_history()
+        api = _api_copy(history)
+        # Same string object — no byte copy of large payloads.
+        assert (
+            api[1]["tool_calls"][1]["function"]["arguments"]
+            is history[1]["tool_calls"][1]["function"]["arguments"]
+        )
+        # Different container objects at every level.
+        assert api[1] is not history[1]
+        assert api[1]["tool_calls"] is not history[1]["tool_calls"]
+        assert api[1]["tool_calls"][0] is not history[1]["tool_calls"][0]
+        assert (
+            api[1]["tool_calls"][0]["function"]
+            is not history[1]["tool_calls"][0]["function"]
+        )
+
+    def test_non_dict_messages_pass_through(self):
+        sentinel = object()
+        assert cl._clone_message_for_send(sentinel) is sentinel
+        assert cl._clone_message_for_send("plain") == "plain"
+        assert cl._clone_message_for_send(None) is None
+
+
+class TestSendPathBuildIsWiredToTheClone:
+    """The api_messages build must actually USE the structural clone.
+
+    The isolation tests above exercise ``_clone_message_for_send`` directly,
+    so they cannot notice the build site quietly reverting to the shallow
+    ``msg.copy()`` (the mutation that recreates #80498). This AST contract
+    pins the wiring: inside ``run_conversation_loop``'s api_messages build,
+    the per-message copy expression must be a ``_clone_message_for_send``
+    call, and no ``.copy()``-shaped fallback may reappear on the history
+    iteration variable.
+    """
+
+    def test_history_build_calls_the_clone(self):
+        import ast
+        import inspect
+
+        source = inspect.getsource(cl)
+        tree = ast.parse(source)
+
+        clone_calls = []
+        shallow_copies = []
+        for node in ast.walk(tree):
+            if isinstance(node, ast.Call):
+                fn = node.func
+                if isinstance(fn, ast.Name) and fn.id == "_clone_message_for_send":
+                    if (
+                        node.args
+                        and isinstance(node.args[0], ast.Name)
+                        and node.args[0].id in ("msg", "pfm")
+                    ):
+                        clone_calls.append(node.args[0].id)
+                if (
+                    isinstance(fn, ast.Attribute)
+                    and fn.attr == "copy"
+                    and isinstance(fn.value, ast.Name)
+                    and fn.value.id in ("msg", "pfm")
+                ):
+                    shallow_copies.append(fn.value.id)
+
+        assert "msg" in clone_calls, (
+            "the api_messages history build no longer clones via "
+            "_clone_message_for_send(msg) — shallow aliasing (#80498) is back"
+        )
+        assert "pfm" in clone_calls, (
+            "the prefill insert no longer clones via "
+            "_clone_message_for_send(pfm)"
+        )
+        assert not shallow_copies, (
+            f"shallow .copy() reappeared on send-path message variables: "
+            f"{shallow_copies} — nested containers alias the persisted history"
+        )