From c7bae88fcee34bf8cf50c7a0eb72b1a135958a05 Mon Sep 17 00:00:00 2001
From: Teknium <127238744+teknium1@users.noreply.github.com>
Date: Wed, 2 Sep 2026 22:18:08 -0700
Subject: [PATCH 1/6] =?UTF-8?q?refactor(tools):=20tool=5Fsearch=20?=
 =?UTF-8?q?=E2=80=94=20drop=20dead=20build=5Fcatalog=5Flisting/=5Fdescribe?=
 =?UTF-8?q?=5Fclassification/=5Fsafe=5Fint,=20fold=20describe=20classifica?=
 =?UTF-8?q?tion=20into=20is=5Fdeferrable=5Ftool=5Fname,=20compact=20docstr?=
 =?UTF-8?q?ings?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 tools/tool_search.py            | 187 ++++++++------------------------
 tools/tool_search_catalog.py    |  55 ++--------
 tools/tool_search_validation.py |  41 ++-----
 3 files changed, 68 insertions(+), 215 deletions(-)

diff --git a/tools/tool_search.py b/tools/tool_search.py
index 64d1efaba4..a10952f1e4 100644
--- a/tools/tool_search.py
+++ b/tools/tool_search.py
@@ -1,19 +1,14 @@
-"""Progressive tool disclosure ("tool search"): MCP/plugin tools (and a
-curated set of event-triggered core tools) are replaced in the model-visible
-array by three bridge tools — tool_search / tool_describe / tool_call.
+"""Progressive tool disclosure ("tool search"): MCP/plugin tools and a curated set
+of event-triggered core tools are replaced in the model-visible array by three
+bridge tools — tool_search / tool_describe / tool_call.
 
-Invariants:
-* Working-set core tools (``toolsets._HERMES_CORE_TOOLS``) and session-gated GUI
-  toolsets never defer unless named in the ``defer`` list.
-* Tiered disclosure: ANY deferrable tool activates the bridge (tier 1 = bridge +
-  listing that fits ``min(threshold_pct% of context, listing_max_tokens)``,
-  tier 2 = bare bridge + per-server summary). The listing scales, not activation.
-* The catalog is stateless — rebuilt from the live tool-defs list on every
-  assembly. A session-keyed catalog drifts from the registry and silently
-  drops tools.
-* Bridge calls route through ``model_tools.handle_function_call`` so guardrails,
-  hooks, approvals, and truncation fire identically; display/trajectory unwrap
-  always shows the underlying tool.
+Invariants: working-set core tools (``toolsets._HERMES_CORE_TOOLS``) and
+session-gated GUI toolsets never defer unless named in ``defer``; ANY deferrable
+tool activates the bridge (the listing scales with budget, not activation); the
+catalog is stateless — rebuilt from the live tool-defs on every assembly (a
+session-keyed catalog drifts from the registry and silently drops tools); bridge
+calls route through ``model_tools.handle_function_call`` so guardrails, hooks,
+approvals and truncation fire identically.
 """
 
 from __future__ import annotations
@@ -23,7 +18,7 @@ import logging
 import math
 from collections import Counter
 from dataclasses import dataclass
-from typing import Any, Dict, Iterable, List, Literal, Optional, Tuple
+from typing import Any, Dict, Iterable, List, Optional, Tuple
 
 from tools.registry import tool_error
 from tools.tool_search_names import (  # noqa: F401 — re-exported public names
@@ -35,7 +30,6 @@ from tools.tool_search_names import (  # noqa: F401 — re-exported public names
 from tools.tool_search_catalog import (  # noqa: F401 — re-exported public/test names
     CHARS_PER_TOKEN,
     CatalogEntry,
-    _classify_source,
     _corpus_stats,
     _entry_search_text,
     _fn,
@@ -45,30 +39,18 @@ from tools.tool_search_catalog import (  # noqa: F401 — re-exported public/tes
     _stem,
     _tokenize,
     build_catalog,
-    build_catalog_listing,
     build_catalog_listing_with_form,
     search_catalog,
 )
-
-from tools.tool_search_validation import (  # noqa: F401 — re-exported public/test names
-    _schema_for_local_validation,
-    _schema_has_external_ref,
-    validate_deferred_call_args,
-)
+from tools.tool_search_validation import validate_deferred_call_args  # noqa: F401
 
 logger = logging.getLogger("tools.tool_search")
 
-# Bound the work one tool_search bridge call can request.
+# Bound the work one bridge call can request.
 _MAX_QUERIES_PER_CALL = 10
-# Bound the work one tool_describe bridge call can request.
 _MAX_DESCRIBE_NAMES_PER_CALL = 10
 
 
-# ---------------------------------------------------------------------------
-# Configuration plumbing
-# ---------------------------------------------------------------------------
-
-
 @dataclass(frozen=True)
 class ToolSearchConfig:
     """Resolved, validated tool-search configuration for a single assembly."""
@@ -79,9 +61,8 @@ class ToolSearchConfig:
     threshold_pct: float  # 0..100
     search_default_limit: int
     max_search_limit: int
-    # Embedded name + short-description manifest so deferred tools stay
-    # DISCOVERABLE (like the skills listing) while full schemas stay deferred.
-    # "auto"/"on" = include when it fits (names-only, then bare); "off" = bare bridge.
+    # Embedded name + short-description manifest keeping deferred tools
+    # DISCOVERABLE. "auto"/"on" = include when it fits; "off" = bare bridge.
     listing: str = "auto"  # "auto" | "on" | "off"
     # Effective budget = min(listing_max_tokens, threshold_pct% of context).
     listing_max_tokens: int = 4000
@@ -95,13 +76,11 @@ class ToolSearchConfig:
 
     @classmethod
     def from_raw(cls, raw: Any) -> "ToolSearchConfig":
-        """Build a config from a raw dict / legacy bool / None. Every field is
-        validated and clamped; unknown values fall back to safe defaults rather
-        than raising, so a typo in user config cannot break the agent."""
+        """Build from a raw dict / legacy bool / None. Every field is clamped and
+        unknown values fall back to safe defaults — a config typo must not break the agent."""
         if not isinstance(raw, dict):
             return cls(enabled="off" if raw is False else "auto", threshold_pct=5.0,
                        search_default_limit=5, max_search_limit=25)
-
         max_search_limit = _clamped_int(raw.get("max_search_limit"), 25, 1, 50)
         defer_raw = raw.get("defer")
         return cls(
@@ -130,16 +109,13 @@ def _tri_state(value: Any) -> str:
     return text if text in ("auto", "on", "off") else "auto"
 
 
-def _safe_int(value: Any, fallback: int) -> int:
-    try:
-        return int(value)
-    except (TypeError, ValueError):
-        return fallback
-
-
 def _clamped_int(value: Any, fallback: int, lo: int, hi: int) -> int:
-    """``_safe_int`` clamped to ``[lo, hi]`` (the fallback is clamped too)."""
-    return max(lo, min(hi, _safe_int(value, fallback)))
+    """``int(value)`` (or ``fallback`` when unparseable) clamped to ``[lo, hi]``."""
+    try:
+        value = int(value)
+    except (TypeError, ValueError):
+        value = fallback
+    return max(lo, min(hi, value))
 
 
 def _safe_float(value: Any, fallback: float) -> float:
@@ -170,11 +146,6 @@ def load_config_readonly() -> ToolSearchConfig:
     return _config_from_loader("load_config_readonly")
 
 
-# ---------------------------------------------------------------------------
-# Tool classification
-# ---------------------------------------------------------------------------
-
-
 def _core_tool_names() -> frozenset[str]:
     """Tool names that never defer by default (lazy import: ``toolsets`` imports
     ``tools.registry``, so a module-level import would be a cycle)."""
@@ -190,9 +161,8 @@ def _core_tool_names() -> frozenset[str]:
 _DIRECT_SURFACE_TOOLSETS = frozenset({"desktop_ui", "project"})
 
 # Curated event-triggered core tools deferred BY DEFAULT — reached for when
-# something specific happens, not every-turn working-set tools, so a catalog
-# stub suffices. ``tools.tool_search.defer`` replaces this list wholesale
-# ([] = legacy everything-eager). Names are POST-rename.
+# something specific happens, so a catalog stub suffices. ``tools.tool_search.defer``
+# replaces this list wholesale ([] = legacy everything-eager). Names are POST-rename.
 # ``clarify`` is deliberately NOT here: A/B showed deferring it collapsed
 # structured-clarify usage (18/18 -> 7/18) — the ask-the-user affordance must
 # be ambient to fire; a stub is not enough.
@@ -225,25 +195,6 @@ def is_deferrable_tool_name(name: str, defer_tools: Optional[frozenset] = None)
         return False
 
 
-def _describe_classification(
-    name: str,
-    defer_tools: Optional[frozenset] = None,
-) -> Literal["available", "not_found", "not_deferrable"]:
-    """Classify a describe name without treating unknown names as errors."""
-    entry = _registry_entry(name)
-    if entry is None:
-        return "not_found"
-    if defer_tools is not None and name in defer_tools:
-        return "available"
-    if (
-        name in BRIDGE_TOOL_NAMES
-        or name in _core_tool_names()
-        or entry.toolset in _DIRECT_SURFACE_TOOLSETS
-    ):
-        return "not_deferrable"
-    return "available"
-
-
 def _tool_def_names(tool_defs: Iterable[Dict[str, Any]]) -> Iterable[str]:
     """Function names of a tool-defs list (``""`` for a nameless def)."""
     return (_fn(td).get("name", "") for td in tool_defs)
@@ -270,11 +221,6 @@ def _deferrable_in(tool_defs: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
     return classify_tools(tool_defs, load_config_readonly().effective_defer_tools)[1]
 
 
-# ---------------------------------------------------------------------------
-# Token estimation and threshold gate
-# ---------------------------------------------------------------------------
-
-
 def estimate_tokens_from_schemas(tool_defs: Iterable[Dict[str, Any]]) -> int:
     """Token cost of a tool-defs list via the chars/4 rule (order-of-magnitude
     precision is all the activation gate needs)."""
@@ -295,15 +241,12 @@ def should_activate(
     """``"off"`` never activates; ``"on"``/``"auto"`` activate whenever any
     deferrable tool exists. ``"auto"`` is an alias of ``"on"`` reserved for a
     future budget-gated mode — do not distinguish them without that design.
-    ``context_length`` is kept in the signature for caller compatibility; the
+    ``context_length`` stays in the signature for caller compatibility; the
     threshold governs the listing budget, not activation."""
     return config.enabled != "off" and deferrable_tokens > 0
 
 
-def listing_token_budget(
-    config: ToolSearchConfig,
-    context_length: Optional[int],
-) -> int:
+def listing_token_budget(config: ToolSearchConfig, context_length: Optional[int]) -> int:
     """``min(listing_max_tokens, threshold_pct% of context)``; unknown context
     uses a 10K percentage leg (5% of a typical 200K window)."""
     if context_length and context_length > 0:
@@ -313,11 +256,6 @@ def listing_token_budget(
     return max(0, min(config.listing_max_tokens, pct_leg))
 
 
-# ---------------------------------------------------------------------------
-# Catalog + BM25 retrieval
-# ---------------------------------------------------------------------------
-
-
 def _bridge_schema(name: str, description: str, properties: Dict[str, Any],
                    required: List[str]) -> Dict[str, Any]:
     """One OpenAI-style function schema (key order is part of the frozen bytes)."""
@@ -375,10 +313,10 @@ def bridge_tool_schemas(
     listing_form: str = "",
 ) -> List[Dict[str, Any]]:
     """Bridge tool schemas injected in place of deferred tools. Kept short —
