diff --git a/tools/web_result_cache.py b/tools/web_result_cache.py
index 4efbe757a3..8bcc23cbf8 100644
--- a/tools/web_result_cache.py
+++ b/tools/web_result_cache.py
@@ -1,13 +1,11 @@
-"""Result caching for web_search / web_extract.
-
-Two caches, both TTL-bounded (default 20 min, ``web.cache_ttl_minutes``; disable with
-``web.cache_enabled: false``). Only successful responses cache.
-* **Search memo** — in-memory, per-process, single-flighted: concurrent identical queries share
-  one paid request. Limits are bucketed to 10/20/50/100 so near-identical requests share an entry.
+"""Result caching for web_search / web_extract; both caches TTL-bounded (default 20 min,
+``web.cache_ttl_minutes``; disable with ``web.cache_enabled: false``), only successful responses cache.
+* **Search memo** — in-memory, per-process, single-flighted: concurrent identical queries share one
+  paid request. Limits bucket to 10/20/50/100 so near-identical requests share an entry.
 * **Extract cache** — disk-backed under ``cache/web`` (cross-process) with a JSON sidecar index:
   URL digest → (file, fetched_at, title). Hits re-run the normal truncate pipeline.
-Lives here, not in generic tool dispatch, so hits sit *after* every safety check (secret-in-URL,
-SSRF, policy) and skip only the vendor call.
+Lives here, not in generic tool dispatch, so hits sit *after* every safety check (secret-in-URL, SSRF,
+policy) and skip only the vendor call.
 """
 
 import hashlib
@@ -86,11 +84,9 @@ def _deep_copy(response: dict) -> dict:
 # ---------------------------------------------------------------------------
 
 class SearchMemo:
-    """TTL memo + single-flight coalescer for search responses.
-
-    Thread-safe: the parallel tool-dispatch pool and subagents share this process, so identical
-    queries genuinely race. Per-key locks make the losers wait for (and share) the winner's response.
-    """
+    """TTL memo + single-flight coalescer for search responses. Thread-safe: the parallel tool-dispatch pool
+    and subagents share this process, so identical queries genuinely race; per-key locks make the losers
+    wait for (and share) the winner's response."""
 
     def __init__(self) -> None:
         self._store: Dict[tuple, Tuple[float, dict]] = {}
@@ -181,8 +177,7 @@ def _cache_dir() -> Optional[Path]:
 
 
 def _index_path() -> Optional[Path]:
-    d = _cache_dir()
-    return (d / _INDEX_FILENAME) if d else None
+    return (d / _INDEX_FILENAME) if (d := _cache_dir()) else None
 
 
 def _load_index() -> dict:
@@ -215,18 +210,14 @@ def _save_index(index: dict) -> None:
 
 
 def _url_digest(url: str, format: Optional[str], provider: str = "") -> str:
-    # format AND provider are part of the key: an html extract is not a
-    # markdown one, and one backend's rendering is not another's.
+    # format AND provider are part of the key: html != markdown, and one backend's rendering is not another's.
     raw = f"{url}\n{format or 'markdown'}\n{provider or ''}"
     return hashlib.sha256(raw.encode("utf-8")).hexdigest()[:16]
 
 
 def _entry_file_path(url: str, format: Optional[str], provider: str) -> Optional[Path]:
-    """Dedicated cache file per (url, format, provider).
-
-    Deliberately NOT the truncate-store file (keyed on URL alone), which
-    html/markdown or two providers' copies of one URL would overwrite.
-    """
+    """Dedicated cache file per (url, format, provider) — deliberately NOT the truncate-store file
+    (keyed on URL alone), which html/markdown or two providers' copies of one URL would overwrite."""
     d = _cache_dir()
     if d is None:
         return None
@@ -263,11 +254,10 @@ def _is_cache_exempt_host(url: str) -> bool:
 
 
 def _is_local_dev_url(url: str) -> bool:
-    """True for loopback/private/LAN URLs — never cached.
+    """True for loopback/private/LAN URLs — never cached: they are the user's own fast-changing dev servers.
 
