feat(jina): add opt-in response byte budget (#6373)

* feat(jina): add opt-in response byte budget

Signed-off-by: 97three <2212371308@qq.com>

* test(jina): clarify response limit assertions

Signed-off-by: 97three <2212371308@qq.com>

---------

Signed-off-by: 97three <2212371308@qq.com>

README.md

@@ -167,6 +167,8 @@ For Google's official Gemini OpenAI-compatible endpoint, use the
 
    Jina fetches support opt-in bounded retries via `max_retries` (default `0`) and `retry_budget_seconds` (default `30`) in the tool configuration. Valid `Retry-After` hints set a minimum wait for HTTP 429/503; 429 without a valid hint stays terminal. Hints that cannot fit the remaining budget stop retries. Local backoff remains randomized. Retries may increase upstream requests and cost; see [Jina fetch retries](backend/docs/CONFIGURATION.md#jina-fetch-retries).
 
+   Jina also accepts an opt-in `max_response_bytes` tool setting (positive integer; omitted/null disables it). It stops oversized decoded responses before extraction, with no partial success or retry. This leaves the 4096-character output cap unchanged and does not bound HTTPX decompressor allocations or wire bandwidth; see [response budget](backend/docs/CONFIGURATION.md#jina-response-byte-budget).
+
    Run `make doctor` at any time to verify your setup and get actionable fix hints.
    If you are opening a GitHub issue about a local setup or runtime problem, run
    `make support-bundle`. The command prints reporter next steps, writes a

backend/docs/CONFIGURATION.md

@@ -656,6 +656,27 @@ cannot cancel work already started by Jina. Enabling retries can send up to `1 +
 and incur additional cost. Successful content and final `Error:` results retain
 the existing contract.
 
+#### Jina response byte budget
+
+On the same Jina `web_fetch` tool entry, optionally set `max_response_bytes: 1048576`
+(for example, 1 MiB). This uses existing tool configuration extras; no model-facing
+argument is added. Omitted or `null` preserves the buffered default. An enabled
+value must be a positive integer; booleans, strings, fractions, zero and negative
+values return `Error:` before HTTP client creation or network activity.
+
+Enabled fetches stream and count actual content-decoded bytes (after decompression,
+before text decoding), ignoring `Content-Length`. Exactly the limit is accepted.
+The first chunk exceeding it stops consumption and closes the response, returning
+an explicit size `Error:` without body content, partial success or readability
+extraction. This applies to all statuses, including 502/503/504, and oversize never
+retries. Each retry response has its own counter within the existing shared time
+budget. Streams close on success, errors, cancellation and read failures. Responses
+within the limit retain charset decoding and existing status/retry handling.
+
+This limits response retention/consumption, not wire-byte bandwidth or allocations
+inside HTTPX's decompressor; it is not a hard process-memory bound. The final
+Markdown truncation at 4096 characters is unchanged and independent of this option.
+
 Serper `web_search` also accepts the optional model argument
 `time_range: "day" | "week" | "month" | "year"`. For example,
 `{"query": "Python releases", "time_range": "week"}` sends `tbs: "qdr:w"`

backend/packages/harness/deerflow/community/jina_ai/AGENTS.md

@@ -8,3 +8,11 @@ zero. Wait max(server floor, existing budget-capped jittered 0.5–4s backoff).
 Never reduce server floors; unfit waits return the HTTP error. Reset hints each
 attempt. Auth/payment errors stay terminal. Tests: `test_jina_retries.py` and
 `test_jina_retry_after.py` (offline, not hosted-provider validation).
+
+`max_response_bytes`: null/omitted keeps buffered POST; otherwise positive int
+(excluding bool), validated before client creation. Stream `aiter_bytes`, count
+content-decoded bytes before text decoding; exact limit passes, excess closes
+and returns a body-free terminal Error before extraction, for every status.
+Keep one retry loop/deadline and reset the counter per response. Never re-decode
+compression or mutate HTTPX internals. Decoder allocations/wire bytes are outside
+the cap. Offline transports: `tests/test_jina_response_limit.py`.

backend/packages/harness/deerflow/community/jina_ai/jina_client.py

@@ -61,7 +61,9 @@ def _retry_after(value: str | None) -> float | None:
 
 
 class JinaClient:
-    async def crawl(self, url: str, return_format: str = "html", timeout: int = 10, proxy: str | None = None, trust_env: bool = True, *, max_retries: int = 0, retry_budget_seconds: float = 30.0) -> str:
+    async def crawl(
+        self, url: str, return_format: str = "html", timeout: int = 10, proxy: str | None = None, trust_env: bool = True, *, max_retries: int = 0, retry_budget_seconds: float = 30.0, max_response_bytes: int | None = None
+    ) -> str:
         """Fetch with optional bounded retries; cancellation always propagates."""
         global _api_key_warned
         headers = {
@@ -76,6 +78,8 @@ class JinaClient:
             logger.warning("Jina API key is not set. Provide your own key to access a higher rate limit. See https://jina.ai/reader for more information.")
         data = {"url": url}
         try:
+            if max_response_bytes is not None and (isinstance(max_response_bytes, bool) or not isinstance(max_response_bytes, int) or max_response_bytes <= 0):
+                raise ValueError("max_response_bytes must be a positive integer or null")
             if isinstance(max_retries, bool) or not isinstance(max_retries, int) or max_retries < 0:
                 raise ValueError("max_retries must be a non-negative integer")
             if isinstance(retry_budget_seconds, bool) or not isinstance(retry_budget_seconds, (int, float)) or not math.isfinite(retry_budget_seconds) or retry_budget_seconds <= 0:
@@ -97,14 +101,28 @@ class JinaClient:
                         request_timeout = min(timeout, remaining) if remaining is not None else timeout
                         server_floor = None
                         try:
-                            response = await client.post("https://r.jina.ai/", headers=headers, json=data, timeout=request_timeout)
+                            if max_response_bytes is None:
+                                response = await client.post("https://r.jina.ai/", headers=headers, json=data, timeout=request_timeout)
+                                response_text = response.text
+                            else:
+                                async with client.stream("POST", "https://r.jina.ai/", headers=headers, json=data, timeout=request_timeout) as response:
+                                    content = bytearray()
+                                    # aiter_bytes decodes Content-Encoding once. Check before
+                                    # retaining each chunk; HTTPX decoder allocations are outside this cap.
+                                    async for chunk in response.aiter_bytes():
+                                        if len(content) + len(chunk) > max_response_bytes:
+                                            return f"Error: Jina API response exceeds max_response_bytes ({max_response_bytes})"
+                                        content.extend(chunk)
+                                    # Match HTTPX text semantics without decompressing again or
+                                    # mutating the response's private buffered-content state.
+                                    response_text = content.decode(response.encoding or "utf-8", errors="replace")
                         except (httpx.ConnectError, httpx.ConnectTimeout):
                             if attempt == max_retries:
                                 raise
                         else:
                             if response.status_code == 200:
-                                if response.text and response.text.strip():
-                                    return response.text
+                                if response_text and response_text.strip():
+                                    return response_text
                                 error_message = "Jina API returned empty response"
                                 logger.error(error_message)
                                 return f"Error: {error_message}"
@@ -112,21 +130,21 @@ class JinaClient:
                                 server_floor = _retry_after(response.headers.get("Retry-After"))
                             retryable = response.status_code in {502, 503, 504} or (response.status_code == 429 and server_floor is not None)
                             if not retryable or attempt == max_retries:
-                                error_message = f"Jina API returned status {response.status_code}: {response.text}"
+                                error_message = f"Jina API returned status {response.status_code}: {response_text}"
                                 logger.error(error_message)
                                 return f"Error: {error_message}"
 
                         # Only allowlisted failures reach the non-blocking wait.
                         remaining = deadline - asyncio.get_running_loop().time()
                         if server_floor is not None and server_floor >= remaining:
-                            return f"Error: Jina API returned status {response.status_code}: {response.text}"
+                            return f"Error: Jina API returned status {response.status_code}: {response_text}"
                         if remaining <= 0:
                             raise TimeoutError
                         wait = min(delay, remaining) * random.uniform(0.5, 1.0)
                         if server_floor is not None:
                             wait = max(wait, server_floor)
                             if wait >= remaining:
-                                return f"Error: Jina API returned status {response.status_code}: {response.text}"
+                                return f"Error: Jina API returned status {response.status_code}: {response_text}"
                         await asyncio.sleep(wait)
                         delay = min(delay * 2, 4.0)
         except Exception as e:

backend/packages/harness/deerflow/community/jina_ai/tools.py

@@ -56,14 +56,14 @@ async def web_fetch_tool(url: str) -> str:
     timeout = 10
     proxy = None
     trust_env = True
-    retry_options = {}
+    request_options = {}
     config = get_app_config().get_tool_config("web_fetch")
     if config is not None:
         timeout = _coerce_timeout(config.model_extra.get("timeout"), timeout)
         proxy = _coerce_proxy(config.model_extra.get("proxy"))
         trust_env = _coerce_bool(config.model_extra.get("trust_env"), trust_env)
-        retry_options = {key: config.model_extra[key] for key in ("max_retries", "retry_budget_seconds") if key in config.model_extra}
-    html_content = await jina_client.crawl(url, return_format="html", timeout=timeout, proxy=proxy, trust_env=trust_env, **retry_options)
+        request_options = {key: config.model_extra[key] for key in ("max_retries", "retry_budget_seconds", "max_response_bytes") if key in config.model_extra}
+    html_content = await jina_client.crawl(url, return_format="html", timeout=timeout, proxy=proxy, trust_env=trust_env, **request_options)
     if isinstance(html_content, str) and html_content.startswith("Error:"):
         return html_content
     article = await asyncio.to_thread(readability_extractor.extract_article, html_content, url=url)

backend/tests/test_jina_response_limit.py

@@ -0,0 +1,236 @@
+"""Offline response budgets exercised through HTTPX's real streaming decoder."""
+
+import asyncio
+import gzip
+import json
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, Mock
+
+import httpx
+import pytest
+
+from deerflow.community.jina_ai.jina_client import JinaClient
+
+pytestmark = pytest.mark.anyio
+
+
+@pytest.fixture
+def anyio_backend():
+    return "asyncio"
+
+
+class Stream(httpx.AsyncByteStream):
+    def __init__(self, chunks, failure=None, block=False):
+        self.chunks = chunks
+        self.failure = failure
+        self.block = block
+        self.entered = asyncio.Event()
+        self.reads = 0
+        self.closed = False
+
+    async def __aiter__(self):
+        for chunk in self.chunks:
+            self.reads += 1
+            yield chunk
+        self.entered.set()
+        if self.failure:
+            raise self.failure
+        if self.block:
+            await asyncio.Event().wait()
+
+    async def aclose(self):
+        self.closed = True
+
+
+@pytest.fixture
+def transport(monkeypatch):
+    real_client = httpx.AsyncClient
+    responses, requests, options = [], [], []
+
+    def handle(request):
+        requests.append(request)
+        return responses.pop(0)
+
+    def client(**kwargs):
+        options.append(kwargs)
+        # Record proxy configuration but keep all traffic on the offline transport.
+        return real_client(transport=httpx.MockTransport(handle), trust_env=False)
+
+    monkeypatch.setattr(httpx, "AsyncClient", client)
+    monkeypatch.setenv("JINA_API_KEY", "dummy-test-key")
+    return responses, requests, options
+
+
+def respond(transport, chunks, status=200, headers=None, **kwargs):
+    stream = Stream(chunks, **kwargs)
+    transport[0].append(httpx.Response(status, headers=headers, stream=stream))
+    return stream
+
+
+@pytest.mark.parametrize("options", [{}, {"max_response_bytes": None}, {"max_response_bytes": 5}])
+async def test_default_and_request_compatibility(transport, options):
+    stream = respond(transport, [b"hello"])
+    assert await JinaClient().crawl("https://example.com", timeout=7, proxy="http://proxy.invalid", trust_env=False, **options) == "hello"
+    request = transport[1][0]
+    assert request.method == "POST" and str(request.url) == "https://r.jina.ai/"
+    assert json.loads(request.content) == {"url": "https://example.com"}
+    assert request.headers["authorization"] == "Bearer dummy-test-key"
+    assert request.headers["x-return-format"] == "html"
+    assert request.headers["x-timeout"] == "7"
+    assert request.headers["content-type"] == "application/json"
+    assert request.extensions["timeout"]["read"] == 7
+    assert transport[2] == [{"proxy": "http://proxy.invalid", "trust_env": False}]
+    assert stream.closed
+
+
+@pytest.mark.parametrize("limit", [True, False, 0, -1, "5", 1.5, 5.0])
+async def test_invalid_before_client_creation(transport, limit):
+    result = await JinaClient().crawl("https://example.com", max_response_bytes=limit)
+    assert result.startswith("Error:") and "max_response_bytes" in result
+    assert not transport[1] and not transport[2]
+
+
+@pytest.mark.parametrize("length", [None, "1", "999999"])
+@pytest.mark.parametrize("status", [200, 400, 429, 502, 503, 504])
+async def test_overflow_stops_and_never_retries(transport, length, status):
+    headers = {} if length is None else {"content-length": length}
+    stream = respond(transport, [b"ab", b"cd", b"e", b"must not read"], status, headers)
+    result = await JinaClient().crawl("https://example.com", max_response_bytes=4, max_retries=2)
+    assert result.startswith("Error:") and "max_response_bytes" in result
+    assert "abcd" not in result and "must not read" not in result
+    assert stream.reads == 3 and stream.closed
+    assert len(transport[1]) == 1
+
+
+@pytest.mark.parametrize(
+    "chunks,headers,expected",
+    [
+        ([b"ab", b"cd"], {"content-length": "999"}, "abcd"),
+        ([b"\xe2", b"\x82\xac"], {}, "€"),
+        ([b"caf\xe9"], {"content-type": "text/html; charset=iso-8859-1"}, "café"),
+        ([b"a\xff"], {}, "a<EFBFBD>"),
+        ([b"a\xff"], {"content-type": "text/html; charset=invalid"}, "a<EFBFBD>"),
+    ],
+)
+async def test_exact_limit_and_encoding(transport, chunks, headers, expected):
+    stream = respond(transport, chunks, headers=headers)
+    assert await JinaClient().crawl("https://example.com", max_response_bytes=sum(map(len, chunks))) == expected
+    assert stream.closed
+
+
+@pytest.mark.parametrize("limit", [999, 1000])
+async def test_gzip_counts_decoded_bytes(transport, limit):
+    body = b"x" * 1000
+    compressed = gzip.compress(body)
+    stream = respond(transport, [compressed[:10], compressed[10:]], headers={"content-encoding": "gzip", "content-length": str(len(compressed))})
+    result = await JinaClient().crawl("https://example.com", max_response_bytes=limit)
+    if limit == 1000:
+        assert result == body.decode()
+    else:
+        assert result.startswith("Error:")
+        assert "max_response_bytes" in result
+    assert stream.closed
+
+
+@pytest.mark.parametrize("status,body,expected", [(400, b"bad", "status 400: bad"), (429, b"busy", "status 429: busy"), (200, b"  ", "empty response"), (200, b"", "empty response")])
+async def test_within_limit_errors(transport, status, body, expected):
+    stream = respond(transport, [body], status)
+    result = await JinaClient().crawl("https://example.com", max_response_bytes=4, max_retries=2)
+    assert result.startswith("Error:") and expected in result
+    assert stream.closed and len(transport[1]) == 1
+
+
+async def test_retry_resets_counter(transport, monkeypatch):
+    monkeypatch.setattr(asyncio, "sleep", AsyncMock())
+    streams = [respond(transport, [b"busy"], 503), respond(transport, [b"done"])]
+    assert await JinaClient().crawl("https://example.com", max_response_bytes=4, max_retries=1) == "done"
+    assert all(stream.closed for stream in streams)
+    assert len(transport[1]) == 2
+
+
+@pytest.mark.parametrize("failure", [httpx.ReadError("offline"), httpx.ReadTimeout("offline")])
+async def test_read_failure_closes_without_retry(transport, failure):
+    stream = respond(transport, [b"a"], failure=failure)
+    result = await JinaClient().crawl("https://example.com", max_response_bytes=4, max_retries=1)
+    assert result.startswith("Error:") and type(failure).__name__ in result
+    assert stream.closed and len(transport[1]) == 1
+
+
+async def test_cancellation_closes(transport):
+    stream = respond(transport, [b"a"], block=True)
+    task = asyncio.create_task(JinaClient().crawl("https://example.com", max_response_bytes=4, max_retries=1))
+    try:
+        await asyncio.wait_for(stream.entered.wait(), 1)
+        task.cancel()
+        with pytest.raises(asyncio.CancelledError):
+            await task
+        assert stream.closed
+    finally:
+        task.cancel()
+        await asyncio.gather(task, return_exceptions=True)
+
+
+async def test_shared_deadline_covers_retry_stream(transport, monkeypatch):
+    monkeypatch.setattr(asyncio, "sleep", AsyncMock())
+    first = respond(transport, [b"busy"], 503)
+    second = respond(transport, [b"a"], block=True)
+    result = await asyncio.wait_for(JinaClient().crawl("https://example.com", max_response_bytes=4, max_retries=2, retry_budget_seconds=0.05), 1)
+    assert "retry time budget exhausted" in result
+    assert first.closed and second.closed and second.entered.is_set()
+    timeouts = [request.extensions["timeout"]["read"] for request in transport[1]]
+    assert len(timeouts) == 2 and 0 < timeouts[1] < timeouts[0] <= 0.05
+
+
+@pytest.mark.parametrize("limit", [4, "4"])
+async def test_tool_error_skips_extraction(transport, monkeypatch, limit):
+    from deerflow.community.jina_ai import tools
+
+    config = SimpleNamespace(model_extra={"max_response_bytes": limit})
+    monkeypatch.setattr(tools, "get_app_config", lambda: SimpleNamespace(get_tool_config=lambda name: config))
+    extract = Mock(side_effect=AssertionError("must not extract"))
+    monkeypatch.setattr(tools.readability_extractor, "extract_article", extract)
+    stream = respond(transport, [b"oversized", b"unread"])
+    result = await tools.web_fetch_tool.ainvoke({"url": "https://example.com"})
+    assert result.startswith("Error:") and "max_response_bytes" in result
+    extract.assert_not_called()
+    assert set(tools.web_fetch_tool.args) == {"url"}
+    if isinstance(limit, int):
+        assert stream.closed and stream.reads == 1
+    else:
+        assert not transport[2]
+
+
+async def test_tool_real_extraction_preserves_output_cap(transport, monkeypatch):
+    from deerflow.community.jina_ai import tools
+
+    monkeypatch.setattr("readabilipy.simple_json.have_node", lambda: False)
+    body = ("<html><title>Example</title><body><article><p>" + "Useful article text. " * 500 + "</p></article></body></html>").encode()
+    config = SimpleNamespace(model_extra={"max_response_bytes": len(body)})
+    monkeypatch.setattr(tools, "get_app_config", lambda: SimpleNamespace(get_tool_config=lambda name: config))
+    stream = respond(transport, [body])
+    result = await tools.web_fetch_tool.ainvoke({"url": "https://example.com"})
+    assert "Useful article text." in result and len(result) == 4096
+    assert stream.closed
+
+
+@pytest.mark.parametrize("status", [429, 503])
+@pytest.mark.parametrize("budget,limit", [(30, 4), (5, 4), (30, 3)])
+async def test_streaming_preserves_retry_after_policy(transport, monkeypatch, status, budget, limit):
+    sleep = AsyncMock()
+    monkeypatch.setattr(asyncio, "sleep", sleep)
+    first = respond(transport, [b"busy"], status, {"Retry-After": "7"})
+    second = respond(transport, [b"done"])
+    result = await JinaClient().crawl("https://example.com", max_response_bytes=limit, max_retries=1, retry_budget_seconds=budget)
+    assert first.closed
+    if limit == 3:
+        assert "max_response_bytes" in result
+        sleep.assert_not_awaited()
+        assert len(transport[1]) == 1
+    elif budget == 5:
+        assert result == f"Error: Jina API returned status {status}: busy"
+        sleep.assert_not_awaited()
+        assert len(transport[1]) == 1
+    else:
+        assert result == "done"
+        sleep.assert_awaited_once_with(7)
+        assert second.closed and len(transport[1]) == 2