diff --git a/agent/bounded_response.py b/agent/bounded_response.py new file mode 100644 index 00000000000..e5177bc8a2b --- /dev/null +++ b/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 diff --git a/agent/gemini_native_adapter.py b/agent/gemini_native_adapter.py index c254bf61311..9d3b1eb324d 100644 --- a/agent/gemini_native_adapter.py +++ b/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): diff --git a/tests/agent/test_bounded_response.py b/tests/agent/test_bounded_response.py new file mode 100644 index 00000000000..dfc93adcc5f --- /dev/null +++ b/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