fix(community): resolve SSRF-guarded hostnames off the event loop (#6140)

* test(blocking-io): catch socket.getaddrinfo on the event loop

Blockbuster wraps socket.socket methods but not the module-level
resolver, so a synchronous hostname lookup on the loop passes the strict
gate. The SSRF guard in deerflow.community.url_safety resolves with
socket.getaddrinfo from async web tools, and the anchors added next stay
green against that unfixed code without this rule.

Add socket.getaddrinfo to _PROJECT_BLOCKING_RULES, scoped to app.* and
deerflow.* callers. loop.getaddrinfo and asyncio.to_thread resolve in a
worker thread and are not flagged. The full make test-blocking-io suite
stays green with the rule active (187 passed).

* fix(community): resolve SSRF-guarded hostnames off the event loop

validate_public_http_url resolves hostnames with blocking
socket.getaddrinfo. The crawl4ai and Browserless web_fetch, web_capture,
browser_navigate, the Gateway browser navigate route, and the Live
stream's navigate input and seed all called it directly from async code,
so a slow DNS answer for a model- or user-chosen URL stalled the Gateway
event loop for the whole lookup.

Run the guard through asyncio.to_thread at each of those call sites, the
same way personal_network already offloads it. The screening result is
unchanged. fastcrw's web_fetch is a sync tool and personal MCP validation
already runs in a worker, so neither needed a change.

* docs(changelog): reference #6140 in the SSRF guard event-loop entry

* fix(browser): run the Playwright request guard off the shared browser loop

BrowserSession._install_request_guard screens every redirect hop, iframe
and subresource by calling the SSRF guard inside the async route
callback. The guard resolves hostnames with blocking socket.getaddrinfo,
and that callback runs on the process-wide Playwright loop, which also
pumps Live frames and input for every browser session, so one slow
lookup stalled all of them.

Run the guard through asyncio.to_thread there as well, and pin it with a
strict-gate anchor that drives the real route handler. Also note in the
url_safety docstring that the screen is check-time only: a caller that
re-resolves at connect time is not pinned to the vetted IPs.

---------

Co-authored-by: Willem Jiang <willem.jiang@gmail.com>

CHANGELOG.md

@@ -460,6 +460,20 @@ This release closes that milestone with **301 merged pull requests**.
 
 ### Fixed
 
+- **community:** The SSRF URL guard no longer stalls the Gateway event loop
+  while it resolves a hostname. `validate_public_http_url` resolves with
+  blocking `socket.getaddrinfo`, and the crawl4ai and Browserless `web_fetch`,
+  `web_capture`, `browser_navigate`, the Gateway browser navigate route, and
+  the Live stream's navigate input and seed called it directly from async code,
+  so a slow DNS answer for a model- or user-chosen URL froze every other request
+  and stream for the length of the lookup. The Playwright request guard, which
+  screens every redirect hop and subresource, did the same on the shared browser
+  loop and stalled Live frames and input for every browser session. These
+  callers now run the guard
+  through `asyncio.to_thread`; what it allows and rejects is unchanged. The
+  strict blocking-IO gate gains a `socket.getaddrinfo` rule, because
+  Blockbuster's defaults wrap socket methods but not the module-level resolver.
+  ([#6140])
 - **agents:** Loop-detection integer thresholds now reject YAML booleans instead
   of coercing `true` to `1`. A configuration such as `warn_threshold: true`
   and `hard_limit: true` previously made the first tool-call set meet the hard
@@ -7561,4 +7575,5 @@ with **180 merged pull requests** since the first 2.0 milestone tag.
 [#6132]: https://github.com/bytedance/deer-flow/pull/6132
 [#6134]: https://github.com/bytedance/deer-flow/pull/6134
 [#6135]: https://github.com/bytedance/deer-flow/pull/6135
+[#6140]: https://github.com/bytedance/deer-flow/pull/6140
 

CHANGELOG_zh.md

@@ -419,6 +419,15 @@
 
 ### 修复
 
+- **社区工具：** SSRF URL 校验在解析主机名时不再阻塞 Gateway 事件循环。
+  `validate_public_http_url` 通过阻塞的 `socket.getaddrinfo` 解析主机名，而
+  crawl4ai 与 Browserless 的 `web_fetch`、`web_capture`、`browser_navigate`、
+  Gateway 浏览器导航路由以及 Live 流的导航输入和 seed 都在异步代码中直接调用它，
+  因此模型或用户选择的 URL 一旦遇到缓慢的 DNS 响应，整个查询期间其他请求和流都会停滞。
+  逐个检查重定向和子资源的 Playwright 请求守卫也在共享的浏览器事件循环上同步解析，
+  会让所有浏览器会话的 Live 画面与输入一同停滞。这些调用方现在通过 `asyncio.to_thread` 运行校验，放行与拒绝的结果不变。
+  严格的阻塞 IO 检测新增 `socket.getaddrinfo` 规则，因为 Blockbuster
+  默认只包装 socket 方法，不包装模块级解析函数。([#6140])
 - **智能体：** 循环检测的整数阈值现在会拒绝 YAML 布尔值，而不是把
   `true` 静默转换为 `1`。此前若配置 `warn_threshold: true` 和
   `hard_limit: true`，第一组工具调用就会达到硬上限并强制终止智能体；
@@ -6329,3 +6338,4 @@ DeerFlow 2.0 是围绕"超级智能体"框架的彻底重写，核心包含子
 [#6132]: https://github.com/bytedance/deer-flow/pull/6132
 [#6134]: https://github.com/bytedance/deer-flow/pull/6134
 [#6135]: https://github.com/bytedance/deer-flow/pull/6135
+[#6140]: https://github.com/bytedance/deer-flow/pull/6140

backend/app/gateway/routers/browser.py

@@ -456,7 +456,7 @@ async def browser_stream(websocket: WebSocket, thread_id: ThreadId) -> None:
                     # SSRF-screen client-driven navigations with the same policy
                     # the agent tools enforce; reject rather than dispatch.
                     url = event.get("url")
-                    reason = validate_browser_url(url) if isinstance(url, str) else "Error: invalid navigation URL"
+                    reason = await asyncio.to_thread(validate_browser_url, url) if isinstance(url, str) else "Error: invalid navigation URL"
                     if reason is not None:
                         await _send_payload({"type": "nav_rejected", "url": url, "message": reason})
                         continue
@@ -480,7 +480,7 @@ async def browser_stream(websocket: WebSocket, thread_id: ThreadId) -> None:
         # page differs from the latest visible browser artifact, align Live with
         # what the user expects instead of requiring an off/on reconnect.
         seed = websocket.query_params.get("seed")
-        if seed and validate_browser_url(seed) is None:
+        if seed and await asyncio.to_thread(validate_browser_url, seed) is None:
             with contextlib.suppress(Exception):
                 current = await session.current_url()
                 if _should_apply_browser_seed(current, seed):

backend/packages/harness/deerflow/community/browser_automation/session.py

@@ -461,7 +461,8 @@ class BrowserSession:
             url = ""
             with contextlib.suppress(Exception):
                 url = route.request.url
-            if url.startswith(("http://", "https://")) and guard(url) is not None:
+            # The guard resolves hostnames; keep that off the shared browser loop.
+            if url.startswith(("http://", "https://")) and await asyncio.to_thread(guard, url) is not None:
                 logger.warning("browser request blocked by SSRF guard: %s", redact_browser_url(url))
                 with contextlib.suppress(Exception):
                     await route.abort("blockedbyclient")

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

@@ -242,7 +242,7 @@ async def navigate_and_capture(*, thread_id: str | None, url: str, outputs_path:
     Returns ``{"screenshot": virtual_path|None, "url": str, "title": str}``.
     Raises :class:`ValueError` when the URL fails SSRF validation.
     """
-    url_error = _validate_url("browser_navigate", url)
+    url_error = await asyncio.to_thread(_validate_url, "browser_navigate", url)
     if url_error:
         raise ValueError(url_error)
     cfg = _get_tool_config("browser_navigate")
@@ -290,7 +290,7 @@ async def browser_navigate_tool(runtime: Runtime, url: str, tool_call_id: Annota
         url: The http(s) URL to open.
     """
     try:
-        url_error = _validate_url("browser_navigate", url)
+        url_error = await asyncio.to_thread(_validate_url, "browser_navigate", url)
         if url_error:
             return _tool_message(url_error, tool_call_id)
         with _resolve_session(runtime, "browser_navigate") as session:

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

@@ -257,7 +257,8 @@ async def web_fetch_tool(url: str) -> str:
     try:
         cfg = _get_tool_config("web_fetch") or {}
         allow_private_addresses = _as_bool(cfg.get("allow_private_addresses"), False)
-        url_error = validate_public_http_url(
+        url_error = await asyncio.to_thread(
+            validate_public_http_url,
             url,
             allow_private_addresses=allow_private_addresses,
             resolver=_resolve_host_addresses,
@@ -328,7 +329,7 @@ async def web_capture_tool(
         cfg = _get_tool_config("web_capture") or {}
         allow_private_addresses = _as_bool(cfg.get("allow_private_addresses"), False)
 
-        url_error = _validate_capture_url(url, allow_private_addresses=allow_private_addresses)
+        url_error = await asyncio.to_thread(_validate_capture_url, url, allow_private_addresses=allow_private_addresses)
         if url_error:
             return _tool_message(url_error, tool_call_id)
 

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

@@ -1,3 +1,4 @@
+import asyncio
 import logging
 
 from langchain.tools import tool
@@ -100,7 +101,7 @@ async def web_fetch_tool(url: str) -> str:
     try:
         cfg = _get_tool_config("web_fetch")  # read config once; pass the values down
         allow_private_addresses = _coerce_bool(cfg.get("allow_private_addresses") if cfg is not None else None, False)
-        url_error = validate_public_http_url(url, allow_private_addresses=allow_private_addresses)
+        url_error = await asyncio.to_thread(validate_public_http_url, url, allow_private_addresses=allow_private_addresses)
         if url_error:
             return url_error
         filter_mode = _coerce_filter(cfg.get("filter") if cfg is not None else None)

backend/packages/harness/deerflow/community/url_safety.py

@@ -11,7 +11,11 @@ _BLOCKED_HOSTNAMES = {"localhost", "metadata.google.internal"}
 
 
 def resolve_host_addresses(hostname: str) -> list[ipaddress._BaseAddress]:
-    """Resolve a hostname to all IP addresses for SSRF screening."""
+    """Resolve a hostname to all IP addresses for SSRF screening.
+
+    Blocking: this is a synchronous DNS lookup, so async callers must run it
+    (or :func:`validate_public_http_url`) via ``asyncio.to_thread``.
+    """
     addresses: list[ipaddress._BaseAddress] = []
     try:
         infos = socket.getaddrinfo(hostname, None)
@@ -44,6 +48,14 @@ def validate_public_http_url(
     ``None`` when the caller may proceed.  The check is intentionally conservative
     for self-hosted fetch/render services because those services run inside the
     deployment network and can otherwise reach cloud metadata or private hosts.
+
+    A hostname URL is resolved synchronously; from a coroutine, call this via
+    ``asyncio.to_thread`` so a slow DNS answer cannot stall the event loop.
+
+    The check runs at validation time only. A caller that connects later
+    resolves the name again, so a rebinding DNS server can still hand that
+    connect a private address unless the connection is pinned to the vetted
+    IPs, as ``deerflow.mcp.personal_network`` does.
     """
     parsed = urlparse(url)
     if parsed.scheme not in {"http", "https"} or not parsed.netloc:

backend/packages/harness/deerflow/tools/AGENTS.md

@@ -59,6 +59,8 @@ Durable MCP task-management tools are added only while the process-local task su
 - `browser_automation/` - Agentic browser control (stateful `navigate → observe → click/type` loop) via Playwright, distinct from the read-only `web_fetch`/`web_capture` tools. Tools: `browser_navigate`, `browser_snapshot`, `browser_click`, `browser_type`, `browser_get_text`, `browser_back`, `browser_screenshot`, `browser_close` (config `group: browser`). A process-local `BrowserSessionManager` owns one private, loop-affine Playwright event-loop thread (same pattern as the BoxLite provider) so a per-thread browser session survives across turns regardless of the caller's loop (Gateway / TUI / test). Each action returns a fresh page snapshot whose interactive elements are addressed by a stable numeric `[ref]` index (stamped as `data-df-ref` during snapshot), so the model acts on what it just observed instead of holding stale handles or guessing selectors. URLs are SSRF-screened via the shared `validate_public_http_url` (opt-out `allow_private_addresses` only for intentional internal targets). CDP attachment cannot install the request guard on an existing Chrome context, so `cdp_url` fails closed unless the operator explicitly sets `allow_unguarded_cdp: true` for a trusted local browser. Browser REST/Live access also requires an exact non-NULL thread owner, rather than the general legacy shared-thread policy, because retained pages may contain authenticated state. Session admission is a hard `max_sessions` cap: pinned Live/operation sessions are never evicted, and a new thread is rejected when no unpinned session can be closed; one Live viewer owns a session at a time. Optional dependency: `cd backend && uv sync --extra browser && uv run playwright install chromium`; `scripts/detect_uv_extras.py` preserves the extra when `config.yaml` enables `browser_navigate`, and Gateway startup fails fast if configured browser control cannot import Playwright. Tests: `tests/test_browser_automation.py` (mocked tools + a real-Chromium integration test guarded by `importorskip`); `tests/manual_browser_live_check.py` is a manual DeepSeek-driven end-to-end check (not collected by pytest).
   Live UI input dispatch is kept independent from JPEG capture: non-move actions start a rate-limited background refresh loop, so pointer, wheel, or keyboard input stays responsive while continuous gestures still produce frames throughout the interaction.
 
+The shared SSRF guard (`community/url_safety.py`) resolves hostnames with blocking `socket.getaddrinfo`, so async tools, Gateway routes, and the Playwright request guard (`BrowserSession._install_request_guard`, which runs on the shared browser loop) call `validate_public_http_url` (or a wrapper such as `validate_browser_url`) through `asyncio.to_thread`. The guard checks at validation time only; fetches that re-resolve later are not pinned to the vetted IPs (only `mcp/personal_network.py` pins). The strict blocking-IO gate carries a `socket.getaddrinfo` rule for this; `tests/blocking_io/test_web_tool_url_validation.py` drives each web tool entry point and the request guard.
+
 Additional providers also live here (`boxlite`, `brave`, `browserless`, `crawl4ai`, `ddg_search`, `e2b_sandbox`, `exa`, `fastcrw`, `groundroute`, `infoquest`, `searxng`, `serper`, `serply`, `sofya`, `tencent_wsa`, `tenki`, `unbrowse`); see each subpackage for specifics. `tencent_wsa` uses Tencent Cloud Web Search's service API key endpoint (`TENCENTCLOUD_WSA_APIKEY`), not Tencent Cloud SecretId/SecretKey signing. Its `max_results` is capped at 50; requests above 10 use Tencent's optional `Cnt` parameter, which needs a Tencent Cloud plan that supports it. E2B bootstrap is required. If it fails, the provider kills and closes the unusable remote sandbox. New sandbox creation raises an error. Warm-pool reclaim and remote discovery discard the sandbox and continue acquisition. E2B mounts remain optional.
 
 E2B output sync records remote file versions and actual host file metadata in a thread-local manifest. The manifest binds to the remote sandbox ID. A complete output listing removes entries for deleted files. This avoids repeat downloads when the host filesystem rounds modification times. A single release-time sync pass is bounded by aggregate ceilings (`_MAX_SYNC_TOTAL_BYTES`, `_MAX_SYNC_FILES`, `_SYNC_DEADLINE_SECONDS`) on top of the per-file `_MAX_DOWNLOAD_SIZE` cap, so a pathological outputs tree cannot make release download unboundedly; a truncated pass logs what it dropped and leaves the manifest un-pruned (only entries observed in that pass are reconciled), so files it never reached are retried on the next release rather than being forgotten.

backend/tests/blocking_io/test_gate_smoke.py

@@ -33,6 +33,15 @@ async def test_gate_catches_unoffloaded_blocking_io_in_deerflow_module(tmp_path:
         ensure_sqlite_parent_dir(str(db_file))
 
 
+async def test_gate_catches_hostname_resolution_on_the_loop() -> None:
+    # Blockbuster's default rules miss module-level socket.getaddrinfo; the
+    # project rule must catch a synchronous SSRF-guard lookup on the loop.
+    from deerflow.community.url_safety import resolve_host_addresses
+
+    with pytest.raises(BlockingError):
+        resolve_host_addresses("127.1")
+
+
 async def test_gate_restores_blockbuster_patches_after_exceptions() -> None:
     original_stat = os.stat
 

backend/tests/blocking_io/test_web_tool_url_validation.py

@@ -0,0 +1,116 @@
+"""Web tools must resolve SSRF-screened hostnames off the agent event loop.
+
+``http://127.1/`` is not an ``ipaddress`` literal, so the URL guard hands it to
+the real ``socket.getaddrinfo``, which expands it to 127.0.0.1 without any DNS
+traffic. The strict gate's ``socket.getaddrinfo`` rule fails each test if the
+tool resolves on the loop instead of in a worker thread.
+"""
+
+from types import SimpleNamespace
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from deerflow.community.browser_automation import tools as browser_tools
+from deerflow.community.browserless import tools as browserless_tools
+from deerflow.community.crawl4ai import tools as crawl4ai_tools
+
+pytestmark = pytest.mark.asyncio
+
+_UNRESOLVED_LOOPBACK_URL = "http://127.1/"
+
+
+async def test_crawl4ai_web_fetch_resolves_off_loop() -> None:
+    with (
+        patch.object(crawl4ai_tools, "_get_tool_config", return_value={}),
+        patch.object(crawl4ai_tools, "_build_client") as build_client,
+    ):
+        result = await crawl4ai_tools.web_fetch_tool.ainvoke({"url": _UNRESOLVED_LOOPBACK_URL})
+
+    assert result == "Error: Refusing to fetch a private, loopback, or metadata address"
+    build_client.assert_not_called()
+
+
+async def test_browserless_web_fetch_resolves_off_loop() -> None:
+    with (
+        patch.object(browserless_tools, "_get_tool_config", return_value={}),
+        patch.object(browserless_tools, "_get_browserless_client") as get_client,
+    ):
+        result = await browserless_tools.web_fetch_tool.ainvoke({"url": _UNRESOLVED_LOOPBACK_URL})
+
+    assert result == "Error: Refusing to fetch a private, loopback, or metadata address"
+    get_client.assert_not_called()
+
+
+async def test_browserless_web_capture_resolves_off_loop() -> None:
+    with (
+        patch.object(browserless_tools, "_get_tool_config", return_value={}),
+        patch.object(browserless_tools, "_get_browserless_client") as get_client,
+    ):
+        result = await browserless_tools.web_capture_tool.coroutine(
+            runtime=SimpleNamespace(context={}, state={}),
+            url=_UNRESOLVED_LOOPBACK_URL,
+            tool_call_id="call-1",
+        )
+
+    assert result.update["messages"][0].content == "Error: Refusing to capture a private, loopback, or metadata address"
+    get_client.assert_not_called()
+
+
+async def test_browser_navigate_tool_resolves_off_loop() -> None:
+    manager = MagicMock()
+    with (
+        patch.object(browser_tools, "_get_tool_config", return_value={}),
+        patch.object(browser_tools, "get_browser_session_manager", return_value=manager),
+    ):
+        result = await browser_tools.browser_navigate_tool.coroutine(
+            runtime=SimpleNamespace(context={"thread_id": "thread-1"}, state={}),
+            url=_UNRESOLVED_LOOPBACK_URL,
+            tool_call_id="call-1",
+        )
+
+    assert result.update["messages"][0].content == "Error: Refusing to browse a private, loopback, or metadata address"
+    manager.acquire_session.assert_not_called()
+
+
+async def test_browser_navigate_and_capture_resolves_off_loop(tmp_path) -> None:
+    manager = MagicMock()
+    with (
+        patch.object(browser_tools, "_get_tool_config", return_value={}),
+        patch.object(browser_tools, "get_browser_session_manager", return_value=manager),
+        pytest.raises(ValueError, match="Refusing to browse a private, loopback, or metadata address"),
+    ):
+        await browser_tools.navigate_and_capture(thread_id="thread-1", url=_UNRESOLVED_LOOPBACK_URL, outputs_path=tmp_path)
+
+    manager.acquire_session.assert_not_called()
+
+
+async def test_browser_request_guard_resolves_off_loop() -> None:
+    # The context route guard screens every redirect hop and subresource on the
+    # shared Playwright loop, which also pumps Live frames for every session.
+    from deerflow.community.browser_automation.session import BrowserSession
+
+    captured = {}
+
+    class _FakeContext:
+        async def route(self, _pattern, handler):
+            captured["handler"] = handler
+
+    class _FakeRoute:
+        request = SimpleNamespace(url=_UNRESOLVED_LOOPBACK_URL)
+        aborted_with = None
+
+        async def abort(self, error_code):
+            self.aborted_with = error_code
+
+        async def continue_(self):
+            raise AssertionError("a loopback request must not continue")
+
+    session = BrowserSession(MagicMock(), headless=True, timeout_ms=1000, viewport={"width": 1000, "height": 500}, url_guard=browser_tools.validate_browser_url)
+    session._context = _FakeContext()
+    route = _FakeRoute()
+    with patch.object(browser_tools, "_get_tool_config", return_value={}):
+        await session._install_request_guard()
+        await captured["handler"](route)
+
+    assert route.aborted_with == "blockedbyclient"

backend/tests/support/detectors/blocking_io_runtime.py

@@ -21,7 +21,11 @@ _SCANNED_MODULES: tuple[str, ...] = ("app", "deerflow")
 # Add DeerFlow-local rules here only when Blockbuster's default rule set misses
 # a generic blocking primitive used by production code. If a path is invisible
 # because no test exercises it, add a production-path runtime anchor instead.
-_PROJECT_BLOCKING_RULES: tuple[tuple[str, BlockBusterFunction], ...] = ()
+_PROJECT_BLOCKING_RULES: tuple[tuple[str, BlockBusterFunction], ...] = (
+    # Blockbuster wraps socket.socket methods but not the module-level resolver,
+    # and the SSRF guard in deerflow.community.url_safety resolves hostnames.
+    ("socket.getaddrinfo", BlockBusterFunction(None, "socket.getaddrinfo", scanned_modules=list(_SCANNED_MODULES))),
+)
 
 
 def _install_project_rules(bb: BlockBuster) -> None:

backend/tests/test_browser_router.py

@@ -1,3 +1,4 @@
+import asyncio
 import ipaddress
 import json
 import logging
@@ -343,3 +344,48 @@ def test_validate_browser_url_rejects_private_and_non_http(monkeypatch):
     # A normal public URL passes (returns None = allowed).
     monkeypatch.setattr(browser_tools, "_resolve_host_addresses", lambda _host: [ipaddress.ip_address("93.184.215.14")])
     assert validate_browser_url("https://github.com/bytedance/deer-flow") is None
+
+
+def test_browser_stream_resolves_seed_and_navigate_urls_off_the_loop():
+    store = MagicMock()
+    store.check_access = AsyncMock(return_value=True)
+    store.get = AsyncMock(return_value={"thread_id": "thread-1", "user_id": "browser-user"})
+    app = _browser_ws_app(store)
+    session = MagicMock()
+    for method in ("current_url", "navigate", "start_screencast", "stop_screencast", "tabs", "dispatch_input"):
+        setattr(session, method, AsyncMock())
+    session.current_url.return_value = "about:blank"
+    session.tabs.return_value = []
+    manager = MagicMock()
+    manager.acquire_session.return_value.__enter__.return_value = session
+    resolved_on_loop: dict[str, bool] = {}
+
+    def resolve(host: str) -> list[ipaddress._BaseAddress]:
+        try:
+            asyncio.get_running_loop()
+        except RuntimeError:
+            resolved_on_loop[host] = False
+        else:
+            resolved_on_loop[host] = True
+        return [ipaddress.ip_address("127.0.0.1" if host == "internal.example" else "93.184.215.14")]
+
+    from deerflow.community.browser_automation import tools as browser_tools
+
+    with (
+        patch.object(browser_router, "_authenticate_ws", AsyncMock(return_value=_user())),
+        patch.object(browser_router, "_browser_tools_enabled", return_value=True),
+        patch("deerflow.config.get_app_config", return_value=SimpleNamespace(get_tool_config=lambda _name: None)),
+        patch("deerflow.community.browser_automation.get_browser_session_manager", return_value=manager),
+        patch.object(browser_tools, "_get_tool_config", return_value={}),
+        patch.object(browser_tools, "_resolve_host_addresses", side_effect=resolve),
+        TestClient(app) as client,
+        client.websocket_connect("/api/threads/thread-1/browser/stream?seed=https://seed.example/") as ws,
+    ):
+        ws.send_text(json.dumps({"type": "navigate", "url": "https://internal.example/"}))
+        while (payload := ws.receive_json())["type"] != "nav_rejected":
+            pass
+
+    assert payload["url"] == "https://internal.example/"
+    session.navigate.assert_awaited_once_with("https://seed.example/")
+    session.dispatch_input.assert_not_awaited()
+    assert resolved_on_loop == {"seed.example": False, "internal.example": False}