From 51a33c79b340bc724937a3017247f3fc98c8b2b1 Mon Sep 17 00:00:00 2001 From: creed <115765389+97Three@users.noreply.github.com> Date: Wed, 7 Oct 2026 10:10:36 +0800 Subject: [PATCH] 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 | 2 + backend/docs/CONFIGURATION.md | 21 ++ .../deerflow/community/jina_ai/AGENTS.md | 8 + .../deerflow/community/jina_ai/jina_client.py | 32 ++- .../deerflow/community/jina_ai/tools.py | 6 +- backend/tests/test_jina_response_limit.py | 236 ++++++++++++++++++ 6 files changed, 295 insertions(+), 10 deletions(-) create mode 100644 backend/tests/test_jina_response_limit.py diff --git a/README.md b/README.md index fc5285fc32..43a767eae1 100644 --- a/README.md +++ b/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 diff --git a/backend/docs/CONFIGURATION.md b/backend/docs/CONFIGURATION.md index be8ac6e8d7..5478841957 100644 --- a/backend/docs/CONFIGURATION.md +++ b/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"` diff --git a/backend/packages/harness/deerflow/community/jina_ai/AGENTS.md b/backend/packages/harness/deerflow/community/jina_ai/AGENTS.md index d3f2452083..de5e0a4312 100644 --- a/backend/packages/harness/deerflow/community/jina_ai/AGENTS.md +++ b/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`. diff --git a/backend/packages/harness/deerflow/community/jina_ai/jina_client.py b/backend/packages/harness/deerflow/community/jina_ai/jina_client.py index e2673b5823..856a96c33b 100644 --- a/backend/packages/harness/deerflow/community/jina_ai/jina_client.py +++ b/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: diff --git a/backend/packages/harness/deerflow/community/jina_ai/tools.py b/backend/packages/harness/deerflow/community/jina_ai/tools.py index 5f80653203..fd1e5c27df 100644 --- a/backend/packages/harness/deerflow/community/jina_ai/tools.py +++ b/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) diff --git a/backend/tests/test_jina_response_limit.py b/backend/tests/test_jina_response_limit.py new file mode 100644 index 0000000000..0fbd668f69 --- /dev/null +++ b/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�"), + ([b"a\xff"], {"content-type": "text/html; charset=invalid"}, "a�"), + ], +) +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 = ("Example

" + "Useful article text. " * 500 + "

").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