fix(llms): read streamed error bodies before raise_for_status in Anthropic and Gemini adapters (#1959)

omnigent/llms/adapters/anthropic.py

@@ -652,6 +652,12 @@ async def _stream_request(
             json=payload,
         ) as resp,
     ):
+        # Buffer error bodies before raising: a streamed response is
+        # unread, so exc.response.text would raise ResponseNotRead and
+        # error classification (e.g. context-overflow detection) would
+        # never see the provider's message.
+        if resp.status_code >= 400:
+            await resp.aread()
         resp.raise_for_status()
         async for chunk in _stream_to_chat_chunks(
             resp.aiter_lines(),

omnigent/llms/adapters/gemini.py

@@ -194,6 +194,13 @@ class GeminiAdapter(BaseAdapter):
                 json=payload,
             ) as resp,
         ):
+            # Buffer error bodies before raising: a streamed response
+            # is unread, so exc.response.text would raise
+            # ResponseNotRead and error classification (e.g.
+            # context-overflow detection) would never see the
+            # provider's message.
+            if resp.status_code >= 400:
+                await resp.aread()
             resp.raise_for_status()
             async for line in resp.aiter_lines():
                 if not line.startswith("data: "):

tests/llms/conftest.py

@@ -0,0 +1,55 @@
+"""Shared fixtures for LLM adapter tests."""
+
+from collections.abc import AsyncIterator, Callable
+from typing import Any
+
+import httpx
+import pytest
+
+
+class _UnreadBodyStream(httpx.AsyncByteStream):
+    """
+    Response stream whose body is NOT buffered at construction.
+
+    ``httpx.Response(content=...)`` eagerly reads the body into
+    ``_content``, which would make ``response.text`` work even on a
+    streamed response that was never read — hiding the exact bug the
+    adapters guard against. Passing ``stream=`` keeps the body unread
+    until ``aread()`` is called, matching a live streaming connection.
+    """
+
+    def __init__(self, data: bytes) -> None:
+        self._data = data
+
+    async def __aiter__(self) -> AsyncIterator[bytes]:
+        yield self._data
+
+
+@pytest.fixture
+def serve_streamed_response(
+    monkeypatch: pytest.MonkeyPatch,
+) -> Callable[[int, bytes], None]:
+    """
+    Patch ``httpx.AsyncClient`` so any streamed request receives a
+    canned response whose body stays unread until ``aread()``.
+
+    Returns an installer: call it with ``(status_code, body)`` before
+    invoking the adapter under test.
+    """
+    real_async_client = httpx.AsyncClient
+
+    def _install(status_code: int, body: bytes) -> None:
+        transport = httpx.MockTransport(
+            lambda request: httpx.Response(
+                status_code,
+                stream=_UnreadBodyStream(body),
+            )
+        )
+
+        def _factory(**kwargs: Any) -> httpx.AsyncClient:
+            kwargs["transport"] = transport
+            return real_async_client(**kwargs)
+
+        monkeypatch.setattr(httpx, "AsyncClient", _factory)
+
+    return _install

tests/llms/test_anthropic_adapter.py

@@ -1,8 +1,10 @@
 """Tests for llms.adapters.anthropic — translation logic."""
 
+import asyncio
 import base64
 import json
 
