fix(community): build remaining search clients off the event loop (#7933)

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

backend/AGENTS.md

@@ -137,12 +137,10 @@ 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 several community providers, including RAGFlow, LightRAG,
-You.com, GroundRoute, Webz, Crawl4AI, Browserless, InfoQuest and Jsonify, use
-`asyncio.to_thread` for synchronous TLS setup; requests and cleanup stay on-loop,
-and Browserless and Jsonify own construction as a task so cancellation during
-setup still closes the client it builds.
-The list is non-exhaustive: SearXNG and Jina still construct their clients on
-the loop and should follow the same rule when touched.
+You.com, GroundRoute, Webz, Crawl4AI, Browserless, InfoQuest, Jsonify, SearXNG
+and Jina, use `asyncio.to_thread` for synchronous TLS setup; requests and cleanup
+stay on-loop, and Browserless and Jsonify own construction as a task so
+cancellation during setup still closes the client it builds.
 
 ### Harness / App Split
 

backend/packages/harness/deerflow/community/jina_ai/jina_client.py

@@ -128,7 +128,7 @@ async def _protect_response_cleanup(response: httpx.Response):
 async def _admission_client(options):
     # Keep one client across retries, matching the default transport behavior.
     # Cleanup is still shielded so cancellation always propagates after close.
-    client = httpx.AsyncClient(**options)
+    client = await asyncio.to_thread(httpx.AsyncClient, **options)
     try:
         client.event_hooks["response"].append(_protect_response_cleanup)
         yield client
@@ -136,6 +136,13 @@ async def _admission_client(options):
         await _finish_cleanup(client.aclose())
 
 
+@asynccontextmanager
+async def _plain_client(options):
+    """Build the default client off the loop; keep requests and cleanup on it."""
+    async with await asyncio.to_thread(httpx.AsyncClient, **options) as client:
+        yield client
+
+
 class JinaClient:
     async def crawl(
         self,
@@ -192,7 +199,7 @@ class JinaClient:
                     client_kwargs: dict[str, object] = {"trust_env": trust_env, "follow_redirects": True}
                     if proxy:
                         client_kwargs["proxy"] = proxy
-                    async with _admission_client(client_kwargs) if admission else httpx.AsyncClient(**client_kwargs) as client:
+                    async with _admission_client(client_kwargs) if admission else _plain_client(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

backend/packages/harness/deerflow/community/searxng/searxng_client.py

@@ -1,3 +1,4 @@
+import asyncio
 import logging
 from typing import Any
 
@@ -141,7 +142,7 @@ class SearxngClient:
 
         logger.debug(f"Searching SearXNG at {self.base_url} with query: {query} (page {pageno})")
         try:
-            async with httpx.AsyncClient(timeout=30) as client:
+            async with await asyncio.to_thread(httpx.AsyncClient, timeout=30) as client:
                 resp = await client.get(
                     f"{self.base_url}/search",
                     params=params,

backend/tests/blocking_io/test_search_client_ssl.py

@@ -110,3 +110,46 @@ async def test_infoquest_client_tls_setup_runs_off_loop(operation, monkeypatch):
     assert contexts
     assert len(clients) == 1
     assert clients[0].is_closed
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("provider", ["searxng", "jina"])
+async def test_remaining_search_clients_build_tls_off_loop(provider, 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)
+        if provider == "searxng":
+            assert request.url.host == "searx.example"
+            payload = {"results": [{"url": "https://example.com", "title": "Synthetic", "content": "Synthetic text"}]}
+            return httpx.Response(200, request=request, json=payload)
+        assert request.url.host == "r.jina.ai"
+        return httpx.Response(200, request=request, text="Synthetic text")
+
+    monkeypatch.setattr(ssl, "create_default_context", create_context)
+    monkeypatch.setattr(httpx.AsyncClient, "send", send)
+
+    if provider == "searxng":
+        from deerflow.community.searxng.searxng_client import SearxngClient
+
+        result = await SearxngClient("https://searx.example").search("synthetic query", max_results=1)
+        assert result and result[0]["url"] == "https://example.com"
+    else:
+        from deerflow.community.jina_ai.jina_client import JinaClient
+
+        result = await JinaClient().crawl("https://example.com", max_retries=0)
+        assert result == "Synthetic text"
+
+    assert contexts
+    assert len(clients) == 1
+    assert clients[0].is_closed

backend/tests/test_jina_request_admission.py

@@ -2,6 +2,7 @@
 
 import asyncio
 import json
+import ssl
 import threading
 from concurrent.futures import ThreadPoolExecutor
 from types import SimpleNamespace
@@ -531,6 +532,37 @@ async def test_enabled_retries_reuse_one_client(monkeypatch):
     assert len(clients) == 1
 
 
+async def test_admitted_client_builds_tls_off_loop(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
+
+    class RecordingClient(httpx.AsyncClient):
+        def __init__(self, **kwargs):
+            super().__init__(**kwargs)
+            clients.append(self)
+
+    async def send(self, request, **kwargs):
+        assert threading.get_ident() == loop_thread
+        return httpx.Response(200, text="ok", request=request)
+
+    monkeypatch.setattr(ssl, "create_default_context", create_context)
+    monkeypatch.setattr(httpx, "AsyncClient", RecordingClient)
+    monkeypatch.setattr(RecordingClient, "send", send)
+
+    assert await crawl(max_retries=0) == "ok"
+    assert contexts
+    assert len(clients) == 1
+    assert clients[0].is_closed
+
+
 async def test_client_pool_close_does_not_hold_attempt_permit(monkeypatch):
     original_client = httpx.AsyncClient
     closing, release = asyncio.Event(), asyncio.Event()