fix(compressor): age out stale tool-result images during compaction

_strip_historical_media anchors on the newest image-bearing USER message and
returns the list untouched when that anchor is index 0 or does not exist. A
session whose images arrive from tools rather than attachments therefore has
nothing to be "before": twenty vision_analyze results keep multi-MB of base64
in every request body, the provider answers 413, and the 413 handler's
recovery compaction lands right back in this function and frees nothing. The
reporter saw seven compactions in thirteen minutes, all below 200K tokens.

Age tool-result images on their own timeline: keep the newest one, since that
is the image the model is reasoning about, and strip every older one wherever
it sits, including inside the protected tail. The tail exists to preserve
conversational continuity, not to pin bytes the model has already moved past.

User-message images keep today's treatment exactly. The user anchor is
checked first, so a tool result that is the newest of its kind but still sits
before that anchor is stripped as it always has been, and the anchor message
itself is still kept byte-for-byte - test_compressor_zero_user_guard depends
on that.

Refs #89938

agent/context_compressor.py

@@ -1863,9 +1863,15 @@ def _strip_historical_media(messages: List[Dict[str, Any]]) -> List[Dict[str, An
     placeholder so the outgoing request stops re-shipping the same multi-MB
     base-64 image blobs on every turn.
 
-    If no user message carries images, the list is returned unchanged.
-    If the only user message with images is the very first one (nothing
-    earlier to strip), the list is returned unchanged.
+    Tool results carry their own images (``vision_analyze`` and friends) and
+    are aged out on their own timeline: every image-bearing tool message
+    except the newest one is stripped, wherever it sits. Without that, a
+    session whose images arrive from tools rather than attachments has no
+    anchor to be "before" and keeps every blob forever (#89938).
+
+    If no message carries images at all, the list is returned unchanged. So
+    is a list whose only image-bearing user message is the very first one and
+    which has no tool-result images (nothing to strip in either rule).
 
     Shallow copies of touched messages only; input is never mutated.
     Port of Kilo-Org/kilocode#9434 (adapted for the OpenAI-style message
@@ -1889,15 +1895,48 @@ def _strip_historical_media(messages: List[Dict[str, Any]]) -> List[Dict[str, An
             anchor = i
             break
 
-    if anchor <= 0:
-        # No image-bearing user message, or it's the very first message —
-        # nothing before it to strip.
+    # Newest tool message carrying an image. Tool-result images
+    # (``vision_analyze``, screenshot-returning tools) accumulate on their own
+    # timeline and the user anchor never protects the stale ones: a session
+    # whose only image-bearing user message is the FIRST one leaves
+    # ``anchor <= 0`` and strips nothing at all, so twenty tool results keep
+    # multi-MB of base64 in every request body until the provider answers 413
+    # -- and the 413 handler's recovery compaction lands right back here and
+    # frees nothing, which is the wedge in #89938. Keep the newest tool image,
+    # since that is the one the model is reasoning about, and drop every older
+    # one wherever it sits.
+    tool_anchor = -1
+    for i in range(len(messages) - 1, -1, -1):
+        msg = messages[i]
+        if not isinstance(msg, dict):
+            continue
+        if msg.get("role") != "tool":
+            continue
+        if _content_has_images(msg.get("content")):
+            tool_anchor = i
+            break
+
+    if anchor <= 0 and tool_anchor < 0:
+        # No image-bearing user message (or it is the very first, with nothing
+        # earlier to strip), and no tool-result images to age out either.
         return messages
 
+    def _is_stale(index: int, message: Dict[str, Any]) -> bool:
+        # Rule 1 (unchanged): everything before the newest image-bearing user
+        # message. Checked first so a tool result that is the newest of its
+        # kind but still sits before that anchor keeps today's behaviour.
+        if 0 < anchor and index < anchor:
+            return True
+        # Rule 2: a tool result whose image has been superseded by a newer
+        # one. Applies inside the protected tail as well -- the tail exists to
+        # preserve conversational continuity, not to pin bytes the model has
+        # already moved past.
+        return message.get("role") == "tool" and index != tool_anchor
+
     changed = False
     result: List[Dict[str, Any]] = []
     for i, msg in enumerate(messages):
-        if i >= anchor or not isinstance(msg, dict):
+        if not isinstance(msg, dict) or not _is_stale(i, msg):
             result.append(msg)
             continue
         content = msg.get("content")

tests/agent/test_compressor_historical_media.py

@@ -107,6 +107,114 @@ class TestStripHistoricalMedia:
         # Second pass is a no-op — no images left before the anchor.
         assert second is first
 
+    def test_strips_stale_tool_result_images_when_no_user_image_exists(self):
+        """#89938: vision_analyze results are the only images in the session.
+
+        Before this rule the anchor stayed at -1 and the list came back
+        untouched, so every base64 blob rode along on every request.
+        """
+        msgs = [
+            {"role": "user", "content": "look at these"},
+            {"role": "tool", "tool_call_id": "a", "content": [TEXT, IMG_URL]},
+            {"role": "tool", "tool_call_id": "b", "content": [TEXT, IMG_URL]},
+            {"role": "tool", "tool_call_id": "c", "content": [TEXT, IMG_URL]},
+        ]
+        out = _strip_historical_media(msgs)
+
+        assert not _content_has_images(out[1]["content"])
+        assert not _content_has_images(out[2]["content"])
+        # The newest tool image is what the model is reasoning about.
+        assert _content_has_images(out[3]["content"])
+
+    def test_first_message_user_image_no_longer_blocks_tool_stripping(self):
+        """#89938's other half: ``anchor <= 0`` used to return early.
+
+        One attachment on the opening message plus a run of vision tool calls
+        is the exact reproduction in the report - and the old early return
+        meant the 413 recovery compaction freed nothing.
+        """
+        msgs = [
+            {"role": "user", "content": [TEXT, IMG_URL]},
+            {"role": "tool", "tool_call_id": "a", "content": [TEXT, IMG_URL]},
+            {"role": "tool", "tool_call_id": "b", "content": [TEXT, IMG_URL]},
+        ]
+        out = _strip_historical_media(msgs)
+
+        # The opening attachment keeps today's treatment: nothing precedes it.
+        assert _content_has_images(out[0]["content"])
+        assert not _content_has_images(out[1]["content"])
+        assert _content_has_images(out[2]["content"])
+
+    def test_newest_tool_image_survives_inside_the_protected_tail(self):
+        msgs = [
+            {"role": "user", "content": "hi"},
+            {"role": "user", "content": [TEXT, IMG_URL]},
+            {"role": "tool", "tool_call_id": "a", "content": [TEXT, IMG_URL]},
+            {"role": "tool", "tool_call_id": "b", "content": [TEXT, IMG_URL]},
+        ]
+        out = _strip_historical_media(msgs)
+
+        # The user anchor is index 1, so nothing before it changes and the
+        # anchor itself is kept byte-for-byte (test_compressor_zero_user_guard
+        # depends on that).
+        assert out[1] is msgs[1]
+        assert not _content_has_images(out[2]["content"])
+        assert _content_has_images(out[3]["content"])
+
+    def test_tool_image_before_the_user_anchor_is_still_stripped(self):
+        """Rule 1 keeps precedence over rule 2 where the two disagree."""
+        msgs = [
+            {"role": "tool", "tool_call_id": "a", "content": [TEXT, IMG_URL]},
+            {"role": "assistant", "content": "ok"},
+            {"role": "user", "content": [TEXT, IMG_URL]},
+        ]
+        out = _strip_historical_media(msgs)
+
+        # Index 0 is the newest *tool* image, but it sits before the user
+        # anchor, which has always stripped it. That must not regress.
+        assert not _content_has_images(out[0]["content"])
+        assert _content_has_images(out[2]["content"])
+
+    def test_unchanged_when_nothing_carries_images(self):
+        msgs = [
+            {"role": "user", "content": [TEXT]},
+            {"role": "tool", "tool_call_id": "a", "content": [TEXT]},
+        ]
+        assert _strip_historical_media(msgs) is msgs
+
+    def test_single_tool_image_is_left_alone(self):
+        msgs = [
+            {"role": "user", "content": "look"},
+            {"role": "tool", "tool_call_id": "a", "content": [TEXT, IMG_URL]},
+        ]
+        assert _strip_historical_media(msgs) is msgs
+
+    def test_idempotent_over_tool_images(self):
+        msgs = [
+            {"role": "tool", "tool_call_id": "a", "content": [TEXT, IMG_URL]},
+            {"role": "tool", "tool_call_id": "b", "content": [TEXT, IMG_URL]},
+        ]
+        first = _strip_historical_media(msgs)
+        assert first is not msgs
+        assert _strip_historical_media(first) is first
+
+    def test_stripped_tool_message_drops_its_api_content_sidecar(self):
+        """Replaying the sidecar would resend the bytes the strip removed."""
+        msgs = [
+            {
+                "role": "tool",
+                "tool_call_id": "a",
+                "content": [TEXT, IMG_URL],
+                "api_content": "the exact multimodal bytes sent last turn",
+            },
+            {"role": "tool", "tool_call_id": "b", "content": [TEXT, IMG_URL]},
+        ]
+        out = _strip_historical_media(msgs)
+
+        assert "api_content" not in out[0]
+        # The input list is never mutated.
+        assert "api_content" in msgs[0]
+
     def test_non_dict_messages_pass_through(self):
         msgs = [
             "not-a-dict",  # shouldn't crash
@@ -166,3 +274,56 @@ class TestCompressIntegration:
             assert not _content_has_images(m.get("content")), (
                 f"Stale image in {m.get('role')!r} message after compression"
             )
+
+    def test_compress_frees_stale_vision_tool_results(self, compressor):
+        """#89938 end to end: the 413 recovery compaction must free bytes.
+
+        The reported session had one attachment on the opening message and a
+        run of ``vision_analyze`` results after it. ``anchor <= 0`` made this
+        pass a no-op, so every recovery compaction returned a body that was
+        still multi-MB and the provider answered 413 again - seven times in
+        thirteen minutes.
+        """
+
+        def call(idx: str):
+            return [
+                {
+                    "role": "assistant",
+                    "content": None,
+                    "tool_calls": [
+                        {
+                            "id": idx,
+                            "type": "function",
+                            "function": {"name": "vision_analyze", "arguments": "{}"},
+                        }
+                    ],
+                },
+                {"role": "tool", "tool_call_id": idx, "content": [TEXT, IMG_URL]},
+            ]
+
+        msgs = [
+            {"role": "system", "content": "sys"},
+            {"role": "user", "content": [TEXT, IMG_URL]},  # the ONLY user image, first
+            *call("a"),
+            *call("b"),
+            {"role": "user", "content": "and this one?"},
+            *call("c"),
+        ]
+        with patch.object(compressor, "_generate_summary", return_value="SUMMARY TEXT"):
+            out = compressor.compress(msgs, current_tokens=60_000)
+
+        with_images = [m for m in out if isinstance(m, dict) and _content_has_images(m.get("content"))]
+        # Exactly one survivor: the newest vision result. The opening
+        # attachment is protect_first_n material and may or may not survive
+        # the summary window, so assert on what must NOT be there instead.
+        assert len(with_images) <= 2
+        stale_tool_images = [
+            m for m in out
+            if isinstance(m, dict)
+            and m.get("role") == "tool"
+            and _content_has_images(m.get("content"))
+            and m.get("tool_call_id") != "c"
+        ]
+        assert stale_tool_images == [], (
+            "vision_analyze results a, b still carry base64 after compression"
+        )