-    every byte is paid on every turn. ``listing`` is embedded in the
-    tool_search description; ``listing_form`` picks the framing (per-tool forms
-    say "skip search when you see the exact name", the "groups" summary says
-    which domains exist and that search is mandatory)."""
+    every byte is paid on every turn. ``listing`` is embedded in the tool_search
+    description; ``listing_form`` picks the framing (per-tool forms say "skip
+    search when you see the exact name", "groups" says which domains exist and
+    that search is mandatory)."""
     return [
         _bridge_schema(
             TOOL_SEARCH_NAME,
@@ -427,14 +365,9 @@ def bridge_tool_schemas(
     ]
 
 
-# ---------------------------------------------------------------------------
-# Public entry point: assemble tool-defs with optional tool search
-# ---------------------------------------------------------------------------
-
-
 @dataclass
 class AssemblyResult:
-    """Outcome of one assembly. Useful for tests and observability."""
+    """Outcome of one assembly (tests and observability)."""
 
     tool_defs: List[Dict[str, Any]]
     activated: bool
@@ -442,8 +375,7 @@ class AssemblyResult:
     deferred_tokens: int = 0
     threshold_tokens: int = 0
     # 0 = passthrough; 1 = bridge + per-tool listing (full/names/mixed);
-    # 2 = bare bridge / server-summary only — domains stay visible but
-    # individual tools are reachable only via tool_search.
+    # 2 = bare bridge / server-summary only (tools reachable only via tool_search).
     tier: int = 0
     listing_form: str = "none"  # "full" | "names" | "mixed" | "groups" | "none"
 
@@ -459,10 +391,8 @@ def assemble_tool_defs(
     tools. Idempotent — bridge tools already present are stripped first."""
     if config is None:
         config = load_config()
-
     incoming = [td for td, name in zip(tool_defs, _tool_def_names(tool_defs))
                 if name not in BRIDGE_TOOL_NAMES]
-
     visible, deferrable = classify_tools(incoming, config.effective_defer_tools)
     if not deferrable:
         return AssemblyResult(tool_defs=incoming, activated=False)
@@ -470,9 +400,7 @@ def assemble_tool_defs(
     deferrable_tokens = estimate_tokens_from_schemas(deferrable)
     if not should_activate(config, deferrable_tokens, context_length):
         return AssemblyResult(
-            tool_defs=incoming,
-            activated=False,
-            deferred_count=len(deferrable),
+            tool_defs=incoming, activated=False, deferred_count=len(deferrable),
             deferred_tokens=deferrable_tokens,
             threshold_tokens=int((context_length or 0) * (config.threshold_pct / 100.0)),
             tier=0,
@@ -483,33 +411,20 @@ def assemble_tool_defs(
     if config.listing != "off":
         listing, listing_form = build_catalog_listing_with_form(
             deferrable, max_tokens=listing_budget)
-    bridge = bridge_tool_schemas(len(deferrable), listing=listing,
-                                 listing_form=listing_form)
+    bridge = bridge_tool_schemas(len(deferrable), listing=listing, listing_form=listing_form)
     tier = 1 if listing_form in ("full", "names", "mixed") else 2
-
     logger.info(
         "tool_search activated (tier %d): %d core/visible tools kept, %d deferred "
         "(~%d tokens), listing %s (budget ~%d tokens)",
-        tier, len(visible), len(deferrable), deferrable_tokens,
-        listing_form, listing_budget,
+        tier, len(visible), len(deferrable), deferrable_tokens, listing_form, listing_budget,
     )
-
     return AssemblyResult(
-        tool_defs=visible + bridge,
-        activated=True,
-        deferred_count=len(deferrable),
-        deferred_tokens=deferrable_tokens,
-        threshold_tokens=listing_budget,
-        tier=tier,
-        listing_form=listing_form,
+        tool_defs=visible + bridge, activated=True, deferred_count=len(deferrable),
+        deferred_tokens=deferrable_tokens, threshold_tokens=listing_budget,
+        tier=tier, listing_form=listing_form,
     )
 
 
-# ---------------------------------------------------------------------------
-# Bridge tool dispatch
-# ---------------------------------------------------------------------------
-
-
 def is_bridge_tool(name: str) -> bool:
     return name in BRIDGE_TOOL_NAMES
 
@@ -534,8 +449,7 @@ def _shared_tool_record(entry: CatalogEntry) -> Dict[str, Any]:
 
 def _available_source_summary(catalog: List[CatalogEntry]) -> List[Dict[str, Any]]:
     """Deterministic ``[{name, tool_count}]`` of connected sources, attached to
-    empty query groups so a lexical miss is not read as a missing capability
-    (adds nothing to the fixed per-turn prompt)."""
+    empty query groups so a lexical miss is not read as a missing capability."""
     counts = Counter(_listing_group_label(entry.source_name) for entry in catalog)
     return [{"name": name, "tool_count": counts[name]} for name in sorted(counts)]
 
@@ -545,8 +459,7 @@ def _string_list_arg(
 ) -> Tuple[Optional[List[str]], Optional[str]]:
     """Read a list-of-strings bridge argument -> ``(items, error_json)``. A bare
     string (a common model slip) counts as a one-item list. Rejects non-list
-    input, an empty list (after stripping blanks), and more than ``max_items``
-    (bounds the work one bridge call can request)."""
+    input, an empty list (after stripping blanks), and more than ``max_items``."""
     raw = args.get(key)
     if isinstance(raw, str):
         raw = [raw]
@@ -581,21 +494,16 @@ def dispatch_tool_search(args: Dict[str, Any],
     """
     if config is None:
         config = load_config()
-
     queries, err = _string_list_arg(
         args, "queries", dedupe=False, max_items=_MAX_QUERIES_PER_CALL,
         retry_hint="Retry with fewer, more targeted queries.")
     if err:
         return err
-
     raw_limit = args.get("limit")
-    if raw_limit is None:
-        limit = config.search_default_limit
-    else:
-        limit = _clamped_int(raw_limit, config.search_default_limit, 1, config.max_search_limit)
+    limit = (config.search_default_limit if raw_limit is None
+             else _clamped_int(raw_limit, config.search_default_limit, 1, config.max_search_limit))
 
     catalog = build_catalog(_deferrable_in(current_tool_defs))
-
     results: List[Dict[str, Any]] = []
     tools_map: Dict[str, Dict[str, Any]] = {}
     corpus_stats = _corpus_stats(catalog)
@@ -614,7 +522,6 @@ def dispatch_tool_search(args: Dict[str, Any],
                 "object before concluding the capability is unavailable."
             )
         results.append(group)
-
     return json.dumps({
         "queries": queries,
         "total_available": len(catalog),
@@ -637,7 +544,6 @@ def dispatch_tool_describe(args: Dict[str, Any],
     """
     if config is None:
         config = load_config_readonly()
-
     names, err = _string_list_arg(
         args, "names", dedupe=True, max_items=_MAX_DESCRIBE_NAMES_PER_CALL,
         retry_hint="Retry with fewer names per call.")
@@ -646,7 +552,6 @@ def dispatch_tool_describe(args: Dict[str, Any],
 
     deferrable = _deferrable_in(current_tool_defs)
     by_name = {name: _fn(td) for td, name in zip(deferrable, _tool_def_names(deferrable)) if name}
-
     tools: Dict[str, Dict[str, Any]] = {}
     not_found: List[str] = []
     errors: Dict[str, str] = {}
@@ -657,9 +562,10 @@ def dispatch_tool_describe(args: Dict[str, Any],
                 "description": fn.get("description", ""),
                 "parameters": fn.get("parameters", {}),
             }
-        elif _describe_classification(
+        elif _registry_entry(name) is not None and not is_deferrable_tool_name(
             name, load_config_readonly().effective_defer_tools
-        ) == "not_deferrable":
+        ):
+            # Registered but bridge/core/GUI-surface: a real name, wrong door.
             errors[name] = (
                 f"'{name}' is not a deferrable tool. If you see it in the tools list "
                 "already, call it directly; otherwise check the spelling against tool_search."
@@ -729,7 +635,6 @@ __all__ = [
     "estimate_tokens_from_schemas",
     "should_activate",
     "build_catalog",
-    "build_catalog_listing",
     "build_catalog_listing_with_form",
     "listing_token_budget",
     "search_catalog",
diff --git a/tools/tool_search_catalog.py b/tools/tool_search_catalog.py
index d745fd2bed..f371d047f9 100644
--- a/tools/tool_search_catalog.py
+++ b/tools/tool_search_catalog.py
@@ -42,8 +42,7 @@ _thread_local = threading.local()
 def _stemmer() -> Any:
     st = getattr(_thread_local, "stemmer", None)
     if st is None:
-        st = snowballstemmer.stemmer("english")
-        _thread_local.stemmer = st
+        st = _thread_local.stemmer = snowballstemmer.stemmer("english")
     return st
 
 
@@ -54,10 +53,8 @@ def _stem(token: str) -> str:
 
 
 def _tokenize(text: str) -> List[str]:
-    """Lowercase alphanumeric tokens, Snowball-stemmed (English).
-
-    Shared by the index path and the query path so "issues" matches ``create_issue``.
-    """
+    """Lowercase alphanumeric tokens, Snowball-stemmed (English). Shared by the
+    index and query paths so "issues" matches ``create_issue``."""
     if not text:
         return []
     return [_stem(token.lower()) for token in _TOKEN_RE.findall(text)]
@@ -84,8 +81,7 @@ def _entry_search_text(td: Dict[str, Any], source_label: str = "") -> str:
     top-level parameter names. Schema bodies are excluded (noise, no recall
     gain). The ``mcp__`` prefix is dropped — it is in every MCP document, so
     its IDF is ~0. The source label lets a service-name query ("linear") reach
-    a tool whose own name omits the vendor.
-    """
+    a tool whose own name omits the vendor."""
     fn = _fn(td)
     name = fn.get("name", "")
     if name.startswith("mcp__"):
@@ -135,8 +131,8 @@ def _bm25_score(query_tokens: List[str], doc_tokens: List[str],
                 doc_lengths: List[int], avg_dl: float,
                 doc_freq: Dict[str, int], n_docs: int,
                 k1: float = 1.5, b: float = 0.75) -> float:
-    """Standard BM25 score for one query against one document (inlined; the
-    catalog is bounded — typically < 500 tools — so a dependency is not worth it)."""
+    """Standard BM25 for one query against one document (inlined; the catalog
+    is bounded — typically < 500 tools — so a dependency is not worth it)."""
     if not doc_tokens:
         return 0.0
     score = 0.0
@@ -144,12 +140,10 @@ def _bm25_score(query_tokens: List[str], doc_tokens: List[str],
     doc_tf = Counter(doc_tokens)
     for q in query_tokens:
         df = doc_freq.get(q, 0)
-        if df == 0:
+        tf = doc_tf.get(q, 0)
+        if df == 0 or tf == 0:
             continue
         idf = math.log(1 + (n_docs - df + 0.5) / (df + 0.5))
-        tf = doc_tf.get(q, 0)
-        if tf == 0:
-            continue
         norm = tf * (k1 + 1) / (tf + k1 * (1 - b + b * dl / max(avg_dl, 1.0)))
         score += idf * norm
     return score
@@ -178,17 +172,14 @@ def search_catalog(
     """Top-``limit`` catalog entries for ``query`` by BM25 (exact name match
     ranks first). Falls back to a name-substring match only when NO query
     token appears in any document (e.g. "hub" vs ``github_*``); the IDF
-    variant is strictly positive, so a hit anywhere suppresses the fallback.
-    """
+    variant is strictly positive, so a hit anywhere suppresses the fallback."""
     if not catalog or limit <= 0:
         return []
     query_tokens = _tokenize(query)
     if not query_tokens:
         return []
-
     if corpus_stats is None:
         corpus_stats = _corpus_stats(catalog)
-    doc_lengths, avg_dl, doc_freq, n_docs = corpus_stats
 
     scored: List[Tuple[float, CatalogEntry]] = []
     exact_name = query.strip().lower()
@@ -196,17 +187,12 @@ def search_catalog(
         if entry.name.lower() == exact_name:
             scored.append((float("inf"), entry))
             continue
-        s = _bm25_score(query_tokens, entry._tokens, doc_lengths, avg_dl,
-                        doc_freq, n_docs)
+        s = _bm25_score(query_tokens, entry._tokens, *corpus_stats)
         if s > 0:
             scored.append((s, entry))
-
     if not scored:
         ql = query.lower()
-        for entry in catalog:
-            if ql in entry.name.lower():
-                scored.append((0.1, entry))
-
+        scored = [(0.1, entry) for entry in catalog if ql in entry.name.lower()]
     scored.sort(key=lambda x: x[0], reverse=True)
     return [e for _, e in scored[:limit]]
 
@@ -239,16 +225,6 @@ def _listing_group_label(source_name: str) -> str:
     return label[4:] if label.startswith("mcp-") else label
 
 
-def build_catalog_listing(
-    deferrable: List[Dict[str, Any]],
-    *,
-    max_tokens: int = 4000,
-) -> Optional[str]:
-    """Render the deferred-catalog manifest; text only (see
-    :func:`build_catalog_listing_with_form` for the degradation ladder)."""
-    return build_catalog_listing_with_form(deferrable, max_tokens=max_tokens)[0]
-
-
 def build_catalog_listing_with_form(
     deferrable: List[Dict[str, Any]],
     *,
@@ -267,9 +243,6 @@ def build_catalog_listing_with_form(
     Degradation is PER SERVER (largest first): one huge server must not cost
     a small co-attached server its listing.
     """
-    if not deferrable:
-        return None, "none"
-
     groups: Dict[str, List[Tuple[str, str]]] = {}
     for td in deferrable:
         fn = _fn(td)
@@ -280,7 +253,6 @@ def build_catalog_listing_with_form(
         # ``_listing_group_label("")`` is "other", so one call covers both.
         label = _listing_group_label(_classify_source(name)[1])
         groups.setdefault(label, []).append((name, _short_desc(fn.get("description", ""))))
-
     if not groups:
         return None, "none"
 
@@ -310,10 +282,8 @@ def build_catalog_listing_with_form(
         text = assemble_if_fits(modes)
         if text is not None:
             return text, mode
-
     # 3. Per-server degradation: collapse the LARGEST rendered groups first
-    #    (deterministic: size then label) so one oversized server does not
-    #    cost a small co-attached server its per-tool names.
+    #    (deterministic: size then label).
     by_size = sorted(groups, key=lambda lbl: (-len(render_group(lbl, "names")), lbl))
     for lbl in by_size:
         modes[lbl] = "summary"
@@ -321,5 +291,4 @@ def build_catalog_listing_with_form(
         if text is not None:
             form = "groups" if all(m == "summary" for m in modes.values()) else "mixed"
             return text, form
-
     return None, "none"
diff --git a/tools/tool_search_validation.py b/tools/tool_search_validation.py
index c5ee9ba401..39d8658253 100644
--- a/tools/tool_search_validation.py
+++ b/tools/tool_search_validation.py
@@ -22,20 +22,13 @@ def _schema_for_local_validation(node: Any) -> Any:
         return [_schema_for_local_validation(item) for item in node]
     if not isinstance(node, dict):
         return node
-
-    normalized = {}
-    for key, value in node.items():
-        if key == "nullable":
-            continue
-        # Literal keywords hold instance data, not schemas: copy byte-for-byte.
-        normalized[key] = (
-            copy.deepcopy(value)
-            if key in _SCHEMA_LITERAL_KEYS
-            else _schema_for_local_validation(value)
-        )
+    # Literal keywords hold instance data, not schemas: copy byte-for-byte.
+    normalized = {
+        key: copy.deepcopy(value) if key in _SCHEMA_LITERAL_KEYS else _schema_for_local_validation(value)
+        for key, value in node.items() if key != "nullable"
+    }
     if node.get("nullable") is not True:
         return normalized
-
     schema_type = normalized.get("type")
     if isinstance(schema_type, str):
         if schema_type != "null":
@@ -45,7 +38,6 @@ def _schema_for_local_validation(node: Any) -> Any:
         if "null" not in schema_type:
             normalized["type"] = [*schema_type, "null"]
         return normalized
-
     # No ``type`` to extend ($ref/combinator): wrap so local refs still resolve
     # from the root while null stays an explicit alternative.
     return {"anyOf": [normalized, {"type": "null"}]}
@@ -83,10 +75,7 @@ def _validation_path(error: Any) -> str:
 
 def _validation_error(message: str, *, path: str, constraint: str, parameters: Any) -> str:
     return tool_error(
-        message,
-        path=path,
-        constraint=constraint,
-        parameters=parameters,
+        message, path=path, constraint=constraint, parameters=parameters,
         hint="Retry tool_call with 'arguments' matching the parameters schema above.",
     )
 
@@ -122,18 +111,14 @@ def validate_deferred_call_args(name: str, args: Dict[str, Any]) -> Optional[str
 
         validation_schema = _schema_for_local_validation(params)
         if _schema_has_external_ref(validation_schema):
-            logger.debug(
-                "Skipping local deferred-argument validation for %s: external $ref",
-                name,
-            )
+            logger.debug("Skipping local deferred-argument validation for %s: external $ref", name)
             return None
 
         # Validate the repaired shape dispatch will see; copy because
         # coerce_tool_args may normalize in place (dispatch re-coerces canonically).
-        candidate_args = dict(args)
         try:
             from model_tools import coerce_tool_args
-            candidate_args = coerce_tool_args(name, candidate_args)
+            candidate_args = coerce_tool_args(name, dict(args))
         except Exception:
             logger.debug("Deferred-argument coercion failed for %s", name, exc_info=True)
             candidate_args = dict(args)
@@ -142,20 +127,14 @@ def validate_deferred_call_args(name: str, args: Dict[str, Any]) -> Optional[str
             from jsonschema.exceptions import best_match
             from jsonschema.validators import validator_for
         except ImportError:
-            logger.debug(
-                "jsonschema unavailable; keeping required-only validation for %s",
-                name,
-            )
+            logger.debug("jsonschema unavailable; keeping required-only validation for %s", name)
             return None
 
         validator_cls = validator_for(validation_schema)
         validator_cls.check_schema(validation_schema)
-        validation_error = best_match(
-            validator_cls(validation_schema).iter_errors(candidate_args)
-        )
+        validation_error = best_match(validator_cls(validation_schema).iter_errors(candidate_args))
         if validation_error is None:
             return None
-
         path = _validation_path(validation_error)
         constraint = str(getattr(validation_error, "validator", None) or "schema")
         detail = re.sub(r"\s+", " ", str(validation_error.message)).strip()

From a0c2336c1da84f4cfddc5194f1654cbf64118a1e Mon Sep 17 00:00:00 2001
From: Teknium <127238744+teknium1@users.noreply.github.com>
Date: Wed, 2 Sep 2026 22:29:22 -0700
Subject: [PATCH 2/6] =?UTF-8?q?refactor(tools):=20tool=5Fresult=5Fstorage/?=
 =?UTF-8?q?backend=5Fhelpers/output=5Flimits/spill=5Fsafety=20=E2=80=94=20?=
 =?UTF-8?q?unify=20.env=20lookup=20(=5Fdotenv=5Fvalue),=20alias=20normaliz?=
 =?UTF-8?q?e=5Fmodal=5Fmode,=20collapse=20branches,=20compact=20docstrings?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 tools/spill_safety.py         |  35 +++----
 tools/tool_backend_helpers.py |  89 ++++++------------
 tools/tool_output_limits.py   |  21 ++---
 tools/tool_result_storage.py  | 170 +++++++++++-----------------------
 4 files changed, 103 insertions(+), 212 deletions(-)

diff --git a/tools/spill_safety.py b/tools/spill_safety.py
index 13d26f1075..62fa4dea78 100644
--- a/tools/spill_safety.py
+++ b/tools/spill_safety.py
@@ -1,21 +1,16 @@
 """Symlink-safe creation helpers for spill/cache files.
 
-Spill files (terminal output, hook context, subagent summaries, web_extract
-text) live in predictable directories under ``~/.hermes``; a plain
+Spill files live in predictable directories under ``~/.hermes``; a plain
 ``open(path, "w")`` there would follow a pre-planted symlink and let a local
-process redirect the write onto ``~/.bashrc``, ``authorized_keys``, etc.
-
-Every helper refuses symlinks by construction: new files use
-``O_CREAT | O_EXCL`` (fails on ANY existing path, including a dangling link);
-overwrites ``lstat`` + ``unlink`` the existing path first (removes the link,
-never its target) and then create exclusively, so the pair can't be raced.
-
-Privacy tiers: ``private=True`` (default) forces ``0o700`` dirs / ``0o600``
-files for spills that may hold pre-redaction secrets; ``private=False`` keeps
-umask-default perms for cache dirs bind-mounted into remote terminal backends
-(``credential_files._CACHE_DIRS``), where a non-root container UID must read them.
-
-Disk failures are the caller's concern: helpers raise ``OSError``.
+process redirect the write onto ``~/.bashrc``, ``authorized_keys``, etc. Every
+helper refuses symlinks by construction: new files use ``O_CREAT | O_EXCL``
+(fails on ANY existing path, including a dangling link); overwrites ``lstat`` +
+``unlink`` the existing path first (removes the link, never its target), then
+create exclusively, so the pair can't be raced. ``private=True`` (default)
+forces ``0o700`` dirs / ``0o600`` files for spills that may hold pre-redaction
+secrets; ``private=False`` keeps umask perms for cache dirs bind-mounted into
+remote backends (``credential_files._CACHE_DIRS``) where a non-root container
+UID must read them. Disk failures are the caller's concern: helpers raise ``OSError``.
 """
 
 from __future__ import annotations
@@ -38,10 +33,8 @@ _O_NOFOLLOW = getattr(os, "O_NOFOLLOW", 0)
 
 def ensure_spill_dir(path: Path, *, private: bool = True) -> Path:
     """Create ``path`` (and parents) as a directory, refusing symlinks.
-
-    ``private=True`` creates the leaf ``0o700`` and tightens an existing leaf to
-    ``0o700``. Raises ``OSError`` if the leaf is not a real directory.
-    """
+    ``private=True`` creates the leaf ``0o700`` and tightens an existing leaf.
+    Raises ``OSError`` if the leaf is not a real directory."""
     path = Path(path)
     path.mkdir(mode=0o700 if private else 0o777, parents=True, exist_ok=True)
     st = os.lstat(path)
@@ -61,11 +54,9 @@ def open_exclusive(
     errors: str = "strict",
 ) -> IO[str]:
     """Open ``path`` for writing via exclusive create; never follows a link.
-
     ``overwrite=True`` first unlinks an existing path (``lstat``-checked, so only
     the link itself is removed and directories are refused), then creates
-    exclusively — the overwrite path cannot be redirected through a symlink either.
-    """
+    exclusively — the overwrite path cannot be redirected through a symlink either."""
     path = Path(path)
     if overwrite:
         try:
diff --git a/tools/tool_backend_helpers.py b/tools/tool_backend_helpers.py
index 603a052a81..09152df819 100644
--- a/tools/tool_backend_helpers.py
+++ b/tools/tool_backend_helpers.py
@@ -19,12 +19,10 @@ _VALID_MODAL_MODES = {"auto", "direct", "managed"}
 
 def managed_nous_tools_enabled(*, force_fresh: bool = False) -> bool:
     """True when the user is entitled to the Nous Tool Gateway (coarse gate:
-    paid Portal service access OR a live free tool pool — "is ANY managed tool usable").
-
-    Fails closed on unknown/error entitlement — never blocks startup.
-    Per-category coverage is narrowed by callers via ``tool_gateway_entitled_for``.
-    ``force_fresh=True`` is for interactive flows that must see a just-purchased grant.
-    """
+    paid Portal service access OR a live free tool pool). Fails closed on
+    unknown/error entitlement — never blocks startup. Per-category coverage is
+    narrowed by callers via ``tool_gateway_entitled_for``. ``force_fresh=True``
+    is for interactive flows that must see a just-purchased grant."""
     try:
         from hermes_cli.nous_account import get_nous_portal_account_info
 
@@ -50,10 +48,7 @@ def nous_tool_gateway_unavailable_message(
         )
 
         account_info = get_nous_portal_account_info(force_fresh=force_fresh)
-        message = format_nous_portal_entitlement_message(
-            account_info,
-            capability=capability,
-        )
+        message = format_nous_portal_entitlement_message(account_info, capability=capability)
         if message:
             return message
     except Exception:
@@ -76,9 +71,7 @@ def coerce_modal_mode(value: object | None) -> str:
     return mode if mode in _VALID_MODAL_MODES else _DEFAULT_MODAL_MODE
 
 
-def normalize_modal_mode(value: object | None) -> str:
-    """Return a normalized modal execution mode."""
-    return coerce_modal_mode(value)
+normalize_modal_mode = coerce_modal_mode
 
 
 def has_direct_modal_credentials() -> bool:
@@ -105,8 +98,6 @@ def resolve_modal_backend_state(
     requested_mode = coerce_modal_mode(modal_mode)
     if managed_enabled is None:
         managed_enabled = managed_nous_tools_enabled()
-    managed_mode_blocked = requested_mode == "managed" and not managed_enabled
-
     managed_ok = managed_enabled and managed_ready
     if requested_mode == "managed":
         selected_backend = "managed" if managed_ok else None
@@ -114,13 +105,12 @@ def resolve_modal_backend_state(
         selected_backend = "direct" if has_direct else None
     else:
         selected_backend = "managed" if managed_ok else "direct" if has_direct else None
-
     return {
         "requested_mode": requested_mode,
         "mode": requested_mode,
         "has_direct": has_direct,
         "managed_ready": managed_ready,
-        "managed_mode_blocked": managed_mode_blocked,
+        "managed_mode_blocked": requested_mode == "managed" and not managed_enabled,
         "selected_backend": selected_backend,
     }
 
@@ -137,6 +127,16 @@ def _scoped_credential(name: str) -> str:
         return (os.getenv(name, "") or "").strip()
 
 
+def _dotenv_value(env_var: str) -> str:
+    """``.env`` value via ``hermes_cli.config.get_env_value`` (``""`` when unavailable)."""
+    try:
+        from hermes_cli.config import get_env_value
+
+        return str(get_env_value(env_var) or "").strip()
+    except Exception:  # pragma: no cover — config is in-repo
+        return ""
+
+
 def resolve_provider_secret(
     env_var: str,
     provider_id: str,
@@ -154,11 +154,9 @@ def resolve_provider_secret(
     value = str(config_value or "").strip()
     if value:
         return value
-
     key = _scoped_credential(env_var)
     if key:
         return key
-
     try:
         from agent.secret_scope import is_multiplex_active
 
@@ -166,21 +164,9 @@ def resolve_provider_secret(
             return ""
     except Exception:  # pragma: no cover — secret_scope is in-repo
         pass
-
-    if env_getter is not None:
-        key = str(env_getter(env_var) or "").strip()
-    else:
-        try:
-            from hermes_cli.config import get_env_value
-
-            key = str(get_env_value(env_var) or "").strip()
-        except ImportError:  # pragma: no cover — config is in-repo
-            key = ""
-    if key:
+    key = str(env_getter(env_var) or "").strip() if env_getter is not None else _dotenv_value(env_var)
+    if key or not provider_id:
         return key
-
-    if not provider_id:
-        return ""
     try:
         from agent.credential_pool import load_pool
 
@@ -193,9 +179,7 @@ def resolve_provider_secret(
             if entry is None:
                 continue
             key = str(
-                getattr(entry, "runtime_api_key", "")
-                or getattr(entry, "access_token", "")
-                or ""
+                getattr(entry, "runtime_api_key", "") or getattr(entry, "access_token", "") or ""
             ).strip()
             if key:
                 return key
@@ -207,8 +191,7 @@ def resolve_provider_secret(
 def resolve_openai_audio_api_key() -> str:
     """Prefer VOICE_TOOLS_OPENAI_KEY, else OPENAI_API_KEY (scope-aware, with
     credential-pool fallback for the latter). Must go through the secret scope:
-    a raw ``os.environ`` read could bill another profile's account under multiplex.
-    """
+    a raw ``os.environ`` read could bill another profile's account under multiplex."""
     return (
         resolve_provider_secret("VOICE_TOOLS_OPENAI_KEY", "")
         or resolve_provider_secret("OPENAI_API_KEY", "openai-api")
@@ -216,10 +199,7 @@ def resolve_openai_audio_api_key() -> str:
 
 
 def prefers_gateway(config_section: str) -> bool:
-    """Return True when the user opted into the Tool Gateway for this tool.
-
-    Reads ``<section>.use_gateway`` from config.yaml.  Never raises.
-    """
+    """True when ``<section>.use_gateway`` is set in config.yaml. Never raises."""
     try:
         from hermes_cli.config import load_config
         section = (load_config() or {}).get(config_section)
@@ -273,10 +253,8 @@ def read_selection(section: str) -> str | None:
     raw = _raw_section(section)
     if raw is None:
         return None
-
     if "use_gateway" in raw and is_truthy_value(raw.get("use_gateway"), default=False):
         return NOUS_MANAGED_PROVIDER
-
     for key in _SELECTION_NAME_KEYS.get(section, _DEFAULT_NAME_KEYS):
         value = raw.get(key)
         if value is not None:
@@ -302,27 +280,22 @@ def selection_exists(section: str) -> bool:
 
 # Backends that once shipped in-tree but were removed. A config still pointing at
 # one would otherwise fail silently at the FIRST tool call with a generic "no
-# registered provider has that name" — no migration, no startup notice. Consulted
-# by the startup config check and selection_error() so the user gets a real message.
-# Add removals here, never as one-off string checks at call sites, e.g.
+# registered provider has that name". Consulted by the startup config check and
+# selection_error(). Add removals here, never as one-off string checks, e.g.
 #   "web": {"<name>": "the <Name> backend was removed in vX.Y.Z (...)"},
 REMOVED_BACKENDS: Dict[str, Dict[str, str]] = {}
 
 
 def removed_backend_note(section: str, name: str) -> Optional[str]:
     """Explanation for a backend that used to ship in-tree, or None.
-
-    ``name`` tolerates the quoted form callers pass to selection_error().
-    """
+    ``name`` tolerates the quoted form callers pass to selection_error()."""
     normalized = (name or "").strip().strip("'\"").lower()
     return REMOVED_BACKENDS.get(section, {}).get(normalized)
 
 
 def selection_error(section: str, selection_name: str, failure: str) -> str:
     """The uniform honest-error contract for a selected-but-broken provider."""
-    note = removed_backend_note(section, selection_name)
-    if note:
-        failure = note
+    failure = removed_backend_note(section, selection_name) or failure
     return (
         f"{section} is configured to use {selection_name} (set via hermes "
         f"tools), but {failure}. Run 'hermes tools' to change it."
@@ -333,12 +306,4 @@ def fal_key_is_configured() -> bool:
     """True when FAL_KEY is set (scope/env, else ``.env`` for CLI paths that
     run before dotenv loads) to a non-whitespace value — so tool-side and CLI
     setup-time checks agree; whitespace-only counts as unset everywhere."""
-    value = _scoped_credential("FAL_KEY") or None
-    if value is None:
-        try:
-            from hermes_cli.config import get_env_value
-
-            value = get_env_value("FAL_KEY")
-        except Exception:
-            value = None
-    return bool(value and value.strip())
+    return bool(_scoped_credential("FAL_KEY") or _dotenv_value("FAL_KEY"))
diff --git a/tools/tool_output_limits.py b/tools/tool_output_limits.py
index d091a4de61..8b670aa289 100644
--- a/tools/tool_output_limits.py
+++ b/tools/tool_output_limits.py
@@ -1,10 +1,9 @@
 """Configurable tool-output truncation limits (``tool_output`` in config.yaml).
 
-Centralises the caps previously hardcoded in ``terminal_tool`` (``max_bytes``)
-and ``file_operations`` (``max_lines`` / ``max_line_length``). Defaults equal
-the old constants, so behaviour is unchanged when the section is absent, and
-the reader never raises — any config error falls back to the defaults.
-Ported from anomalyco/opencode PR #23770. Example::
+Centralises the caps once hardcoded in ``terminal_tool`` (``max_bytes``) and
+``file_operations`` (``max_lines`` / ``max_line_length``). Defaults equal the
+old constants and the reader never raises, so behaviour is unchanged when the
+section is absent or malformed::
 
     tool_output:
       max_bytes: 100000        # terminal output cap (chars)
@@ -39,11 +38,8 @@ def _coerce_positive_int(value: Any, default: int) -> int:
 
 
 def get_tool_output_limits() -> Dict[str, int]:
-    """Return resolved limits ``{max_bytes, max_lines, max_line_length}``; never raises.
-
-    Cached for the process lifetime — ``_reset_tool_output_limits_cache()``
-    forces a fresh read after config changes.
-    """
+    """Resolved ``{max_bytes, max_lines, max_line_length}``; never raises. Cached
+    for the process — ``_reset_tool_output_limits_cache()`` forces a fresh read."""
     global _cached_limits
     if _cached_limits is not None:
         return _cached_limits
@@ -55,13 +51,10 @@ def get_tool_output_limits() -> Dict[str, int]:
             section = {}
     except Exception:
         section = {}
-
     _cached_limits = {
         "max_bytes": _coerce_positive_int(section.get("max_bytes"), DEFAULT_MAX_BYTES),
         "max_lines": _coerce_positive_int(section.get("max_lines"), DEFAULT_MAX_LINES),
-        "max_line_length": _coerce_positive_int(
-            section.get("max_line_length"), DEFAULT_MAX_LINE_LENGTH
-        ),
+        "max_line_length": _coerce_positive_int(section.get("max_line_length"), DEFAULT_MAX_LINE_LENGTH),
     }
     return _cached_limits
 
diff --git a/tools/tool_result_storage.py b/tools/tool_result_storage.py
index 8dfa8f6afb..58cc91d906 100644
--- a/tools/tool_result_storage.py
+++ b/tools/tool_result_storage.py
@@ -1,23 +1,14 @@
 """Tool result persistence -- preserves large outputs instead of truncating.
 
-Three layers of defense against context-window overflow:
-
-1. **Per-tool output cap** (inside each tool, e.g. search_files pre-truncates).
-2. **Per-result persistence** (``maybe_persist_tool_result``): output over the
-   tool's registered threshold is persisted and replaced in-context by a
-   preview + file path. The canonical home is ALWAYS host-side,
-   ``$HERMES_HOME/cache/spillover/{tool_use_id}.txt``, so it works for sessions
-   that never ran a terminal (MCP-only, cron, gateway). Local backend / no env:
-   the model sees the host path. Remote backends (docker/ssh/modal/daytona):
-   ``cache/spillover`` is in the auto-mounted/synced cache-dir list
-   (tools/credential_files.py), so the model sees the translated in-sandbox
-   path — probed for readability first, falling back to a copy written into
-   the sandbox temp dir (e.g. a persistent container created before spillover
-   joined the mount list). The dir is pruned hourly by gateway housekeeping
-   and once per process on first spill (CLI-only installs never run housekeeping).
-3. **Per-turn aggregate budget** (``enforce_turn_budget``): if all tool results
-   in one turn exceed the turn budget, the largest non-persisted results are
-   spilled until under budget.
+Three layers against context-window overflow: (1) per-tool output caps inside
+each tool; (2) ``maybe_persist_tool_result`` — output over the tool's threshold
+is persisted and replaced in-context by a preview + path. The canonical home is
+ALWAYS host-side ``$HERMES_HOME/cache/spillover/{tool_use_id}.txt`` (works for
+MCP-only/cron/gateway sessions that never ran a terminal); remote backends see
+the translated in-sandbox path (``cache/spillover`` is in the auto-mounted cache
+list), probed for readability, else a copy written into the sandbox temp dir.
+(3) ``enforce_turn_budget`` — spills the largest results of a turn until the
+aggregate fits.
 """
 
 import hashlib
@@ -57,10 +48,7 @@ def get_spillover_dir():
 
 def cleanup_spillover_cache(max_age_hours: int = SPILLOVER_MAX_AGE_HOURS) -> int:
     """Delete spillover files older than *max_age_hours*; returns count removed.
-
-    Same contract as the ``cleanup_*_cache`` helpers in ``gateway.platforms.base``
-    so the gateway housekeeping loop can prune this dir on the hourly cadence.
-    """
+    Same contract as the ``cleanup_*_cache`` helpers the gateway housekeeping loop prunes hourly."""
     cutoff = time.time() - (max_age_hours * 3600)
     removed = 0
     try:
@@ -93,11 +81,9 @@ def _prune_spillover_once() -> None:
 
 
 def _is_host_side_env(env) -> bool:
-    """True when this process should write the spill file directly.
-
-    ``env=None`` (no sandbox yet) and the local backend qualify; remote backends
-    resolve ``read_file`` inside the sandbox, so the spill must be written there.
-    """
+    """True when this process should write the spill file directly: ``env=None``
+    (no sandbox yet) or the local backend. Remote backends resolve ``read_file``
+    inside the sandbox, so the spill must be written there."""
     if env is None:
         return True
     try:
@@ -123,13 +109,10 @@ def _write_to_spillover(content: str, filename: str):
 
 
 def _sandbox_visible_spillover_path(host_path: str, env) -> str | None:
-    """Return the path where a remote backend can read *host_path*, or None.
-
-    Translates via the same helper the image tools use, forces a sync for synced
-    backends, then PROBES readability — a persistent docker container created
-    before spillover joined the mount list lacks the bind mount and must fall
-    back to the in-sandbox write.
-    """
+    """Path where a remote backend can read *host_path*, or None. Translates via
+    the image tools' helper, forces a sync for synced backends, then PROBES
+    readability — a persistent container created before spillover joined the
+    mount list lacks the bind mount and must fall back to the in-sandbox write."""
     try:
         from tools.credential_files import to_agent_visible_cache_path
 
@@ -137,14 +120,12 @@ def _sandbox_visible_spillover_path(host_path: str, env) -> str | None:
     except Exception as exc:
         logger.debug("Spillover path translation failed: %s", exc)
         return None
-
     sync_manager = getattr(env, "_sync_manager", None)
     if sync_manager is not None:
         try:
             sync_manager.sync(force=True)
         except Exception as exc:
             logger.debug("Spillover sync failed: %s", exc)
-
     try:
         result = env.execute(f"test -r {shlex.quote(visible)}", timeout=15)
         if result.get("returncode", 1) == 0:
@@ -156,17 +137,15 @@ def _sandbox_visible_spillover_path(host_path: str, env) -> str | None:
 
 def _resolve_storage_dir(env) -> str:
     """Return the best temp-backed storage dir for this environment."""
-    if env is not None:
-        get_temp_dir = getattr(env, "get_temp_dir", None)
-        if callable(get_temp_dir):
-            try:
-                temp_dir = get_temp_dir()
-            except Exception as exc:
-                logger.debug("Could not resolve env temp dir: %s", exc)
-            else:
-                if temp_dir:
-                    temp_dir = temp_dir.rstrip("/") or "/"
-                    return f"{temp_dir}/hermes-results"
+    get_temp_dir = getattr(env, "get_temp_dir", None)
+    if callable(get_temp_dir):
+        try:
+            temp_dir = get_temp_dir()
+        except Exception as exc:
+            logger.debug("Could not resolve env temp dir: %s", exc)
+        else:
+            if temp_dir:
+                return f"{temp_dir.rstrip('/') or '/'}/hermes-results"
     return STORAGE_DIR
 
 
@@ -175,16 +154,13 @@ def _safe_result_filename(tool_use_id: str) -> str:
     raw_id = str(tool_use_id or "tool_result")
     safe_stem = _UNSAFE_RESULT_FILENAME_CHARS.sub("_", raw_id).strip("._-")
     changed = safe_stem != raw_id
-
     if not safe_stem:
         safe_stem = "tool_result"
         changed = True
-
     if changed or len(safe_stem) > _MAX_RESULT_FILENAME_STEM:
         digest = hashlib.sha256(raw_id.encode("utf-8")).hexdigest()[:12]
         safe_stem = safe_stem[:_MAX_RESULT_FILENAME_STEM].rstrip("._-") or "tool_result"
         safe_stem = f"{safe_stem}_{digest}"
-
     return f"{safe_stem}.txt"
 
 
@@ -201,11 +177,9 @@ def generate_preview(content: str, max_chars: int = DEFAULT_PREVIEW_SIZE_CHARS)
 
 def _write_to_sandbox(content: str, remote_path: str, env) -> bool:
     """Write content into the sandbox via env.execute(). Returns True on success.
-
     Content goes through stdin, not the command string: Linux ``MAX_ARG_STRLEN``
-    caps a single argv element at 128 KB, so a heredoc-in-command silently failed
-    for exactly the oversized results persistence exists to handle.
-    """
+    caps one argv element at 128 KB, so a heredoc-in-command silently failed
+    for exactly the oversized results persistence exists to handle."""
     storage_dir = os.path.dirname(remote_path)
     cmd = f"mkdir -p {shlex.quote(storage_dir)} && cat > {shlex.quote(remote_path)}"
     result = env.execute(cmd, timeout=30, stdin_data=content)
@@ -221,33 +195,27 @@ def _build_persisted_message(
     """Build the <persisted-output> replacement block."""
     size_kb = original_size / 1024
     size_str = f"{size_kb / 1024:.1f} MB" if size_kb >= 1024 else f"{size_kb:.1f} KB"
-
-    msg = f"{PERSISTED_OUTPUT_TAG}\n"
-    msg += f"This tool result was too large ({original_size:,} characters, {size_str}).\n"
-    msg += f"Full output saved to: {file_path}\n"
-    msg += "Use the read_file tool with offset and limit to access specific sections of this output.\n"
-    msg += (
+    return (
+        f"{PERSISTED_OUTPUT_TAG}\n"
+        f"This tool result was too large ({original_size:,} characters, {size_str}).\n"
+        f"Full output saved to: {file_path}\n"
+        "Use the read_file tool with offset and limit to access specific sections of this output.\n"
         "Recovery: page through the saved file with read_file (offset/limit) or "
         "process it with execute_code — do NOT re-request the same data from the "
         "remote API; the full result is already on disk.\n\n"
+        f"Preview (first {len(preview)} chars):\n"
+        + preview + ("\n..." if has_more else "")
+        + f"\n{PERSISTED_OUTPUT_CLOSING_TAG}"
     )
-    msg += f"Preview (first {len(preview)} chars):\n"
-    msg += preview
-    if has_more:
-        msg += "\n..."
-    msg += f"\n{PERSISTED_OUTPUT_CLOSING_TAG}"
-    return msg
 
 
 _PERSISTED_PATH_RE = re.compile(r"^Full output saved to: (.+)$", re.MULTILINE)
 
 
 def extract_persisted_path(content: str) -> str | None:
-    """Return the file path from a <persisted-output> block, or None.
-
-    Lets the result-reference stubbing guard (agent/tool_guardrails.py) carry
-    the spillover path in a stub instead of leaving it dangling.
-    """
+    """Return the file path from a <persisted-output> block, or None. Lets the
+    result-reference stubbing guard (agent/tool_guardrails.py) carry the
+    spillover path in a stub instead of leaving it dangling."""
     if not isinstance(content, str) or PERSISTED_OUTPUT_TAG not in content:
         return None
     match = _PERSISTED_PATH_RE.search(content)
@@ -263,16 +231,10 @@ def maybe_persist_tool_result(
     threshold: int | float | None = None,
 ) -> str:
     """Layer 2: persist an oversized result, return preview + path.
-
     ``threshold`` overrides ``config.resolve_threshold(tool_name)``. Falls back
-    to inline truncation when no write location succeeds.
-    """
+    to inline truncation when no write location succeeds."""
     effective_threshold = threshold if threshold is not None else config.resolve_threshold(tool_name)
-
-    if effective_threshold == float("inf"):
-        return content
-
-    if len(content) <= effective_threshold:
+    if effective_threshold == float("inf") or len(content) <= effective_threshold:
         return content
 
     filename = _safe_result_filename(tool_use_id)
@@ -285,22 +247,19 @@ def maybe_persist_tool_result(
         )
         return _build_persisted_message(preview, has_more, len(content), path)
 
-    # Always persist host-side first: cache/spillover is the single canonical
-    # home for spilled results regardless of backend.
+    # Always persist host-side first: cache/spillover is the single canonical home.
     host_path = _write_to_spillover(content, filename)
-
     if _is_host_side_env(env):
         if host_path is not None:
             return _persisted(host_path)
-    elif env is not None:
-        # Remote backend: reference the mounted/synced path when the sandbox
-        # can actually read it ...
+    else:
+        # Remote backend: reference the mounted/synced path when the sandbox can
+        # actually read it, else write into the sandbox temp dir (pre-existing
+        # containers without the spillover mount, translation/probe failures).
         if host_path is not None:
             visible = _sandbox_visible_spillover_path(host_path, env)
             if visible is not None:
                 return _persisted(visible, f" [host: {host_path}]")
-        # ... else write into the sandbox temp dir (pre-existing containers
-        # without the spillover mount, translation/probe failures).
         remote_path = f"{_resolve_storage_dir(env)}/{filename}"
         try:
             if _write_to_sandbox(content, remote_path, env):
@@ -324,47 +283,30 @@ def enforce_turn_budget(
     env=None,
     config: BudgetConfig = DEFAULT_BUDGET,
 ) -> list[dict]:
-    """Layer 3: enforce the aggregate budget across all tool results in a turn.
-
-    Persists the largest non-persisted results first until under budget.
-    Mutates the list in-place and returns it.
-    """
+    """Layer 3: persist the largest non-persisted results first until the turn's
+    aggregate is under budget. Mutates the list in-place and returns it."""
     candidates = []
     total_size = 0
     for i, msg in enumerate(tool_messages):
-        content = msg.get("content", "")
-        size = len(content)
+        size = len(msg.get("content", ""))
         total_size += size
-        if PERSISTED_OUTPUT_TAG not in content:
+        if PERSISTED_OUTPUT_TAG not in msg.get("content", ""):
             candidates.append((i, size))
-
     if total_size <= config.turn_budget:
         return tool_messages
 
     candidates.sort(key=lambda x: x[1], reverse=True)
-
     for idx, size in candidates:
         if total_size <= config.turn_budget:
             break
-        msg = tool_messages[idx]
-        content = msg["content"]
-        tool_use_id = msg.get("tool_call_id", f"budget_{idx}")
-
+        content = tool_messages[idx]["content"]
+        tool_use_id = tool_messages[idx].get("tool_call_id", f"budget_{idx}")
         replacement = maybe_persist_tool_result(
-            content=content,
-            tool_name=_BUDGET_TOOL_NAME,
-            tool_use_id=tool_use_id,
-            env=env,
-            config=config,
-            threshold=0,
+            content=content, tool_name=_BUDGET_TOOL_NAME, tool_use_id=tool_use_id,
+            env=env, config=config, threshold=0,
         )
         if replacement != content:
-            total_size -= size
-            total_size += len(replacement)
+            total_size += len(replacement) - size
             tool_messages[idx]["content"] = replacement
-            logger.info(
-                "Budget enforcement: persisted tool result %s (%d chars)",
-                tool_use_id, size,
-            )
-
+            logger.info("Budget enforcement: persisted tool result %s (%d chars)", tool_use_id, size)
     return tool_messages

From 8b33fd46be2730233ad65e78379553864e36c510 Mon Sep 17 00:00:00 2001
From: Teknium <127238744+teknium1@users.noreply.github.com>
Date: Wed, 2 Sep 2026 22:36:39 -0700
Subject: [PATCH 3/6] =?UTF-8?q?refactor(tools):=20todo/tour/tip=20?=
 =?UTF-8?q?=E2=80=94=20hoist=20status=20marker=20tables,=20collapse=20sing?=
 =?UTF-8?q?le-use=20locals,=20compact=20handlers=20and=20docstrings?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 tools/tip_tool.py  | 11 +--------
 tools/todo_tool.py | 61 +++++++++++++---------------------------------
 tools/tour_tool.py | 23 ++++++-----------
 3 files changed, 25 insertions(+), 70 deletions(-)

diff --git a/tools/tip_tool.py b/tools/tip_tool.py
index e66d225cdc..c5dd4eb76b 100644
--- a/tools/tip_tool.py
+++ b/tools/tip_tool.py
@@ -21,10 +21,8 @@ def tip_tool(text: str, selector: str, title: str = "", side: str = "") -> str:
     """Show one tip bubble anchored to ``selector``."""
     text = (text or "").strip()
     selector = (selector or "").strip()
-
     if not text:
         return tool_error("tip needs text — the one line the bubble says.")
-
     if not selector:
         return tool_error(
             "tip needs a selector to point at. Call tour(action='targets') to see "
@@ -33,17 +31,14 @@ def tip_tool(text: str, selector: str, title: str = "", side: str = "") -> str:
 
     if side and side not in SIDES:
         return tool_error(f"side must be one of: {', '.join(SIDES)}.")
-
     payload = {"selector": selector, "text": text}
     payload.update({k: v for k, v in (("title", title), ("side", side)) if v})
-
     try:
         ok = desktop_ui.emit("tip.show", payload)
     except Exception as exc:
         return tool_error(f"Failed to show the tip: {exc}")
     if not ok:
         return tool_error("tip is only available in the Hermes desktop app.")
-
     return json.dumps({"success": True, "selector": selector}, ensure_ascii=False)
 
 
@@ -93,11 +88,7 @@ registry.register(
     toolset="desktop_ui",
     schema=TIP_SCHEMA,
     handler=lambda args, **kw: tip_tool(
-        text=args.get("text", ""),
-        selector=args.get("selector", ""),
-        title=args.get("title", ""),
-        side=args.get("side", ""),
-    ),
+        **{k: args.get(k, "") for k in ("text", "selector", "title", "side")}),
     check_fn=check_tips_enabled,
     emoji="💡",
 )
diff --git a/tools/todo_tool.py b/tools/todo_tool.py
index 2543b6da82..b7b993868d 100644
--- a/tools/todo_tool.py
+++ b/tools/todo_tool.py
@@ -27,13 +27,13 @@ _TRUNCATION_MARKER = "… [truncated]"
 TODO_INJECTION_HEADER = (
     "[Your active task list was preserved across context compression]"
 )
+_STATUS_MARKERS = {"completed": "[x]", "in_progress": "[>]", "pending": "[ ]", "cancelled": "[~]"}
+_ACTIVE_STATUSES = {"pending", "in_progress"}
 
 
 class TodoStore:
     """In-memory todo list, one per AIAgent. List position is priority.
-
-    Items: ``{id, content, status, parent?}`` — ``parent`` nests a subtask.
-    """
+    Items: ``{id, content, status, parent?}`` — ``parent`` nests a subtask."""
 
     def __init__(self):
         self._items: List[Dict[str, str]] = []
@@ -46,13 +46,12 @@ class TodoStore:
     def write(self, todos: List[Dict[str, Any]], merge: bool = False) -> List[Dict[str, str]]:
         """Replace the list (default) or merge by id; returns the full list after writing."""
         before = self.read()
-        if not merge:
-            self._items = self._fresh_items(todos)
-        else:
+        if merge:
             self._merge(todos)
+        else:
+            self._items = self._fresh_items(todos)
         # Keep the highest-priority head so a replayed list can't grow re-injection unbounded.
-        if len(self._items) > MAX_TODO_ITEMS:
-            self._items = self._items[:MAX_TODO_ITEMS]
+        del self._items[MAX_TODO_ITEMS:]
         self._sanitize_parents(self._items)
         if self._items != before:
             self._revision += 1
@@ -104,12 +103,7 @@ class TodoStore:
         """Return the full state clients can reconcile atomically."""
         return {"todos": self.read(), "revision": self._revision}
 
-    def restore(
-        self,
-        todos: List[Dict[str, Any]],
-        *,
-        revision: Any = 0,
-    ) -> List[Dict[str, str]]:
+    def restore(self, todos: List[Dict[str, Any]], *, revision: Any = 0) -> List[Dict[str, str]]:
         """Restore a trusted snapshot without manufacturing a new revision."""
         self._items = self._fresh_items(todos)[:MAX_TODO_ITEMS]
         try:
@@ -120,16 +114,11 @@ class TodoStore:
 
     def format_for_injection(self) -> Optional[str]:
         """Render the list for post-compression injection, or None if nothing active.
-
         Only pending/in_progress items are injected — finished ones make the
         model re-do work after compression. A parent is kept (with its real
-        status marker) when any descendant is active so subtasks keep context.
-        """
+        status marker) when any descendant is active so subtasks keep context."""
         if not self._items:
             return None
-
-        markers = {"completed": "[x]", "in_progress": "[>]", "pending": "[ ]", "cancelled": "[~]"}
-        active = {"pending", "in_progress"}
         children: Dict[str, List[Dict[str, str]]] = {}
         roots: List[Dict[str, str]] = []
         for item in self._items:
@@ -144,9 +133,9 @@ class TodoStore:
             has_active_kid = False
             for kid in children.get(item["id"], []):
                 has_active_kid |= render(kid, depth + 1, kid_lines)
-            keep = item["status"] in active or has_active_kid
+            keep = item["status"] in _ACTIVE_STATUSES or has_active_kid
             if keep:
-                marker = markers.get(item["status"], "[?]")
+                marker = _STATUS_MARKERS.get(item["status"], "[?]")
                 out.append(
                     f"{'  ' * depth}- {marker} {item['id']}. "
                     f"{item['content']} ({item['status']})"
@@ -157,17 +146,13 @@ class TodoStore:
         lines = [TODO_INJECTION_HEADER]
         for item in roots:
             render(item, 0, lines)
-        if len(lines) == 1:
-            return None
-
-        return "\n".join(lines)
+        return "\n".join(lines) if len(lines) > 1 else None
 
     @staticmethod
     def _cap_content(content: str) -> str:
         """Truncate to MAX_TODO_CONTENT_CHARS keeping the head (the actionable part) + marker."""
         if len(content) > MAX_TODO_CONTENT_CHARS:
-            keep = MAX_TODO_CONTENT_CHARS - len(_TRUNCATION_MARKER)
-            return content[:keep] + _TRUNCATION_MARKER
+            return content[:MAX_TODO_CONTENT_CHARS - len(_TRUNCATION_MARKER)] + _TRUNCATION_MARKER
         return content
 
     @staticmethod
@@ -175,14 +160,12 @@ class TodoStore:
         """Normalize one item to ``{id, content, status, parent?}`` with placeholders for missing fields."""
         if not isinstance(item, dict):
             return {"id": "?", "content": "(invalid item)", "status": "pending"}
-
         item_id = str(item.get("id", "")).strip() or "?"
         content = str(item.get("content", "")).strip()
         content = TodoStore._cap_content(content) if content else "(no description)"
         status = str(item.get("status", "pending")).strip().lower()
         if status not in VALID_STATUSES:
             status = "pending"
-
         result = {"id": item_id, "content": content, "status": status}
         parent = str(item.get("parent") or "").strip()
         if parent and parent != item_id:
@@ -215,16 +198,13 @@ class TodoStore:
                 # Non-dict items get a synthetic key so _validate can handle them
                 last_index[f"__invalid_{i}"] = i
                 continue
-            item_id = str(item.get("id", "")).strip() or "?"
-            last_index[item_id] = i
+            last_index[str(item.get("id", "")).strip() or "?"] = i
         return [todos[i] for i in sorted(last_index.values())]
 
     @staticmethod
     def _normalize_order(items: List[Dict[str, str]]) -> List[Dict[str, str]]:
         """Lift the in_progress step ahead of any earlier pending placeholder.
-
-        Nested lists keep authored order — reordering would tear a subtask from its siblings.
-        """
+        Nested lists keep authored order — reordering would tear a subtask from its siblings."""
         if any(item.get("parent") for item in items):
             return items
         statuses = [item["status"] for item in items]
@@ -233,11 +213,8 @@ class TodoStore:
         active_index = statuses.index("in_progress")
         if "pending" not in statuses[:active_index]:
             return items
-        pending_index = statuses.index("pending")
-
         normalized = items.copy()
-        active_item = normalized.pop(active_index)
-        normalized.insert(pending_index, active_item)
+        normalized.insert(statuses.index("pending"), normalized.pop(active_index))
         return normalized
 
 
@@ -249,7 +226,6 @@ def todo_tool(
     """Write ``todos`` (replace or ``merge`` by id) or read when None; returns list + summary JSON."""
     if store is None:
         return tool_error("TodoStore not initialized")
-
     if todos is not None:
         if isinstance(todos, str):  # LLMs sometimes send a JSON string instead of a list
             try:
@@ -257,9 +233,7 @@ def todo_tool(
             except (json.JSONDecodeError, TypeError):
                 return tool_error("todos must be a list of objects, got unparseable string")
         if not isinstance(todos, list):
-            return tool_error(
-                f"todos must be a list, got {type(todos).__name__}"
-            )
+            return tool_error(f"todos must be a list, got {type(todos).__name__}")
         items = store.write(todos, merge)
     else:
         items = store.read()
@@ -267,7 +241,6 @@ def todo_tool(
     summary = {"total": len(items)}
     for status in ("pending", "in_progress", "completed", "cancelled"):
         summary[status] = sum(1 for i in items if i["status"] == status)
-
     return json.dumps({
         "todos": items,
         "revision": store.snapshot()["revision"],
diff --git a/tools/tour_tool.py b/tools/tour_tool.py
index 0d075bacd0..0422443d8e 100644
--- a/tools/tour_tool.py
+++ b/tools/tour_tool.py
@@ -1,16 +1,13 @@
 #!/usr/bin/env python3
 """Guided tour (highlight + narrate UI elements) in the Hermes desktop GUI.
 
-Generic, no baked-in tours: the agent discovers targets (``action="targets"``)
+Generic, no baked-in tours: the agent discovers targets (``action="targets"``),
 then highlights by CSS selector one step at a time (``show``) or hands over a
-step list the user pages with Next/Prev (``start``). Surfaces ``app`` (Hermes'
-own DOM) and ``preview`` (the in-app browser page) share the renderer's driver.js.
-Round-trips through the gateway blocking-prompt bridge (``tour.request`` /
-``tour.respond``) so the agent learns whether the selector matched.
-
-Lives in ``desktop_ui`` (GUI sessions only) and withdraws itself when the user
-turns tours off: a tour takes the whole screen, so "off" must mean the model is
-never told the tool exists rather than being offered a call that fails.
+step list the user pages with Next/Prev (``start``). Round-trips through the
+gateway blocking-prompt bridge (``tour.request``/``tour.respond``) so the agent
+learns whether the selector matched. Lives in ``desktop_ui`` and withdraws itself
+when the user turns tours off: a tour takes the whole screen, so "off" must mean
+the model is never told the tool exists rather than offered a call that fails.
 """
 
 import json
@@ -170,13 +167,7 @@ registry.register(
     schema=TOUR_SCHEMA,
     handler=lambda args, **kw: tour_tool(
         action=args.get("action", ""),
-        surface=args.get("surface"),
-        selector=args.get("selector"),
-        title=args.get("title"),
-        text=args.get("text"),
-        side=args.get("side"),
-        steps=args.get("steps"),
-        step_index=args.get("step_index"),
+        **{k: args.get(k) for k in ("surface", "selector", "title", "text", "side", "steps", "step_index")},
         callback=kw.get("callback"),
     ),
     check_fn=check_tours_enabled,

From cffc1df5169f44bb372bcbb8ae317cec81cb68cc Mon Sep 17 00:00:00 2001
From: Teknium <127238744+teknium1@users.noreply.github.com>
Date: Wed, 2 Sep 2026 22:42:44 -0700
Subject: [PATCH 4/6] refactor(tools): group H layout compaction (AST-neutral
 bracket hugging)

---
 tools/spill_safety.py           |  12 +---
 tools/tool_backend_helpers.py   |  39 +++--------
 tools/tool_result_storage.py    |  35 +++-------
 tools/tool_search.py            | 110 +++++++++-----------------------
 tools/tool_search_catalog.py    |  11 +---
 tools/tool_search_validation.py |  15 ++---
 6 files changed, 60 insertions(+), 162 deletions(-)

diff --git a/tools/spill_safety.py b/tools/spill_safety.py
index 62fa4dea78..69b68ec172 100644
--- a/tools/spill_safety.py
+++ b/tools/spill_safety.py
@@ -20,11 +20,7 @@ import stat
 from pathlib import Path
 from typing import IO
 
-__all__ = [
-    "ensure_spill_dir",
-    "open_exclusive",
-    "write_text_exclusive",
-]
+__all__ = ["ensure_spill_dir", "open_exclusive", "write_text_exclusive"]
 
 # O_NOFOLLOW is POSIX-only; on Windows O_EXCL alone already refuses every
 # pre-existing path.
@@ -51,8 +47,7 @@ def open_exclusive(
     private: bool = True,
     overwrite: bool = False,
     encoding: str = "utf-8",
-    errors: str = "strict",
-) -> IO[str]:
+    errors: str = "strict") -> IO[str]:
     """Open ``path`` for writing via exclusive create; never follows a link.
     ``overwrite=True`` first unlinks an existing path (``lstat``-checked, so only
     the link itself is removed and directories are refused), then creates
@@ -83,8 +78,7 @@ def write_text_exclusive(
     private: bool = True,
     overwrite: bool = False,
     encoding: str = "utf-8",
-    errors: str = "strict",
-) -> None:
+    errors: str = "strict") -> None:
     """``Path.write_text`` equivalent that refuses to follow symlinks."""
     with open_exclusive(
         path, private=private, overwrite=overwrite, encoding=encoding, errors=errors
diff --git a/tools/tool_backend_helpers.py b/tools/tool_backend_helpers.py
index 09152df819..5a8aba8fc1 100644
--- a/tools/tool_backend_helpers.py
+++ b/tools/tool_backend_helpers.py
@@ -36,10 +36,7 @@ def managed_nous_tools_enabled(*, force_fresh: bool = False) -> bool:
 
 
 def nous_tool_gateway_unavailable_message(
-    capability: str = "the Nous Tool Gateway",
-    *,
-    force_fresh: bool = False,
-) -> str:
+    capability: str = "the Nous Tool Gateway", *, force_fresh: bool = False) -> str:
     """Return account-aware guidance for an unavailable Nous Tool Gateway path."""
     try:
         from hermes_cli.nous_account import (
@@ -55,8 +52,7 @@ def nous_tool_gateway_unavailable_message(
         pass
     return (
         f"{capability} is unavailable. Run `hermes model` to refresh your "
-        "Nous Portal login and billing status."
-    )
+        "Nous Portal login and billing status.")
 
 
 def normalize_browser_cloud_provider(value: object | None) -> str:
@@ -81,9 +77,7 @@ def has_direct_modal_credentials() -> bool:
     except (PermissionError, OSError):
         modal_file_exists = False
     return bool(
-        (os.getenv("MODAL_TOKEN_ID") and os.getenv("MODAL_TOKEN_SECRET"))
-        or modal_file_exists
-    )
+        (os.getenv("MODAL_TOKEN_ID") and os.getenv("MODAL_TOKEN_SECRET")) or modal_file_exists)
 
 
 def resolve_modal_backend_state(
@@ -91,8 +85,7 @@ def resolve_modal_backend_state(
     *,
     has_direct: bool,
     managed_ready: bool,
-    managed_enabled: bool | None = None,
-) -> Dict[str, Any]:
+    managed_enabled: bool | None = None) -> Dict[str, Any]:
     """Resolve direct vs managed Modal backend: ``direct``/``managed`` are
     exclusive; ``auto`` prefers managed when available, else direct."""
     requested_mode = coerce_modal_mode(modal_mode)
@@ -111,8 +104,7 @@ def resolve_modal_backend_state(
         "has_direct": has_direct,
         "managed_ready": managed_ready,
         "managed_mode_blocked": requested_mode == "managed" and not managed_enabled,
-        "selected_backend": selected_backend,
-    }
+        "selected_backend": selected_backend}
 
 
 def _scoped_credential(name: str) -> str:
@@ -138,11 +130,7 @@ def _dotenv_value(env_var: str) -> str:
 
 
 def resolve_provider_secret(
-    env_var: str,
-    provider_id: str,
-    config_value: str = "",
-    env_getter=None,
-) -> str:
+    env_var: str, provider_id: str, config_value: str = "", env_getter=None) -> str:
     """Resolve a voice-provider API key (single owner for STT/TTS lookup).
 
     Order: explicit ``config_value`` -> profile secret scope / env -> ``.env``
@@ -194,8 +182,7 @@ def resolve_openai_audio_api_key() -> str:
     a raw ``os.environ`` read could bill another profile's account under multiplex."""
     return (
         resolve_provider_secret("VOICE_TOOLS_OPENAI_KEY", "")
-        or resolve_provider_secret("OPENAI_API_KEY", "openai-api")
-    )
+        or resolve_provider_secret("OPENAI_API_KEY", "openai-api"))
 
 
 def prefers_gateway(config_section: str) -> bool:
@@ -215,16 +202,11 @@ def prefers_gateway(config_section: str) -> bool:
 NOUS_MANAGED_PROVIDER = "nous"
 
 # Per-capability keys that also count as "this category has been configured".
-_EXTRA_SELECTION_KEYS = {
-    "web": ("search_backend", "extract_backend"),
-}
+_EXTRA_SELECTION_KEYS = {"web": ("search_backend", "extract_backend")}
 
 # Key(s) carrying the category's provider selection. ``browser.backend`` is the
 # DRIVER choice (browser-use CLI vs built-in), not the cloud provider — excluded.
-_SELECTION_NAME_KEYS = {
-    "browser": ("cloud_provider",),
-    "web": ("backend",),
-}
+_SELECTION_NAME_KEYS = {"browser": ("cloud_provider",), "web": ("backend",)}
 _DEFAULT_NAME_KEYS = ("provider", "backend", "cloud_provider")
 
 
@@ -298,8 +280,7 @@ def selection_error(section: str, selection_name: str, failure: str) -> str:
     failure = removed_backend_note(section, selection_name) or failure
     return (
         f"{section} is configured to use {selection_name} (set via hermes "
-        f"tools), but {failure}. Run 'hermes tools' to change it."
-    )
+        f"tools), but {failure}. Run 'hermes tools' to change it.")
 
 
 def fal_key_is_configured() -> bool:
diff --git a/tools/tool_result_storage.py b/tools/tool_result_storage.py
index 58cc91d906..775fcc4ca2 100644
--- a/tools/tool_result_storage.py
+++ b/tools/tool_result_storage.py
@@ -19,11 +19,7 @@ import shlex
 import threading
 import time
 
-from tools.budget_config import (
-    DEFAULT_PREVIEW_SIZE_CHARS,
-    BudgetConfig,
-    DEFAULT_BUDGET,
-)
+from tools.budget_config import DEFAULT_PREVIEW_SIZE_CHARS, BudgetConfig, DEFAULT_BUDGET
 
 logger = logging.getLogger(__name__)
 PERSISTED_OUTPUT_TAG = "<persisted-output>"
@@ -187,11 +183,7 @@ def _write_to_sandbox(content: str, remote_path: str, env) -> bool:
 
 
 def _build_persisted_message(
-    preview: str,
-    has_more: bool,
-    original_size: int,
-    file_path: str,
-) -> str:
+    preview: str, has_more: bool, original_size: int, file_path: str) -> str:
     """Build the <persisted-output> replacement block."""
     size_kb = original_size / 1024
     size_str = f"{size_kb / 1024:.1f} MB" if size_kb >= 1024 else f"{size_kb:.1f} KB"
@@ -205,8 +197,7 @@ def _build_persisted_message(
         "remote API; the full result is already on disk.\n\n"
         f"Preview (first {len(preview)} chars):\n"
         + preview + ("\n..." if has_more else "")
-        + f"\n{PERSISTED_OUTPUT_CLOSING_TAG}"
-    )
+        + f"\n{PERSISTED_OUTPUT_CLOSING_TAG}")
 
 
 _PERSISTED_PATH_RE = re.compile(r"^Full output saved to: (.+)$", re.MULTILINE)
@@ -228,8 +219,7 @@ def maybe_persist_tool_result(
     tool_use_id: str,
     env=None,
     config: BudgetConfig = DEFAULT_BUDGET,
-    threshold: int | float | None = None,
-) -> str:
+    threshold: int | float | None = None) -> str:
     """Layer 2: persist an oversized result, return preview + path.
     ``threshold`` overrides ``config.resolve_threshold(tool_name)``. Falls back
     to inline truncation when no write location succeeds."""
@@ -243,8 +233,7 @@ def maybe_persist_tool_result(
     def _persisted(path: str, host_suffix: str = "") -> str:
         logger.info(
             "Persisted large tool result: %s (%s, %d chars -> %s%s)",
-            tool_name, tool_use_id, len(content), path, host_suffix,
-        )
+            tool_name, tool_use_id, len(content), path, host_suffix)
         return _build_persisted_message(preview, has_more, len(content), path)
 
     # Always persist host-side first: cache/spillover is the single canonical home.
@@ -269,20 +258,15 @@ def maybe_persist_tool_result(
 
     logger.info(
         "Inline-truncating large tool result: %s (%d chars, no sandbox write)",
-        tool_name, len(content),
-    )
+        tool_name, len(content))
     return (
         f"{preview}\n\n"
         f"[Truncated: tool response was {len(content):,} chars. "
-        f"Full output could not be saved to sandbox.]"
-    )
+        f"Full output could not be saved to sandbox.]")
 
 
 def enforce_turn_budget(
-    tool_messages: list[dict],
-    env=None,
-    config: BudgetConfig = DEFAULT_BUDGET,
-) -> list[dict]:
+    tool_messages: list[dict], env=None, config: BudgetConfig = DEFAULT_BUDGET) -> list[dict]:
     """Layer 3: persist the largest non-persisted results first until the turn's
     aggregate is under budget. Mutates the list in-place and returns it."""
     candidates = []
@@ -303,8 +287,7 @@ def enforce_turn_budget(
         tool_use_id = tool_messages[idx].get("tool_call_id", f"budget_{idx}")
         replacement = maybe_persist_tool_result(
             content=content, tool_name=_BUDGET_TOOL_NAME, tool_use_id=tool_use_id,
-            env=env, config=config, threshold=0,
-        )
+            env=env, config=config, threshold=0)
         if replacement != content:
             total_size += len(replacement) - size
             tool_messages[idx]["content"] = replacement
diff --git a/tools/tool_search.py b/tools/tool_search.py
index a10952f1e4..333683a070 100644
--- a/tools/tool_search.py
+++ b/tools/tool_search.py
@@ -22,25 +22,12 @@ from typing import Any, Dict, Iterable, List, Optional, Tuple
 
 from tools.registry import tool_error
 from tools.tool_search_names import (  # noqa: F401 — re-exported public names
-    BRIDGE_TOOL_NAMES,
-    TOOL_CALL_NAME,
-    TOOL_DESCRIBE_NAME,
-    TOOL_SEARCH_NAME,
+    BRIDGE_TOOL_NAMES, TOOL_CALL_NAME, TOOL_DESCRIBE_NAME, TOOL_SEARCH_NAME,
 )
 from tools.tool_search_catalog import (  # noqa: F401 — re-exported public/test names
-    CHARS_PER_TOKEN,
-    CatalogEntry,
-    _corpus_stats,
-    _entry_search_text,
-    _fn,
-    _listing_group_label,
-    _registry_entry,
-    _short_desc,
-    _stem,
-    _tokenize,
-    build_catalog,
-    build_catalog_listing_with_form,
-    search_catalog,
+    CHARS_PER_TOKEN, CatalogEntry, _corpus_stats, _entry_search_text, _fn,
+    _listing_group_label, _registry_entry, _short_desc, _stem, _tokenize,
+    build_catalog, build_catalog_listing_with_form, search_catalog,
 )
 from tools.tool_search_validation import validate_deferred_call_args  # noqa: F401
 
@@ -94,9 +81,7 @@ class ToolSearchConfig:
             # A list replaces the curated default wholesale; anything else = curated.
             defer_tools=(
                 frozenset(str(n).strip() for n in defer_raw if str(n).strip())
-                if isinstance(defer_raw, (list, tuple, set)) else None
-            ),
-        )
+                if isinstance(defer_raw, (list, tuple, set)) else None))
 
 
 _TRI_STATE_ALIASES = {"true": "on", "1": "on", "yes": "on", "false": "off", "0": "off", "no": "off"}
@@ -202,8 +187,7 @@ def _tool_def_names(tool_defs: Iterable[Dict[str, Any]]) -> Iterable[str]:
 
 def classify_tools(
     tool_defs: List[Dict[str, Any]],
-    defer_tools: Optional[frozenset] = None,
-) -> Tuple[List[Dict[str, Any]], List[Dict[str, Any]]]:
+    defer_tools: Optional[frozenset] = None) -> Tuple[List[Dict[str, Any]], List[Dict[str, Any]]]:
     """Split a tool-defs list into (visible, deferrable). Bridge tools are
     dropped (they are re-added after classification)."""
     visible: List[Dict[str, Any]] = []
@@ -236,8 +220,7 @@ def estimate_tokens_from_schemas(tool_defs: Iterable[Dict[str, Any]]) -> int:
 def should_activate(
     config: ToolSearchConfig,
     deferrable_tokens: int,
-    context_length: Optional[int],
-) -> bool:
+    context_length: Optional[int]) -> bool:
     """``"off"`` never activates; ``"on"``/``"auto"`` activate whenever any
     deferrable tool exists. ``"auto"`` is an alias of ``"on"`` reserved for a
     future budget-gated mode — do not distinguish them without that design.
@@ -280,8 +263,7 @@ def _search_description(deferred_count: int, listing: Optional[str], listing_for
         "tool's description. Follow with "
         f"`{TOOL_DESCRIBE_NAME}` to load full parameter schemas, "
         f"then `{TOOL_CALL_NAME}` to invoke. Tools listed at the top of this "
-        "system prompt are already available and do not need to be searched."
-    )
+        "system prompt are already available and do not need to be searched.")
     if not listing:
         return desc
     if listing_form == "groups":
@@ -290,28 +272,24 @@ def _search_description(deferred_count: int, listing: Optional[str], listing_for
             "through this bridge. For any request in these domains, search "
             "here FIRST — do not claim the capability is unavailable and do "
             "not substitute a generic tool (terminal/browser) without "
-            "searching.\n\n" + listing
-        )
+            "searching.\n\n" + listing)
     desc += (
         "\n\nEvery deferred capability is listed below. If a tool name "
         "appears here, do NOT claim it is unavailable — load it with "
         f"`{TOOL_DESCRIBE_NAME}` (skip `{TOOL_SEARCH_NAME}` when you "
-        "already see the exact name)."
-    )
+        "already see the exact name).")
     if listing_form == "mixed":
         desc += (
             " For servers marked 'names not listed', the tools exist "
             f"too — find them with `{TOOL_SEARCH_NAME}` before "
-            "concluding anything is missing."
-        )
+            "concluding anything is missing.")
     return desc + "\n\n" + listing
 
 
 def bridge_tool_schemas(
     deferred_count: int,
     listing: Optional[str] = None,
-    listing_form: str = "",
-) -> List[Dict[str, Any]]:
+    listing_form: str = "") -> List[Dict[str, Any]]:
     """Bridge tool schemas injected in place of deferred tools. Kept short —
     every byte is paid on every turn. ``listing`` is embedded in the tool_search
     description; ``listing_form`` picks the framing (per-tool forms say "skip
@@ -384,8 +362,7 @@ def assemble_tool_defs(
     tool_defs: List[Dict[str, Any]],
     *,
     context_length: Optional[int] = None,
-    config: Optional[ToolSearchConfig] = None,
-) -> AssemblyResult:
+    config: Optional[ToolSearchConfig] = None) -> AssemblyResult:
     """Return the tool-defs list the model should actually see: passthrough
     when inactive, otherwise deferrable tools replaced by the three bridge
     tools. Idempotent — bridge tools already present are stripped first."""
@@ -403,8 +380,7 @@ def assemble_tool_defs(
             tool_defs=incoming, activated=False, deferred_count=len(deferrable),
             deferred_tokens=deferrable_tokens,
             threshold_tokens=int((context_length or 0) * (config.threshold_pct / 100.0)),
-            tier=0,
-        )
+            tier=0)
 
     listing, listing_form = None, "none"
     listing_budget = listing_token_budget(config, context_length)
@@ -416,13 +392,11 @@ def assemble_tool_defs(
     logger.info(
         "tool_search activated (tier %d): %d core/visible tools kept, %d deferred "
         "(~%d tokens), listing %s (budget ~%d tokens)",
-        tier, len(visible), len(deferrable), deferrable_tokens, listing_form, listing_budget,
-    )
+        tier, len(visible), len(deferrable), deferrable_tokens, listing_form, listing_budget)
     return AssemblyResult(
         tool_defs=visible + bridge, activated=True, deferred_count=len(deferrable),
         deferred_tokens=deferrable_tokens, threshold_tokens=listing_budget,
-        tier=tier, listing_form=listing_form,
-    )
+        tier=tier, listing_form=listing_form)
 
 
 def is_bridge_tool(name: str) -> bool:
@@ -443,8 +417,7 @@ def _shared_tool_record(entry: CatalogEntry) -> Dict[str, Any]:
         "source": entry.source,
         "source_name": entry.source_name,
         "description": (entry.description or "")[:400],  # cap chatty MCP descriptions
-        "required": [r[:64] for r in required if isinstance(r, str)][:32],
-    }
+        "required": [r[:64] for r in required if isinstance(r, str)][:32]}
 
 
 def _available_source_summary(catalog: List[CatalogEntry]) -> List[Dict[str, Any]]:
@@ -519,8 +492,7 @@ def dispatch_tool_search(args: Dict[str, Any],
                 "This query returned no lexical matches, but the sources above "
                 "are connected and their tools remain available. Retry "
                 "tool_search with the service name plus a concrete action or "
-                "object before concluding the capability is unavailable."
-            )
+                "object before concluding the capability is unavailable.")
         results.append(group)
     return json.dumps({
         "queries": queries,
@@ -560,16 +532,13 @@ def dispatch_tool_describe(args: Dict[str, Any],
         if fn is not None:
             tools[name] = {
                 "description": fn.get("description", ""),
-                "parameters": fn.get("parameters", {}),
-            }
+                "parameters": fn.get("parameters", {})}
         elif _registry_entry(name) is not None and not is_deferrable_tool_name(
-            name, load_config_readonly().effective_defer_tools
-        ):
+            name, load_config_readonly().effective_defer_tools):
             # Registered but bridge/core/GUI-surface: a real name, wrong door.
             errors[name] = (
                 f"'{name}' is not a deferrable tool. If you see it in the tools list "
-                "already, call it directly; otherwise check the spelling against tool_search."
-            )
+                "already, call it directly; otherwise check the spelling against tool_search.")
         else:
             not_found.append(name)
 
@@ -590,8 +559,7 @@ def scoped_deferrable_names(tool_defs: List[Dict[str, Any]]) -> frozenset[str]:
     defer_tools = load_config_readonly().effective_defer_tools
     return frozenset(
         name for name in _tool_def_names(tool_defs)
-        if name and is_deferrable_tool_name(name, defer_tools)
-    )
+        if name and is_deferrable_tool_name(name, defer_tools))
 
 
 def resolve_underlying_call(args: Dict[str, Any]) -> Tuple[Optional[str], Dict[str, Any], Optional[str]]:
@@ -616,34 +584,16 @@ def resolve_underlying_call(args: Dict[str, Any]) -> Tuple[Optional[str], Dict[s
     if not is_deferrable_tool_name(name, load_config_readonly().effective_defer_tools):
         return None, {}, (
             f"'{name}' is not a deferrable tool. If it appears in the model-facing tools "
-            "list already, call it directly instead of via tool_call."
-        )
+            "list already, call it directly instead of via tool_call.")
     return name, raw_args, None
 
 
 __all__ = [
-    "TOOL_SEARCH_NAME",
-    "TOOL_DESCRIBE_NAME",
-    "TOOL_CALL_NAME",
-    "BRIDGE_TOOL_NAMES",
-    "ToolSearchConfig",
-    "CatalogEntry",
-    "AssemblyResult",
-    "load_config",
-    "is_deferrable_tool_name",
-    "classify_tools",
-    "estimate_tokens_from_schemas",
-    "should_activate",
-    "build_catalog",
-    "build_catalog_listing_with_form",
-    "listing_token_budget",
-    "search_catalog",
-    "bridge_tool_schemas",
-    "assemble_tool_defs",
-    "is_bridge_tool",
-    "dispatch_tool_search",
-    "dispatch_tool_describe",
-    "resolve_underlying_call",
-    "scoped_deferrable_names",
-    "validate_deferred_call_args",
+    "TOOL_SEARCH_NAME", "TOOL_DESCRIBE_NAME", "TOOL_CALL_NAME", "BRIDGE_TOOL_NAMES",
+    "ToolSearchConfig", "CatalogEntry", "AssemblyResult", "load_config",
+    "is_deferrable_tool_name", "classify_tools", "estimate_tokens_from_schemas",
+    "should_activate", "build_catalog", "build_catalog_listing_with_form", "listing_token_budget",
+    "search_catalog", "bridge_tool_schemas", "assemble_tool_defs", "is_bridge_tool",
+    "dispatch_tool_search", "dispatch_tool_describe", "resolve_underlying_call",
+    "scoped_deferrable_names", "validate_deferred_call_args",
 ]
diff --git a/tools/tool_search_catalog.py b/tools/tool_search_catalog.py
index f371d047f9..986c73f672 100644
--- a/tools/tool_search_catalog.py
+++ b/tools/tool_search_catalog.py
@@ -122,8 +122,7 @@ def build_catalog(tool_defs: List[Dict[str, Any]]) -> List[CatalogEntry]:
             schema=td,
             source=source,
             source_name=source_name,
-            _tokens=_tokenize(_entry_search_text(td, source_label)),
-        ))
+            _tokens=_tokenize(_entry_search_text(td, source_label))))
     return catalog
 
 
@@ -167,8 +166,7 @@ def search_catalog(
     query: str,
     limit: int = 5,
     *,
-    corpus_stats: Optional[_CorpusStats] = None,
-) -> List[CatalogEntry]:
+    corpus_stats: Optional[_CorpusStats] = None) -> List[CatalogEntry]:
     """Top-``limit`` catalog entries for ``query`` by BM25 (exact name match
     ranks first). Falls back to a name-substring match only when NO query
     token appears in any document (e.g. "hub" vs ``github_*``); the IDF
@@ -226,10 +224,7 @@ def _listing_group_label(source_name: str) -> str:
 
 
 def build_catalog_listing_with_form(
-    deferrable: List[Dict[str, Any]],
-    *,
-    max_tokens: int = 4000,
-) -> Tuple[Optional[str], str]:
+    deferrable: List[Dict[str, Any]], *, max_tokens: int = 4000) -> Tuple[Optional[str], str]:
     """Render the skills-style deferred-catalog manifest: ``- name: short desc``
     lines grouped under a heading per source (MCP server / plugin toolset).
 
diff --git a/tools/tool_search_validation.py b/tools/tool_search_validation.py
index 39d8658253..17e3513516 100644
--- a/tools/tool_search_validation.py
+++ b/tools/tool_search_validation.py
@@ -25,8 +25,7 @@ def _schema_for_local_validation(node: Any) -> Any:
     # Literal keywords hold instance data, not schemas: copy byte-for-byte.
     normalized = {
         key: copy.deepcopy(value) if key in _SCHEMA_LITERAL_KEYS else _schema_for_local_validation(value)
-        for key, value in node.items() if key != "nullable"
-    }
+        for key, value in node.items() if key != "nullable"}
     if node.get("nullable") is not True:
         return normalized
     schema_type = normalized.get("type")