-    Private-address pages are the user's own fast-changing dev servers; freshness is the point of
-    fetching them. Hostname heuristics only, no DNS: this is a freshness decision, not a security
-    boundary (SSRF enforcement lives in tools/url_safety.py, which blocks these by default anyway).
+    Hostname heuristics only, no DNS: this is a freshness decision, not a security boundary (SSRF
+    enforcement lives in tools/url_safety.py, which blocks these by default anyway).
     """
     try:
         host = _url_host(url).lower()
@@ -301,8 +291,7 @@ def extract_cache_get(url: str, format: Optional[str] = None, provider: str = ""
     if not entry or (time.time() - float(entry.get("fetched_at", 0))) >= ttl_seconds():
         return None
     try:
-        file_path = Path(entry["file"])
-        cache_root = _cache_dir()
+        file_path, cache_root = Path(entry["file"]), _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
@@ -314,11 +303,8 @@ def extract_cache_get(url: str, format: Optional[str] = None, provider: str = ""
 
 
 def extract_cache_put(url: str, content: str, title: str = "", format: Optional[str] = None, provider: str = "") -> None:
-    """Store one successful extraction's full clean text for TTL reuse.
-
-    Pages over the truncate-store ceiling are not cached: serving a capped copy
-    back as if whole would silently lose the tail.
-    """
+    """Store one successful extraction's full clean text for TTL reuse; pages over the truncate-store
+    ceiling are not cached (serving a capped copy back as if whole would silently lose the tail)."""
     if not content or not _cacheable(url):
         return
     try:
diff --git a/tools/web_tools.py b/tools/web_tools.py
index 592b5c4fc6..44b87f019e 100644
--- a/tools/web_tools.py
+++ b/tools/web_tools.py
@@ -15,20 +15,16 @@ import os
 from typing import List, Dict, Any, Optional
 import httpx  # noqa: F401 — kept at module top so tests can patch tools.web_tools.httpx
 
-# Vendor helpers re-exported so external code and unit-test patches of
-# ``tools.web_tools.<name>`` keep working after the plugin migration.
+# Vendor helpers re-exported so external code and test patches of ``tools.web_tools.<name>`` keep working.
 from plugins.web.firecrawl.provider import (  # noqa: F401 — backward-compat names
     Firecrawl, _firecrawl_backend_help_suffix, _get_firecrawl_client, _get_firecrawl_gateway_url,
     _is_tool_gateway_ready, check_firecrawl_api_key,
 )
-from plugins.web.tavily.provider import (  # noqa: F401 — backward-compat names
-    _normalize_tavily_documents, _normalize_tavily_search_results, _tavily_request,
-)
+from plugins.web.tavily.provider import _normalize_tavily_documents, _normalize_tavily_search_results, _tavily_request  # noqa: F401
 from plugins.web.parallel.provider import _get_async_parallel_client, _get_parallel_client  # noqa: F401
 from plugins.web.exa.provider import _get_exa_client  # noqa: F401
 
-# Per-vendor client cache slots. Plugins read/write these via tools.web_tools so
-# tests that reset ``tools.web_tools._<vendor>_client = None`` keep working.
+# Per-vendor client cache slots; plugins read/write these via tools.web_tools (tests reset them to None).
 _firecrawl_client: Optional[Any] = None
 _firecrawl_client_config: Optional[Any] = None
 _parallel_client: Optional[Any] = None
@@ -70,9 +66,7 @@ def _env_value(name: str) -> str:
         val = get_env_value(name)
     except Exception:
         val = None
-    if val is None:
-        val = os.getenv(name, "")
-    return (val or "").strip()
+    return ((os.getenv(name, "") if val is None else val) or "").strip()
 
 
 def _has_env(name: str) -> bool:
@@ -93,25 +87,27 @@ def _configured_backend(key: str = "backend") -> str:
     return (_load_web_config().get(key) or "").lower().strip()
 
 
-# Built-in backends probed by the hardcoded checks in _BUILTIN_AVAILABILITY. Any other name is a
-# plugin-registered provider resolved via the registry's ``is_available()``. Includes ``xai`` (probed via
-# has_xai_credentials(), not a registered provider) even though the registry's _LEGACY_PREFERENCE omits it —
-# if xai ever ships as a registered provider, drop it here.
+# Built-in backends probed by _BUILTIN_AVAILABILITY; any other name is a plugin-registered provider resolved
+# via the registry's ``is_available()``. Includes ``xai`` (probed via has_xai_credentials(), not a registered
+# provider) though the registry's _LEGACY_PREFERENCE omits it — drop it here if xai ever registers.
 _LEGACY_WEB_BACKENDS = frozenset(
     {"parallel", "firecrawl", "tavily", "exa", "searxng", "brave-free", "ddgs", "xai", "keenable"}
 )
 
 
-def _registered_web_provider(backend: str):
-    """Plugin-registered web provider by name, or ``None`` (registry lookups are never fatal)."""
-    if not backend:
-        return None
+def _registry_call(func_name: str, default, *args):
+    """``agent.web_search_registry.<func_name>(*args)``, or *default* if it raised (registry is optional, never fatal)."""
     try:
-        from agent.web_search_registry import get_provider
-        return get_provider(backend)
+        import agent.web_search_registry as registry_mod
+        return getattr(registry_mod, func_name)(*args)
     except Exception as exc:  # noqa: BLE001 — registry optional; never fatal
-        logger.debug("web provider registry lookup failed for %r: %s", backend, exc)
-        return None
+        logger.debug("web provider registry %s%r failed: %s", func_name, args, exc)
+        return default
+
+
+def _registered_web_provider(backend: str):
+    """Plugin-registered web provider by name, or ``None``."""
+    return _registry_call("get_provider", None, backend) if backend else None
 
 
 def _probe(provider, method: str, context: str = "") -> Optional[bool]:
@@ -128,37 +124,32 @@ def _probe(provider, method: str, context: str = "") -> Optional[bool]:
 
 def _list_registered_web_providers():
     """All plugin-registered web providers (empty list on failure)."""
-    try:
-        from agent.web_search_registry import list_providers
-        return list_providers()
-    except Exception as exc:  # noqa: BLE001 — registry optional; never fatal
-        logger.debug("web provider registry list failed: %s", exc)
-        return []
+    return _registry_call("list_providers", [])
 
 
 def _get_backend() -> str:
     """Shared web backend name.
 
-    A stored ``web.backend`` is returned as-is — no availability probe, no fallback — so a broken
-    selection surfaces the vendor's honest error rather than silently rerouting. The autodetect
-    ladder runs ONLY when no web selection has ever been stored.
+    A stored ``web.backend`` is returned as-is — no availability probe, no fallback — so a broken selection
+    surfaces the vendor's honest error rather than silently rerouting. Autodetect runs ONLY when no web
+    selection has ever been stored.
     """
     configured = _configured_backend()
     if configured:
-        # "nous" (managed subscription) is serviced by the firecrawl provider,
-        # whose client resolver routes it through the managed Tool Gateway.
+        # "nous" (managed subscription) is serviced by the firecrawl provider, whose client
+        # resolver routes it through the managed Tool Gateway.
         from tools.tool_backend_helpers import NOUS_MANAGED_PROVIDER
         return "firecrawl" if configured == NOUS_MANAGED_PROVIDER else configured
 
     from tools.tool_backend_helpers import selection_exists
     if selection_exists("web"):
-        # Selection exists (use_gateway / per-capability keys) but no shared
-        # name: keep the firecrawl default rather than credential-laddering.
+        # Selection exists (use_gateway / per-capability keys) but no shared name: keep the
+        # firecrawl default rather than credential-laddering.
         return "firecrawl"
 
-    # Never-configured install. Explicit user credentials beat the managed-gateway probe (a Nous
-    # OAuth token's tier may not grant web access, and the gateway then fails at runtime with no
-    # fallback). Free tiers trail paid.
+    # Never-configured install. Explicit user credentials beat the managed-gateway probe (a Nous OAuth
+    # token's tier may not grant web access; the gateway then fails at runtime with no fallback).
+    # Free tiers trail paid.
     backend_candidates = (
         ("tavily", _has_env("TAVILY_API_KEY")),
         ("exa", _has_env("EXA_API_KEY")),
@@ -210,8 +201,7 @@ def _tavily_explicitly_configured() -> bool:
 
 
 def _xai_available() -> bool:
-    # Cheap probe only (env var OR auth.json OAuth). resolve_xai_http_credentials()
-    # can trigger a network token refresh and this runs on every dispatch.
+    # Cheap probe only (env var OR auth.json OAuth): resolve_xai_http_credentials() may refresh over the network.
     try:
         from tools.xai_http import has_xai_credentials
         return has_xai_credentials()
@@ -228,8 +218,8 @@ def _ddgs_package_importable() -> bool:
         return False
 
 
-# Availability probes for the built-in backends (see _LEGACY_WEB_BACKENDS). Lambdas so tests patching
-# module-level helpers (e.g. _ddgs_package_importable, check_firecrawl_api_key) are honored at call time.
+# Built-in availability probes (see _LEGACY_WEB_BACKENDS). Lambdas so test patches of module-level helpers
+# (e.g. _ddgs_package_importable, check_firecrawl_api_key) are honored at call time.
 _BUILTIN_AVAILABILITY = {
     "exa": lambda: _has_env("EXA_API_KEY"),
     "parallel": lambda: _has_env("PARALLEL_API_KEY"),
@@ -261,9 +251,8 @@ def _is_backend_available(backend: str) -> bool:
 def _web_requires_env() -> list[str]:
     """Tool-registry metadata env vars for the web backends.
 