+import httpx
 import pytest
 
 from omnigent.llms.adapters.anthropic import (
@@ -10,8 +12,11 @@ from omnigent.llms.adapters.anthropic import (
     _chat_to_anthropic,
     _convert_tool_choice,
     _convert_tools,
+    _stream_request,
     _translate_part_to_anthropic,
 )
+from omnigent.llms.errors import ContextWindowExceededError
+from omnigent.runtime.llm_retry import classify_llm_error
 
 
 def test_system_messages_extracted() -> None:
@@ -601,3 +606,56 @@ def test_max_completion_tokens_alias() -> None:
     messages = [{"role": "user", "content": "Hi"}]
     payload = _chat_to_anthropic(messages, "claude-test", None, {"max_completion_tokens": 2048})
     assert payload["max_tokens"] == 2048
+
+
+# ── Streaming error body buffering ───────────────────────
+
+
+def test_streamed_400_overflow_classified_as_context_window_exceeded(
+    serve_streamed_response,
+) -> None:
+    """
+    A streamed HTTP 400 buffers the error body before raising so
+    ``classify_llm_error`` can detect context-window overflow.
+
+    Without the ``aread()`` guard in ``_stream_request`` the body of a
+    streamed error response is never read; ``exc.response.text`` then
+    raises ``ResponseNotRead``, degrades to
+    ``"<unreadable response body>"``, and a genuine overflow 400 is
+    misclassified as a plain ``PermanentLLMError`` — the workflow's
+    compact-and-retry path never fires.
+
+    Failure meaning: the guard has been removed and streaming
+    Anthropic overflow errors no longer trigger compaction.
+    """
+    overflow_body = json.dumps(
+        {
+            "type": "error",
+            "error": {
+                "type": "invalid_request_error",
+                "message": ("prompt is too long: 210141 tokens > 200000 maximum"),
+            },
+        }
+    ).encode()
+    serve_streamed_response(400, overflow_body)
+
+    async def _run() -> Exception:
+        gen = _stream_request(
+            headers={},
+            payload={"model": "claude-test", "stream": True},
+            base_url="https://fake-host/v1",
+        )
+        try:
+            async for _ in gen:
+                pass
+        except httpx.HTTPStatusError as exc:
+            return classify_llm_error(exc, [429, 500, 502, 503])
+        raise AssertionError("Expected HTTPStatusError was not raised")
+
+    err = asyncio.run(_run())
+
+    assert isinstance(err, ContextWindowExceededError)
+    assert err.max_context_tokens == 200000
+    assert err.actual_tokens == 210141
+    assert err.detail is not None
+    assert "prompt is too long" in (err.detail.response_body or "")

tests/llms/test_gemini_adapter.py

@@ -1,11 +1,14 @@
 """Tests for llms.adapters.gemini — translation logic."""
 
+import asyncio
 import json
 
+import httpx
 import pytest
 
 from omnigent.llms._responses_to_chat import chat_stream_to_response_events
 from omnigent.llms.adapters.gemini import (
+    GeminiAdapter,
     _chat_to_gemini,
     _convert_tools,
     _extract_usage,
@@ -14,7 +17,9 @@ from omnigent.llms.adapters.gemini import (
     _normalize_finish_reason,
     _translate_part_to_gemini,
 )
+from omnigent.llms.errors import ContextWindowExceededError
 from omnigent.llms.types import FunctionCallOutput
+from omnigent.runtime.llm_retry import classify_llm_error
 
 # ── Request translation ──────────────────────────────────
 
@@ -562,3 +567,62 @@ def test_file_data_without_data_uri_raises() -> None:
     }
     with pytest.raises(ValueError, match="data: URI"):
         _translate_part_to_gemini(part)
+
+
+# ── Streaming error body buffering ───────────────────────
+
+
+def test_streamed_400_overflow_classified_as_context_window_exceeded(
+    serve_streamed_response,
+) -> None:
+    """
+    A streamed HTTP 400 buffers the error body before raising so
+    ``classify_llm_error`` can detect context-window overflow.
+
+    Without the ``aread()`` guard in ``_stream_request`` the body of a
+    streamed error response is never read; ``exc.response.text`` then
+    raises ``ResponseNotRead``, degrades to
+    ``"<unreadable response body>"``, and a genuine overflow 400 is
+    misclassified as a plain ``PermanentLLMError`` — the workflow's
+    compact-and-retry path never fires.
+
+    Failure meaning: the guard has been removed and streaming Gemini
+    (and Vertex, which inherits ``_stream_request``) overflow errors
+    no longer trigger compaction.
+    """
+    overflow_body = json.dumps(
+        {
+            "error": {
+                "code": 400,
+                "message": (
+                    "The input token count (1194139) exceeds the maximum"
+                    " number of tokens allowed (1048576)."
+                ),
+                "status": "INVALID_ARGUMENT",
+            }
+        }
+    ).encode()
+    serve_streamed_response(400, overflow_body)
+
+    adapter = GeminiAdapter()
+
+    async def _run() -> Exception:
+        gen = adapter._stream_request(
+            "https://fake-host/v1beta/models/gemini-test:streamGenerateContent",
+            {},
+            {"contents": []},
+        )
+        try:
+            async for _ in gen:
+                pass
+        except httpx.HTTPStatusError as exc:
+            return classify_llm_error(exc, [429, 500, 502, 503])
+        raise AssertionError("Expected HTTPStatusError was not raised")
+
+    err = asyncio.run(_run())
+
+    assert isinstance(err, ContextWindowExceededError)
+    assert err.max_context_tokens == 1048576
+    assert err.actual_tokens == 1194139
+    assert err.detail is not None
+    assert "input token count" in (err.detail.response_body or "")

tests/llms/test_openai_adapter.py

@@ -586,3 +586,65 @@ def test_parse_responses_event_non_native_item_done_returns_none() -> None:
 
     event = _parse_responses_event("response.output_item.done", {"item": {"type": "message"}})
     assert event is None
+
+
+def test_streamed_400_overflow_classified_as_context_window_exceeded(
+    serve_streamed_response,
+) -> None:
+    """
+    A streamed HTTP 400 buffers the error body before raising so
+    ``classify_llm_error`` can detect context-window overflow.
+
+    End-to-end companion to the ``aread()`` test above: a genuine
+    OpenAI overflow body must survive the streaming error path and
+    classify as ``ContextWindowExceededError`` (not a plain
+    ``PermanentLLMError``) so the workflow can compact and retry.
+
+    Failure meaning: the ``aread()`` guard has been removed and
+    streaming OpenAI overflow errors no longer trigger compaction.
+    """
+    import asyncio
+
+    import httpx
+
+    from omnigent.llms.errors import ContextWindowExceededError
+    from omnigent.runtime.llm_retry import classify_llm_error
+
+    adapter = OpenAICompatibleAdapter(base_url="https://fake-host/v1", api_key_env=None)
+    overflow_body = json.dumps(
+        {
+            "error": {
+                "message": (
+                    "This model's maximum context length is 128000 tokens."
+                    " However, you requested 131015 tokens (127015 in the"
+                    " messages, 4000 in the completion). Please reduce the"
+                    " length of the messages or completion."
+                ),
+                "type": "invalid_request_error",
+                "param": "messages",
+                "code": "context_length_exceeded",
+            }
+        }
+    ).encode()
+    serve_streamed_response(400, overflow_body)
+
+    async def _run() -> Exception:
+        gen = adapter._stream_request(
+            "https://fake-host/v1/chat/completions",
+            {},
+            {"model": "dummy", "stream": True, "messages": []},
+        )
+        try:
+            async for _ in gen:
+                pass
+        except httpx.HTTPStatusError as exc:
+            return classify_llm_error(exc, [429, 500, 502, 503])
+        raise AssertionError("Expected HTTPStatusError was not raised")
+
+    err = asyncio.run(_run())
+
+    assert isinstance(err, ContextWindowExceededError)
+    assert err.max_context_tokens == 128000
+    assert err.actual_tokens == 131015
+    assert err.detail is not None
+    assert "maximum context length" in (err.detail.response_body or "")