diff --git a/backend/AGENTS.md b/backend/AGENTS.md index 72b8fb8d29..edb2e81977 100644 --- a/backend/AGENTS.md +++ b/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 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 039dad7434..09d498cdd3 100644 --- a/backend/packages/harness/deerflow/community/jina_ai/jina_client.py +++ b/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 diff --git a/backend/packages/harness/deerflow/community/searxng/searxng_client.py b/backend/packages/harness/deerflow/community/searxng/searxng_client.py index a23bf10e30..0652de6e3a 100644 --- a/backend/packages/harness/deerflow/community/searxng/searxng_client.py +++ b/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, diff --git a/backend/tests/blocking_io/test_search_client_ssl.py b/backend/tests/blocking_io/test_search_client_ssl.py index e603047b31..6c8a2ebf11 100644 --- a/backend/tests/blocking_io/test_search_client_ssl.py +++ b/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 diff --git a/backend/tests/test_jina_request_admission.py b/backend/tests/test_jina_request_admission.py index 2873e6793d..9ae272b9ce 100644 --- a/backend/tests/test_jina_request_admission.py +++ b/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()