@@ -56,8 +55,7 @@ def _schema_has_external_ref(node: Any) -> bool:
     return any(
         _schema_has_external_ref(value)
         for key, value in node.items()
-        if key not in _SCHEMA_LITERAL_KEYS
-    )
+        if key not in _SCHEMA_LITERAL_KEYS)
 
 
 def _validation_path(error: Any) -> str:
@@ -76,8 +74,7 @@ def _validation_path(error: Any) -> str:
 def _validation_error(message: str, *, path: str, constraint: str, parameters: Any) -> str:
     return tool_error(
         message, path=path, constraint=constraint, parameters=parameters,
-        hint="Retry tool_call with 'arguments' matching the parameters schema above.",
-    )
+        hint="Retry tool_call with 'arguments' matching the parameters schema above.")
 
 
 def validate_deferred_call_args(name: str, args: Dict[str, Any]) -> Optional[str]:
@@ -106,8 +103,7 @@ def validate_deferred_call_args(name: str, args: Dict[str, Any]) -> Optional[str
                 return _validation_error(
                     f"tool_call to '{name}' is missing required argument(s): "
                     f"{', '.join(missing)}. The tool was NOT invoked.",
-                    path="arguments", constraint="required", parameters=params,
-                )
+                    path="arguments", constraint="required", parameters=params)
 
         validation_schema = _schema_for_local_validation(params)
         if _schema_has_external_ref(validation_schema):
