diff --git a/README.md b/README.md
index ae0bc5000d..da209bfb2d 100644
--- a/README.md
+++ b/README.md
@@ -142,6 +142,8 @@ It is disabled by default; see the linked guide to enable it.
 
    Jina, Browserless, and InfoQuest web fetches resolve relative links and image sources using the requested page URL (or a usable HTML base URL), so returned Markdown includes complete destinations. Link resolution preserves the surrounding HTML source, including malformed-page formatting.
 
+   Jina fetches support opt-in bounded retries via `max_retries` (default `0`) and `retry_budget_seconds` (default `30`) in the tool configuration. Retry waits are randomized within the time budget. Retries may increase upstream requests and cost; see [Jina fetch retries](backend/docs/CONFIGURATION.md#jina-fetch-retries).
+
    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 ac78610dc1..ef6a735dda 100644
--- a/backend/docs/CONFIGURATION.md
+++ b/backend/docs/CONFIGURATION.md
@@ -575,6 +575,41 @@ empty or omitted `include_domains`. These filters compose with `max_results`
 and the model's optional `time_range`. The model-visible arguments remain `query`
 and `time_range`; the filters do not apply to `web_fetch` or other search providers.
 
+#### Jina fetch retries
+
+Jina's `web_fetch` keeps one attempt by default. Configure retries on its existing
+`tools` entry; model-facing arguments remain unchanged:
+
+```yaml
+tools:
+  - name: web_fetch
+    group: web
+    use: deerflow.community.jina_ai.tools:web_fetch_tool
+    timeout: 10
+    max_retries: 2              # Additional attempts; default 0 (disabled)
+    retry_budget_seconds: 30   # Total request + backoff budget; default 30
+```
+
+`max_retries` must be a non-negative integer and `retry_budget_seconds` a finite,
+positive number (YAML numbers, not strings or booleans). Invalid values return an
+`Error:` without sending a request. The budget applies only when retries are enabled,
+starts before HTTP client creation, and covers all attempts and waits. Each HTTP
+request timeout is capped by the remaining budget; the outer deadline also bounds
+responses that keep delivering data. The existing `timeout` remains Jina's
+`X-Timeout` header and the per-request HTTP timeout limit.
+
+Only HTTP 502/503/504 and HTTPX connection-establishment errors (`ConnectError`,
+`ConnectTimeout`) are retried. Authentication/client errors, 429, other statuses,
+empty successful responses, read/write timeouts and arbitrary exceptions are not
+retried. `Retry-After` is not interpreted. Backoff ceilings start at 0.5 seconds,
+double to 1 and 2 seconds, then stay at 4 seconds. Each asynchronous wait caps its
+ceiling by the remaining budget and independently samples a uniform factor from
+0.5 to 1.0, reducing synchronized retries without increasing the wait cap.
+Cancellation propagates during requests and waits. This stops local work; it
+cannot cancel work already started by Jina. Enabling retries can send up to `1 + max_retries` upstream requests
+and incur additional cost. Successful content and final `Error:` results retain
+the existing contract.
+
 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
new file mode 100644
index 0000000000..a7987ab2a2
--- /dev/null
+++ b/backend/packages/harness/deerflow/community/jina_ai/AGENTS.md
@@ -0,0 +1,9 @@
+# Jina web fetch
+
+Jina opt-in retries stay provider-local: `max_retries=0` preserves one attempt;
+`retry_budget_seconds` bounds enabled request sequences and asynchronous backoff
+with one asyncio deadline. Only 502/503/504 and connection-establishment failures
+retry; cancellation propagates. Backoff ceilings double from 0.5 to 4 seconds.
+Each wait caps the ceiling by the remaining budget, then multiplies it by a fresh
+uniform random factor from 0.5 to 1.0 to avoid synchronized retries.
+Offline coverage: `tests/test_jina_retries.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 8c79e05694..bcf0053f15 100644
--- a/backend/packages/harness/deerflow/community/jina_ai/jina_client.py
+++ b/backend/packages/harness/deerflow/community/jina_ai/jina_client.py
@@ -1,5 +1,8 @@
+import asyncio
 import logging
+import math
 import os
+import random
 
 import httpx
 
@@ -9,7 +12,8 @@ _api_key_warned = False
 
 
 class JinaClient:
-    async def crawl(self, url: str, return_format: str = "html", timeout: int = 10, proxy: str | None = None, trust_env: bool = True) -> 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) -> str:
+        """Fetch with optional bounded retries; cancellation always propagates."""
         global _api_key_warned
         headers = {
             "Content-Type": "application/json",
@@ -23,24 +27,52 @@ 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:
-            client_kwargs: dict[str, object] = {"trust_env": trust_env}
-            if proxy:
-                client_kwargs["proxy"] = proxy
-            async with httpx.AsyncClient(**client_kwargs) as client:
-                response = await client.post("https://r.jina.ai/", headers=headers, json=data, timeout=timeout)
+            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:
+                raise ValueError("retry_budget_seconds must be a finite positive number")
 
-            if response.status_code != 200:
-                error_message = f"Jina API returned status {response.status_code}: {response.text}"
-                logger.error(error_message)
-                return f"Error: {error_message}"
+            # HTTPX timeouts are per network phase, so use an outer deadline to
+            # bound the complete request sequence (including waits and cleanup).
+            deadline = asyncio.get_running_loop().time() + retry_budget_seconds if max_retries else None
+            async with asyncio.timeout_at(deadline):
+                client_kwargs: dict[str, object] = {"trust_env": trust_env}
+                if proxy:
+                    client_kwargs["proxy"] = proxy
+                async with httpx.AsyncClient(**client_kwargs) as client:
+                    delay = 0.5
+                    for attempt in range(max_retries + 1):
+                        remaining = deadline - asyncio.get_running_loop().time() if deadline is not None else None
+                        if remaining is not None and remaining <= 0:
+                            raise TimeoutError
+                        request_timeout = min(timeout, remaining) if remaining is not None else timeout
+                        try:
+                            response = await client.post("https://r.jina.ai/", headers=headers, json=data, timeout=request_timeout)
+                        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
+                                error_message = "Jina API returned empty response"
+                                logger.error(error_message)
+                                return f"Error: {error_message}"
+                            if response.status_code not in {502, 503, 504} or attempt == max_retries:
+                                error_message = f"Jina API returned status {response.status_code}: {response.text}"
+                                logger.error(error_message)
+                                return f"Error: {error_message}"
 
-            if not response.text or not response.text.strip():
-                error_message = "Jina API returned empty response"
-                logger.error(error_message)
-                return f"Error: {error_message}"
-
-            return response.text
+                        # Only allowlisted failures reach the non-blocking wait.
+                        remaining = deadline - asyncio.get_running_loop().time()
+                        if remaining <= 0:
+                            raise TimeoutError
+                        await asyncio.sleep(min(delay, remaining) * random.uniform(0.5, 1.0))
+                        delay = min(delay * 2, 4.0)
         except Exception as e:
-            error_message = f"Request to Jina API failed: {type(e).__name__}: {e}"
+            if isinstance(e, TimeoutError) and max_retries and asyncio.get_running_loop().time() >= deadline:
+                error_message = "Request to Jina API failed: retry time budget exhausted"
+            else:
+                error_message = f"Request to Jina API failed: {type(e).__name__}: {e}"
             logger.warning(error_message)
             return f"Error: {error_message}"
diff --git a/backend/packages/harness/deerflow/community/jina_ai/tools.py b/backend/packages/harness/deerflow/community/jina_ai/tools.py
index 913f420bb7..5f80653203 100644
--- a/backend/packages/harness/deerflow/community/jina_ai/tools.py
+++ b/backend/packages/harness/deerflow/community/jina_ai/tools.py
@@ -56,12 +56,14 @@ async def web_fetch_tool(url: str) -> str:
     timeout = 10
     proxy = None
     trust_env = True
+    retry_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)
-    html_content = await jina_client.crawl(url, return_format="html", timeout=timeout, proxy=proxy, 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)
     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_agent_guidance_check.py b/backend/tests/test_agent_guidance_check.py
index 7f41635012..6213a76fa3 100644
--- a/backend/tests/test_agent_guidance_check.py
+++ b/backend/tests/test_agent_guidance_check.py
@@ -19,6 +19,7 @@ EXPECTED_GUIDANCE_PATHS = {
     "backend/packages/harness/deerflow/agents/middlewares/AGENTS.md",
     "backend/packages/harness/deerflow/agents/memory/AGENTS.md",
     "backend/packages/harness/deerflow/agents/task_continuity/AGENTS.md",
+    "backend/packages/harness/deerflow/community/jina_ai/AGENTS.md",
     "backend/packages/harness/deerflow/community/ragflow/AGENTS.md",
     "backend/packages/harness/deerflow/community/tavily/AGENTS.md",
     "backend/packages/harness/deerflow/community/e2b_sandbox/AGENTS.md",
diff --git a/backend/tests/test_browser_automation.py b/backend/tests/test_browser_automation.py
index db9dd2a746..83a302a05e 100644
--- a/backend/tests/test_browser_automation.py
+++ b/backend/tests/test_browser_automation.py
@@ -9,6 +9,7 @@ from __future__ import annotations
 
 import asyncio
 import gc
+import ipaddress
 import sys
 from types import ModuleType, SimpleNamespace
 from unittest.mock import AsyncMock, MagicMock, patch
@@ -27,6 +28,17 @@ from deerflow.community.browser_automation.session import (
     SnapshotElement,
 )
 
+
+@pytest.fixture
+def public_dns():
+    """Keep mocked tool calls offline while exercising the real URL safety check."""
+    with patch(
+        "deerflow.community.browser_automation.tools._resolve_host_addresses",
+        return_value=[ipaddress.ip_address("93.184.216.34")],
+    ):
+        yield
+
+
 PlaywrightTimeoutError = type(
     "TimeoutError",
     (TimeoutError,),
@@ -70,6 +82,7 @@ class TestBrowserTools:
         manager.get_session.return_value = session
         return patch.object(tools, "get_browser_session_manager", return_value=manager), manager
 
+    @pytest.mark.usefixtures("public_dns")
     async def test_navigate_returns_snapshot(self):
         session = MagicMock()
         session.navigate = AsyncMock(return_value=_snapshot())
@@ -86,6 +99,7 @@ class TestBrowserTools:
         session.navigate.assert_awaited_once_with("https://example.com")
         manager.get_session.assert_called_once()
 
+    @pytest.mark.usefixtures("public_dns")
     async def test_navigate_emits_screenshot_artifact_and_browser_view(self, tmp_path):
         outputs = tmp_path / "outputs"
         outputs.mkdir()
@@ -113,6 +127,7 @@ class TestBrowserTools:
         assert meta["url"] == "https://example.com/"
         assert meta["title"] == "Example"
 
+    @pytest.mark.usefixtures("public_dns")
     async def test_navigate_screenshot_failure_does_not_break_action(self, tmp_path):
         outputs = tmp_path / "outputs"
         outputs.mkdir()
diff --git a/backend/tests/test_crawl4ai_tools.py b/backend/tests/test_crawl4ai_tools.py
index ff6388aca0..4f97ea8ff8 100644
--- a/backend/tests/test_crawl4ai_tools.py
+++ b/backend/tests/test_crawl4ai_tools.py
@@ -9,6 +9,16 @@ import pytest
 from deerflow.community.crawl4ai.crawl4ai_client import Crawl4AiClient
 
 
+@pytest.fixture
+def public_dns():
+    """Keep mocked tool calls offline while exercising the real URL safety check."""
+    with patch(
+        "deerflow.community.url_safety.resolve_host_addresses",
+        return_value=[ipaddress.ip_address("93.184.216.34")],
+    ):
+        yield
+
+
 class AsyncMock(MagicMock):
     """Mock that supports async call."""
 
@@ -164,6 +174,7 @@ class TestCrawl4AiTools:
     """Tests for the Crawl4AI tool functions."""
 
     @patch("deerflow.community.crawl4ai.tools._build_client")
+    @pytest.mark.usefixtures("public_dns")
     async def test_web_fetch_tool_success(self, mock_build):
         from deerflow.community.crawl4ai import tools
 
@@ -178,6 +189,7 @@ class TestCrawl4AiTools:
         assert "Error:" not in result
 
     @patch("deerflow.community.crawl4ai.tools._build_client")
+    @pytest.mark.usefixtures("public_dns")
     async def test_web_fetch_tool_truncates_to_4096(self, mock_build):
         from deerflow.community.crawl4ai import tools
 
@@ -191,6 +203,7 @@ class TestCrawl4AiTools:
         assert len(result) == 4096
 
     @patch("deerflow.community.crawl4ai.tools._build_client")
+    @pytest.mark.usefixtures("public_dns")
     async def test_web_fetch_tool_error_passthrough(self, mock_build):
         from deerflow.community.crawl4ai import tools
 
@@ -202,8 +215,10 @@ class TestCrawl4AiTools:
             result = await tools.web_fetch_tool.ainvoke("https://example.com")
 
         assert result.startswith("Error:")
+        mock_client.fetch_markdown.assert_called_once()
 
     @patch("deerflow.community.crawl4ai.tools._build_client")
+    @pytest.mark.usefixtures("public_dns")
     async def test_web_fetch_tool_exception(self, mock_build):
         from deerflow.community.crawl4ai import tools
 
@@ -215,6 +230,7 @@ class TestCrawl4AiTools:
             result = await tools.web_fetch_tool.ainvoke("https://example.com")
 
         assert result.startswith("Error:")
+        mock_client.fetch_markdown.assert_called_once()
 
     @patch("deerflow.community.crawl4ai.tools._build_client")
     async def test_web_fetch_tool_rejects_metadata_ip(self, mock_build):
@@ -255,6 +271,7 @@ class TestCrawl4AiTools:
         mock_client.fetch_markdown.assert_called_once()
 
     @patch("deerflow.community.crawl4ai.tools._build_client")
+    @pytest.mark.usefixtures("public_dns")
     async def test_web_fetch_tool_reads_config_once(self, mock_build):
         """Config is read exactly once per invocation (no split read on hot-reload)."""
         from deerflow.community.crawl4ai import tools
@@ -269,6 +286,7 @@ class TestCrawl4AiTools:
         mock_cfg.assert_called_once_with("web_fetch")
 
     @patch("deerflow.community.crawl4ai.tools._build_client")
+    @pytest.mark.usefixtures("public_dns")
     async def test_web_fetch_tool_passes_configured_filter(self, mock_build):
         from deerflow.community.crawl4ai import tools
 
@@ -283,6 +301,7 @@ class TestCrawl4AiTools:
         assert mock_client.fetch_markdown.call_args.kwargs.get("filter_mode") == "raw"
 
     @patch("deerflow.community.crawl4ai.tools._build_client")
+    @pytest.mark.usefixtures("public_dns")
     async def test_web_fetch_tool_invalid_filter_falls_back_to_fit(self, mock_build):
         from deerflow.community.crawl4ai import tools
 
diff --git a/backend/tests/test_extension_manager.py b/backend/tests/test_extension_manager.py
index b809aa2fb9..f04bead15a 100644
--- a/backend/tests/test_extension_manager.py
+++ b/backend/tests/test_extension_manager.py
@@ -92,7 +92,7 @@ packages = ["demo_extension"]
     )
 
 
-def _write_host_project(root: Path) -> None:
+def _write_host_project(root: Path, *, no_index: bool = False) -> None:
     backend = root / "backend"
     backend.mkdir()
     (backend / "pyproject.toml").write_text(
@@ -111,6 +111,10 @@ default-groups = ["extensions"]
 """,
         encoding="utf-8",
     )
+    if no_index:
+        # Use uv's supported project setting; UV_NO_INDEX is not recognized.
+        with (backend / "pyproject.toml").open("a", encoding="utf-8") as project:
+            project.write("no-index = true\n")
     (root / "config.yaml").write_text("config_version: 1\n", encoding="utf-8")
 
 
@@ -820,10 +824,9 @@ def test_install_rejects_a_pypi_requirement_resolved_from_an_external_local_whee
     root = tmp_path / "deer-flow"
     wheels = tmp_path / "wheels"
     root.mkdir()
-    _write_host_project(root)
+    _write_host_project(root, no_index=True)
     _write_demo_wheel(wheels)
     monkeypatch.setenv("UV_FIND_LINKS", str(wheels))
-    monkeypatch.setenv("UV_NO_INDEX", "1")
     pyproject_path = root / "backend" / "pyproject.toml"
     config_path = root / "config.yaml"
     before = (pyproject_path.read_bytes(), config_path.read_bytes())
@@ -843,12 +846,11 @@ def test_install_rejects_a_local_wheel_directory_ignored_by_the_docker_context(
 ) -> None:
     root = tmp_path / "deer-flow"
     root.mkdir()
-    _write_host_project(root)
+    _write_host_project(root, no_index=True)
     wheels = root / "backend" / relative_wheels
     wheels.parent.mkdir(parents=True, exist_ok=True)
     _write_demo_wheel(wheels)
     monkeypatch.setenv("UV_FIND_LINKS", str(wheels))
-    monkeypatch.setenv("UV_NO_INDEX", "1")
     pyproject_path = root / "backend" / "pyproject.toml"
     config_path = root / "config.yaml"
     before = (pyproject_path.read_bytes(), config_path.read_bytes())
@@ -866,13 +868,12 @@ def test_install_rejects_a_relative_find_links_wheelhouse_outside_the_build_cont
 ) -> None:
     root = tmp_path / "deer-flow"
     root.mkdir()
-    _write_host_project(root)
+    _write_host_project(root, no_index=True)
     _write_demo_wheel(root / "backend" / "wheelhouse")
     # uv resolves a relative UV_FIND_LINKS against its working directory (the
     # backend project), so the lock records a relative registry that only
     # exists on this host.
     monkeypatch.setenv("UV_FIND_LINKS", "wheelhouse")
-    monkeypatch.setenv("UV_NO_INDEX", "1")
     pyproject_path = root / "backend" / "pyproject.toml"
     config_path = root / "config.yaml"
     before = (pyproject_path.read_bytes(), config_path.read_bytes())
diff --git a/backend/tests/test_fastcrw_tools.py b/backend/tests/test_fastcrw_tools.py
index 1248d71f09..00afab9e7b 100644
--- a/backend/tests/test_fastcrw_tools.py
+++ b/backend/tests/test_fastcrw_tools.py
@@ -4,6 +4,18 @@ import ipaddress
 import json
 from unittest.mock import MagicMock, patch
 
+import pytest
+
+
+@pytest.fixture
+def public_dns():
+    """Keep mocked tool calls offline while exercising the real URL safety check."""
+    with patch(
+        "deerflow.community.url_safety.resolve_host_addresses",
+        return_value=[ipaddress.ip_address("93.184.216.34")],
+    ):
+        yield
+
 
 class TestWebSearchTool:
     @patch.dict("os.environ", {}, clear=True)
@@ -69,6 +81,7 @@ class TestWebFetchTool:
     @patch.dict("os.environ", {}, clear=True)
     @patch("deerflow.community.fastcrw.tools.FirecrawlApp")
     @patch("deerflow.community.fastcrw.tools.get_app_config")
+    @pytest.mark.usefixtures("public_dns")
     def test_fetch_uses_web_fetch_config(self, mock_get_app_config, mock_fastcrw_cls):
         fetch_config = MagicMock()
         fetch_config.model_extra = {"api_key": "fastcrw-fetch-key", "base_url": "http://localhost:3000"}
@@ -100,6 +113,7 @@ class TestWebFetchTool:
     @patch.dict("os.environ", {}, clear=True)
     @patch("deerflow.community.fastcrw.tools.FirecrawlApp")
     @patch("deerflow.community.fastcrw.tools.get_app_config")
+    @pytest.mark.usefixtures("public_dns")
     def test_fetch_returns_error_when_no_content(self, mock_get_app_config, mock_fastcrw_cls):
         mock_get_app_config.return_value.get_tool_config.return_value = None
 
@@ -115,6 +129,7 @@ class TestWebFetchTool:
     @patch.dict("os.environ", {}, clear=True)
     @patch("deerflow.community.fastcrw.tools.FirecrawlApp")
     @patch("deerflow.community.fastcrw.tools.get_app_config")
+    @pytest.mark.usefixtures("public_dns")
     def test_fetch_returns_error_string_on_exception(self, mock_get_app_config, mock_fastcrw_cls):
         mock_get_app_config.return_value.get_tool_config.return_value = None
         mock_fastcrw_cls.return_value.scrape.side_effect = RuntimeError("scrape failed")
diff --git a/backend/tests/test_jina_retries.py b/backend/tests/test_jina_retries.py
new file mode 100644
index 0000000000..9a37d565c4
--- /dev/null
+++ b/backend/tests/test_jina_retries.py
@@ -0,0 +1,181 @@
+"""Offline coverage of Jina's opt-in retry policy."""
+
+import asyncio
+import random
+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"
+
+
+@pytest.fixture
+def requests(monkeypatch):
+    post = AsyncMock()
+    client = AsyncMock()
+    client.__aenter__.return_value = client
+    client.post = post
+    monkeypatch.setattr(httpx, "AsyncClient", lambda **kwargs: client)
+    monkeypatch.setenv("JINA_API_KEY", "test-key")
+    return post
+
+
+async def test_default_single_attempt(requests, monkeypatch):
+    jitter = Mock()
+    monkeypatch.setattr(random, "uniform", jitter)
+    requests.return_value = httpx.Response(503, text="unavailable")
+    assert "503" in await JinaClient().crawl("https://example.com")
+    assert requests.await_count == 1
+    jitter.assert_not_called()
+
+
+@pytest.mark.parametrize("failure", [httpx.Response(502), httpx.Response(503), httpx.Response(504), httpx.ConnectError("offline"), httpx.ConnectTimeout("offline")])
+async def test_transient_recovers(requests, monkeypatch, failure):
+    monkeypatch.setattr(asyncio, "sleep", AsyncMock())
+    requests.side_effect = [failure, httpx.Response(200, text="success")]
+    assert await JinaClient().crawl("https://example.com", max_retries=1) == "success"
+    assert requests.await_count == 2
+
+
+@pytest.mark.parametrize("failure", [httpx.Response(code) for code in (400, 401, 403, 404, 429, 500)] + [httpx.Response(200, text="  "), httpx.ReadTimeout("slow"), httpx.WriteError("write"), RuntimeError("unexpected")])
+async def test_permanent_failure(requests, failure):
+    requests.side_effect = [failure, httpx.Response(200, text="unexpected")]
+    assert (await JinaClient().crawl("https://example.com", max_retries=2)).startswith("Error:")
+    assert requests.await_count == 1
+
+
+@pytest.mark.parametrize(
+    ("jitter_factor", "expected_waits"),
+    [(0.5, [0.25, 0.5, 1, 2, 2, 2]), (0.75, [0.375, 0.75, 1.5, 3, 3, 3]), (1.0, [0.5, 1, 2, 4, 4, 4])],
+)
+async def test_exhaustion_and_capped_backoff(requests, monkeypatch, jitter_factor, expected_waits):
+    sleep = AsyncMock()
+    jitter = Mock(return_value=jitter_factor)
+    monkeypatch.setattr(asyncio, "sleep", sleep)
+    monkeypatch.setattr(random, "uniform", jitter)
+    requests.return_value = httpx.Response(503, text="unavailable")
+    assert "503" in await JinaClient().crawl("https://example.com", max_retries=6)
+    assert requests.await_count == 7
+    assert [call.args[0] for call in sleep.await_args_list] == expected_waits
+    assert jitter.call_count == 6
+    assert all(call.args == (0.5, 1.0) for call in jitter.call_args_list)
+
+
+async def test_each_retry_samples_new_jitter(requests, monkeypatch):
+    sleep = AsyncMock()
+    jitter = Mock(side_effect=[0.5, 1.0, 0.75])
+    monkeypatch.setattr(asyncio, "sleep", sleep)
+    monkeypatch.setattr(random, "uniform", jitter)
+    requests.return_value = httpx.Response(503, text="unavailable")
+
+    assert "503" in await JinaClient().crawl("https://example.com", max_retries=3)
+
+    assert requests.await_count == 4
+    assert [call.args[0] for call in sleep.await_args_list] == [0.25, 1, 1.5]
+    assert jitter.call_count == 3
+
+
+async def test_backoff_jitter_is_capped_by_remaining_budget(requests, monkeypatch):
+    sleep = AsyncMock()
+    monkeypatch.setattr(asyncio, "sleep", sleep)
+    monkeypatch.setattr(random, "uniform", lambda low, high: 0.5)
+    requests.side_effect = [httpx.Response(503), httpx.Response(200, text="success")]
+
+    assert await JinaClient().crawl("https://example.com", max_retries=1, retry_budget_seconds=0.1) == "success"
+
+    assert requests.await_count == 2
+    sleep.assert_awaited_once()
+    assert 0 < sleep.await_args.args[0] <= 0.05
+
+
+@pytest.mark.parametrize("during_backoff", [False, True])
+async def test_shared_deadline(requests, monkeypatch, during_backoff):
+    monkeypatch.setattr(random, "uniform", lambda low, high: 1.0)
+
+    async def post(*args, **kwargs):
+        assert 0 < kwargs["timeout"] <= 0.05
+        if not during_backoff:
+            await asyncio.Event().wait()
+        return httpx.Response(503)
+
+    requests.side_effect = post
+    result = await asyncio.wait_for(JinaClient().crawl("https://example.com", max_retries=2, retry_budget_seconds=0.05), timeout=1)
+    assert result.startswith("Error:")
+    assert "budget" in result
+    assert requests.await_count == 1
+
+
+async def test_budget_is_shared_across_attempts(requests, monkeypatch):
+    timeouts = []
+    real_sleep = asyncio.sleep
+
+    async def post(*args, **kwargs):
+        timeouts.append(kwargs["timeout"])
+        if len(timeouts) == 1:
+            await real_sleep(0.02)
+            return httpx.Response(503)
+        await asyncio.Event().wait()
+
+    monkeypatch.setattr(asyncio, "sleep", AsyncMock())
+    requests.side_effect = post
+    result = await asyncio.wait_for(JinaClient().crawl("https://example.com", max_retries=2, retry_budget_seconds=0.1), 1)
+    assert "budget" in result
+    assert len(timeouts) == 2
+    assert timeouts[1] < timeouts[0]
+
+
+@pytest.mark.parametrize("during_backoff", [False, True])
+async def test_cancellation(requests, monkeypatch, during_backoff):
+    entered = asyncio.Event()
+    cancelled = asyncio.Event()
+
+    async def block(*args, **kwargs):
+        entered.set()
+        try:
+            await asyncio.Event().wait()
+        finally:
+            cancelled.set()
+
+    if during_backoff:
+        requests.return_value = httpx.Response(503)
+        monkeypatch.setattr(asyncio, "sleep", block)
+    else:
+        requests.side_effect = block
+    task = asyncio.create_task(JinaClient().crawl("https://example.com", max_retries=2))
+    try:
+        await asyncio.wait_for(entered.wait(), 1)
+        task.cancel()
+        with pytest.raises(asyncio.CancelledError):
+            await asyncio.wait_for(task, 1)
+        assert cancelled.is_set()
+        assert requests.await_count == 1
+    finally:
+        task.cancel()
+        await asyncio.gather(task, return_exceptions=True)
+
+
+@pytest.mark.parametrize("options", [{"max_retries": value} for value in (-1, True, 1.5, "2", None)] + [{"retry_budget_seconds": value} for value in (0, -1, True, float("nan"), float("inf"), "10", None)])
+async def test_invalid_settings_do_not_send(requests, options):
+    assert (await JinaClient().crawl("https://example.com", **options)).startswith("Error:")
+    requests.assert_not_awaited()
+
+
+async def test_tool_forwards_settings(monkeypatch):
+    from deerflow.community.jina_ai import tools
+
+    crawl = AsyncMock(return_value="Error: test")
+    config = SimpleNamespace(model_extra={"max_retries": 2, "retry_budget_seconds": 12.5})
+    monkeypatch.setattr(tools, "get_app_config", lambda: SimpleNamespace(get_tool_config=lambda name: config))
+    monkeypatch.setattr(JinaClient, "crawl", crawl)
+    assert await tools.web_fetch_tool.ainvoke("https://example.com") == "Error: test"
+    assert crawl.call_args.kwargs["max_retries"] == 2
+    assert crawl.call_args.kwargs["retry_budget_seconds"] == 12.5
diff --git a/config.example.yaml b/config.example.yaml
index 6cf19e4cec..58fad20c9e 100644
--- a/config.example.yaml
+++ b/config.example.yaml
@@ -1164,6 +1164,9 @@ tools:
     group: web
     use: deerflow.community.jina_ai.tools:web_fetch_tool
     timeout: 10
+    # Opt-in retries may increase upstream requests and cost.
+    # max_retries: 0
+    # retry_budget_seconds: 30
     # Optional proxy for restricted networks / Docker / WSL.
     # Use host.docker.internal instead of 127.0.0.1 when the proxy runs on the host.
     # proxy: $HTTPS_PROXY