-    Gateway vars are always listed: gating them on ``managed_nous_tools_enabled()`` cost a synchronous
-    portal HTTP refresh at every CLI startup. Contract: set var -> tool sees it; not-logged-in users
-    simply lack the vars, so extras are harmless.
+    Gateway vars are always listed: gating them on ``managed_nous_tools_enabled()`` cost a synchronous portal
+    HTTP refresh at every CLI startup. Contract: set var -> tool sees it; extras are harmless for the not-logged-in.
     """
     return [
         "EXA_API_KEY", "PARALLEL_API_KEY", "TAVILY_API_KEY", "KEENABLE_API_KEY", "FIRECRAWL_API_KEY",
@@ -280,9 +269,8 @@ _debug = DebugSession("web_tools", env_var="WEB_TOOLS_DEBUG")
 def _ensure_web_plugins_loaded() -> None:
     """Idempotently run plugin discovery so the web registry is populated.
 
-    Dispatch is reachable from contexts that never triggered discovery (subprocess agent runs,
-    delegate children, scripts); without it the registry is empty and a configured backend yields
-    a misleading "No web ... provider configured" error.
+    Dispatch is reachable from contexts that never triggered discovery (subprocess agent runs, delegate
+    children, scripts); without it a configured backend yields a misleading "No web ... provider" error.
     """
     try:
         from hermes_cli.plugins import _ensure_plugins_discovered
@@ -308,22 +296,16 @@ def _debug_error(call_name: str, debug_call_data: dict, error_msg: str) -> str:
 def web_search_tool(query: str, limit: int = 5) -> str:
     """Search the web via the configured backend.
 
-    Returns a JSON string ``{"success": bool, "data": {"web": [{"title", "url", "description",
-    "position"}, ...]}}`` (metadata only — use web_extract_tool for page content) or
-    ``{"success": false, "error": ...}``.
+    Returns a JSON string ``{"success": bool, "data": {"web": [{"title", "url", "description", "position"},
+    ...]}}`` (metadata only — use web_extract_tool for page content) or ``{"success": false, "error": ...}``.
     """
     try:
-        limit = int(limit)
+        limit = min(max(int(limit), 1), 100)
     except (TypeError, ValueError):
         limit = 5
-    limit = min(max(limit, 1), 100)
-
     debug_call_data = {
-        "parameters": {"query": query, "limit": limit},
-        "error": None,
-        "results_count": 0,
-        "original_response_size": 0,
-        "final_response_size": 0,
+        "parameters": {"query": query, "limit": limit}, "error": None, "results_count": 0,
+        "original_response_size": 0, "final_response_size": 0,
     }
 
     try:
@@ -334,7 +316,6 @@ def web_search_tool(query: str, limit: int = 5) -> str:
         # Sync only — every provider's search() is sync.
         _ensure_web_plugins_loaded()
         from agent.web_search_registry import get_active_search_provider, get_provider as _wsp_get_provider
-
         backend = _get_search_backend()
         provider = _wsp_get_provider(backend) if backend else None
         if provider is None or not provider.supports_search():
@@ -347,12 +328,8 @@ def web_search_tool(query: str, limit: int = 5) -> str:
             provider = get_active_search_provider()
 
         if provider is None:
-            response_data = {
-                "success": False,
-                "error": _no_provider_error(
-                    "search", "No web search provider configured. Run `hermes tools` to set one up."
-                ),
-            }
+            error_text = _no_provider_error("search", "No web search provider configured. Run `hermes tools` to set one up.")
+            response_data = {"success": False, "error": error_text}
         else:
             logger.info("Web search via %s: '%s' (limit: %d)", provider.name, query, limit)
             response_data = _memoized_search(provider, query, limit)
@@ -370,13 +347,11 @@ def web_search_tool(query: str, limit: int = 5) -> str:
 def _memoized_search(provider, query: str, limit: int) -> dict:
     """TTL memo + single-flight around the paid vendor call (tools/web_result_cache.py).
 
-    Sits after every safety/config check. The provider is asked for the BUCKETED count so
-    near-identical limits share an entry; the caller's count is sliced out. Only successful,
-    non-rescued responses are cached — caching a rescue would make the one-shot ring fallback
-    sticky for a whole TTL.
+    Sits after every safety/config check. The provider is asked for the BUCKETED count so near-identical
+    limits share an entry; the caller's count is sliced out. Only successful, non-rescued responses are
+    cached — caching a rescue would make the one-shot ring fallback sticky for a whole TTL.
     """
     from tools.web_result_cache import bucket_limit, search_memo, slice_search_response
-
     def _paid_search() -> tuple[dict, bool]:
         fetch_limit = bucket_limit(limit)
         try:
@@ -404,29 +379,21 @@ def _memoized_search(provider, query: str, limit: int) -> dict:
 async def web_extract_tool(urls: List[Any], format: str = None, char_limit: Optional[int] = None) -> str:
     """Extract clean page content (no LLM) from URLs via the configured backend.
 