@@ -143,8 +139,7 @@ def validate_deferred_call_args(name: str, args: Dict[str, Any]) -> Optional[str
         return _validation_error(
             f"tool_call to '{name}' failed argument validation at {path} "
             f"({constraint}): {detail}. The tool was NOT invoked.",
-            path=path, constraint=constraint, parameters=params,
-        )
+            path=path, constraint=constraint, parameters=params)
     except Exception:  # pragma: no cover — never block dispatch on validator bugs
         logger.debug("validate_deferred_call_args failed for %s", name, exc_info=True)
         return None

From 5db38e3a4e618c06f9fec942581c3a2e397c7eef Mon Sep 17 00:00:00 2001
From: Teknium <127238744+teknium1@users.noreply.github.com>
Date: Wed, 2 Sep 2026 22:51:44 -0700
Subject: [PATCH 5/6] =?UTF-8?q?refactor(tools):=20group=20H=20=E2=80=94=20?=
 =?UTF-8?q?compact=20module/function=20docstrings=20by=20hand=20(keep=20WH?=
 =?UTF-8?q?Y/invariants)?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 tools/spill_safety.py         | 22 +++++---------
 tools/tip_tool.py             | 13 +++------
 tools/tool_backend_helpers.py | 14 ++++-----
 tools/tool_output_limits.py   | 17 ++++-------
 tools/tool_result_storage.py  | 16 ++++------
 tools/tool_search.py          | 55 ++++++++++++++---------------------
 tools/tour_tool.py            | 17 +++++------
 7 files changed, 57 insertions(+), 97 deletions(-)

