fix(agents): bound streaming error-response body reads

Port from openclaw/openclaw#95108: an unbounded response.read() on a
non-OK *streaming* response can balloon memory (huge body) or hang the
agent forever (body opens then stalls with no further bytes). The
diagnostic body is only ever shown truncated, so reading megabytes or
blocking indefinitely buys nothing.

Add agent/bounded_response.read_streaming_error_body() which caps the
read at a byte limit and enforces a hard wall-clock deadline (run on a
worker thread so it can interrupt a socket read that stalls mid-chunk,
which a between-chunk wall-clock check cannot). Wire it into all three
streaming error-body sites that previously did a bare response.read():
native Gemini, Gemini Cloud Code, and Antigravity Cloud Code. The
existing error builders now accept an optional pre-read body_text so
classification (status code, RESOURCE_EXHAUSTED, free-tier guidance,
Retry-After) is preserved unchanged.

Tests use a real in-process socket server (no mocks): oversize body is
capped, stalled body hits the deadline with partial text preserved,
normal error envelope reads intact and parses.

agent/bounded_response.py

@@ -0,0 +1,148 @@
+"""Bounded reads of HTTP error response bodies.
+
+When a provider returns a non-OK status on a *streaming* request, Hermes reads
+the response body to build a useful diagnostic error. A bare ``response.read()``
+on a streaming httpx response is unbounded in two dangerous ways:
+
+1. A server can declare (or stream) an arbitrarily large body, so the read can
+   balloon memory.
+2. A server can open the body and then stall forever (no ``Content-Length``,
+   no further bytes), so the read hangs the agent indefinitely.
+
+Both are realistic against a misbehaving proxy, a hijacked endpoint, or a
+provider having a bad day. The diagnostic body is only ever shown to the user
+truncated to a few hundred characters, so reading megabytes — or blocking
+forever — buys nothing.
+
+``read_streaming_error_body`` bounds the read to a byte cap and enforces a
+hard wall-clock deadline, returning the decoded text snippet. Callers pass the
+returned text into their existing error builders instead of touching
+``response.text`` (which would be unbounded / would raise after a partial
+stream read).
+
+A subtlety the implementation must respect: ``httpx``'s ``iter_bytes()`` blocks
+*inside* the C/socket read while waiting for the next chunk. A wall-clock check
+placed only between yielded chunks cannot interrupt a server that opens the
+body and then stalls mid-chunk — control never returns to Python until httpx's
+own (often 30s+) read timeout fires. To guarantee a bounded stop regardless of
+socket behavior, the read runs on a daemon worker thread and the caller waits
+on it with a hard deadline; on timeout we close the response (which unblocks /
+cancels the read) and return whatever partial bytes were collected.
+
+Ported and adapted from openclaw/openclaw#95108 ("bound Anthropic error
+streams"), generalized to cover Hermes's three streaming error-body sites
+(native Gemini, Gemini Cloud Code, Antigravity Cloud Code).
+"""
+
+from __future__ import annotations
+
+import logging
+import threading
+from typing import List, Optional
+
+import httpx
+
+logger = logging.getLogger(__name__)
+
+# Defaults chosen to comfortably hold any real provider error envelope (Google
+# RPC error JSON, Anthropic error JSON) while rejecting pathological bodies.
+DEFAULT_ERROR_BODY_MAX_BYTES = 64 * 1024
+# Hard wall-clock deadline for the whole bounded read. A streaming error body
+# that does not finish within this window is abandoned and the connection is
+# closed; we keep whatever partial bytes arrived.
+DEFAULT_ERROR_BODY_TIMEOUT_S = 10.0
+
+
+def read_streaming_error_body(
+    response: httpx.Response,
+    *,
+    max_bytes: int = DEFAULT_ERROR_BODY_MAX_BYTES,
+    timeout_s: float = DEFAULT_ERROR_BODY_TIMEOUT_S,
+) -> str:
+    """Read a non-OK streaming response body with a byte cap and a hard deadline.
+
+    Returns the decoded body text (UTF-8, errors replaced), truncated to
+    ``max_bytes``. Never raises: any transport error, stall, or oversize
+    condition is swallowed and the best-effort partial text (or an empty
+    string) is returned, because this runs on the error path and must not
+    mask the original HTTP failure with a read error.
+
+    The byte cap protects against huge bodies; the wall-clock deadline (enforced
+    via a worker thread so it can interrupt a socket read that stalls mid-chunk)
+    protects against bodies that open and then hang.
+    """
+    chunks: List[bytes] = []
+    state = {"truncated": False}
+    done = threading.Event()
+
+    def _drain() -> None:
+        total = 0
+        try:
+            for chunk in response.iter_bytes():
+                if not chunk:
+                    continue
+                remaining = max_bytes - total
+                if remaining <= 0:
+                    state["truncated"] = True
+                    break
+                if len(chunk) > remaining:
+                    chunks.append(chunk[:remaining])
+                    total += remaining
+                    state["truncated"] = True
+                    break
+                chunks.append(chunk)
+                total += len(chunk)
+        except Exception as exc:  # noqa: BLE001 - error path must not raise
+            logger.debug("bounded error-body read failed: %s", exc)
+        finally:
+            done.set()
+
+    worker = threading.Thread(
+        target=_drain, name="bounded-error-body-read", daemon=True
+    )
+    worker.start()
+    finished = done.wait(timeout=timeout_s)
+
+    if not finished:
+        logger.debug(
+            "bounded error-body read: hard timeout after %.1fs (%d bytes so far)",
+            timeout_s,
+            sum(len(c) for c in chunks),
+        )
+        # Closing the response cancels the in-flight socket read, letting the
+        # worker thread unwind. We do not join (it is a daemon and may be
+        # blocked in C); the partial `chunks` collected so far are returned.
+        _safe_close(response)
+    else:
+        _safe_close(response)
+
+    if state["truncated"]:
+        logger.debug(
+            "bounded error-body read: capped at %d bytes (max=%d)",
+            sum(len(c) for c in chunks),
+            max_bytes,
+        )
+    return b"".join(chunks).decode("utf-8", errors="replace")
+
+
+def _safe_close(response: httpx.Response) -> None:
+    try:
+        response.close()
+    except Exception:  # noqa: BLE001
+        pass
+
+
+def read_error_body_or_default(
+    response: httpx.Response,
+    *,
+    max_bytes: int = DEFAULT_ERROR_BODY_MAX_BYTES,
+    timeout_s: float = DEFAULT_ERROR_BODY_TIMEOUT_S,
+) -> Optional[str]:
+    """Like ``read_streaming_error_body`` but returns ``None`` on empty body.
+
+    Convenience for callers that distinguish "no body" from "empty string".
+    """
+    text = read_streaming_error_body(
+        response, max_bytes=max_bytes, timeout_s=timeout_s
+    )
+    return text or None