-    Pages over ``char_limit`` (default web.extract_char_limit or 15000) are head+tail truncated with
-    a footer pointing at the stored full text. Inline base64 images become ``[IMAGE: alt]``
-    placeholders. URLs carrying secrets are refused before any fetch; private-network URLs are
-    blocked per entry. Returns a JSON string with a ``results`` list of url/title/content/error.
+    Pages over ``char_limit`` (default web.extract_char_limit or 15000) are head+tail truncated with a footer
+    pointing at the stored full text; inline base64 images become ``[IMAGE: alt]``. URLs carrying secrets are
+    refused before any fetch; private-network URLs are blocked per entry. Returns JSON ``{"results": [...]}``.
     """
     normalized_urls, normalized_indices, invalid_urls, blocked = _validate_extract_urls(urls)
     if blocked is not None:
         return blocked
-
     debug_call_data = {
-        "parameters": {"urls": normalized_urls, "format": format, "char_limit": char_limit},
-        "error": None,
-        "pages_extracted": 0,
-        "pages_truncated": 0,
-        "original_response_size": 0,
-        "final_response_size": 0,
-        "truncation_metrics": [],
-        "processing_applied": [],
+        "parameters": {"urls": normalized_urls, "format": format, "char_limit": char_limit}, "error": None,
+        "pages_extracted": 0, "pages_truncated": 0, "original_response_size": 0, "final_response_size": 0,
+        "truncation_metrics": [], "processing_applied": [],
     }
 
     try:
         logger.info("Extracting content from %d URL(s)", len(normalized_urls))
-
         # SSRF protection — filter private/internal URLs before any backend.
         safe_urls, safe_indices = [], []
         ssrf_blocked: Dict[int, Dict[str, Any]] = {}
@@ -447,15 +414,13 @@ async def web_extract_tool(urls: List[Any], format: str = None, char_limit: Opti
                 return error_json
             results = await _extract_safe_urls(provider, safe_urls, format)
 
-        # Reconstruct input order across invalid, blocked, and provider entries
-        # (providers preserve the order of the safe URL list they receive).
+        # Reconstruct input order across invalid, blocked, and provider entries (providers preserve safe-list order).
         if invalid_urls or ssrf_blocked:
             results = _merge_in_order(len(urls), {**ssrf_blocked, **invalid_urls}, safe_indices, safe_urls, results)
 
         logger.info("Extracted content from %d pages", len(results))
         debug_call_data["pages_extracted"] = len(results)
         debug_call_data["original_response_size"] = len(json.dumps({"results": results}))
-
         debug_call_data["processing_applied"].append("truncate_and_store")
         _truncate_results(results, _effective_char_limit(char_limit), debug_call_data)
         trimmed = _trim_results(results)
@@ -463,11 +428,8 @@ async def web_extract_tool(urls: List[Any], format: str = None, char_limit: Opti
             result_json = tool_error("Content was inaccessible or not found")
         else:
             result_json = json.dumps({"results": trimmed}, indent=2, ensure_ascii=False)
-
-        # Belt-and-suspenders sweep over the serialized JSON in case a provider
-        # tucked a base64 blob somewhere unexpected (e.g. metadata).
+        # Belt-and-suspenders sweep over the serialized JSON in case a provider tucked a base64 blob in metadata.
         cleaned_result = convert_base64_images_to_links(result_json)
-
         debug_call_data["final_response_size"] = len(cleaned_result)
         debug_call_data["processing_applied"].append("base64_image_conversion")
         _finish_debug("web_extract_tool", debug_call_data)
@@ -480,10 +442,9 @@ async def web_extract_tool(urls: List[Any], format: str = None, char_limit: Opti
 def _provider_is_ready(provider) -> bool:
     """True when *provider* is keyed-available OR keyless-capable, without raising.
 
-    ``get_active_*_provider()`` returns an explicitly configured backend even when ``is_available()``
-    is False (so dispatch can emit a precise error), so readiness gates (tool check_fn, ``hermes
-    doctor``) must probe for real. Keyless mode (Exa/Parallel free tier) is a working state, not a
-    misconfig.
+    ``get_active_*_provider()`` returns an explicitly configured backend even when ``is_available()`` is False
+    (so dispatch can emit a precise error), so readiness gates (tool check_fn, ``hermes doctor``) must probe
+    for real. Keyless mode (Exa/Parallel free tier) is a working state, not a misconfig.
     """
     if provider is None:
         return False
@@ -499,8 +460,8 @@ def _provider_is_ready(provider) -> bool:
 def check_web_api_key() -> bool:
     """``check_fn`` gate for web_search / web_extract: is any web backend available?
 
-    A plugin-registered provider reporting ``is_available()`` must light the tools up even with no
-    built-in credentials; resolution funnels through :func:`_is_backend_available`.
+    A plugin-registered provider reporting ``is_available()`` must light the tools up even with no built-in
+    credentials; resolution funnels through :func:`_is_backend_available`.
     """
     configured = _configured_backend()
     if configured and _is_backend_available(configured):
@@ -508,8 +469,7 @@ def check_web_api_key() -> bool:
     # Boolean OR over built-ins — probe order is irrelevant here.
     if any(_is_backend_available(backend) for backend in _LEGACY_WEB_BACKENDS):
         return True
-    # Plugin path. Discovery must run first: check_fn fires at tool-registration
-    # time, before any dispatch has populated the registry.
+    # Plugin path. Discovery must run first: check_fn fires at tool-registration time, before any dispatch.
     try:
         _ensure_web_plugins_loaded()
         from agent.web_search_registry import get_active_search_provider, get_active_extract_provider
diff --git a/tools/web_tools_extract.py b/tools/web_tools_extract.py
index 75ea2f802c..249ec09821 100644
--- a/tools/web_tools_extract.py
+++ b/tools/web_tools_extract.py
@@ -24,8 +24,8 @@ _EXTRACT_BACKENDS_HINT = "firecrawl, tavily, keenable, exa, or parallel."
 def _web_extract_url(value: Any) -> Optional[str]:
     """URL from a model-supplied extract item (str, or dict with ``url``/``href``); None if unusable.
 
