diff --git a/README.md b/README.md index 0793a01056..7c2a221c0d 100644 --- a/README.md +++ b/README.md @@ -254,7 +254,7 @@ fences, including unclosed fences, in complete and streamed replies. Brave web search preserves valid entries in mixed result lists. Malformed response containers or lists containing no usable entries return a structured format error; missing, null, or empty results keep the existing "No results found" response. Format errors also log the malformed container's path and type, or the absence of usable result objects, without including search queries, credentials, or payload values. - Several providers, including You.com, GroundRoute, Webz and InfoQuest, initialize HTTPX clients in a worker thread so synchronous TLS trust loading does not block the agent event loop; requests and session cleanup remain asynchronous. + Several providers, including You.com, GroundRoute, Webz, InfoQuest and Jsonify, initialize HTTPX clients in a worker thread so synchronous TLS trust loading does not block the agent event loop; requests and session cleanup remain asynchronous. GroundRoute web search and fetch contain malformed HTTP-success payloads the same way: a non-object payload, a `results` container that is not a list, or a non-empty list with no object entries returns the provider's format error, while non-object entries in an otherwise valid list are skipped in order. Missing, null, or empty results keep the existing "No results found" response. diff --git a/backend/AGENTS.md b/backend/AGENTS.md index 8a655f174b..3a3312ecfa 100644 --- a/backend/AGENTS.md +++ b/backend/AGENTS.md @@ -137,12 +137,12 @@ 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 and InfoQuest, use +You.com, GroundRoute, Webz, Crawl4AI, Browserless, InfoQuest and Jsonify, 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 and Jina still construct their -clients on the loop and should follow the same rule when touched. +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. ### Harness / App Split diff --git a/backend/packages/harness/deerflow/community/jsonify/tools.py b/backend/packages/harness/deerflow/community/jsonify/tools.py index 47c99cfb30..591a6947ce 100644 --- a/backend/packages/harness/deerflow/community/jsonify/tools.py +++ b/backend/packages/harness/deerflow/community/jsonify/tools.py @@ -31,6 +31,7 @@ from pydantic import AnyHttpUrl, BaseModel, ConfigDict, Field, ValidationError, from deerflow.community.provider_credentials import redact_secret, reject_url_userinfo from deerflow.community.url_safety import validate_public_http_url from deerflow.config import get_app_config +from deerflow.utils.file_io import await_drained logger = logging.getLogger(__name__) @@ -101,6 +102,27 @@ def _build_params( return params +async def _open_client(timeout: float) -> httpx.AsyncClient: + # HTTPX loads TLS trust synchronously. Own the construction task so a + # cancellation can still close the client that the worker returns. + construction = asyncio.create_task(asyncio.to_thread(httpx.AsyncClient, timeout=timeout)) + try: + return await asyncio.shield(construction) + except asyncio.CancelledError: + + async def close_after_construction() -> None: + try: + client = await construction + await client.aclose() + except Exception as exc: + logger.error("Jsonify HTTP client cleanup after cancellation failed (%s)", type(exc).__name__) + + # Report failures inside the owned task before a repeated cancellation + # makes await_drained discard its outcome. + await await_drained(close_after_construction()) + raise + + async def _read_rows(settings: _JsonifySettings, args: _RowsArgs) -> dict[str, Any]: serve_url = str(settings.datasets[args.dataset]) url_error = await asyncio.to_thread(validate_public_http_url, serve_url, action="read") @@ -113,7 +135,7 @@ async def _read_rows(settings: _JsonifySettings, args: _RowsArgs) -> dict[str, A offset = 0 version: int | None = None truncated = False - async with httpx.AsyncClient(timeout=settings.timeout) as client: + async with await _open_client(settings.timeout) as client: while True: limit = min(_PAGE_SIZE, max_rows - len(rows)) response = await client.get( diff --git a/backend/tests/blocking_io/test_jsonify_client_ssl.py b/backend/tests/blocking_io/test_jsonify_client_ssl.py new file mode 100644 index 0000000000..8d7904896b --- /dev/null +++ b/backend/tests/blocking_io/test_jsonify_client_ssl.py @@ -0,0 +1,167 @@ +"""Exercise Jsonify with real HTTPX initialization and no provider requests.""" + +import asyncio +import logging +import ssl +import threading +from types import SimpleNamespace + +import httpx +import pytest + +from deerflow.community.jsonify import tools + + +@pytest.fixture +def read_rows(monkeypatch): + monkeypatch.setattr(tools, "validate_public_http_url", lambda *args, **kwargs: None) + settings = tools._JsonifySettings(datasets={"demo": "https://jsonify.invalid/serve"}) + args = tools._RowsArgs(dataset="demo") + + async def read(): + return await tools._read_rows(settings, args) + + return read + + +@pytest.mark.asyncio +async def test_jsonify_loads_tls_trust_off_loop(read_rows, monkeypatch): + loop_thread = threading.get_ident() + create_default_context = ssl.create_default_context + tls_threads = [] + clients = [] + + def create_context(*args, **kwargs): + tls_threads.append(threading.get_ident()) + return create_default_context(*args, **kwargs) + + async def send(self, request, **kwargs): + assert threading.get_ident() == loop_thread + assert request.url.host == "jsonify.invalid" + clients.append(self) + return httpx.Response(200, request=request, json={"rows": [{"name": "demo"}], "has_more": False}) + + monkeypatch.setattr(ssl, "create_default_context", create_context) + monkeypatch.setattr(httpx.AsyncClient, "send", send) + + result = await read_rows() + + assert result["rows"] == [{"name": "demo"}] + assert tls_threads and loop_thread not in tls_threads + assert len(clients) == 1 + assert clients[0].is_closed + + +@pytest.mark.asyncio +async def test_jsonify_closes_client_if_cancelled_during_setup(read_rows, monkeypatch): + loop = asyncio.get_running_loop() + loop_thread = threading.get_ident() + started = asyncio.Event() + finished = asyncio.Event() + release = threading.Event() + real_client = httpx.AsyncClient + clients = [] + + def construct(**kwargs): + assert threading.get_ident() != loop_thread, "the constructor must not run on the event loop" + loop.call_soon_threadsafe(started.set) + try: + assert release.wait(10), "the constructor was never released" + client = real_client(**kwargs) + clients.append(client) + return client + finally: + loop.call_soon_threadsafe(finished.set) + + async def unexpected_send(self, request, **kwargs): + pytest.fail("a cancelled construction must not send a request") + + monkeypatch.setattr(httpx, "AsyncClient", construct) + monkeypatch.setattr(real_client, "send", unexpected_send) + pending = asyncio.create_task(read_rows()) + start_waiter = asyncio.create_task(started.wait()) + try: + done, _ = await asyncio.wait({pending, start_waiter}, timeout=10, return_when=asyncio.FIRST_COMPLETED) + if pending in done: + await pending + assert start_waiter in done, "the constructor never started" + + pending.cancel() + await asyncio.sleep(0) # Deliver cancellation while construction is still gated. + release.set() + + with pytest.raises(asyncio.CancelledError): + await pending + await asyncio.wait_for(finished.wait(), timeout=10) + assert len(clients) == 1 + assert clients[0].is_closed + finally: + release.set() + pending.cancel() + start_waiter.cancel() + await asyncio.gather(pending, start_waiter, return_exceptions=True) + if started.is_set(): + await asyncio.wait_for(finished.wait(), timeout=10) + for client in clients: + if not client.is_closed: + await client.aclose() + + +@pytest.mark.asyncio +@pytest.mark.parametrize("failure_phase", ["construction", "close"]) +async def test_jsonify_logs_failures_during_repeated_cancellation(read_rows, monkeypatch, caplog, failure_phase): + loop = asyncio.get_running_loop() + loop_thread = threading.get_ident() + started, finished = asyncio.Event(), asyncio.Event() + release = threading.Event() + close_threads = [] + secret = "synthetic-sensitive-value" + caplog.set_level(logging.ERROR) + + async def close(): + close_threads.append(threading.get_ident()) + raise RuntimeError(secret) + + def construct(**kwargs): + assert threading.get_ident() != loop_thread + loop.call_soon_threadsafe(started.set) + try: + assert release.wait(10), "the constructor was never released" + if failure_phase == "construction": + raise RuntimeError(secret) + return SimpleNamespace(aclose=close) + finally: + loop.call_soon_threadsafe(finished.set) + + monkeypatch.setattr(httpx, "AsyncClient", construct) + pending = asyncio.create_task(read_rows()) + start_waiter = asyncio.create_task(started.wait()) + try: + done, _ = await asyncio.wait({pending, start_waiter}, timeout=10, return_when=asyncio.FIRST_COMPLETED) + if pending in done: + await pending + assert start_waiter in done, "the constructor never started" + + pending.cancel() + await asyncio.sleep(0) # Enter cleanup while construction is still gated. + pending.cancel() + await asyncio.sleep(0) # Cancel the drain before its worker fails. + release.set() + + with pytest.raises(asyncio.CancelledError): + await pending + await asyncio.wait_for(finished.wait(), timeout=10) + + assert close_threads == ([loop_thread] if failure_phase == "close" else []) + errors = [record for record in caplog.records if record.levelno == logging.ERROR] + assert len(errors) == 1 + assert errors[0].levelno == logging.ERROR + assert "RuntimeError" in errors[0].getMessage() + assert secret not in errors[0].getMessage() + finally: + release.set() + pending.cancel() + start_waiter.cancel() + await asyncio.gather(pending, start_waiter, return_exceptions=True) + if started.is_set(): + await asyncio.wait_for(finished.wait(), timeout=10)