agent/gemini_native_adapter.py

@@ -27,6 +27,7 @@ from typing import Any, Dict, Iterator, List, Optional
 
 import httpx
 
+from agent.bounded_response import read_streaming_error_body
 from agent.gemini_schema import sanitize_gemini_tool_parameters
 
 logger = logging.getLogger(__name__)
@@ -742,14 +743,17 @@ def translate_stream_event(event: Dict[str, Any], model: str, tool_call_indices:
     return chunks
 
 
-def gemini_http_error(response: httpx.Response) -> GeminiAPIError:
+def gemini_http_error(
+    response: httpx.Response, *, body_text: Optional[str] = None
+) -> GeminiAPIError:
     status = response.status_code
-    body_text = ""
     body_json: Dict[str, Any] = {}
-    try:
-        body_text = response.text
-    except Exception:
-        body_text = ""
+    if body_text is None:
+        try:
+            body_text = response.text
+        except Exception:
+            body_text = ""
+    body_text = body_text or ""
     if body_text:
         try:
             parsed = json.loads(body_text)
@@ -968,8 +972,8 @@ class GeminiNativeClient:
             try:
                 with self._http.stream("POST", url, json=request, headers=stream_headers, timeout=timeout) as response:
                     if response.status_code != 200:
-                        response.read()
-                        raise gemini_http_error(response)
+                        body_text = read_streaming_error_body(response)
+                        raise gemini_http_error(response, body_text=body_text)
                     tool_call_indices: Dict[str, Dict[str, Any]] = {}
                     for event in _iter_sse_events(response):
                         for chunk in translate_stream_event(event, model, tool_call_indices):

tests/agent/test_bounded_response.py

@@ -0,0 +1,154 @@
+"""Tests for bounded reads of streaming HTTP error response bodies.
+
+Exercises the real ``httpx`` streaming path against an in-process socket server
+(no mocks) so the byte-cap and hard-deadline contracts are validated end to end,
+the way they behave against a real misbehaving provider.
+
+Covers the bug class ported from openclaw/openclaw#95108: an unbounded
+``response.read()`` on a non-OK streaming response can balloon memory (huge
+body) or hang forever (body opens then stalls).
+"""
+
+from __future__ import annotations
+
+import http.server
+import json
+import socketserver
+import threading
+import time
+
+import httpx
+import pytest
+
+from agent.bounded_response import (
+    read_error_body_or_default,
+    read_streaming_error_body,
+)
+
+
+class _ThreadingServer(socketserver.ThreadingTCPServer):
+    daemon_threads = True
+    allow_reuse_address = True
+
+
+def _make_handler():
+    class _Handler(http.server.BaseHTTPRequestHandler):
+        def log_message(self, format, *args):  # noqa: A002 - http.server API
+            pass
+
+        def do_POST(self):  # noqa: N802 - http.server API
+            if self.path == "/oversize":
+                # ~128 MiB if read unbounded; no Content-Length.
+                self.send_response(500)
+                self.send_header("Content-Type", "text/plain")
+                self.end_headers()
+                try:
+                    for _ in range(2000):
+                        self.wfile.write(b"x" * 65536)
+                        self.wfile.flush()
+                except Exception:
+                    pass
+            elif self.path == "/stall":
+                # Send a little, then stall forever (no further bytes).
+                self.send_response(500)
+                self.send_header("Content-Type", "text/plain")
+                self.end_headers()
+                self.wfile.write(b"partial failure detail")
+                self.wfile.flush()
+                time.sleep(60)
+            elif self.path == "/normal":
+                body = json.dumps(
+                    {
+                        "error": {
+                            "code": 429,
+                            "message": "quota exceeded",
+                            "status": "RESOURCE_EXHAUSTED",
+                        }
+                    }
+                ).encode()
+                self.send_response(429)
+                self.send_header("Content-Type", "application/json")
+                self.send_header("Content-Length", str(len(body)))
+                self.end_headers()
+                self.wfile.write(body)
+            elif self.path == "/empty":
+                self.send_response(500)
+                self.send_header("Content-Length", "0")
+                self.end_headers()
+
+    return _Handler
+
+
+@pytest.fixture()
+def server_base():
+    httpd = _ThreadingServer(("127.0.0.1", 0), _make_handler())
+    port = httpd.server_address[1]
+    thread = threading.Thread(target=httpd.serve_forever, daemon=True)
+    thread.start()
+    try:
+        yield f"http://127.0.0.1:{port}"
+    finally:
+        httpd.shutdown()
+
+
+@pytest.fixture()
+def client():
+    # Generous read timeout so the bounding is provably done by our helper,
+    # not by httpx's own timeout.
+    c = httpx.Client(
+        timeout=httpx.Timeout(connect=5.0, read=45.0, write=5.0, pool=5.0)
+    )
+    try:
+        yield c
+    finally:
+        c.close()
+
+
+def test_oversize_body_is_capped(server_base, client):
+    start = time.monotonic()
+    with client.stream("POST", server_base + "/oversize") as response:
+        text = read_streaming_error_body(
+            response, max_bytes=64 * 1024, timeout_s=10.0
+        )
+    elapsed = time.monotonic() - start
+    assert 0 < len(text) <= 64 * 1024
+    # Capping must return promptly, not after draining the whole body.
+    assert elapsed < 9.0
+
+
+def test_stalled_body_hits_hard_deadline(server_base, client):
+    start = time.monotonic()
+    with client.stream("POST", server_base + "/stall") as response:
+        text = read_streaming_error_body(
+            response, max_bytes=64 * 1024, timeout_s=2.0
+        )
+    elapsed = time.monotonic() - start
+    # Partial bytes that arrived before the stall are preserved.
+    assert "partial failure detail" in text
+    # The hard deadline bounds the read; we must not wait for the server stall.
+    assert elapsed < 5.0
+
+
+def test_normal_error_body_read_intact(server_base, client):
+    with client.stream("POST", server_base + "/normal") as response:
+        text = read_streaming_error_body(response)
+    parsed = json.loads(text)
+    assert parsed["error"]["status"] == "RESOURCE_EXHAUSTED"
+
+
+def test_empty_body_returns_empty_string(server_base, client):
+    with client.stream("POST", server_base + "/empty") as response:
+        text = read_streaming_error_body(response)
+    assert text == ""
+
+
+def test_or_default_returns_none_on_empty(server_base, client):
+    with client.stream("POST", server_base + "/empty") as response:
+        result = read_error_body_or_default(response)
+    assert result is None
+
+
+def test_or_default_returns_text_when_present(server_base, client):
+    with client.stream("POST", server_base + "/normal") as response:
+        result = read_error_body_or_default(response)
+    assert result is not None and "RESOURCE_EXHAUSTED" in result