-    Models sometimes forward a whole search result instead of its URL, hence the
-    dict form. Never stringify arbitrary objects into misleading fetch targets.
+    Models sometimes forward a whole search result instead of its URL, hence the dict form. Never
+    stringify arbitrary objects into misleading fetch targets.
     """
     if isinstance(value, dict):
         value = value.get("url") or value.get("href")
@@ -52,7 +52,6 @@ def _strict_selection_error(capability: str, backend: str) -> str:
     """
     from agent.web_search_registry import _disabled_web_plugin_for
     from tools.tool_backend_helpers import selection_error
-
     disabled_key = _disabled_web_plugin_for(capability=capability)
     if disabled_key:
         return _disabled_plugin_error(capability, disabled_key)
@@ -62,7 +61,6 @@ def _strict_selection_error(capability: str, backend: str) -> str:
 def _no_provider_error(capability: str, fallback: str) -> str:
     """Error when no provider resolved: point at a disabled bundled plugin if that is the real cause."""
     from agent.web_search_registry import _disabled_web_plugin_for
-
     disabled_key = _disabled_web_plugin_for(capability=capability)
     return _disabled_plugin_error(capability, disabled_key) if disabled_key else fallback
 
@@ -75,6 +73,11 @@ def _extract_error_json(error: str) -> str:
     return json.dumps({"success": False, "error": error}, ensure_ascii=False)
 
 
+def _refuse_all(error: str):
+    """Whole-call refusal tuple for ``_validate_extract_urls`` (exfiltration prevention)."""
+    return None, None, None, json.dumps({"success": False, "error": error})
+
+
 def _merge_in_order(
     total: int, fixed: Dict[int, dict], fetch_positions: List[int], fetch_urls: List[str], results: List[dict]
 ) -> List[dict]:
@@ -95,7 +98,6 @@ def _validate_extract_urls(urls: List[Any]):
     """
     from agent.redact import _PREFIX_RE
     from urllib.parse import unquote
-
     normalized_urls: List[str] = []
     normalized_indices: List[int] = []
     invalid_urls: Dict[int, Dict[str, Any]] = {}
@@ -103,32 +105,22 @@ def _validate_extract_urls(urls: List[Any]):
         _url = _web_extract_url(item)
         if _url is None:
             invalid_urls[index] = _result_entry(
-                "",
-                f"Invalid URL item at index {index}: expected a URL string "
-                "or an object with a string 'url' or 'href' field",
+                "", f"Invalid URL item at index {index}: expected a URL string or an object with a string 'url' or 'href' field"
             )
             continue
         normalized_url = normalize_url_for_request(_url)
-        if any(
-            _PREFIX_RE.search(candidate)
-            for candidate in (_url, unquote(_url), normalized_url, unquote(normalized_url))
-        ):
-            return None, None, None, json.dumps({
-                "success": False,
-                "error": "Blocked: URL contains what appears to be an API key or token. "
-                         "Secrets must not be sent in URLs.",
-            })
+        if any(_PREFIX_RE.search(c) for c in (_url, unquote(_url), normalized_url, unquote(normalized_url))):
+            return _refuse_all(
+                "Blocked: URL contains what appears to be an API key or token. Secrets must not be sent in URLs."
+            )
         sensitive_query_key = sensitive_query_param_name(normalized_url)
         if sensitive_query_key:
-            return None, None, None, json.dumps({
-                "success": False,
-                "error": (
-                    "Blocked: URL contains a credential-like query parameter "
-                    f"({sensitive_query_key}). Web extract backends are third-party "
-                    "readers; remove the sensitive query parameter or use a local "
-                    "browser session when this access is explicitly required."
-                ),
-            })
+            return _refuse_all(
+                "Blocked: URL contains a credential-like query parameter "
+                f"({sensitive_query_key}). Web extract backends are third-party "
+                "readers; remove the sensitive query parameter or use a local "
+                "browser session when this access is explicitly required."
+            )
         normalized_urls.append(normalized_url)
         normalized_indices.append(index)
     return normalized_urls, normalized_indices, invalid_urls, None
@@ -137,12 +129,10 @@ def _validate_extract_urls(urls: List[Any]):
 def _resolve_extract_provider(backend: str):
     """Resolve the extract provider for *backend*; returns ``(provider, error_json)``.
 
-    A registered search-only backend is a typed error (never a silent switch). An unregistered
-    name with a stored web selection is a strict-selection error; with no selection, fall
-    through to the availability walk.
+    A registered search-only backend is a typed error (never a silent switch). An unregistered name with a
+    stored web selection is a strict-selection error; with no selection, fall through to the availability walk.
     """
     from agent.web_search_registry import get_active_extract_provider, get_provider as _wsp_get_provider
-
     provider = _wsp_get_provider(backend) if backend else None
     if provider is not None and provider.supports_extract():
         return provider, None
@@ -153,7 +143,6 @@ def _resolve_extract_provider(backend: str):
             "Set web.extract_backend to " + _EXTRACT_BACKENDS_HINT
         )
     from tools.tool_backend_helpers import selection_exists
-
     if backend and selection_exists("web"):
         return None, _extract_error_json(_strict_selection_error("extract", backend))
     provider = get_active_extract_provider()
@@ -167,12 +156,11 @@ def _resolve_extract_provider(backend: str):
 async def _dispatch_extract(provider, fetch_urls: List[str], format: Optional[str]) -> List[dict]:
     """Call ``provider.extract`` (async or sync-in-thread), with one-shot keyless rescue.
 
