fix(recovery): revive the reactive-recovery lane, which never ran

src/query/query.py

@@ -1321,14 +1321,36 @@ def _do_provider_call():
             err_msg._api_error = "max_output_tokens"  # type: ignore[attr-defined]
             return [err_msg], []
 
-        # Ch5/B.1 — tag media-size errors so the loop can withhold them
-        # and (in B.2) route through reactive-compact recovery. Mirrors
-        # TS `isWithheldMediaSizeError` at query.ts:892. `is_media_size_error`
+        # Ch5/B.1 — tag media errors so the loop can withhold them and (in
+        # B.2) route through media recovery. Mirrors TS
+        # `isWithheldMediaSizeError` at query.ts:892. `is_media_size_error`
         # expects a str (substring match), so pass error_str explicitly.
-        from ..services.api.errors import is_media_size_error
-        if is_media_size_error(error_str):
+        #
+        # RETRYABLE ERRORS FIRST. This branch RETURNS a tagged message rather
+        # than re-raising, which takes the request out of the retry lane
+        # entirely (``categorize_retryable_api_error`` only ever sees
+        # exceptions that propagate out of this function). So a 429 or a 5xx
+        # whose body happens to mention images — "Rate limit reached for
+        # images: ..." is real provider wording — would be converted from
+        # "back off and retry" into a non-retryable media terminal. Classify
+        # on transport/status BEFORE matching on prose.
+        from ..services.api.errors import (
+            is_media_size_error,
+            is_overloaded_error,
+            is_rate_limit_error,
+        )
+
+        _status = getattr(e, "status", getattr(e, "status_code", None))
+        _retryable = (
+            is_rate_limit_error(e)
+            or is_overloaded_error(e)
+            or (isinstance(_status, int) and _status >= 500)
+        )
+        if is_media_size_error(error_str) and not _retryable:
             err_msg = _create_assistant_api_error_message(
-                f"Media too large: {error_str}",
+                # "too large" was wrong for a COUNT rejection, which is what
+                # this branch most often sees; the operator reads this string.
+                f"Media rejected: {error_str}",
                 error="media_size",
             )
             err_msg._api_error = "media_size"  # type: ignore[attr-defined]
@@ -2012,21 +2034,86 @@ def _marking_chunk_cb(text: str) -> None:
                 )
                 from ..services.api.errors import PromptTooLongError
 
