From 60400fb753b347a5e88d92c34c095f8bfb560966 Mon Sep 17 00:00:00 2001 From: embwl0x Date: Tue, 11 Aug 2026 03:52:20 -0500 Subject: [PATCH 1/2] fix(web): serialize provider client initialization --- plugins/web/exa/provider.py | 45 +++--- plugins/web/parallel/provider.py | 61 ++++---- .../web/test_provider_client_concurrency.py | 133 ++++++++++++++++++ 3 files changed, 196 insertions(+), 43 deletions(-) create mode 100644 tests/plugins/web/test_provider_client_concurrency.py diff --git a/plugins/web/exa/provider.py b/plugins/web/exa/provider.py index 17ce665dc18a5..4c7603fb56314 100644 --- a/plugins/web/exa/provider.py +++ b/plugins/web/exa/provider.py @@ -26,6 +26,7 @@ import logging import os +import threading from typing import Any, Dict, List from agent.web_search_provider import WebSearchProvider @@ -36,6 +37,7 @@ # :mod:`tools.web_tools` so tests that do ``tools.web_tools._exa_client = # None`` between cases see fresh state. The plugin reads/writes through # that public module (see :func:`_get_exa_client`). +_client_lock = threading.Lock() def _get_exa_client() -> Any: @@ -51,30 +53,35 @@ def _get_exa_client() -> Any: if cached is not None: return cached - from agent.web_search_provider import get_provider_env + with _client_lock: + cached = getattr(_wt, "_exa_client", None) + if cached is not None: + return cached - api_key = get_provider_env("EXA_API_KEY") - if not api_key: - raise ValueError( - "EXA_API_KEY environment variable not set. " - "Get your API key at https://exa.ai" - ) + from agent.web_search_provider import get_provider_env + + api_key = get_provider_env("EXA_API_KEY") + if not api_key: + raise ValueError( + "EXA_API_KEY environment variable not set. " + "Get your API key at https://exa.ai" + ) - try: - from tools.lazy_deps import ensure as _lazy_ensure + try: + from tools.lazy_deps import ensure as _lazy_ensure - _lazy_ensure("search.exa", prompt=False) - except ImportError: - pass - except Exception as exc: # noqa: BLE001 — lazy_deps surfaces install hints - raise ImportError(str(exc)) + _lazy_ensure("search.exa", prompt=False) + except ImportError: + pass + except Exception as exc: # noqa: BLE001 — lazy_deps surfaces install hints + raise ImportError(str(exc)) - from exa_py import Exa # noqa: WPS433 — deliberately lazy + from exa_py import Exa # noqa: WPS433 — deliberately lazy - client = Exa(api_key=api_key) - client.headers["x-exa-integration"] = "hermes-agent" - _wt._exa_client = client - return client + client = Exa(api_key=api_key) + client.headers["x-exa-integration"] = "hermes-agent" + _wt._exa_client = client + return client def _reset_client_for_tests() -> None: diff --git a/plugins/web/parallel/provider.py b/plugins/web/parallel/provider.py index 028f5df3fc37b..2d772348cbcb7 100644 --- a/plugins/web/parallel/provider.py +++ b/plugins/web/parallel/provider.py @@ -30,6 +30,7 @@ import logging import os +import threading from typing import Any, Dict, List from agent.web_search_provider import WebSearchProvider @@ -41,6 +42,8 @@ # ``tools.web_tools._parallel_client = None`` between cases see fresh state. # The plugin reads/writes through that public module (see # :func:`_get_sync_client` / :func:`_get_async_client`). +_sync_client_lock = threading.Lock() +_async_client_lock = threading.Lock() def _ensure_parallel_sdk_installed() -> None: @@ -73,21 +76,26 @@ def _get_sync_client() -> Any: if cached is not None: return cached - from agent.web_search_provider import get_provider_env + with _sync_client_lock: + cached = getattr(_wt, "_parallel_client", None) + if cached is not None: + return cached - api_key = get_provider_env("PARALLEL_API_KEY") - if not api_key: - raise ValueError( - "PARALLEL_API_KEY environment variable not set. " - "Get your API key at https://parallel.ai" - ) + from agent.web_search_provider import get_provider_env + + api_key = get_provider_env("PARALLEL_API_KEY") + if not api_key: + raise ValueError( + "PARALLEL_API_KEY environment variable not set. " + "Get your API key at https://parallel.ai" + ) - _ensure_parallel_sdk_installed() - from parallel import Parallel # noqa: WPS433 — deliberately lazy + _ensure_parallel_sdk_installed() + from parallel import Parallel # noqa: WPS433 — deliberately lazy - client = Parallel(api_key=api_key) - _wt._parallel_client = client - return client + client = Parallel(api_key=api_key) + _wt._parallel_client = client + return client def _get_async_client() -> Any: @@ -101,21 +109,26 @@ def _get_async_client() -> Any: if cached is not None: return cached - from agent.web_search_provider import get_provider_env + with _async_client_lock: + cached = getattr(_wt, "_async_parallel_client", None) + if cached is not None: + return cached - api_key = get_provider_env("PARALLEL_API_KEY") - if not api_key: - raise ValueError( - "PARALLEL_API_KEY environment variable not set. " - "Get your API key at https://parallel.ai" - ) + from agent.web_search_provider import get_provider_env + + api_key = get_provider_env("PARALLEL_API_KEY") + if not api_key: + raise ValueError( + "PARALLEL_API_KEY environment variable not set. " + "Get your API key at https://parallel.ai" + ) - _ensure_parallel_sdk_installed() - from parallel import AsyncParallel # noqa: WPS433 — deliberately lazy + _ensure_parallel_sdk_installed() + from parallel import AsyncParallel # noqa: WPS433 — deliberately lazy - client = AsyncParallel(api_key=api_key) - _wt._async_parallel_client = client - return client + client = AsyncParallel(api_key=api_key) + _wt._async_parallel_client = client + return client def _reset_clients_for_tests() -> None: diff --git a/tests/plugins/web/test_provider_client_concurrency.py b/tests/plugins/web/test_provider_client_concurrency.py new file mode 100644 index 0000000000000..eb898e0c68ace --- /dev/null +++ b/tests/plugins/web/test_provider_client_concurrency.py @@ -0,0 +1,133 @@ +from concurrent.futures import ThreadPoolExecutor +import sys +import threading +import types + +import pytest + +import tools.web_tools as web_tools +from plugins.web.exa import provider as exa_provider +from plugins.web.parallel import provider as parallel_provider + + +@pytest.mark.parametrize( + ( + "provider_module", + "lock_name", + "getter_name", + "cache_name", + "env_name", + "sdk_module_name", + "constructor_name", + "needs_headers", + ), + [ + pytest.param( + parallel_provider, + "_sync_client_lock", + "_get_sync_client", + "_parallel_client", + "PARALLEL_API_KEY", + "parallel", + "Parallel", + False, + id="parallel-sync", + ), + pytest.param( + parallel_provider, + "_async_client_lock", + "_get_async_client", + "_async_parallel_client", + "PARALLEL_API_KEY", + "parallel", + "AsyncParallel", + False, + id="parallel-async", + ), + pytest.param( + exa_provider, + "_client_lock", + "_get_exa_client", + "_exa_client", + "EXA_API_KEY", + "exa_py", + "Exa", + True, + id="exa", + ), + ], +) +def test_concurrent_first_use_constructs_one_client( + monkeypatch, + provider_module, + lock_name, + getter_name, + cache_name, + env_name, + sdk_module_name, + constructor_name, + needs_headers, +): + workers = 8 + callers_ready = threading.Barrier(workers, timeout=5) + release_constructor = threading.Event() + state_changed = threading.Condition() + lock_attempts = [0] + constructed = [] + + class ObservedLock: + def __init__(self): + self._lock = threading.Lock() + + def __enter__(self): + with state_changed: + lock_attempts[0] += 1 + state_changed.notify_all() + self._lock.acquire() + return self + + def __exit__(self, *_args): + self._lock.release() + + def constructor(*_args, **_kwargs): + client = types.SimpleNamespace(headers={}) if needs_headers else object() + with state_changed: + constructed.append(client) + state_changed.notify_all() + assert release_constructor.wait(timeout=5) + return client + + sdk_module = types.ModuleType(sdk_module_name) + setattr(sdk_module, constructor_name, constructor) + monkeypatch.setitem(sys.modules, sdk_module_name, sdk_module) + monkeypatch.setenv(env_name, "test-key") + monkeypatch.setattr(web_tools, cache_name, None) + monkeypatch.setattr(provider_module, lock_name, ObservedLock()) + if provider_module is parallel_provider: + monkeypatch.setattr( + parallel_provider, "_ensure_parallel_sdk_installed", lambda: None + ) + else: + from tools import lazy_deps + + monkeypatch.setattr(lazy_deps, "ensure", lambda *_args, **_kwargs: None) + + getter = getattr(provider_module, getter_name) + + def get_client(_index): + callers_ready.wait() + return getter() + + with ThreadPoolExecutor(max_workers=workers) as pool: + futures = [pool.submit(get_client, index) for index in range(workers)] + with state_changed: + reached_boundary = state_changed.wait_for( + lambda: lock_attempts[0] == workers or len(constructed) == workers, + timeout=5, + ) + release_constructor.set() + returned = [future.result(timeout=5) for future in futures] + + assert reached_boundary + assert len(constructed) == 1 + assert all(client is returned[0] for client in returned) From a442a7704e65547c24190d4fb86dad46bb504d0b Mon Sep 17 00:00:00 2001 From: embwl0x Date: Sat, 15 Aug 2026 18:30:14 -0500 Subject: [PATCH 2/2] fix(web): keep async Parallel lifecycle out of singleton lock --- plugins/web/exa/provider.py | 5 ++- plugins/web/parallel/provider.py | 35 +++++++++---------- .../web/test_provider_client_concurrency.py | 11 ------ 3 files changed, 20 insertions(+), 31 deletions(-) diff --git a/plugins/web/exa/provider.py b/plugins/web/exa/provider.py index 4c7603fb56314..92904947586b5 100644 --- a/plugins/web/exa/provider.py +++ b/plugins/web/exa/provider.py @@ -49,6 +49,8 @@ def _get_exa_client() -> Any: """ import tools.web_tools as _wt + # Intentionally lock-free after publication; the slow path double-checks + # under the lock before constructing the singleton. cached = getattr(_wt, "_exa_client", None) if cached is not None: return cached @@ -88,7 +90,8 @@ def _reset_client_for_tests() -> None: """Drop the cached Exa client so tests can re-instantiate cleanly.""" import tools.web_tools as _wt - _wt._exa_client = None + with _client_lock: + _wt._exa_client = None class ExaWebSearchProvider(WebSearchProvider): diff --git a/plugins/web/parallel/provider.py b/plugins/web/parallel/provider.py index 2d772348cbcb7..f513a8bd1d8d3 100644 --- a/plugins/web/parallel/provider.py +++ b/plugins/web/parallel/provider.py @@ -43,7 +43,6 @@ # The plugin reads/writes through that public module (see # :func:`_get_sync_client` / :func:`_get_async_client`). _sync_client_lock = threading.Lock() -_async_client_lock = threading.Lock() def _ensure_parallel_sdk_installed() -> None: @@ -72,6 +71,8 @@ def _get_sync_client() -> Any: """ import tools.web_tools as _wt + # Intentionally lock-free after publication; the slow path double-checks + # under the lock before constructing the singleton. cached = getattr(_wt, "_parallel_client", None) if cached is not None: return cached @@ -109,26 +110,21 @@ def _get_async_client() -> Any: if cached is not None: return cached - with _async_client_lock: - cached = getattr(_wt, "_async_parallel_client", None) - if cached is not None: - return cached + from agent.web_search_provider import get_provider_env - from agent.web_search_provider import get_provider_env + api_key = get_provider_env("PARALLEL_API_KEY") + if not api_key: + raise ValueError( + "PARALLEL_API_KEY environment variable not set. " + "Get your API key at https://parallel.ai" + ) - api_key = get_provider_env("PARALLEL_API_KEY") - if not api_key: - raise ValueError( - "PARALLEL_API_KEY environment variable not set. " - "Get your API key at https://parallel.ai" - ) - - _ensure_parallel_sdk_installed() - from parallel import AsyncParallel # noqa: WPS433 — deliberately lazy + _ensure_parallel_sdk_installed() + from parallel import AsyncParallel # noqa: WPS433 — deliberately lazy - client = AsyncParallel(api_key=api_key) - _wt._async_parallel_client = client - return client + client = AsyncParallel(api_key=api_key) + _wt._async_parallel_client = client + return client def _reset_clients_for_tests() -> None: @@ -139,7 +135,8 @@ def _reset_clients_for_tests() -> None: """ import tools.web_tools as _wt - _wt._parallel_client = None + with _sync_client_lock: + _wt._parallel_client = None _wt._async_parallel_client = None diff --git a/tests/plugins/web/test_provider_client_concurrency.py b/tests/plugins/web/test_provider_client_concurrency.py index eb898e0c68ace..8d3d039a1ac57 100644 --- a/tests/plugins/web/test_provider_client_concurrency.py +++ b/tests/plugins/web/test_provider_client_concurrency.py @@ -33,17 +33,6 @@ False, id="parallel-sync", ), - pytest.param( - parallel_provider, - "_async_client_lock", - "_get_async_client", - "_async_parallel_client", - "PARALLEL_API_KEY", - "parallel", - "AsyncParallel", - False, - id="parallel-async", - ), pytest.param( exa_provider, "_client_lock",