-    Rescue fires on a raised exception or when the WHOLE batch failed (backend
-    outage, not per-page problems). Rescued batches are never cached.
+    Rescue fires on a raised exception or when the WHOLE batch failed (backend outage, not per-page
+    problems). Rescued batches are never cached.
     """
     import inspect
     from tools.web_result_cache import extract_cache_put
-
     try:
         if inspect.iscoroutinefunction(provider.extract):
             results = await provider.extract(fetch_urls, format=format)
@@ -199,15 +187,13 @@ async def _dispatch_extract(provider, fetch_urls: List[str], format: Optional[st
 async def _extract_safe_urls(provider, safe_urls: List[str], format: Optional[str]) -> List[dict]:
     """Serve cache hits, fetch the rest, and merge back in ``safe_urls`` order.
 
-    The disk cache (tools/web_result_cache.py) sits AFTER the secret-URL gate, SSRF gate, and
-    provider resolution, and is gated per-URL on the website policy — a hit skips only the vendor
-    call, never a control. Policy-blocked URLs are cache misses so dispatch handles them exactly
-    as without a cache. Keys include provider and format, so switching either within the TTL
-    never serves the other's content.
+    The disk cache (tools/web_result_cache.py) sits AFTER the secret-URL gate, SSRF gate, and provider
+    resolution, and is gated per-URL on the website policy — a hit skips only the vendor call, never a control.
+    Policy-blocked URLs are cache misses so dispatch handles them exactly as without a cache. Keys include
+    provider and format, so switching either within the TTL never serves the other's content.
     """
     from tools.web_result_cache import extract_cache_get
     from tools.website_policy import check_website_access as _check_site
-
     cached_results: Dict[int, Dict[str, Any]] = {}
     fetch_urls: List[str] = []
     fetch_positions: List[int] = []
diff --git a/tools/web_tools_rescue.py b/tools/web_tools_rescue.py
index 0a0dd9f00e..5c9dde021e 100644
--- a/tools/web_tools_rescue.py
+++ b/tools/web_tools_rescue.py
@@ -20,12 +20,10 @@ _RING_KEY_VARS = {
 def _keyless_rescue_enabled() -> bool:
     """``web.keyless_rescue`` (default on), implicitly off when the keyless tier is disabled."""
     from tools.web_tools import _load_web_config
-
     if not _load_web_config().get("keyless_rescue", True):
         return False
     try:
         from agent.web_search_registry import _keyless_tier_enabled
-
         return _keyless_tier_enabled()
     except Exception as exc:  # noqa: BLE001 — registry optional
         logger.debug("keyless rescue tier check failed: %s", exc)
@@ -42,15 +40,12 @@ def _rescue_eligible(provider) -> bool:
         return False
     try:
         from plugins.web.keyless_mcp import _KEYLESS_RING, use_keyless
-
         name = getattr(provider, "name", "")
-        if name in _KEYLESS_RING:
-            from agent.web_search_provider import get_provider_env
-
-            key_var = _RING_KEY_VARS.get(name, "")
-            api_key = get_provider_env(key_var) if key_var else ""
-            return not use_keyless(name, api_key)
-        return True
+        if name not in _KEYLESS_RING:
+            return True
+        from agent.web_search_provider import get_provider_env
+        key_var = _RING_KEY_VARS.get(name, "")
+        return not use_keyless(name, get_provider_env(key_var) if key_var else "")
     except Exception as exc:  # noqa: BLE001 — rescue is best-effort
         logger.debug("rescue eligibility check failed: %s", exc)
         return False
@@ -59,17 +54,14 @@ def _rescue_eligible(provider) -> bool:
 def _rescue_search(provider_name: str, original_error: str, query: str, limit: int) -> dict:
     """Rescue a failed search via the ring; annotate the result with the original failure."""
     from plugins.web.keyless_mcp import search_with_failover
-
     logger.warning("web_search backend '%s' failed (%s); one-shot keyless rescue", provider_name, (original_error or "")[:200])
     rescued = search_with_failover(provider_name, query, limit)
     if rescued.get("success"):
         data = rescued.setdefault("data", {})
         data["rescued_from"] = provider_name
         data["backend_error"] = (
-            f"Configured backend '{provider_name}' failed this call "
-            f"({(original_error or 'unknown error')[:300]}); result served "
-            "by the keyless free tier. The next call will use "
-            f"'{provider_name}' again."
+            f"Configured backend '{provider_name}' failed this call ({(original_error or 'unknown error')[:300]}); "
+            f"result served by the keyless free tier. The next call will use '{provider_name}' again."
         )
         return rescued
     # Ring also failed: the ORIGINAL error names the user's setup, so lead with it.
@@ -92,7 +84,6 @@ def _rescue_extract(provider_name: str, urls: list, results: list) -> list:
     list replaces the batch wholesale.
     """
     from plugins.web.keyless_mcp import extract_with_failover
-
     parity = len(results) == len(urls)
     rescue_idx = [i for i, r in enumerate(results) if not parity or not _policy_blocked_result(r)]
     if not rescue_idx:
@@ -108,11 +99,10 @@ def _rescue_extract(provider_name: str, urls: list, results: list) -> list:
     if rescued and all(r.get("error", "") for r in rescued):
         return results  # rescue also failed everywhere: keep original errors
     for r in rescued:
-        if not r.get("error"):
-            meta = r.setdefault("metadata", {})
-            if isinstance(meta, dict):
-                meta["rescued_from"] = provider_name
-                meta["backend_error"] = (original_error or "")[:300]
+        meta = None if r.get("error") else r.setdefault("metadata", {})
+        if isinstance(meta, dict):
+            meta["rescued_from"] = provider_name
+            meta["backend_error"] = (original_error or "")[:300]
     if parity and len(rescued) == len(rescue_idx):
         merged = list(results)
         for pos, i in enumerate(rescue_idx):
diff --git a/tools/web_tools_truncate.py b/tools/web_tools_truncate.py
index e46800a31a..47d09e0543 100644
--- a/tools/web_tools_truncate.py
+++ b/tools/web_tools_truncate.py
@@ -14,12 +14,11 @@ from typing import Any, List, Optional
 
 logger = logging.getLogger("tools.web_tools")
 
-# Per-page char budget sent to the model (override: web.extract_char_limit).
-# Larger pages are head+tail truncated and the full text stored on disk.
+# Per-page char budget sent to the model (override: web.extract_char_limit); larger pages are
+# head+tail truncated and the full text stored on disk.
 DEFAULT_EXTRACT_CHAR_LIMIT = 15000