diff --git a/tools/spill_safety.py b/tools/spill_safety.py
index 69b68ec172..0f2d12ff85 100644
--- a/tools/spill_safety.py
+++ b/tools/spill_safety.py
@@ -1,17 +1,11 @@
-"""Symlink-safe creation helpers for spill/cache files.
-
-Spill files live in predictable directories under ``~/.hermes``; a plain
-``open(path, "w")`` there would follow a pre-planted symlink and let a local
-process redirect the write onto ``~/.bashrc``, ``authorized_keys``, etc. Every
-helper refuses symlinks by construction: new files use ``O_CREAT | O_EXCL``
-(fails on ANY existing path, including a dangling link); overwrites ``lstat`` +
-``unlink`` the existing path first (removes the link, never its target), then
-create exclusively, so the pair can't be raced. ``private=True`` (default)
-forces ``0o700`` dirs / ``0o600`` files for spills that may hold pre-redaction
-secrets; ``private=False`` keeps umask perms for cache dirs bind-mounted into
-remote backends (``credential_files._CACHE_DIRS``) where a non-root container
-UID must read them. Disk failures are the caller's concern: helpers raise ``OSError``.
-"""
+"""Symlink-safe creation helpers for spill/cache files under ``~/.hermes``, where a
+plain ``open(path, "w")`` would follow a pre-planted symlink onto ``~/.bashrc`` etc.
+New files use ``O_CREAT | O_EXCL`` (fails on ANY existing path, even a dangling
+link); overwrites ``lstat`` + ``unlink`` first (removes the link, never its target)
+then create exclusively, so the pair can't be raced. ``private=True`` (default) =
+``0o700`` dirs / ``0o600`` files for spills that may hold pre-redaction secrets;
+``private=False`` keeps umask perms for cache dirs bind-mounted into remote backends
+(``credential_files._CACHE_DIRS``). Disk failures raise ``OSError`` to the caller."""
 
 from __future__ import annotations
 
