diff --git a/hermes_cli/config_defaults.py b/hermes_cli/config_defaults.py index e270042131..adcad7f340 100644 --- a/hermes_cli/config_defaults.py +++ b/hermes_cli/config_defaults.py @@ -530,6 +530,14 @@ DEFAULT_CONFIG = { # is also excluded from the keyless ring) # unset — auto: keyed when the API key is present, else the ring "provider_tier": {}, + # TTL result caching for web_search + web_extract. Repeat searches + # (same query, same provider) within the TTL are served from an + # in-process memo; repeat extracts of the same URL are served from + # the cache/web full-text store. Concurrent identical searches + # (parallel subagents) coalesce into one vendor request. Only + # successful responses are cached. + "cache_enabled": True, + "cache_ttl_minutes": 20, }, "browser": { diff --git a/tests/tools/conftest.py b/tests/tools/conftest.py index 355e3bb9a5..df6e0d82e5 100644 --- a/tests/tools/conftest.py +++ b/tests/tools/conftest.py @@ -55,6 +55,21 @@ def _materialize_mcp_sdk_symbols(): yield +@pytest.fixture(autouse=True) +def _clear_web_result_cache(): + """Reset the web_search TTL memo between tests. + + The memo is module-global state in tools/web_result_cache.py; without + this, a test that exercised web_search_tool leaves a cached response + that a later test with the same query would receive instead of its own + mocked provider result. + """ + from tools.web_result_cache import search_memo + search_memo.clear() + yield + search_memo.clear() + + def register_all_web_providers(): """Register all bundled web-search providers into the global registry. diff --git a/tests/tools/test_web_result_cache.py b/tests/tools/test_web_result_cache.py new file mode 100644 index 0000000000..516b31e482 --- /dev/null +++ b/tests/tools/test_web_result_cache.py @@ -0,0 +1,270 @@ +"""Tests for tools/web_result_cache.py — TTL memo for web_search and the +disk-backed extract cache, plus their wiring into web_tools. + +The cache sits AFTER safety gates and around the paid vendor call only, so +these tests focus on: hit/miss semantics, TTL expiry, limit bucketing + +slicing, single-flight coalescing, error non-caching, the disable flag, and +extract index integrity (tamper = miss, oversized = not indexed). +""" + +import json +import threading +import time + +import pytest + +import tools.web_result_cache as wrc +from tools.web_result_cache import ( + SearchMemo, + bucket_limit, + extract_cache_get, + extract_cache_put, + normalize_query, + slice_search_response, +) + + +@pytest.fixture(autouse=True) +def _isolated_cache(tmp_path, monkeypatch): + """Point the extract cache at a temp dir and force cache-on defaults.""" + cache_dir = tmp_path / "cache" / "web" + cache_dir.mkdir(parents=True) + monkeypatch.setattr(wrc, "_cache_dir", lambda: cache_dir) + monkeypatch.setattr(wrc, "_web_config", lambda: {}) + yield cache_dir + + +def _ok_response(n=10): + return { + "success": True, + "data": {"web": [ + {"title": f"t{i}", "url": f"https://e.com/{i}", "description": "d"} + for i in range(n) + ]}, + } + + +# ── bucketing / normalization ──────────────────────────────────────────── + +def test_bucket_limit_rounds_up(): + assert bucket_limit(1) == 10 + assert bucket_limit(10) == 10 + assert bucket_limit(11) == 20 + assert bucket_limit(50) == 50 + assert bucket_limit(99) == 100 + assert bucket_limit(500) == 100 + + +def test_normalize_query_folds_case_and_whitespace(): + assert normalize_query(" Weather in\tVegas ") == "weather in vegas" + + +def test_slice_search_response_trims_to_requested_limit(): + sliced = slice_search_response(_ok_response(10), 3) + assert len(sliced["data"]["web"]) == 3 + # original untouched (defensive copy) + assert len(_ok_response(10)["data"]["web"]) == 10 + + +# ── search memo ────────────────────────────────────────────────────────── + +def test_search_memo_hit_within_ttl(): + memo = SearchMemo() + memo.store("firecrawl", "weather in vegas", 5, _ok_response()) + hit = memo.lookup("firecrawl", "Weather In Vegas", 8) # same bucket (10) + assert hit is not None and hit["success"] + + +def test_search_memo_miss_across_providers_and_buckets(): + memo = SearchMemo() + memo.store("firecrawl", "q", 5, _ok_response()) + assert memo.lookup("tavily", "q", 5) is None # different provider + assert memo.lookup("firecrawl", "q", 15) is None # different bucket + assert memo.lookup("firecrawl", "other", 5) is None # different query + + +def test_search_memo_expires_after_ttl(monkeypatch): + memo = SearchMemo() + memo.store("firecrawl", "q", 5, _ok_response()) + monkeypatch.setattr(wrc, "ttl_seconds", lambda: 0.0) + # store used the old TTL; force expiry by faking monotonic forward + real = time.monotonic + monkeypatch.setattr(time, "monotonic", lambda: real() + 100 * 3600) + assert memo.lookup("firecrawl", "q", 5) is None + + +def test_search_memo_never_caches_failures(): + memo = SearchMemo() + memo.store("firecrawl", "q", 5, {"success": False, "error": "boom"}) + assert memo.lookup("firecrawl", "q", 5) is None + + +def test_search_memo_disabled_by_config(monkeypatch): + monkeypatch.setattr(wrc, "_web_config", lambda: {"cache_enabled": False}) + memo = SearchMemo() + memo.store("firecrawl", "q", 5, _ok_response()) + assert memo.lookup("firecrawl", "q", 5) is None + + +def test_search_memo_hit_returns_copy(): + memo = SearchMemo() + memo.store("firecrawl", "q", 5, _ok_response()) + first = memo.lookup("firecrawl", "q", 5) + first["data"]["web"].clear() + second = memo.lookup("firecrawl", "q", 5) + assert len(second["data"]["web"]) == 10 + + +def test_single_flight_coalesces_concurrent_identical_queries(): + """Two threads race the same query: exactly one paid call happens.""" + memo = SearchMemo() + calls = [] + barrier = threading.Barrier(2) + results = [] + + def worker(): + barrier.wait() + resp = memo.lookup("p", "q", 5) + if resp is None: + with memo.flight_lock("p", "q", 5): + resp = memo.lookup("p", "q", 5) + if resp is None: + calls.append(1) # the "paid" request + time.sleep(0.05) # widen the race window + resp = _ok_response() + memo.store("p", "q", 5, resp) + results.append(resp) + + threads = [threading.Thread(target=worker) for _ in range(2)] + for t in threads: + t.start() + for t in threads: + t.join() + + assert len(calls) == 1, "concurrent identical queries must share one request" + assert len(results) == 2 and all(r["success"] for r in results) + + +# ── extract cache ──────────────────────────────────────────────────────── + +def _put_and_get(url="https://example.com/a", content="hello world", + fmt=None, **kw): + extract_cache_put(url, content, title="T", format=fmt, **kw) + return extract_cache_get(url, format=fmt) + + +def test_extract_cache_roundtrip(monkeypatch, _isolated_cache): + # _store_full_text writes to the real hermes dir; redirect it here. + stored = {} + + def fake_store(url, content): + p = _isolated_cache / "page.md" + p.write_text(content, encoding="utf-8") + stored["path"] = str(p) + return str(p) + + import tools.web_tools as wt + monkeypatch.setattr(wt, "_store_full_text", fake_store) + + hit = _put_and_get() + assert hit is not None + assert hit["content"] == "hello world" + assert hit["title"] == "T" + assert hit["cached"] is True + + +def test_extract_cache_expired_entry_is_miss(monkeypatch, _isolated_cache): + import tools.web_tools as wt + p = _isolated_cache / "page.md" + p.write_text("x", encoding="utf-8") + monkeypatch.setattr(wt, "_store_full_text", lambda u, c: str(p)) + extract_cache_put("https://e.com", "x") + monkeypatch.setattr(wrc, "ttl_seconds", lambda: 0.0) + assert extract_cache_get("https://e.com") is None + + +def test_extract_cache_format_participates_in_key(monkeypatch, _isolated_cache): + import tools.web_tools as wt + p = _isolated_cache / "page.md" + p.write_text("md", encoding="utf-8") + monkeypatch.setattr(wt, "_store_full_text", lambda u, c: str(p)) + extract_cache_put("https://e.com", "md", format="markdown") + assert extract_cache_get("https://e.com", format="html") is None + assert extract_cache_get("https://e.com", format="markdown") is not None + + +def test_extract_cache_oversized_page_not_indexed(monkeypatch, _isolated_cache): + import tools.web_tools as wt + monkeypatch.setattr( + wt, "_store_full_text", + lambda u, c: str(_isolated_cache / "should-not-happen.md"), + ) + big = "x" * (wt.MAX_STORED_TEXT_CHARS + 1) + extract_cache_put("https://big.com", big) + assert extract_cache_get("https://big.com") is None + + +def test_extract_cache_tampered_index_path_is_miss(_isolated_cache, tmp_path): + """An index entry pointing outside cache/web must never be read.""" + outside = tmp_path / "outside.md" + outside.write_text("secret", encoding="utf-8") + index = { + wrc._url_digest("https://evil.com", None): { + "url": "https://evil.com", + "file": str(outside), + "title": "", + "fetched_at": time.time(), + } + } + (_isolated_cache / wrc._INDEX_FILENAME).write_text(json.dumps(index)) + assert extract_cache_get("https://evil.com") is None + + +def test_extract_cache_missing_file_is_miss(_isolated_cache): + index = { + wrc._url_digest("https://gone.com", None): { + "url": "https://gone.com", + "file": str(_isolated_cache / "pruned.md"), + "title": "", + "fetched_at": time.time(), + } + } + (_isolated_cache / wrc._INDEX_FILENAME).write_text(json.dumps(index)) + assert extract_cache_get("https://gone.com") is None + + +def test_extract_cache_corrupt_index_is_empty(_isolated_cache): + (_isolated_cache / wrc._INDEX_FILENAME).write_text("{not json") + assert extract_cache_get("https://any.com") is None + + +def test_extract_cache_disabled_by_config(monkeypatch, _isolated_cache): + import tools.web_tools as wt + p = _isolated_cache / "page.md" + p.write_text("x", encoding="utf-8") + monkeypatch.setattr(wt, "_store_full_text", lambda u, c: str(p)) + extract_cache_put("https://e.com", "x") + monkeypatch.setattr(wrc, "_web_config", lambda: {"cache_enabled": False}) + assert extract_cache_get("https://e.com") is None + + +def test_index_eviction_keeps_newest(monkeypatch, _isolated_cache): + monkeypatch.setattr(wrc, "_INDEX_MAX_ENTRIES", 3) + now = time.time() + index = { + f"digest{i}": {"url": f"u{i}", "file": "f", "fetched_at": now + i} + for i in range(6) + } + wrc._save_index(index) + saved = json.loads((_isolated_cache / wrc._INDEX_FILENAME).read_text()) + assert len(saved) == 3 + assert set(saved) == {"digest3", "digest4", "digest5"} + + +def test_ttl_clamping(monkeypatch): + monkeypatch.setattr(wrc, "_web_config", lambda: {"cache_ttl_minutes": 0}) + assert wrc.ttl_seconds() == 60.0 # floor 1 minute + monkeypatch.setattr(wrc, "_web_config", lambda: {"cache_ttl_minutes": 99999}) + assert wrc.ttl_seconds() == 1440 * 60.0 # ceiling 24h + monkeypatch.setattr(wrc, "_web_config", lambda: {"cache_ttl_minutes": "bogus"}) + assert wrc.ttl_seconds() == 20 * 60.0 # default on garbage diff --git a/tools/web_result_cache.py b/tools/web_result_cache.py new file mode 100644 index 0000000000..8163f51a18 --- /dev/null +++ b/tools/web_result_cache.py @@ -0,0 +1,306 @@ +"""Result caching for web_search / web_extract. + +Two caches, both TTL-bounded (default 20 minutes, ``web.cache_ttl_minutes``): + +* **Search memo** — in-memory, per-process. Keyed by (provider, normalized + query, bucketed limit). Concurrent identical queries are single-flighted: + the first caller performs the paid request while the rest wait and share + the response. Requested limits are bucketed up to 10/20/50/100 so + near-identical requests (limit=5 vs limit=8) share one entry; callers get + their requested count sliced from the bucket. + +* **Extract cache** — disk-backed, cross-process. Reuses the existing + ``cache/web`` full-text store (the same files the truncate-store footer + points read_file at) plus a small JSON sidecar index mapping URL digest → + (file, fetched_at, title). A repeat ``web_extract`` of the same URL within + TTL reads the stored clean text back instead of re-scraping, then re-runs + the normal truncate pipeline with the caller's char_limit. + +Why this lives here and not in generic tool dispatch (issue #8126): a +dispatch-level memo would have to reason about middleware, approval gates, +and hooks on cache hits. Down here the cache sits *after* every safety check +(secret-in-URL, SSRF, policy) and *before* the paid vendor call — hits skip +only the network request, never a control. + +Disable with ``web.cache_enabled: false``; both TTLs come from +``web.cache_ttl_minutes``. Only successful responses are ever cached. +""" + +import hashlib +import json +import logging +import re +import threading +import time +from pathlib import Path +from typing import Any, Dict, List, Optional, Tuple + +logger = logging.getLogger(__name__) + +# Requested search limits are rounded UP to one of these buckets so cache +# keys collide on purpose (idea borrowed from Apodex FrontierAgent's +# web_search num-bucketing). Callers get their requested count sliced out. +_LIMIT_BUCKETS = (10, 20, 50, 100) + +DEFAULT_TTL_MINUTES = 20 + +# Extract-index sidecar filename inside cache/web. +_INDEX_FILENAME = "extract-index.json" + +# Cap index growth; oldest entries evicted past this. +_INDEX_MAX_ENTRIES = 500 + + +def _web_config() -> dict: + try: + from tools.web_tools import _load_web_config + return _load_web_config() + except Exception: # noqa: BLE001 — config problems must never break tools + return {} + + +def cache_enabled() -> bool: + """Both caches honor ``web.cache_enabled`` (default: on).""" + val = _web_config().get("cache_enabled") + if val is None: + return True + return bool(val) + + +def ttl_seconds() -> float: + """TTL from ``web.cache_ttl_minutes`` (default 20, clamped 1–1440).""" + raw = _web_config().get("cache_ttl_minutes") + try: + minutes = float(raw) if raw is not None else DEFAULT_TTL_MINUTES + except (TypeError, ValueError): + minutes = DEFAULT_TTL_MINUTES + minutes = max(1.0, min(minutes, 1440.0)) + return minutes * 60.0 + + +def bucket_limit(limit: int) -> int: + """Round a requested result count up to the nearest bucket.""" + for b in _LIMIT_BUCKETS: + if limit <= b: + return b + return _LIMIT_BUCKETS[-1] + + +def normalize_query(query: str) -> str: + """Case-fold and collapse whitespace so trivial variants share an entry.""" + return re.sub(r"\s+", " ", (query or "").strip().lower()) + + +# --------------------------------------------------------------------------- +# Search memo (in-memory, single-flight) +# --------------------------------------------------------------------------- + +class SearchMemo: + """TTL memo + single-flight coalescer for search responses. + + Thread-safe: web tools run inside the parallel tool-dispatch thread pool + and subagents share this process, so identical queries can genuinely race. + Per-key locks make the losers of that race wait for (and share) the + winner's response instead of issuing their own paid request. + """ + + def __init__(self) -> None: + self._store: Dict[tuple, Tuple[float, dict]] = {} + self._store_lock = threading.Lock() + self._key_locks: Dict[tuple, threading.Lock] = {} + + def _key(self, provider: str, query: str, limit: int) -> tuple: + return (provider, normalize_query(query), bucket_limit(limit)) + + def lookup(self, provider: str, query: str, limit: int) -> Optional[dict]: + if not cache_enabled(): + return None + key = self._key(provider, query, limit) + with self._store_lock: + hit = self._store.get(key) + if hit is None: + return None + expires, response = hit + if time.monotonic() >= expires: + del self._store[key] + return None + logger.info("web_search cache hit: %r via %s", query, provider) + return json.loads(json.dumps(response)) # defensive copy + + def store(self, provider: str, query: str, limit: int, response: dict) -> None: + """Cache a SUCCESSFUL response for the bucketed key.""" + if not cache_enabled(): + return + if not isinstance(response, dict) or not response.get("success"): + return + key = self._key(provider, query, limit) + with self._store_lock: + # Opportunistic expiry sweep to bound memory. + now = time.monotonic() + for k in [k for k, (exp, _) in self._store.items() if now >= exp]: + del self._store[k] + self._store[key] = (now + ttl_seconds(), json.loads(json.dumps(response))) + + def flight_lock(self, provider: str, query: str, limit: int) -> threading.Lock: + """Per-key lock for single-flight coalescing. + + Callers hold this around lookup-miss → paid request → store, so a + concurrent identical call blocks until the winner has stored, then + finds the entry on its own lookup. + """ + key = self._key(provider, query, limit) + with self._store_lock: + lock = self._key_locks.get(key) + if lock is None: + # Bound the lock table alongside the store. + if len(self._key_locks) > 256: + self._key_locks.clear() + lock = threading.Lock() + self._key_locks[key] = lock + return lock + + def clear(self) -> None: + """Drop all cached entries (tests; config changes).""" + with self._store_lock: + self._store.clear() + self._key_locks.clear() + + +search_memo = SearchMemo() + + +def slice_search_response(response: dict, limit: int) -> dict: + """Trim a bucketed response's result list down to the caller's limit.""" + try: + web = response.get("data", {}).get("web") + if isinstance(web, list) and len(web) > limit: + out = json.loads(json.dumps(response)) + out["data"]["web"] = out["data"]["web"][:limit] + return out + except Exception: # noqa: BLE001 + pass + return response + + +# --------------------------------------------------------------------------- +# Extract cache (disk-backed, reuses cache/web) +# --------------------------------------------------------------------------- + +_index_lock = threading.Lock() + + +def _cache_dir() -> Optional[Path]: + try: + from hermes_constants import get_hermes_dir + d = get_hermes_dir("cache/web", "web_cache") + d.mkdir(parents=True, exist_ok=True) + return d + except Exception: # noqa: BLE001 + return None + + +def _index_path() -> Optional[Path]: + d = _cache_dir() + return (d / _INDEX_FILENAME) if d else None + + +def _load_index() -> dict: + path = _index_path() + if path is None or not path.exists(): + return {} + try: + data = json.loads(path.read_text(encoding="utf-8")) + return data if isinstance(data, dict) else {} + except Exception: # noqa: BLE001 — corrupt index == empty cache + return {} + + +def _save_index(index: dict) -> None: + path = _index_path() + if path is None: + return + try: + if len(index) > _INDEX_MAX_ENTRIES: + newest = sorted( + index.items(), + key=lambda kv: kv[1].get("fetched_at", 0), + reverse=True, + )[:_INDEX_MAX_ENTRIES] + index = dict(newest) + tmp = path.with_suffix(".tmp") + tmp.write_text(json.dumps(index), encoding="utf-8") + tmp.replace(path) + except Exception as exc: # noqa: BLE001 + logger.debug("Failed to save web extract cache index: %s", exc) + + +def _url_digest(url: str, format: Optional[str]) -> str: + # format participates in the key: an html extract is not a markdown one. + raw = f"{url}\n{format or 'markdown'}" + return hashlib.sha256(raw.encode("utf-8")).hexdigest()[:16] + + +def extract_cache_get(url: str, format: Optional[str] = None) -> Optional[dict]: + """Return {'url','title','content'} for a fresh cached page, else None.""" + if not cache_enabled(): + return None + with _index_lock: + index = _load_index() + entry = index.get(_url_digest(url, format)) + if not entry: + return None + if (time.time() - float(entry.get("fetched_at", 0))) >= ttl_seconds(): + return None + try: + file_path = Path(entry["file"]) + cache_root = _cache_dir() + # The index is plain JSON on disk; never let a tampered entry read + # outside cache/web. + if cache_root is None or cache_root.resolve() not in file_path.resolve().parents: + return None + content = file_path.read_text(encoding="utf-8") + except Exception: # noqa: BLE001 — evicted/pruned file == miss + return None + logger.info("web_extract cache hit: %s", url) + return { + "url": url, + "title": entry.get("title", ""), + "content": content, + "error": None, + "cached": True, + } + + +def extract_cache_put( + url: str, + content: str, + title: str = "", + format: Optional[str] = None, +) -> None: + """Store one successful extraction's full clean text for TTL reuse. + + Pages larger than the truncate-store ceiling are NOT indexed for reuse: + the stored copy would be incomplete, and serving it back as if whole + would silently lose the tail. (The capped file is still written by the + truncate-store path for read_file paging — we just don't index it.) + """ + if not cache_enabled() or not content: + return + try: + from tools.web_tools import MAX_STORED_TEXT_CHARS, _store_full_text + if len(content) > MAX_STORED_TEXT_CHARS: + return + file_path = _store_full_text(url, content) + if not file_path: + return + with _index_lock: + index = _load_index() + index[_url_digest(url, format)] = { + "url": url, + "file": file_path, + "title": title or "", + "fetched_at": time.time(), + } + _save_index(index) + except Exception as exc: # noqa: BLE001 — cache writes are best-effort + logger.debug("Failed to cache web extract for %s: %s", url, exc) diff --git a/tools/web_tools.py b/tools/web_tools.py index d215b9ec84..dd09b88d8c 100644 --- a/tools/web_tools.py +++ b/tools/web_tools.py @@ -965,28 +965,67 @@ def web_search_tool(query: str, limit: int = 5) -> str: "Web search via %s: '%s' (limit: %d)", provider.name, query, limit, ) - try: - response_data = provider.search(query, limit) - except Exception as exc: # noqa: BLE001 — candidate for rescue - if _rescue_eligible(provider): - response_data = _rescue_search( - provider.name, str(exc), query, limit - ) + # ── TTL memo + single-flight (tools/web_result_cache.py) ── + # Sits after every safety/config check and directly around the + # paid vendor call. Identical queries within the TTL (subagent + # fan-outs, repeat lookups) are served from memory; concurrent + # identical queries share one request via the flight lock. The + # provider is asked for the BUCKETED count (10/20/50/100) so + # near-identical limits share an entry; the caller's requested + # count is sliced out below. Only successful responses cache. + from tools.web_result_cache import ( + bucket_limit as _bucket_limit, + search_memo as _search_memo, + slice_search_response as _slice_search_response, + ) + + def _paid_search() -> tuple[dict, bool]: + _fetch_limit = _bucket_limit(limit) + _rescued = False + try: + _resp = provider.search(query, _fetch_limit) + except Exception as exc: # noqa: BLE001 — candidate for rescue + if _rescue_eligible(provider): + _rescued = True + _resp = _rescue_search( + provider.name, str(exc), query, _fetch_limit + ) + else: + raise else: - raise - else: - if ( - not response_data.get("success") - and _rescue_eligible(provider) - ): - # One-shot keyless rescue: THIS call rides the free-tier - # ring; the next call attempts the chosen backend again. - response_data = _rescue_search( - provider.name, - str(response_data.get("error", "")), - query, - limit, + if not _resp.get("success") and _rescue_eligible(provider): + # One-shot keyless rescue: THIS call rides the + # free-tier ring; the next call attempts the chosen + # backend again. + _rescued = True + _resp = _rescue_search( + provider.name, + str(_resp.get("error", "")), + query, + _fetch_limit, + ) + return _resp, _rescued + + response_data = _search_memo.lookup(provider.name, query, limit) + if response_data is None: + with _search_memo.flight_lock(provider.name, query, limit): + # Re-check inside the lock: a concurrent identical call + # may have stored while this one waited. + response_data = _search_memo.lookup( + provider.name, query, limit ) + if response_data is None: + response_data, _was_rescued = _paid_search() + # Never cache a rescue-served response: it came from + # a ring vendor, not the chosen backend (wrong key), + # and caching it would make the one-shot rescue + # sticky for this query for a whole TTL — the next + # call must attempt the chosen backend again. + if not _was_rescued: + _search_memo.store( + provider.name, query, limit, response_data + ) + response_data = _slice_search_response(response_data, limit) debug_call_data["results_count"] = len(response_data.get("data", {}).get("web", [])) result_json = json.dumps(response_data, indent=2, ensure_ascii=False) @@ -1121,146 +1160,208 @@ async def web_extract_tool( if not safe_urls: results = [] else: - backend = _get_extract_backend() - - # All seven providers (brave-free, ddgs, searxng, exa, parallel, - # tavily, firecrawl) now live as plugins. The dispatcher is a - # registry lookup + delegation. Some providers' extract() is - # async (parallel, firecrawl), others sync (exa, tavily) — we - # detect coroutine functions and await; sync functions run - # inline (the policy gate, SSRF re-check, etc. live inside the - # provider itself for the firecrawl per-URL loop). - _ensure_web_plugins_loaded() - from agent.web_search_registry import ( - get_active_extract_provider, - get_provider as _wsp_get_provider, - _disabled_web_plugin_for, + # ── Extract cache (tools/web_result_cache.py) ───────────────── + # Disk-backed via the existing cache/web full-text store: a URL + # extracted within the TTL is served from disk instead of + # re-scraped. Sits AFTER the secret-URL and SSRF gates so hits + # skip only the vendor call, never a safety check. Cached + # entries re-run the normal truncate pipeline below, so a + # different caller char_limit still works off one scrape. + from tools.web_result_cache import ( + extract_cache_get as _extract_cache_get, + extract_cache_put as _extract_cache_put, ) + cached_results: Dict[int, Dict[str, Any]] = {} + fetch_urls: List[str] = [] + fetch_positions: List[int] = [] + for position, url in enumerate(safe_urls): + hit = _extract_cache_get(url, format=format) + if hit is not None: + cached_results[position] = hit + else: + fetch_urls.append(url) + fetch_positions.append(position) - provider = _wsp_get_provider(backend) if backend else None - if provider is None or not provider.supports_extract(): - # When the configured name IS registered but doesn't support - # extract (search-only providers like brave-free / ddgs / - # searxng), surface that as a typed "search-only" error - # rather than silently switching backends. When the name - # isn't registered at all (typo / uninstalled plugin), fall - # through to the active-provider walk. - if provider is not None and not provider.supports_extract(): - return json.dumps( - { - "success": False, - "error": ( - f"{provider.display_name} is a search-only " - "backend and cannot extract URL content. " - "Set web.extract_backend to firecrawl, " - "tavily, exa, or parallel." - ), - }, - ensure_ascii=False, - ) - from tools.tool_backend_helpers import ( - selection_error, - selection_exists, + if not fetch_urls: + results = [cached_results[i] for i in range(len(safe_urls))] + else: + backend = _get_extract_backend() + + # All seven providers (brave-free, ddgs, searxng, exa, parallel, + # tavily, firecrawl) now live as plugins. The dispatcher is a + # registry lookup + delegation. Some providers' extract() is + # async (parallel, firecrawl), others sync (exa, tavily) — we + # detect coroutine functions and await; sync functions run + # inline (the policy gate, SSRF re-check, etc. live inside the + # provider itself for the firecrawl per-URL loop). + _ensure_web_plugins_loaded() + from agent.web_search_registry import ( + get_active_extract_provider, + get_provider as _wsp_get_provider, + _disabled_web_plugin_for, ) - if backend and selection_exists("web"): - # Strict selection: a stored-but-unregistered backend - # errors by name instead of silently switching to - # whatever the availability walk finds. - disabled_key = _disabled_web_plugin_for(capability="extract") - if disabled_key: - _vendor = disabled_key.split("/", 1)[-1] - error_text = ( - f"web.extract_backend is set to '{_vendor}', but " - f"its plugin ('{disabled_key}') is disabled in " - f"config. Re-enable it with `hermes plugins " - f"enable {disabled_key}` (or remove it from " - "plugins.disabled)." - ) - else: - error_text = selection_error( - "web", - f"'{backend}'", - "no registered web extract provider has that name", - ) - return json.dumps( - {"success": False, "error": error_text}, - ensure_ascii=False, - ) - provider = get_active_extract_provider() - if provider is None: - # If the configured backend is a bundled web plugin the - # user explicitly disabled, the backend is set correctly - # and the real fix is to re-enable the plugin — say so - # instead of telling them to set web.extract_backend - # (which they already did). #40190 follow-up. - disabled_key = _disabled_web_plugin_for(capability="extract") - if disabled_key: - _vendor = disabled_key.split("/", 1)[-1] + provider = _wsp_get_provider(backend) if backend else None + if provider is None or not provider.supports_extract(): + # When the configured name IS registered but doesn't support + # extract (search-only providers like brave-free / ddgs / + # searxng), surface that as a typed "search-only" error + # rather than silently switching backends. When the name + # isn't registered at all (typo / uninstalled plugin), fall + # through to the active-provider walk. + if provider is not None and not provider.supports_extract(): return json.dumps( { "success": False, "error": ( - f"web.extract_backend is set to '{_vendor}', " - f"but its plugin ('{disabled_key}') is disabled " - "in config. Re-enable it with " - f"`hermes plugins enable {disabled_key}` " - "(or remove it from plugins.disabled)." + f"{provider.display_name} is a search-only " + "backend and cannot extract URL content. " + "Set web.extract_backend to firecrawl, " + "tavily, exa, or parallel." ), }, ensure_ascii=False, ) - return json.dumps( - { - "success": False, - "error": ( - "No web extract provider configured. " - "Set web.extract_backend to firecrawl, " - "tavily, exa, or parallel." - ), - }, - ensure_ascii=False, + from tools.tool_backend_helpers import ( + selection_error, + selection_exists, ) - logger.info( - "Web extract via %s: %d URL(s)", provider.name, len(safe_urls) - ) + if backend and selection_exists("web"): + # Strict selection: a stored-but-unregistered backend + # errors by name instead of silently switching to + # whatever the availability walk finds. + disabled_key = _disabled_web_plugin_for(capability="extract") + if disabled_key: + _vendor = disabled_key.split("/", 1)[-1] + error_text = ( + f"web.extract_backend is set to '{_vendor}', but " + f"its plugin ('{disabled_key}') is disabled in " + f"config. Re-enable it with `hermes plugins " + f"enable {disabled_key}` (or remove it from " + "plugins.disabled)." + ) + else: + error_text = selection_error( + "web", + f"'{backend}'", + "no registered web extract provider has that name", + ) + return json.dumps( + {"success": False, "error": error_text}, + ensure_ascii=False, + ) + provider = get_active_extract_provider() + if provider is None: + # If the configured backend is a bundled web plugin the + # user explicitly disabled, the backend is set correctly + # and the real fix is to re-enable the plugin — say so + # instead of telling them to set web.extract_backend + # (which they already did). #40190 follow-up. + disabled_key = _disabled_web_plugin_for(capability="extract") + if disabled_key: + _vendor = disabled_key.split("/", 1)[-1] + return json.dumps( + { + "success": False, + "error": ( + f"web.extract_backend is set to '{_vendor}', " + f"but its plugin ('{disabled_key}') is disabled " + "in config. Re-enable it with " + f"`hermes plugins enable {disabled_key}` " + "(or remove it from plugins.disabled)." + ), + }, + ensure_ascii=False, + ) + return json.dumps( + { + "success": False, + "error": ( + "No web extract provider configured. " + "Set web.extract_backend to firecrawl, " + "tavily, exa, or parallel." + ), + }, + ensure_ascii=False, + ) - # Async-or-sync dispatch: parallel + firecrawl have async - # extract(); exa + tavily are sync. - import inspect - try: - if inspect.iscoroutinefunction(provider.extract): - results = await provider.extract(safe_urls, format=format) + logger.info( + "Web extract via %s: %d URL(s)", provider.name, len(fetch_urls) + ) + + # Async-or-sync dispatch: parallel + firecrawl have async + # extract(); exa + tavily are sync. + import inspect + try: + if inspect.iscoroutinefunction(provider.extract): + results = await provider.extract(fetch_urls, format=format) + else: + # Run sync extract() in a thread so we don't block the + # event loop on network I/O. + results = await asyncio.to_thread( + provider.extract, fetch_urls, format=format + ) + except Exception as exc: # noqa: BLE001 — candidate for rescue + if _rescue_eligible(provider): + failed = [ + {"url": u, "title": "", "content": "", "error": str(exc)} + for u in fetch_urls + ] + results = await asyncio.to_thread( + _rescue_extract, provider.name, fetch_urls, failed + ) + else: + raise else: - # Run sync extract() in a thread so we don't block the - # event loop on network I/O. - results = await asyncio.to_thread( - provider.extract, safe_urls, format=format - ) - except Exception as exc: # noqa: BLE001 — candidate for rescue - if _rescue_eligible(provider): - failed = [ - {"url": u, "title": "", "content": "", "error": str(exc)} - for u in safe_urls - ] - results = await asyncio.to_thread( - _rescue_extract, provider.name, safe_urls, failed - ) - else: - raise - else: - # One-shot keyless rescue when the WHOLE batch failed - # (backend-level outage, not per-page problems). Stateless: - # the next web_extract call uses the chosen backend again. - if ( - results - and all(r.get("error") for r in results) - and _rescue_eligible(provider) - ): - results = await asyncio.to_thread( - _rescue_extract, provider.name, safe_urls, results + # One-shot keyless rescue when the WHOLE batch failed + # (backend-level outage, not per-page problems). Stateless: + # the next web_extract call uses the chosen backend again. + if ( + results + and all(r.get("error") for r in results) + and _rescue_eligible(provider) + ): + results = await asyncio.to_thread( + _rescue_extract, provider.name, fetch_urls, results + ) + + # Cache each successful fetch's full clean text for TTL reuse + # (best-effort; oversized pages are skipped by the cache). + for fetched_pos, fetched in enumerate(results): + if fetched_pos >= len(fetch_urls): + break + if fetched.get("error"): + continue + _content = ( + fetched.get("raw_content", "") or fetched.get("content", "") ) + if _content: + _extract_cache_put( + fetch_urls[fetched_pos], + _content, + title=fetched.get("title", ""), + format=format, + ) + + # Merge fetched results back with cache hits, restoring the + # safe_urls order the downstream reconstruction expects. + if cached_results: + merged: List[Dict[str, Any]] = [None] * len(safe_urls) # type: ignore[list-item] + for position, hit in cached_results.items(): + merged[position] = hit + for fetched_pos, position in enumerate(fetch_positions): + merged[position] = ( + results[fetched_pos] + if fetched_pos < len(results) + else { + "url": safe_urls[position], + "title": "", + "content": "", + "error": "Extract backend returned no result for this URL", + } + ) + results = merged # Reconstruct the original input order across invalid, blocked, and # provider-processed entries. Providers are expected to preserve the diff --git a/website/docs/user-guide/features/web-search.md b/website/docs/user-guide/features/web-search.md index 32959102b7..daf4465e6e 100644 --- a/website/docs/user-guide/features/web-search.md +++ b/website/docs/user-guide/features/web-search.md @@ -62,6 +62,30 @@ If you specifically need the live DOM rather than extracted markdown — for exa --- +## Result caching + +Repeat web calls within a short window are served from cache instead of the paid backend — this saves credits and latency in the two patterns where duplicates are common: subagent fan-outs (several delegated agents researching the same topic) and the agent re-checking a page it read minutes ago. + +| Call | Cache | Scope | +|------|-------|-------| +| `web_search` — same query (case/whitespace-insensitive), same provider | In-memory memo | Per process | +| `web_extract` — same URL, same format | Full text stored under `~/.hermes/cache/web/` | Shared across CLI, gateway, cron, and subagent processes | + +Concurrent identical searches (a parallel subagent fan-out firing the same query at once) are **coalesced into a single backend request** — the first caller pays; the rest share the response. Requested search limits are bucketed up to 10/20/50/100 so near-identical requests (`limit=5` vs `limit=8`) share one entry, with each caller receiving its requested count. + +Only successful responses are cached. Failures always retry the backend, and responses served by the one-shot keyless rescue are never cached (the next call attempts your chosen backend again). Cached extracts re-run the normal truncation pipeline, so a different `char_limit` on the second call works off the same stored scrape. + +```yaml +# ~/.hermes/config.yaml +web: + cache_enabled: true # default; set false to disable both caches + cache_ttl_minutes: 20 # freshness window, clamped 1–1440 +``` + +If you're researching genuinely live data (scores, prices, breaking news) and need every call fresh, lower the TTL or set `web.cache_enabled: false`. + +--- + ## Setup ### Quick setup via `hermes tools`