-
-# Ceiling on the full-text file written to cache/web so a multi-MB page can't
-# write unbounded bytes on every extract; the model only ever sees char_limit.
+# Ceiling on the full-text file written to cache/web so a multi-MB page can't write unbounded bytes
+# on every extract; the model only ever sees char_limit.
 MAX_STORED_TEXT_CHARS = 2_000_000
 
 _CHAR_LIMIT_FLOOR, _CHAR_LIMIT_CEILING = 2000, 500_000
@@ -36,7 +35,6 @@ def _clamp_char_limit(value: Any) -> int:
 def _get_extract_char_limit() -> int:
     """``web.extract_char_limit`` clamped to a sane range, else the default."""
     from tools.web_tools import _load_web_config  # lazy: tests patch tools.web_tools._load_web_config
-
     try:
         configured = _load_web_config().get("extract_char_limit")
         if configured is not None:
@@ -49,9 +47,8 @@ def _get_extract_char_limit() -> int:
 def convert_base64_images_to_links(text: str) -> str:
     """Replace inline base64 image blobs (token bombs) with ``[IMAGE: alt]`` placeholders.
 
-    Handles markdown images (alt text kept), parenthesised blobs, and bare
-    ``data:image/...;base64,`` payloads. Real http(s) markdown image links are
-    left untouched so the agent can ``web_extract`` / ``vision_analyze`` them.
+    Handles markdown images (alt text kept), parenthesised blobs, and bare ``data:image/...;base64,`` payloads.
+    Real http(s) markdown image links are left untouched so the agent can ``web_extract`` / ``vision_analyze`` them.
     """
     def _md_repl(m: "re.Match[str]") -> str:
         alt = (m.group("alt") or "").strip()
@@ -74,20 +71,15 @@ def _store_full_text(url: str, content: str) -> Optional[str]:
         import hashlib
         from hermes_constants import get_hermes_dir
         from tools.web_result_cache import _host_slug
-
         cache_dir = get_hermes_dir("cache/web", "web_cache")
         cache_dir.mkdir(parents=True, exist_ok=True)
-
-        digest = hashlib.sha256(url.encode("utf-8")).hexdigest()[:10]
-        path = cache_dir / f"{_host_slug(url)}-{digest}.md"
+        path = cache_dir / f"{_host_slug(url)}-{hashlib.sha256(url.encode('utf-8')).hexdigest()[:10]}.md"
         if len(content) > MAX_STORED_TEXT_CHARS:
-            content = (
-                content[:MAX_STORED_TEXT_CHARS]
-                + f"\n\n[... stored copy truncated at {MAX_STORED_TEXT_CHARS:,} chars "
+            content = content[:MAX_STORED_TEXT_CHARS] + (
+                f"\n\n[... stored copy truncated at {MAX_STORED_TEXT_CHARS:,} chars "
                 f"of {len(content):,}; re-extract a more specific URL for the rest ...]"
             )
         from tools.spill_safety import write_text_exclusive
-
         # Deterministic name in a well-known dir: refuse symlinks (lstat-unlink + exclusive create);
         # same-URL re-extraction legitimately overwrites. Not private: cache/web is bind-mounted
         # into remote backends' container UID.
@@ -107,12 +99,9 @@ def _truncate_with_footer(content: str, url: str, char_limit: int) -> tuple[str,
     """
     if len(content) <= char_limit:
         return content, False
-
     head_budget = int(char_limit * 0.75)
     tail_budget = char_limit - head_budget
-
-    head = content[:head_budget]
-    tail = content[-tail_budget:]
+    head, tail = content[:head_budget], content[-tail_budget:]
     # Snap both cuts to line boundaries (head back, tail forward) so we never slice mid-line.
     nl = head.rfind("\n")
     if nl > head_budget * 0.5:
@@ -125,18 +114,15 @@ def _truncate_with_footer(content: str, url: str, char_limit: int) -> tuple[str,
     footer_lines = [
         "",
         "─" * 8 + " [TRUNCATED] " + "─" * 8,
-        f"Showing {len(head):,} chars (head) + {len(tail):,} chars (tail) "
-        f"of {len(content):,} total clean characters.",
+        f"Showing {len(head):,} chars (head) + {len(tail):,} chars (tail) of {len(content):,} total clean characters.",
     ]
     if stored_path:
         # read_file is 1-indexed; +2 lands on the first line after the shown head.
-        middle_start_line = head.count("\n") + 2
-        footer_lines.append(f"Full text saved to: {stored_path}")
-        footer_lines.append(
-            f'To read the omitted middle: read_file path="{stored_path}" '
-            f"offset={middle_start_line} limit=200  (the file is the complete page; "
-            f"raise/lower offset to page through it)."
-        )
+        footer_lines += [
+            f"Full text saved to: {stored_path}",
+            f'To read the omitted middle: read_file path="{stored_path}" offset={head.count(chr(10)) + 2} limit=200  '
+            "(the file is the complete page; raise/lower offset to page through it).",
+        ]
     else:
         footer_lines.append(
             "Full text could not be stored; re-run web_extract on a more "
@@ -182,10 +168,7 @@ def _trim_results(results: List[dict]) -> List[dict]:
     """Keep only url/title/content/error per entry (+ blocked_by_policy when present)."""
     return [
         {
-            "url": r.get("url", ""),
-            "title": r.get("title", ""),
-            "content": r.get("content", ""),
-            "error": r.get("error"),
+            "url": r.get("url", ""), "title": r.get("title", ""), "content": r.get("content", ""), "error": r.get("error"),
             **({"blocked_by_policy": r["blocked_by_policy"]} if "blocked_by_policy" in r else {}),
         }
         for r in results
diff --git a/tools/website_policy.py b/tools/website_policy.py
index 4efbf13a84..49da31f37a 100644
--- a/tools/website_policy.py
+++ b/tools/website_policy.py
@@ -1,9 +1,8 @@
 """Website access policy helpers for URL-capable tools.
 
-Loads a user-managed website blocklist (``security.website_blocklist`` in ~/.hermes/config.yaml
-plus optional shared list files) without pulling in the heavier CLI config stack. The parsed
-policy is cached with a short TTL so config edits take effect quickly without re-parsing YAML on
-every URL check.
+Loads a user-managed website blocklist (``security.website_blocklist`` in ~/.hermes/config.yaml plus
+optional shared list files) without the heavier CLI config stack. The parsed policy is cached with a
+short TTL so config edits take effect quickly without re-parsing YAML on every URL check.
 """
 
 from __future__ import annotations
@@ -31,10 +30,6 @@ _cached_policy_path: Optional[str] = None
 _cached_policy_time: float = 0.0
 
 
-def _get_default_config_path() -> Path:
-    return get_hermes_home() / "config.yaml"
-
-
 class WebsitePolicyError(Exception):
     """Raised when a website policy file is malformed."""
 
@@ -86,7 +81,6 @@ def _load_policy_config(config_path: Path) -> Dict[str, Any]:
     except ImportError:
         logger.debug("PyYAML not installed — website blocklist disabled")
         return dict(_DEFAULT_WEBSITE_BLOCKLIST)
-
     try:
         with open(config_path, encoding="utf-8") as f:
             config = yaml.safe_load(f) or {}
@@ -113,24 +107,16 @@ def _require_type(policy: Dict[str, Any], key: str, kind: type, default: Any) ->
 
 
 def load_website_blocklist(config_path: Optional[Path] = None) -> Dict[str, Any]:
-    """Load and return the parsed website blocklist policy (``{"enabled", "rules"}``).
-
-    Cached for ``_CACHE_TTL_SECONDS`` for the default config path only; an explicit
-    ``config_path`` (tests) always bypasses and never populates the cache.
-    """
+    """Parsed website blocklist policy (``{"enabled", "rules"}``); cached for ``_CACHE_TTL_SECONDS`` for the
+    default config path only — an explicit ``config_path`` (tests) always bypasses and never populates the cache."""
     global _cached_policy, _cached_policy_path, _cached_policy_time
-
-    default_path = _get_default_config_path()
+    default_path = get_hermes_home() / "config.yaml"
     resolved_path = str(config_path or default_path)
     now = time.monotonic()
-
     if config_path is None:
         with _cache_lock:
-            if (
-                _cached_policy is not None
-                and _cached_policy_path == resolved_path
-                and (now - _cached_policy_time) < _CACHE_TTL_SECONDS
-            ):
+            fresh = _cached_policy_path == resolved_path and (now - _cached_policy_time) < _CACHE_TTL_SECONDS
+            if _cached_policy is not None and fresh:
                 return _cached_policy
 
     config_path = config_path or default_path
@@ -139,26 +125,16 @@ def load_website_blocklist(config_path: Optional[Path] = None) -> Dict[str, Any]
     raw_shared_files = _require_type(policy, "shared_files", list, [])
     enabled = _require_type(policy, "enabled", bool, True)
 
-    rules: List[Dict[str, str]] = []
-    seen: set[Tuple[str, str]] = set()
-
-    def _add(pattern: Optional[str], source: str) -> None:
-        if pattern and (source, pattern) not in seen:
-            rules.append({"pattern": pattern, "source": source})
-            seen.add((source, pattern))
-
-    for raw_rule in raw_domains:
-        _add(_normalize_rule(raw_rule), "config")
+    pairs: List[Tuple[str, str]] = [(p, "config") for p in map(_normalize_rule, raw_domains) if p]
     for shared_file in raw_shared_files:
         if not isinstance(shared_file, str) or not shared_file.strip():
             continue
         path = Path(shared_file).expanduser()
         if not path.is_absolute():
             path = (get_hermes_home() / path).resolve()
-        for normalized in _iter_blocklist_file_rules(path):
-            _add(normalized, str(path))
-
-    result = {"enabled": enabled, "rules": rules}
+        pairs += [(normalized, str(path)) for normalized in _iter_blocklist_file_rules(path)]
+    # dict.fromkeys dedupes (pattern, source) while keeping first-seen order.
+    result = {"enabled": enabled, "rules": [{"pattern": p, "source": s} for p, s in dict.fromkeys(pairs)]}
     if config_path == default_path:  # explicit paths are tests — never cache them
         with _cache_lock:
             _cached_policy, _cached_policy_path, _cached_policy_time = result, resolved_path, now
@@ -179,17 +155,16 @@ def _extract_host_from_urlish(url: str) -> str:
     parsed = urlparse(url)
     host = _normalize_host(parsed.hostname or parsed.netloc)
     if not host and "://" not in url:
-        schemeless = urlparse(f"//{url}")
-        host = _normalize_host(schemeless.hostname or schemeless.netloc)
+        parsed = urlparse(f"//{url}")
+        host = _normalize_host(parsed.hostname or parsed.netloc)
     return host
 
 
 def check_website_access(url: str, config_path: Optional[Path] = None) -> Optional[Dict[str, str]]:
-    """Check whether a URL is allowed by the website blocklist policy.
+    """``None`` if the URL is allowed by the blocklist policy, else block metadata (host/rule/source/message).
 
-    Returns ``None`` if allowed, else block metadata (``host``, ``rule``, ``source``, ``message``).
-    Fails open on policy errors (warn + ``None``) so a config typo can't break all web tools —
-    except with an explicit ``config_path`` (tests), where errors propagate.
+    Fails open on policy errors (warn + ``None``) so a config typo can't break all web tools — except with
+    an explicit ``config_path`` (tests), where errors propagate.
     """
     # Fast path: cached policy disabled/empty → no YAML read, no host extraction.
     if config_path is None:
@@ -220,10 +195,7 @@ def check_website_access(url: str, config_path: Optional[Path] = None) -> Option
             source = rule.get("source", "config")
             logger.info("Blocked URL %s — matched rule '%s' from %s", url, pattern, source)
             return {
-                "url": url,
-                "host": host,
-                "rule": pattern,
-                "source": source,
+                "url": url, "host": host, "rule": pattern, "source": source,
                 "message": f"Blocked by website policy: '{host}' matched rule '{pattern}' from {source}",
             }
     return None