diff --git a/tools/tip_tool.py b/tools/tip_tool.py
index c5dd4eb76b..c9edcf40c1 100644
--- a/tools/tip_tool.py
+++ b/tools/tip_tool.py
@@ -1,13 +1,8 @@
 #!/usr/bin/env python3
-"""Point at something in the Hermes desktop GUI and say one line about it.
-
-The quiet sibling of ``tour``: same ``data-tour`` handles and discovery call,
-but an arrow bubble with no scrim/spotlight/paging. Fire-and-forget — a tip is
-not a question, so blocking the turn on a round-trip would stall the reply.
-
-Lives in ``desktop_ui`` (GUI sessions only) and withdraws itself when the user
-turns tips off, so the model is never offered a tool whose call would fail.
-"""
+"""Point at something in the Hermes desktop GUI and say one line about it — the quiet
+sibling of ``tour`` (same ``data-tour`` handles) with no scrim/spotlight/paging.
+Fire-and-forget: a tip is not a question, so blocking on a round-trip would stall the
+reply. Lives in ``desktop_ui`` and withdraws itself when the user turns tips off."""
 
 import json
 
diff --git a/tools/tool_backend_helpers.py b/tools/tool_backend_helpers.py
index 5a8aba8fc1..e7bbc0036c 100644
--- a/tools/tool_backend_helpers.py
+++ b/tools/tool_backend_helpers.py
@@ -18,18 +18,14 @@ _VALID_MODAL_MODES = {"auto", "direct", "managed"}
 
 
 def managed_nous_tools_enabled(*, force_fresh: bool = False) -> bool:
-    """True when the user is entitled to the Nous Tool Gateway (coarse gate:
-    paid Portal service access OR a live free tool pool). Fails closed on
-    unknown/error entitlement — never blocks startup. Per-category coverage is
-    narrowed by callers via ``tool_gateway_entitled_for``. ``force_fresh=True``
-    is for interactive flows that must see a just-purchased grant."""
+    """Coarse gate: entitled to the Nous Tool Gateway (paid Portal access OR a live free
+    pool). Fails closed on unknown/error — never blocks startup. Callers narrow per category
+    via ``tool_gateway_entitled_for``; ``force_fresh`` is for flows needing a just-bought grant."""
     try:
         from hermes_cli.nous_account import get_nous_portal_account_info
 
-        if force_fresh:
-            account_info = get_nous_portal_account_info(force_fresh=True)
-        else:
-            account_info = get_nous_portal_account_info()
+        account_info = (get_nous_portal_account_info(force_fresh=True) if force_fresh
+                        else get_nous_portal_account_info())
         return bool(account_info.logged_in) and account_info.tool_gateway_entitled
     except Exception:
         return False
diff --git a/tools/tool_output_limits.py b/tools/tool_output_limits.py
index 8b670aa289..a2fc8500ca 100644
--- a/tools/tool_output_limits.py
+++ b/tools/tool_output_limits.py
@@ -1,15 +1,8 @@
-"""Configurable tool-output truncation limits (``tool_output`` in config.yaml).
-
-Centralises the caps once hardcoded in ``terminal_tool`` (``max_bytes``) and
-``file_operations`` (``max_lines`` / ``max_line_length``). Defaults equal the
-old constants and the reader never raises, so behaviour is unchanged when the
-section is absent or malformed::
-
-    tool_output:
-      max_bytes: 100000        # terminal output cap (chars)
-      max_lines: 5000          # read_file pagination + truncation cap
-      max_line_length: 2000    # per-line cap before '... [truncated]'
-"""
+"""Configurable tool-output truncation limits (``tool_output`` in config.yaml):
+``max_bytes`` (terminal output cap), ``max_lines`` (read_file pagination cap),
+``max_line_length`` (per-line cap before '... [truncated]'). Defaults equal the
+constants once hardcoded in terminal_tool / file_operations and the reader never
+raises, so behaviour is unchanged when the section is absent or malformed."""
 
 from __future__ import annotations
 
