fix(browserless): construct the fetch and screenshot clients off the event loop (#7913)

* fix(browserless): construct the fetch and screenshot clients off the event loop

* fix(browserless): close the client a cancelled construction still builds

Keeping the constructor off the loop is right, but awaiting
`asyncio.to_thread(httpx.AsyncClient, ...)` directly is not cancellation-safe:
the await can be cancelled while the worker thread is still inside the
constructor. The await is abandoned, the thread runs on and hands back a live
client, and execution never enters the `async with` that would have closed it,
so `is_closed` stays False and the connection pool leaks on that cancellation.

Own the construction as a task instead. When the caller is cancelled, wait for
the task with the shield loop this package already uses for cleanup, close
whatever client it produced, and only then let the cancellation propagate. A
failure while closing is logged rather than raised, so it cannot mask the
cancellation the caller asked for.

The new regression test gates the constructor on a thread event, cancels the
operation while it is parked there, and releases the worker afterwards. Against
the previous commit it fails for both fetch_html_with_status and
capture_screenshot with "the client built after the cancellation was leaked".

---------

Co-authored-by: inchang-ing <197932532+inchang-ing@users.noreply.github.com>

backend/AGENTS.md

@@ -136,9 +136,11 @@ Read the nearest scoped `AGENTS.md` for subsystem ownership.
 
 Read-only knowledge: [contract](docs/KNOWLEDGE.md) covers independent retrieval
 tools, scope intersection and extension-owned providers.
-HTTPX clients for RAGFlow, LightRAG, You.com, GroundRoute, Webz and Crawl4AI use
-`asyncio.to_thread` for synchronous TLS setup; requests and cleanup stay on-loop.
-The list is non-exhaustive: jsonify, SearXNG, Browserless, Jina and InfoQuest
+HTTPX clients for RAGFlow, LightRAG, You.com, GroundRoute, Webz, Crawl4AI and
+Browserless use `asyncio.to_thread` for synchronous TLS setup; requests and
+cleanup stay on-loop, and the construction is owned as a task so cancelling
+during setup still closes the client it builds.
+The list is non-exhaustive: jsonify, SearXNG, Jina and InfoQuest
 still construct their clients on the loop and should follow the same rule when
 touched.
 

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

@@ -1,3 +1,4 @@
+import asyncio
 import logging
 from dataclasses import dataclass
 from typing import Any
@@ -30,6 +31,46 @@ def _get_header(headers: Any, name: str) -> str:
     return str(headers.get(name.lower(), ""))
 
 
+async def _settle(future: "asyncio.Future[Any]") -> Any:
+    """Await a future that owns a resource, even if this task is cancelled again.
+
+    Cleanup must not be the thing that gets abandoned: the caller remembers its
+    own cancellation and re-raises it once the resource is released.
+    """
+    while True:
+        try:
+            return await asyncio.shield(future)
+        except asyncio.CancelledError:
+            if future.cancelled():
+                raise
+
+
+async def _open_client(**kwargs: Any) -> httpx.AsyncClient:
+    """Build an ``httpx.AsyncClient`` without loading TLS trust on the event loop.
+
+    The constructor resolves the system trust store synchronously, so it has to
+    run in a thread. That alone is not enough: a cancellation can arrive while
+    the worker is still inside the constructor, and then the await is abandoned
+    while the thread goes on to hand back a live client that nobody closes. So
+    own the construction as a task and, once cancelled, wait for it and close
+    whatever it produced before the cancellation propagates.
+    """
+    construction: asyncio.Task[httpx.AsyncClient] = asyncio.create_task(asyncio.to_thread(httpx.AsyncClient, **kwargs))
+    try:
+        return await asyncio.shield(construction)
+    except asyncio.CancelledError:
+        try:
+            late_client = await _settle(construction)
+        except Exception as error:
+            logger.error(f"Browserless HTTPX client construction failed after cancellation: {error!s}")
+        else:
+            try:
+                await _settle(asyncio.create_task(late_client.aclose()))
+            except Exception as error:
+                logger.error(f"Closing a Browserless HTTPX client built after cancellation failed: {error!s}")
+        raise
+
+
 class BrowserlessClient:
     """Client for Browserless headless Chrome API."""
 
@@ -142,7 +183,9 @@ class BrowserlessClient:
 
         logger.debug(f"Fetching URL via Browserless: {url}")
         try:
-            async with httpx.AsyncClient(timeout=self.timeout_s) as client:
+            # Construction loads TLS trust synchronously, off the loop, and owns the
+            # client even when this coroutine is cancelled mid-construction.
+            async with await _open_client(timeout=self.timeout_s) as client:
                 resp = await client.post(
                     f"{self.base_url}/content",
                     json=payload,
@@ -232,7 +275,9 @@ class BrowserlessClient:
 
         logger.debug(f"Capturing URL screenshot via Browserless: {url}")
         try:
-            async with httpx.AsyncClient(timeout=self.timeout_s) as client:
+            # Construction loads TLS trust synchronously, off the loop, and owns the
+            # client even when this coroutine is cancelled mid-construction.
+            async with await _open_client(timeout=self.timeout_s) as client:
                 resp = await client.post(
                     f"{self.base_url}/screenshot",
                     json=payload,

backend/tests/blocking_io/test_browserless_client_ssl.py

@@ -0,0 +1,119 @@
+"""Browserless must load HTTPX TLS trust off the agent event loop."""
+
+import asyncio
+import ssl
+import threading
+
+import httpx
+import pytest
+
+from deerflow.community.browserless.browserless_client import BrowserlessClient
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("operation", ["fetch_html_with_status", "capture_screenshot"])
+async def test_browserless_client_tls_setup_runs_off_loop(operation, monkeypatch):
+    loop_thread = threading.get_ident()
+    original_context = ssl.create_default_context
+    contexts = []
+    clients = []
+
+    def create_context(*args, **kwargs):
+        assert threading.get_ident() != loop_thread, "TLS trust loading blocks the event loop"
+        context = original_context(*args, **kwargs)
+        contexts.append(context)
+        return context
+
+    async def send(self, request, **kwargs):
+        assert threading.get_ident() == loop_thread
+        clients.append(self)
+        assert request.url.host == "browserless.example"
+        if operation == "capture_screenshot":
+            assert request.url.path == "/screenshot"
+            return httpx.Response(
+                200,
+                request=request,
+                content=b"\x89PNG synthetic",
+                headers={"Content-Type": "image/png", "X-Response-Code": "200"},
+            )
+        assert request.url.path == "/content"
+        return httpx.Response(
+            200,
+            request=request,
+            text="<html>synthetic page</html>",
+            headers={"X-Response-Code": "200", "X-Response-Status": "OK"},
+        )
+
+    monkeypatch.setattr(ssl, "create_default_context", create_context)
+    monkeypatch.setattr(httpx.AsyncClient, "send", send)
+
+    client = BrowserlessClient("https://browserless.example", token="token")
+
+    if operation == "capture_screenshot":
+        result = await client.capture_screenshot("https://example.com")
+        assert result.content == b"\x89PNG synthetic"
+    else:
+        result = await client.fetch_html_with_status("https://example.com")
+        assert result.html == "<html>synthetic page</html>"
+
+    assert contexts
+    assert len(clients) == 1
+    assert clients[0].is_closed
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("operation", ["fetch_html_with_status", "capture_screenshot"])
+async def test_browserless_client_closes_client_built_after_cancellation(operation, monkeypatch):
+    """Cancelling during construction must not leak the client it eventually builds.
+
+    ``httpx.AsyncClient`` cannot be interrupted inside its constructor, so the
+    worker keeps running after the await is cancelled and then hands back a live
+    client. Nothing can enter the ``async with`` for it, so whoever owns the
+    construction task has to close it, and the operation must still end cancelled.
+    """
+    real_client = httpx.AsyncClient
+    started = threading.Event()
+    release = threading.Event()
+    finished = threading.Event()
+    built = []
+    sent = []
+
+    def gated_client(*args, **kwargs):
+        started.set()
+        assert release.wait(10), "the constructor was never released"
+        client = real_client(*args, **kwargs)
+        built.append(client)
+        finished.set()
+        return client
+
+    async def send(self, request, **kwargs):
+        sent.append(request.url)
+        return httpx.Response(200, request=request, text="unreachable")
+
+    monkeypatch.setattr(httpx, "AsyncClient", gated_client)
+    monkeypatch.setattr(real_client, "send", send)
+
+    client = BrowserlessClient("https://browserless.example", token="token")
+    if operation == "capture_screenshot":
+        pending = asyncio.ensure_future(client.capture_screenshot("https://example.com"))
+    else:
+        pending = asyncio.ensure_future(client.fetch_html_with_status("https://example.com"))
+
+    assert await asyncio.to_thread(started.wait, 10), "the constructor never started"
+    # The operation is suspended on the constructor now, so the cancellation lands
+    # there and never reaches the request: release the worker only afterwards.
+    pending.cancel()
+    await asyncio.sleep(0)
+    await asyncio.sleep(0)
+    release.set()
+
+    with pytest.raises(asyncio.CancelledError):
+        await pending
+
+    # The worker thread cannot be stopped, so wait for it before judging what it
+    # built; a leaked client is only visible once the constructor has returned.
+    assert await asyncio.to_thread(finished.wait, 10), "the constructor never finished"
+
+    assert sent == [], "a cancelled construction must not send a request"
+    assert len(built) == 1
+    assert built[0].is_closed, "the client built after the cancellation was leaked"