diff --git a/backend/AGENTS.md b/backend/AGENTS.md index d0d1bd7700..236f67134a 100644 --- a/backend/AGENTS.md +++ b/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. diff --git a/backend/packages/harness/deerflow/community/browserless/browserless_client.py b/backend/packages/harness/deerflow/community/browserless/browserless_client.py index 75c339f250..7640a85a4f 100644 --- a/backend/packages/harness/deerflow/community/browserless/browserless_client.py +++ b/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, diff --git a/backend/tests/blocking_io/test_browserless_client_ssl.py b/backend/tests/blocking_io/test_browserless_client_ssl.py new file mode 100644 index 0000000000..edb70c7059 --- /dev/null +++ b/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="synthetic page", + 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 == "synthetic page" + + 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"