diff --git a/tools/tool_result_storage.py b/tools/tool_result_storage.py
index 775fcc4ca2..91bfef712d 100644
--- a/tools/tool_result_storage.py
+++ b/tools/tool_result_storage.py
@@ -1,15 +1,11 @@
 """Tool result persistence -- preserves large outputs instead of truncating.
 
-Three layers against context-window overflow: (1) per-tool output caps inside
-each tool; (2) ``maybe_persist_tool_result`` — output over the tool's threshold
-is persisted and replaced in-context by a preview + path. The canonical home is
-ALWAYS host-side ``$HERMES_HOME/cache/spillover/{tool_use_id}.txt`` (works for
-MCP-only/cron/gateway sessions that never ran a terminal); remote backends see
-the translated in-sandbox path (``cache/spillover`` is in the auto-mounted cache
-list), probed for readability, else a copy written into the sandbox temp dir.
-(3) ``enforce_turn_budget`` — spills the largest results of a turn until the
-aggregate fits.
-"""
+Layers against context overflow: (1) per-tool caps inside each tool; (2)
+``maybe_persist_tool_result`` — output over the tool's threshold is persisted and
+replaced by a preview + path. Canonical home is ALWAYS host-side
+``$HERMES_HOME/cache/spillover/{id}.txt`` (works for sessions that never ran a
+terminal); remote backends get the translated in-sandbox path (probed for
+readability) else a copy in the sandbox temp dir. (3) ``enforce_turn_budget``."""
 
 import hashlib
 import logging
diff --git a/tools/tool_search.py b/tools/tool_search.py
index 333683a070..3695be2f9e 100644
--- a/tools/tool_search.py
+++ b/tools/tool_search.py
@@ -1,15 +1,11 @@
 """Progressive tool disclosure ("tool search"): MCP/plugin tools and a curated set
 of event-triggered core tools are replaced in the model-visible array by three
-bridge tools — tool_search / tool_describe / tool_call.
-
-Invariants: working-set core tools (``toolsets._HERMES_CORE_TOOLS``) and
-session-gated GUI toolsets never defer unless named in ``defer``; ANY deferrable
-tool activates the bridge (the listing scales with budget, not activation); the
-catalog is stateless — rebuilt from the live tool-defs on every assembly (a
-session-keyed catalog drifts from the registry and silently drops tools); bridge
-calls route through ``model_tools.handle_function_call`` so guardrails, hooks,
-approvals and truncation fire identically.
-"""
+bridge tools — tool_search / tool_describe / tool_call. Invariants: core tools
+(``toolsets._HERMES_CORE_TOOLS``) and session-gated GUI toolsets never defer unless
+named in ``defer``; ANY deferrable tool activates the bridge (the listing scales
+with budget, not activation); the catalog is stateless — rebuilt from the live
+tool-defs every assembly (a session-keyed one drifts and silently drops tools);
+bridge calls route through ``model_tools.handle_function_call`` (same guardrails)."""
 
 from __future__ import annotations
 
@@ -111,6 +107,7 @@ def _safe_float(value: Any, fallback: float) -> float:
 
 
 def _config_from_loader(loader_name: str) -> ToolSearchConfig:
+    """Tool-search config via ``hermes_cli.config.<loader_name>`` (defaults on any failure)."""
     try:
         import hermes_cli.config as _cfg_mod
         cfg = getattr(_cfg_mod, loader_name)() or {}
@@ -122,12 +119,11 @@ def _config_from_loader(loader_name: str) -> ToolSearchConfig:
 
 
 def load_config() -> ToolSearchConfig:
-    """Load tool-search config from the user config file."""
     return _config_from_loader("load_config")
 
 
 def load_config_readonly() -> ToolSearchConfig:
-    """Load tool-search config without copying the cached full config."""
+    """Same as ``load_config`` without copying the cached full config."""
     return _config_from_loader("load_config_readonly")
 
 
@@ -221,11 +217,9 @@ def should_activate(
     config: ToolSearchConfig,
     deferrable_tokens: int,
     context_length: Optional[int]) -> bool:
-    """``"off"`` never activates; ``"on"``/``"auto"`` activate whenever any
-    deferrable tool exists. ``"auto"`` is an alias of ``"on"`` reserved for a
-    future budget-gated mode — do not distinguish them without that design.
-    ``context_length`` stays in the signature for caller compatibility; the
-    threshold governs the listing budget, not activation."""
+    """``"off"`` never activates; ``"on"``/``"auto"`` activate whenever any deferrable
+    tool exists ("auto" is reserved for a future budget-gated mode — do not distinguish
+    them without that design). ``context_length`` is kept for caller compatibility."""
     return config.enabled != "off" and deferrable_tokens > 0
 
 
@@ -363,9 +357,8 @@ def assemble_tool_defs(
     *,
     context_length: Optional[int] = None,
     config: Optional[ToolSearchConfig] = None) -> AssemblyResult:
-    """Return the tool-defs list the model should actually see: passthrough
-    when inactive, otherwise deferrable tools replaced by the three bridge
-    tools. Idempotent — bridge tools already present are stripped first."""
+    """Tool-defs the model should see: passthrough when inactive, else deferrable tools
+    replaced by the three bridge tools. Idempotent — existing bridge tools are stripped first."""
     if config is None:
         config = load_config()
     incoming = [td for td, name in zip(tool_defs, _tool_def_names(tool_defs))
@@ -404,9 +397,8 @@ def is_bridge_tool(name: str) -> bool:
 
 
 def _shared_tool_record(entry: CatalogEntry) -> Dict[str, Any]:
-    """One record for the response's shared ``tools`` map (held once per tool;
-    per-query groups carry names only). ``required`` lets the model attempt a
-    trivial call without a ``tool_describe`` round-trip."""
+    """One record for the shared ``tools`` map (per-query groups carry names only);
+    ``required`` lets the model attempt a trivial call without a ``tool_describe`` round-trip."""
     schema = entry.schema if isinstance(entry.schema, dict) else {}
     fn = schema.get("function")
     params = fn.get("parameters") if isinstance(fn, dict) else None
@@ -430,9 +422,8 @@ def _available_source_summary(catalog: List[CatalogEntry]) -> List[Dict[str, Any
 def _string_list_arg(
     args: Dict[str, Any], key: str, *, dedupe: bool, max_items: int, retry_hint: str,
 ) -> Tuple[Optional[List[str]], Optional[str]]:
-    """Read a list-of-strings bridge argument -> ``(items, error_json)``. A bare
-    string (a common model slip) counts as a one-item list. Rejects non-list
-    input, an empty list (after stripping blanks), and more than ``max_items``."""
+    """Read a list-of-strings bridge argument -> ``(items, error_json)``. A bare string (a
+    common model slip) is a one-item list; rejects non-lists, all-blank lists, > ``max_items``."""
     raw = args.get(key)
     if isinstance(raw, str):
         raw = [raw]
@@ -552,10 +543,9 @@ def dispatch_tool_describe(args: Dict[str, Any],
 
 
 def scoped_deferrable_names(tool_defs: List[Dict[str, Any]]) -> frozenset[str]:
-    """Deferrable tool names in the *pre-assembly* ``tool_defs`` of the session's
-    toolset scope — the universe ``tool_call`` may reach. Gates both bridge
-    dispatch and the executor unwrap so a restricted session cannot invoke an
-    out-of-scope tool via the bridge."""
+    """Deferrable names in the *pre-assembly* ``tool_defs`` of the session scope — the
+    universe ``tool_call`` may reach. Gates bridge dispatch AND the executor unwrap so a
+    restricted session cannot invoke an out-of-scope tool via the bridge."""
     defer_tools = load_config_readonly().effective_defer_tools
     return frozenset(
         name for name in _tool_def_names(tool_defs)
@@ -563,9 +553,8 @@ def scoped_deferrable_names(tool_defs: List[Dict[str, Any]]) -> frozenset[str]:
 
 
 def resolve_underlying_call(args: Dict[str, Any]) -> Tuple[Optional[str], Dict[str, Any], Optional[str]]:
-    """Parse a ``tool_call`` invocation into (underlying_name, args, error_msg);
-    ``(None, {}, msg)`` on parse error. Shared by dispatch, display, and the
-    trajectory recorder so all three agree on the underlying tool."""
+    """Parse a ``tool_call`` invocation -> (underlying_name, args, error_msg); ``(None, {}, msg)``
+    on error. Shared by dispatch, display and the trajectory recorder so all three agree."""
     name = str(args.get("name") or "").strip()
     if not name:
         return None, {}, "tool_call requires a 'name' argument"
diff --git a/tools/tour_tool.py b/tools/tour_tool.py
index 0422443d8e..b1d4022f06 100644
--- a/tools/tour_tool.py
+++ b/tools/tour_tool.py
@@ -1,14 +1,11 @@
 #!/usr/bin/env python3
-"""Guided tour (highlight + narrate UI elements) in the Hermes desktop GUI.
-
-Generic, no baked-in tours: the agent discovers targets (``action="targets"``),
-then highlights by CSS selector one step at a time (``show``) or hands over a
-step list the user pages with Next/Prev (``start``). Round-trips through the
-gateway blocking-prompt bridge (``tour.request``/``tour.respond``) so the agent
-learns whether the selector matched. Lives in ``desktop_ui`` and withdraws itself
-when the user turns tours off: a tour takes the whole screen, so "off" must mean
-the model is never told the tool exists rather than offered a call that fails.
-"""
+"""Guided tour (highlight + narrate UI elements) in the Hermes desktop GUI. Generic:
+the agent discovers targets (``action="targets"``), then highlights one step at a
+time (``show``) or hands over a step list the user pages (``start``). Round-trips
+through the gateway blocking-prompt bridge (``tour.request``/``tour.respond``) so the
+agent learns whether the selector matched. Lives in ``desktop_ui`` and withdraws
+itself when tours are off: a tour takes the whole screen, so "off" must mean the
+model is never told the tool exists rather than offered a call that fails."""
 
 import json
 from typing import Callable, Optional

From 04247b98405b7301d07c8a22c1be2c90336175b1 Mon Sep 17 00:00:00 2001
From: Teknium <127238744+teknium1@users.noreply.github.com>
Date: Wed, 2 Sep 2026 22:55:05 -0700
Subject: [PATCH 6/6] =?UTF-8?q?refactor(tools):=20group=20H=20=E2=80=94=20?=
 =?UTF-8?q?wrap=20introduced=20>100-col=20lines?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 tools/tool_backend_helpers.py   | 3 ++-
 tools/tool_output_limits.py     | 3 ++-
 tools/tool_result_storage.py    | 3 ++-
 tools/tool_search_validation.py | 3 ++-
 4 files changed, 8 insertions(+), 4 deletions(-)

diff --git a/tools/tool_backend_helpers.py b/tools/tool_backend_helpers.py
index e7bbc0036c..fc3cdb7cf8 100644
--- a/tools/tool_backend_helpers.py
+++ b/tools/tool_backend_helpers.py
@@ -148,7 +148,8 @@ def resolve_provider_secret(
             return ""
     except Exception:  # pragma: no cover — secret_scope is in-repo
         pass
-    key = str(env_getter(env_var) or "").strip() if env_getter is not None else _dotenv_value(env_var)
+    key = (str(env_getter(env_var) or "").strip() if env_getter is not None
+           else _dotenv_value(env_var))
     if key or not provider_id:
         return key
     try:
diff --git a/tools/tool_output_limits.py b/tools/tool_output_limits.py
index a2fc8500ca..c13bc78831 100644
--- a/tools/tool_output_limits.py
+++ b/tools/tool_output_limits.py
@@ -47,7 +47,8 @@ def get_tool_output_limits() -> Dict[str, int]:
     _cached_limits = {
         "max_bytes": _coerce_positive_int(section.get("max_bytes"), DEFAULT_MAX_BYTES),
         "max_lines": _coerce_positive_int(section.get("max_lines"), DEFAULT_MAX_LINES),
-        "max_line_length": _coerce_positive_int(section.get("max_line_length"), DEFAULT_MAX_LINE_LENGTH),
+        "max_line_length": _coerce_positive_int(
+            section.get("max_line_length"), DEFAULT_MAX_LINE_LENGTH),
     }
     return _cached_limits
 
diff --git a/tools/tool_result_storage.py b/tools/tool_result_storage.py
index 91bfef712d..c3d714946d 100644
--- a/tools/tool_result_storage.py
+++ b/tools/tool_result_storage.py
@@ -287,5 +287,6 @@ def enforce_turn_budget(
         if replacement != content:
             total_size += len(replacement) - size
             tool_messages[idx]["content"] = replacement
-            logger.info("Budget enforcement: persisted tool result %s (%d chars)", tool_use_id, size)
+            logger.info("Budget enforcement: persisted tool result %s (%d chars)",
+                        tool_use_id, size)
     return tool_messages
diff --git a/tools/tool_search_validation.py b/tools/tool_search_validation.py
index 17e3513516..1379895eb9 100644
--- a/tools/tool_search_validation.py
+++ b/tools/tool_search_validation.py
@@ -24,7 +24,8 @@ def _schema_for_local_validation(node: Any) -> Any:
         return node
     # Literal keywords hold instance data, not schemas: copy byte-for-byte.
     normalized = {
-        key: copy.deepcopy(value) if key in _SCHEMA_LITERAL_KEYS else _schema_for_local_validation(value)
+        key: (copy.deepcopy(value) if key in _SCHEMA_LITERAL_KEYS
+              else _schema_for_local_validation(value))
         for key, value in node.items() if key != "nullable"}
     if node.get("nullable") is not True:
         return normalized