diff --git a/src/query/query.py b/src/query/query.py index 8a56d19d..ab0c90e5 100644 --- a/src/query/query.py +++ b/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 diff --git a/src/services/api/errors.py b/src/services/api/errors.py index bb5bedc1..3e077eb6 100644 --- a/src/services/api/errors.py +++ b/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)) ) diff --git a/src/services/compact/reactive_compact.py b/src/services/compact/reactive_compact.py index 6a0ce649..63b33e38 100644 --- a/src/services/compact/reactive_compact.py +++ b/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 diff --git a/tests/test_api_errors.py b/tests/test_api_errors.py index b42c8745..834b08ce 100644 --- a/tests/test_api_errors.py +++ b/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): diff --git a/tests/test_query_error_recovery.py b/tests/test_query_error_recovery.py index a3e1847b..bc5a0b16 100644 --- a/tests/test_query_error_recovery.py +++ b/tests/test_query_error_recovery.py @@ -16,6 +16,7 @@ QueryParams, StreamEvent, query, + run_query, ) @@ -669,3 +670,233 @@ async def run(): if __name__ == "__main__": unittest.main() + + +class TestReactiveRecoveryActuallyRuns(unittest.TestCase): + """The recovery lane must RETRY, not just relabel the terminal. + + Both lanes were dead. ``query.py`` builds its trigger as + ``PromptTooLongError("withheld during streaming, recovering")`` because + the original provider exception is consumed during streaming and only the + classification needs to survive — but the gate + (``reactive_compact.is_prompt_too_long_error``) was a pure SUBSTRING test + for "prompt is too long" / "prompt_too_long" / "context_length_exceeded". + A typed ``PromptTooLongError`` matched none of them, so + ``reactive_compact`` returned ``compacted=False`` on its first line and + nothing downstream ran: no compaction, no image strip, no retry. + + Measured on main before the fix: ONE provider call for both a + prompt-too-long and a media rejection. + + The lane's existing tests all stub ``reactive_compact`` itself + (``fake_reactive_compact`` returning ``compacted=True``), so the gate was + never exercised — which is how it stayed broken from 2026-05 to 2026-08. + These tests deliberately do NOT stub it: they count provider calls. + """ + + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.workspace = Path(self.tmp.name) + self.registry = build_default_registry() + self.abort = AbortController() + + def tearDown(self): + self.tmp.cleanup() + + def _drive(self, error_text, messages): + from unittest.mock import MagicMock + + calls = {"n": 0, "images": []} + + def side(msgs, *a, **k): + calls["n"] += 1 + n = 0 + for m in msgs: + c = m.get("content") if isinstance(m, dict) else getattr(m, "content", None) + if isinstance(c, list): + for b in c: + t = b.get("type") if isinstance(b, dict) else getattr(b, "type", None) + if t in ("image", "image_url", "document"): + n += 1 + calls["images"].append(n) + raise Exception(error_text) + + provider = MagicMock() + provider.chat_stream_response.side_effect = side + provider.chat.side_effect = side + params = QueryParams( + messages=messages, + system_prompt="s", + tools=self.registry.list_tools(), + tool_registry=self.registry, + tool_use_context=ToolContext(workspace_root=self.workspace), + provider=provider, + abort_controller=self.abort, + max_turns=10, + ) + _msgs, terminal = _run(run_query(params)) + return terminal, calls + + @staticmethod + def _image_convo(n): + from src.types.content_blocks import ImageBlock, TextBlock + + ms = [UserMessage(content=[TextBlock(text="analyse these frames")])] + for i in range(n): + ms.append(UserMessage(content=[ + TextBlock(text=f"frame {i}"), + ImageBlock(source={ + "type": "base64", "media_type": "image/jpeg", "data": "x" * 300, + }), + ])) + return ms + + def test_media_rejection_strips_images_and_retries(self): + """THE regression guard: a retry must happen AND carry fewer images.""" + terminal, calls = self._drive( + "Exceeded maximum number of images (50) allowed in the request.", + self._image_convo(60), + ) + self.assertGreater( + calls["n"], 1, + "no retry happened — the recovery lane is dead again", + ) + self.assertGreater(calls["images"][0], 0, "first attempt should carry images") + self.assertLess( + calls["images"][1], calls["images"][0], + f"the retry must carry FEWER images; got {calls['images']}", + ) + self.assertEqual(terminal.reason, "image_error") + + def test_prompt_too_long_recovery_retries(self): + """The pre-existing lane, dead by the identical bug since 2026-05.""" + terminal, calls = self._drive( + "prompt is too long: 137500 tokens > 135000 maximum", + self._image_convo(40), + ) + self.assertGreater( + calls["n"], 1, + "prompt_too_long recovery never retried — the gate is broken again", + ) + self.assertEqual(terminal.reason, "prompt_too_long") + + def test_the_synthetic_trigger_passes_its_own_gate(self): + """Pins the exact mismatch, so a future edit to either side is caught. + + ``query.py`` constructs this error; ``reactive_compact`` gates on it. + The two live in different modules and drifted apart silently. + """ + from src.services.api.errors import PromptTooLongError + from src.services.compact.reactive_compact import is_prompt_too_long_error + + self.assertTrue( + is_prompt_too_long_error( + PromptTooLongError("withheld during streaming, recovering") + ), + "a typed PromptTooLongError must satisfy the PromptTooLong gate " + "regardless of its message", + ) + # ...and the string arm still works for untyped provider exceptions. + self.assertTrue(is_prompt_too_long_error(Exception("prompt is too long"))) + self.assertFalse(is_prompt_too_long_error(Exception("network error"))) + + def test_retryable_errors_mentioning_images_stay_retryable(self): + """A 429/5xx whose body mentions images must NOT become a media + terminal — that would take it out of the retry lane.""" + from src.services.api.errors import is_media_size_error + + self.assertFalse( + is_media_size_error("Rate limit reached for images: too many images generated") + ) + + +class TestMediaRecoveryWhenTheSummarizerIsDown(unittest.TestCase): + """The media path must not depend on a working summarizer. + + ``reactive_compact``'s emergency fallback drops the OLDEST messages and + accepts the result on a TOKEN test (``tokens_after < tokens_before * + 0.7``). Image count is never consulted. For the case that motivates this + path — an agent reading frames in a loop, so the images sit in the RECENT + tail — dropping old text satisfies that test while leaving the images in + place. Measured directly against the compactor: 200 messages / 60 images + -> 40 messages / 40 images, returned as ``compacted=True``. + + So routing media through the token compactor can "succeed" and still + retry over the cap, burning the one-shot flag. Stripping is deterministic. + + This test forces the summarizer to fail so the two paths diverge — with a + working summarizer both happen to reach zero images, which is why a + simpler test cannot tell them apart. + """ + + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.workspace = Path(self.tmp.name) + self.registry = build_default_registry() + self.abort = AbortController() + + def tearDown(self): + self.tmp.cleanup() + + def test_images_still_dropped_without_a_summarizer(self): + from unittest import mock + from unittest.mock import MagicMock + + from src.types.content_blocks import ImageBlock, TextBlock + + # Text-heavy head, images concentrated in the recent tail. + msgs = [] + for i in range(60): + msgs.append(UserMessage(content=[TextBlock(text=f"step {i} " + "lorem " * 80)])) + for i in range(40): + msgs.append(UserMessage(content=[ + TextBlock(text=f"frame {i}"), + ImageBlock(source={ + "type": "base64", "media_type": "image/jpeg", "data": "x" * 300, + }), + ])) + + seen = [] + + def side(api_msgs, *a, **k): + n = 0 + for m in api_msgs: + c = m.get("content") if isinstance(m, dict) else getattr(m, "content", None) + if isinstance(c, list): + for b in c: + t = b.get("type") if isinstance(b, dict) else getattr(b, "type", None) + if t in ("image", "image_url", "document"): + n += 1 + seen.append(n) + raise Exception("Exceeded maximum number of images (50) allowed in the request.") + + provider = MagicMock() + provider.chat_stream_response.side_effect = side + provider.chat.side_effect = side + + async def _summarizer_down(ctx): + raise RuntimeError("summarizer unavailable") + + params = QueryParams( + messages=msgs, + system_prompt="s", + tools=self.registry.list_tools(), + tool_registry=self.registry, + tool_use_context=ToolContext(workspace_root=self.workspace), + provider=provider, + abort_controller=self.abort, + max_turns=10, + ) + with mock.patch( + "src.services.compact.reactive_compact.compact_conversation", + _summarizer_down, + ): + _run(run_query(params)) + + self.assertGreater(len(seen), 1, "no retry happened") + self.assertGreater(seen[0], 0, "first attempt should carry images") + self.assertEqual( + seen[1], 0, + "the retry must carry NO images even with the summarizer down; " + f"got {seen} — the token-shaped compactor leaves images behind", + )