-                # Synthesize an exception for reactive_compact's
-                # is_prompt_too_long_error check. The withheld
-                # message holds the original error string; we don't
-                # need to round-trip it precisely because
-                # reactive_compact only uses the exception for
-                # classification.
-                synthetic_err = PromptTooLongError(
-                    "withheld during streaming, recovering"
-                )
-                result: ReactiveCompactResult = await reactive_compact(
-                    messages=messages,
-                    error=synthetic_err,
-                    provider=params.provider,
-                    model=config.model,
-                )
+                # A MEDIA rejection is a COUNT/SIZE violation, not a token
+                # one, so fix it directly instead of routing through the
+                # token-shaped compactor.
+                #
+                # reactive_compact's emergency fallback drops the OLDEST
+                # messages and accepts the result when tokens fall 30%
+                # (reactive_compact.py). Image count is never consulted. For
+                # the case that motivates this path — an agent reading frames
+                # in a loop, so the images are all in the RECENT tail —
+                # dropping old text satisfies the token test while leaving
+                # the images in place: measured 200 msgs/60 images -> 40
+                # msgs/40 images, reported as ``compacted=True``. The retry
+                # then hits the same cap with the one-shot flag already
+                # burned.
+                #
+                # Stripping is deterministic, needs no summarizer call, and
+                # keeps the text context that a full compaction would replace
+                # with a summary. Upstream models these as distinct
+                # operations too (reactiveCompact.ts carries a
+                # 'media_unstrippable' outcome); the port had collapsed them.
+                media_result: ReactiveCompactResult | None = None
+                if is_withheld_media:
+                    from ..context_system.microcompact import (
+                        strip_images_from_typed_messages,
+                    )
+
+                    def _n_images(ms: list[Message]) -> int:
+                        n = 0
+                        for m in ms:
+                            c = getattr(m, "content", None)
+                            if isinstance(c, list):
+                                n += sum(
+                                    1 for b in c
+                                    if getattr(b, "type", None) in ("image", "document")
+                                )
+                        return n
+
+                    before_imgs = _n_images(messages)
+                    stripped = strip_images_from_typed_messages(messages)
+                    after_imgs = _n_images(stripped)
+                    if before_imgs and after_imgs < before_imgs:
+                        logger.info(
+                            "media recovery: stripped %d media block(s) from "
+                            "the conversation and retrying",
+                            before_imgs - after_imgs,
+                        )
+                        media_result = ReactiveCompactResult(
+                            compacted=True,
+                            messages=stripped,
+                            tokens_before=0,
+                        )
+                if is_withheld_media and media_result is not None:
+                    result = media_result
+                else:
+                    # Either a prompt-too-long, or a media error with nothing
+                    # strippable in the typed conversation (media referenced
+                    # some other way). Fall back to the general compactor
+                    # rather than giving up — strip is an OPTIMISATION for the
+                    # case it can fix deterministically, not a replacement.
+                    # Synthesize an exception for reactive_compact's
+                    # is_prompt_too_long_error check. The withheld message
+                    # holds the original error string; we don't need to
+                    # round-trip it precisely because reactive_compact only
+                    # uses the exception for classification.
+                    #
+                    # That predicate is TYPE-aware (reactive_compact.py) --
+                    # it has to be, because this message deliberately is not
+                    # the default one. When it was string-only this synthetic
+                    # error failed the gate and the WHOLE lane was dead: no
+                    # compaction, no retry, one provider call, for both the
+                    # PTL and media paths.
+                    synthetic_err = PromptTooLongError(
+                        "withheld during streaming, recovering"
+                    )
+                    result = await reactive_compact(
+                        messages=messages,
+                        error=synthetic_err,
+                        provider=params.provider,
+                        model=config.model,
+                    )
                 if result.compacted:
                     # ReactiveCompactResult.messages is list[Message]
                     # (verified 2026-05-12 against reactive_compact.py

src/services/api/errors.py

@@ -239,9 +239,12 @@ def is_media_size_error(raw: str) -> bool:
         ("image exceeds" in low and "maximum" in low)
         or ("image dimensions exceed" in low and "many-image" in low)
         or bool(re.search(r"maximum of \d+ pdf pages", low))
-        # Too MANY images, as opposed to too large.
+        # Too MANY images, as opposed to too large. All three are anchored
+        # on a count phrase; a bare "too many images" was tried and REMOVED
+        # because it also matches retry-worthy provider prose such as
+        # "Rate limit reached for images: too many images generated", and
+        # this predicate decides routing, not just wording.
         or bool(re.search(r"maximum number of images", low))
-        or bool(re.search(r"too many images", low))
         or bool(re.search(r"exceeds? the maximum of \d+ images", low))
         or bool(re.search(r"at most \d+ images", low))
     )

src/services/compact/reactive_compact.py

@@ -43,6 +43,31 @@ class ReactiveCompactResult:
 
 
 def is_withheld_prompt_too_long(error: Exception) -> bool:
+    """Whether ``error`` means "the request did not fit".
+
+    TYPE FIRST, then the wire text. The type check is not redundant: a
+    ``PromptTooLongError`` carries its reason in ``str(error)`` only when the
+    caller left the default message alone, and ``query.py``'s recovery path
+    deliberately does not — it constructs one with an explanatory message
+    ("withheld during streaming, recovering") because the original provider
+    exception was already consumed during streaming and only the
+    classification needs to survive.
+
+    Without the isinstance arm that synthetic error failed this predicate, so
+    ``reactive_compact`` returned ``compacted=False`` immediately and the
+    ENTIRE reactive-recovery lane was dead: no compaction, no image strip, no
+    retry, for both the prompt-too-long and the media paths. Measured on
+    ``main`` before this change — one provider call, then a terminal. A typed
+    ``PromptTooLongError`` failing the PromptTooLong predicate is a trap, not
+    a contract.
+
+    The lane's own tests did not catch it because they stub
+    ``reactive_compact`` itself, so the gate was never exercised.
+    """
+    from ..api.errors import PromptTooLongError
+
+    if isinstance(error, PromptTooLongError):
+        return True
     error_str = str(error).lower()
     return (
         "prompt_too_long" in error_str

tests/test_api_errors.py

@@ -131,14 +131,61 @@ def test_non_401(self) -> None:
 
 
 class TestIsMediaSizeError(unittest.TestCase):
-    def test_image_exceeds(self) -> None:
-        self.assertTrue(is_media_size_error("image exceeds the maximum allowed"))
-
-    def test_pdf_pages(self) -> None:
-        self.assertTrue(is_media_size_error("maximum of 100 PDF pages"))
-
-    def test_unrelated(self) -> None:
-        self.assertFalse(is_media_size_error("some other error"))
+    """Pin EVERY pattern individually.
+
+    This predicate decides error ROUTING — a match sends the request into
+    media recovery and out of the retry lane — so a pattern nobody pins can
+    be deleted, or a wrong one added, without a test noticing. Testing the
+    patterns as a group hides exactly that: removing three of four count
+    patterns at once left the loop-level tests green.
+    """
+
+    # (string, expected, why)
+    CASES = [
+        # --- size / PDF (pre-existing) ---
+        ("image exceeds the maximum allowed", True, "size"),
+        ("image dimensions exceed the many-image limit", True, "dimensions"),
+        ("maximum of 100 PDF pages", True, "pdf page cap"),
+        # --- count (the video-processing failure) ---
+        ("Exceeded maximum number of images (50) allowed in the request.",
+         True, "OBSERVED on terminal-bench 2.1"),
+        ("Request exceeds the maximum of 100 images", True, "count, anchored"),
+        ("You may include at most 50 images per request", True, "count, anchored"),
+        # --- case-insensitivity ---
+        ("IMAGE EXCEEDS 5MB MAXIMUM", True, "upper"),
+        ("Maximum Of 100 Pdf Pages", True, "mixed; the regex literal is lowercase"),
+        # --- must NOT match: each has a DIFFERENT recovery, or none ---
+        ("prompt is too long: 137500 tokens > 135000 maximum", False, "own lane"),
+        ("context_length_exceeded", False, "own lane"),
+        ("No endpoints found that support image input", False,
+         "stripping does not help; is_image_unsupported_error owns it"),
+        ("Rate limit reached for images: too many images generated", False,
+         "retryable — must not become a media terminal"),
+        ("The server had an error processing your request", False, "no recovery"),
+        ("maximum context length exceeded", False, "token, not media"),
+        ("some other error", False, "unrelated"),
+    ]
+
+    def test_every_pattern_individually(self) -> None:
+        for raw, expected, why in self.CASES:
+            with self.subTest(raw=raw, why=why):
+                self.assertEqual(is_media_size_error(raw), expected, why)
+
+    def test_bare_too_many_images_is_not_a_pattern(self) -> None:
+        """Deliberately NOT matched. It was tried and removed: it also
+        matches retry-worthy prose ("Rate limit reached for images: too many
+        images generated"), and this predicate takes the request out of the
+        retry lane."""
+        self.assertFalse(is_media_size_error("too many images"))
+
+    def test_case_folding_cannot_widen_the_match_set(self) -> None:
+        """Case-insensitivity must only add case variants of strings that
+        already matched — never anything new."""
+        for raw, _expected, _why in self.CASES:
+            with self.subTest(raw=raw):
+                self.assertEqual(
+                    is_media_size_error(raw), is_media_size_error(raw.lower())
+                )
 
 
 class TestIsImageUnsupportedError(unittest.TestCase):