raise on upstream brain errors instead of yielding an empty stream

Author: Zawwarsami16

tests/test_brains_anthropic.py

@@ -102,3 +102,47 @@ def test_anthropic_in_default_registry():
     from zhub.brains import REGISTRY
     names = [c.name for c in REGISTRY]
     assert "anthropic" in names
+
+
+class _FakeErrorStream:
+    def __init__(self, status_code, body):
+        self.status_code = status_code
+        self._body = body.encode()
+
+    async def aiter_lines(self):
+        yield self._body.decode()
+
+    async def aread(self):
+        return self._body
+
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, *exc):
+        return None
+
+
+class _FakeErrorClient:
+    def __init__(self, status_code, body):
+        self._status = status_code
+        self._body = body
+
+    def stream(self, method, url, **kw):
+        return _FakeErrorStream(self._status, self._body)
+
+    async def aclose(self):
+        pass
+
+
+@pytest.mark.asyncio
+async def test_stream_raises_on_upstream_error():
+    """A non-2xx body is a JSON error, not Anthropic SSE — the adapter should
+    raise rather than end the stream silently with no content."""
+    fake = _FakeErrorClient(529, '{"type":"error","error":{"type":"overloaded_error"}}')
+    adapter = AnthropicAdapter(api_key="sk-ant-x", model="claude-sonnet-4-5", http=fake)
+    with pytest.raises(RuntimeError) as ei:
+        async for _ in adapter.stream([{"role": "user", "content": "hi"}]):
+            pass
+    msg = str(ei.value)
+    assert "529" in msg
+    assert "overloaded_error" in msg

tests/test_brains_groq.py

@@ -105,3 +105,48 @@ async def test_stream_ignores_lines_without_data_prefix():
     out = [c async for c in adapter.stream([{"role": "user", "content": "hi"}])]
     assert [c.delta for c in out if c.delta] == ["ok"]
     assert out[-1].done is True
+
+
+class _FakeErrorStream:
+    def __init__(self, status_code, body):
+        self.status_code = status_code
+        self._body = body.encode()
+
+    async def aiter_lines(self):
+        # an error response is not SSE; emit the raw json body as one line
+        yield self._body.decode()
+
+    async def aread(self):
+        return self._body
+
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, *exc):
+        return None
+
+
+class _FakeErrorClient:
+    def __init__(self, status_code, body):
+        self._status = status_code
+        self._body = body
+
+    def stream(self, method, url, **kw):
+        return _FakeErrorStream(self._status, self._body)
+
+    async def aclose(self):
+        pass
+
+
+@pytest.mark.asyncio
+async def test_stream_raises_on_upstream_error():
+    """A 429/4xx/5xx body is not SSE — without a status check it would parse
+    to an empty stream. The adapter should raise so the publisher surfaces it."""
+    fake = _FakeErrorClient(429, '{"error":{"message":"rate limit exceeded"}}')
+    adapter = GroqAdapter(api_key="k", model="llama-3.3-70b-versatile", http=fake)
+    with pytest.raises(RuntimeError) as ei:
+        async for _ in adapter.stream([{"role": "user", "content": "hi"}]):
+            pass
+    msg = str(ei.value)
+    assert "429" in msg
+    assert "rate limit exceeded" in msg

zhub/brains/_openai_compat.py

@@ -73,6 +73,20 @@ async def stream_openai_compat(
         "POST", f"{base_url.rstrip('/')}/chat/completions",
         json=body, headers=headers,
     ) as response:
+        # An upstream error (429 rate limit, 401 bad key, 5xx) comes back as a
+        # JSON body that isn't SSE, so the line loop below would skip every
+        # line and the generator would end yielding nothing — a silent empty
+        # completion. Catch it here and raise so the publisher surfaces a real
+        # error to the caller instead. getattr default keeps test doubles that
+        # don't set status_code working; real httpx responses always have it.
+        status = getattr(response, "status_code", 200)
+        if status >= 400:
+            try:
+                detail = (await response.aread()).decode("utf-8", "replace").strip()
+            except Exception:
+                detail = ""
+            snippet = f": {detail[:300]}" if detail else ""
+            raise RuntimeError(f"upstream returned HTTP {status}{snippet}")
         async for line in response.aiter_lines():
             line = line.rstrip("\r")
             if not line.startswith("data:"):

zhub/brains/anthropic.py

@@ -95,6 +95,19 @@ async def stream(
         async with self._http.stream(
             "POST", f"{_BASE_URL}/messages", json=body, headers=headers,
         ) as response:
+            # Non-2xx (429 overloaded, 401 bad key, 5xx) returns a JSON error
+            # body, not SSE, so the loop below would skip every line and end
+            # yielding nothing. Raise so the publisher surfaces a real error
+            # instead of a silent empty completion. getattr default keeps test
+            # doubles working; real httpx responses always have status_code.
+            status = getattr(response, "status_code", 200)
+            if status >= 400:
+                try:
+                    detail = (await response.aread()).decode("utf-8", "replace").strip()
+                except Exception:
+                    detail = ""
+                snippet = f": {detail[:300]}" if detail else ""
+                raise RuntimeError(f"upstream returned HTTP {status}{snippet}")
             async for line in response.aiter_lines():
                 line = line.rstrip("\r")
                 if not line.startswith("data:"):