From 8d6232878e1902dbbc8d1bc6030b59b4ae61b08b Mon Sep 17 00:00:00 2001
From: Teknium <127238744+teknium1@users.noreply.github.com>
Date: Wed, 2 Sep 2026 22:44:46 -0700
Subject: [PATCH 1/4] refactor(tools): compact file_tools companions; one
 eviction helper in file_state

---
 tools/file_state.py               | 154 ++++++++-----------
 tools/file_tools_paths.py         |  45 ++----
 tools/file_tools_read_tracking.py | 149 ++++++-------------
 tools/file_tools_write_guards.py  | 236 ++++++++++--------------------
 4 files changed, 192 insertions(+), 392 deletions(-)

diff --git a/tools/file_state.py b/tools/file_state.py
index 77dd0577e0..99f3e2d069 100644
--- a/tools/file_state.py
+++ b/tools/file_state.py
@@ -1,24 +1,11 @@
 """Cross-agent file state coordination.
 
 Prevents mangled edits when concurrent subagents (same process, same
-filesystem) touch the same file: subagent B writes a file that subagent A
-already read, so A's next write would clobber B's changes with stale content.
-Complements the single-agent path-overlap check in
-``run_agent._should_parallelize_tool_batch``.
-
-A process-wide ``FileStateRegistry`` tracks, per resolved path:
-  * per-agent read stamps  {task_id: {path: (mtime, read_ts, partial)}}
-  * last writer globally   {path: (task_id, write_ts)}
-  * a per-path ``threading.Lock`` for read->modify->write sections
-
-Hooks used by the file tools: ``record_read`` (read_file), ``note_write``
-(after write_file/patch), ``check_stale`` (BEFORE write_file/patch),
-``lock_path`` (wrap the whole read->modify->write block) and ``writes_since``
-(delegate_tool's subagent-completion reminder).
-
-All methods are no-ops when ``HERMES_DISABLE_FILE_STATE_GUARD=1``. This is
-separate from ``file_tools._read_tracker``, which handles per-task
-consecutive-read loop detection.
+filesystem) touch the same file: B writes a file A already read, so A's next
+write would clobber B's changes. Complements the single-agent path-overlap
+check in ``run_agent._should_parallelize_tool_batch``. A process-wide
+``FileStateRegistry`` tracks per-agent read stamps, the global last writer and
+a per-path lock; every method is a no-op under ``HERMES_DISABLE_FILE_STATE_GUARD=1``.
 """
 from __future__ import annotations
 
@@ -35,12 +22,45 @@ from typing import Dict, Iterable, List, Optional, Tuple
 # so the model re-reads in full.
 ReadStamp = Tuple[float, float, bool]
 
-# Bounded so long sessions don't accumulate unbounded state; oldest by
-# insertion order are dropped on overflow.
+# Bounded so long sessions don't accumulate unbounded state.
 _MAX_PATHS_PER_AGENT = 4096
 _MAX_GLOBAL_WRITERS = 4096
 
 
+def _disabled() -> bool:
+    # Re-read each call so tests can toggle via monkeypatch.setenv.
+    return os.environ.get("HERMES_DISABLE_FILE_STATE_GUARD", "").strip() == "1"
+
+
+def _mtime_or_none(resolved: str) -> Optional[float]:
+    try:
+        return os.path.getmtime(resolved)
+    except OSError:
+        return None
+
+
+def _fmt_ts(ts: float) -> str:
+    # Short wall-clock for warnings; avoids datetime formatting on the hot path.
+    return time.strftime("%H:%M:%S", time.localtime(ts))
+
+
+def _evict_oldest(container, cap: int) -> None:
+    """Pop entries until *container* is within *cap*.
+
+    Sets pop arbitrary entries (they only feed diagnostic summaries); dicts pop
+    oldest by insertion order. An evicted entry costs one redundant re-send
+    (dedup) or one non-mtime staleness check — graceful degradation, not a bug.
+    """
+    for _ in range(len(container) - cap):
+        try:
+            if isinstance(container, set):
+                container.pop()
+            else:
+                container.pop(next(iter(container)))
+        except (StopIteration, KeyError):
+            break
+
+
 class FileStateRegistry:
     """Process-wide coordinator for cross-agent file edits."""
 
@@ -51,50 +71,31 @@ class FileStateRegistry:
         self._meta_lock = threading.Lock()  # guards _path_locks
         self._state_lock = threading.Lock()  # guards _reads + _last_writer
 
-    def _lock_for(self, resolved: str) -> threading.Lock:
-        with self._meta_lock:
-            lock = self._path_locks.get(resolved)
-            if lock is None:
-                lock = threading.Lock()
-                self._path_locks[resolved] = lock
-            return lock
-
     @contextmanager
     def lock_path(self, resolved: str):
         """Per-path lock: threads on the same path serialize, different paths proceed."""
-        lock = self._lock_for(resolved)
-        lock.acquire()
-        try:
+        with self._meta_lock:
+            lock = self._path_locks.setdefault(resolved, threading.Lock())
+        with lock:
             yield
-        finally:
-            lock.release()
 
-    def record_read(
-        self,
-        task_id: str,
-        resolved: str,
-        *,
-        partial: bool = False,
-        mtime: Optional[float] = None,
-    ) -> None:
+    def _stamp(self, task_id: str, resolved: str, mtime: float, now: float, partial: bool) -> None:
+        """Caller holds ``_state_lock``."""
+        agent_reads = self._reads[task_id]
+        agent_reads[resolved] = (float(mtime), now, bool(partial))
+        _evict_oldest(agent_reads, _MAX_PATHS_PER_AGENT)
+
+    def record_read(self, task_id: str, resolved: str, *, partial: bool = False,
+                    mtime: Optional[float] = None) -> None:
         if _disabled():
             return
         mtime = _mtime_or_none(resolved) if mtime is None else mtime
         if mtime is None:
             return
-        now = time.time()
         with self._state_lock:
-            agent_reads = self._reads[task_id]
-            agent_reads[resolved] = (float(mtime), now, bool(partial))
-            _cap_dict(agent_reads, _MAX_PATHS_PER_AGENT)
+            self._stamp(task_id, resolved, mtime, time.time(), partial)
 
-    def note_write(
-        self,
-        task_id: str,
-        resolved: str,
-        *,
-        mtime: Optional[float] = None,
-    ) -> None:
+    def note_write(self, task_id: str, resolved: str, *, mtime: Optional[float] = None) -> None:
         """Record a successful write: global last-writer AND this agent's own
         read stamp (a write is an implicit read of the current content)."""
         if _disabled():
@@ -105,9 +106,8 @@ class FileStateRegistry:
         now = time.time()
         with self._state_lock:
             self._last_writer[resolved] = (task_id, now)
-            _cap_dict(self._last_writer, _MAX_GLOBAL_WRITERS)
-            self._reads[task_id][resolved] = (float(mtime), now, False)
-            _cap_dict(self._reads[task_id], _MAX_PATHS_PER_AGENT)
+            _evict_oldest(self._last_writer, _MAX_GLOBAL_WRITERS)
+            self._stamp(task_id, resolved, mtime, now, False)
 
     def check_stale(self, task_id: str, resolved: str) -> Optional[str]:
         """Model-facing warning if this write would be stale, else ``None``.
@@ -171,12 +171,8 @@ class FileStateRegistry:
             "Read the file first so you can write an informed edit."
         )
 
-    def writes_since(
-        self,
-        exclude_task_id: str,
-        since_ts: float,
-        paths: Iterable[str],
-    ) -> Dict[str, List[str]]:
+    def writes_since(self, exclude_task_id: str, since_ts: float,
+                     paths: Iterable[str]) -> Dict[str, List[str]]:
         """``{writer_task_id: [paths]}`` for writes after ``since_ts`` by agents
         other than ``exclude_task_id`` (delegate_task's "subagent modified files
         you previously read" reminder)."""
@@ -213,36 +209,6 @@ def get_registry() -> FileStateRegistry:
     return _registry
 
 
-def _disabled() -> bool:
-    # Re-read each call so tests can toggle via monkeypatch.setenv.
-    return os.environ.get("HERMES_DISABLE_FILE_STATE_GUARD", "").strip() == "1"
-
-
-def _mtime_or_none(resolved: str) -> Optional[float]:
-    try:
-        return os.path.getmtime(resolved)
-    except OSError:
-        return None
-
-
-def _fmt_ts(ts: float) -> str:
-    # Short wall-clock for warnings; avoids datetime formatting on the hot path.
-    return time.strftime("%H:%M:%S", time.localtime(ts))
-
-
-def _cap_dict(d: dict, limit: int) -> None:
-    """Trim ``d`` to ``limit`` entries by dropping the insertion-order oldest."""
-    over = len(d) - limit
-    if over <= 0:
-        return
-    it = iter(d)
-    for _ in range(over):
-        try:
-            d.pop(next(it))
-        except (StopIteration, KeyError):
-            break
-
-
 # Convenience wrappers (short names used at call sites).
 def record_read(task_id: str, resolved_or_path: str | Path, *, partial: bool = False) -> None:
     _registry.record_read(task_id, str(resolved_or_path), partial=partial)
@@ -260,11 +226,7 @@ def lock_path(resolved_or_path: str | Path):
     return _registry.lock_path(str(resolved_or_path))
 
 
-def writes_since(
-    exclude_task_id: str,
-    since_ts: float,
-    paths: Iterable[str | Path],
-) -> Dict[str, List[str]]:
+def writes_since(exclude_task_id: str, since_ts: float, paths: Iterable[str | Path]) -> Dict[str, List[str]]:
     return _registry.writes_since(exclude_task_id, since_ts, [str(p) for p in paths])
 
 
diff --git a/tools/file_tools_paths.py b/tools/file_tools_paths.py
index bb79f5c8bd..d25b0f6247 100644
--- a/tools/file_tools_paths.py
+++ b/tools/file_tools_paths.py
@@ -47,10 +47,7 @@ def _terminal_env_type_for_task(task_id: str = "default") -> str:
     """Best-effort terminal backend type for path-resolution decisions."""
     try:
         from tools.terminal_tool import (
-            _active_environments,
-            _env_lock,
-            _get_env_config,
-            _resolve_container_task_id,
+            _active_environments, _env_lock, _get_env_config, _resolve_container_task_id,
         )
 
         try:
@@ -61,14 +58,11 @@ def _terminal_env_type_for_task(task_id: str = "default") -> str:
             env = _active_environments.get(container_key) or _active_environments.get(task_id)
         if env is not None:
             name = env.__class__.__name__.lower()
-            for hint in _ENV_CLASS_NAME_HINTS:
-                if hint in name:
-                    return hint
+            hint = next((h for h in _ENV_CLASS_NAME_HINTS if h in name), None)
             stamped = getattr(env, "_hermes_backend_name", None)
-            if isinstance(stamped, str) and stamped:
-                return stamped
-        cfg = _get_env_config()
-        return str(cfg.get("env_type") or os.getenv("TERMINAL_ENV") or "local").lower()
+            if hint or (isinstance(stamped, str) and stamped):
+                return hint or stamped
+        return str(_get_env_config().get("env_type") or os.getenv("TERMINAL_ENV") or "local").lower()
     except Exception:
         return str(os.getenv("TERMINAL_ENV") or "local").lower()
 
@@ -103,9 +97,7 @@ def _sentinel_free_abs_cwd(raw: str | None) -> str | None:
     if raw.lower() in _TERMINAL_CWD_SENTINELS:
         return None
     expanded = _expand_tilde(raw)
-    if not os.path.isabs(expanded):
-        return None
-    return expanded
+    return expanded if os.path.isabs(expanded) else None
 
 
 def _configured_terminal_cwd() -> str | None:
@@ -151,12 +143,7 @@ def _authoritative_workspace_root(task_id: str = "default") -> str | None:
         recorded = get_session_cwd(task_id)
     except Exception:
         recorded = None
-    if recorded:
-        return recorded
-    registered = _registered_task_cwd_override(task_id)
-    if registered:
-        return registered
-    return _configured_terminal_cwd()
+    return recorded or _registered_task_cwd_override(task_id) or _configured_terminal_cwd()
 
 
 def _resolve_base_dir(
@@ -255,16 +242,14 @@ def _path_resolution_warning(filepath: str, resolved: Path, task_id: str = "defa
             root = _normalize_without_host_deref(Path(_expand_tilde(workspace_root)))
         else:
             root = Path(_expand_tilde(workspace_root)).resolve()
-        try:
-            resolved.relative_to(root)
+        if resolved.is_relative_to(root):
             return None
-        except ValueError:
-            return (
-                f"Relative path {filepath!r} resolved to {str(resolved)!r}, which is "
-                f"OUTSIDE the active workspace ({str(root)!r}). The edit will land in "
-                f"a different directory than the terminal's cwd. If this is not "
-                f"intended (e.g. a git-worktree session writing into the main "
-                f"checkout), pass an absolute path under the workspace instead."
-            )
+        return (
+            f"Relative path {filepath!r} resolved to {str(resolved)!r}, which is "
+            f"OUTSIDE the active workspace ({str(root)!r}). The edit will land in "
+            f"a different directory than the terminal's cwd. If this is not "
+            f"intended (e.g. a git-worktree session writing into the main "
+            f"checkout), pass an absolute path under the workspace instead."
+        )
     except Exception:
         return None
diff --git a/tools/file_tools_read_tracking.py b/tools/file_tools_read_tracking.py
index 1789c76bdc..de50abe7af 100644
--- a/tools/file_tools_read_tracking.py
+++ b/tools/file_tools_read_tracking.py
@@ -1,23 +1,12 @@
 """Per-task read/search bookkeeping for the file tools.
 
-Process-lifetime state behind ``read_file`` / ``search_files`` / ``write_file`` /
-``patch``; ``tools.file_tools`` re-imports every name here. Per task_id
-``_read_tracker`` stores:
-
-  last_key / consecutive   most recent read-or-search key and its repeat count
-                           (loop detection; reset by any OTHER tool call).
-  read_history             set of (path, offset, limit) — diagnostic summaries only.
-  dedup                    (resolved_path, offset, limit) -> mtime; skip identical
-                           re-reads of unchanged files. Cleared on context
-                           compression (the original content was summarised away).
-  dedup_hits               per-key count of stub returns, to break stub loops.
-  read_timestamps          resolved_path -> mtime at last read/write by this task;
-                           write/patch warn when the file changed underneath.
-  not_found                (op, resolved_path) -> (monotonic, cached error JSON);
-                           short-TTL negative cache for retried missing paths.
-
-Every container is hard-capped (``_cap_read_tracker_data``) so a long CLI
-session accretes a few hundred KB at most instead of ~1.5MB per 10k reads.
+Process-lifetime state behind read_file/search_files/write_file/patch;
+``tools.file_tools`` re-imports every name here. Per task_id ``_read_tracker``
+stores: ``last_key``/``consecutive`` (loop detection; reset by any OTHER tool
+call), ``read_history`` (diagnostics), ``dedup`` (key -> mtime; cleared on
+context compression), ``dedup_hits`` (stub-loop breaker), ``read_timestamps``
+(staleness warnings) and ``not_found`` (short-TTL negative cache). Every
+container is hard-capped (``_cap_read_tracker_data``) so long sessions stay small.
 """
 
 import logging
@@ -25,6 +14,7 @@ import os
 import threading
 import time
 
+from tools.file_state import _evict_oldest
 from tools.file_tools_paths import _authoritative_workspace_root, _resolve_path_for_task
 
 logger = logging.getLogger("tools.file_tools")
@@ -33,8 +23,7 @@ _read_tracker_lock = threading.Lock()
 _read_tracker: dict = {}
 
 # Consecutive patch failures per (task_id, resolved_path); escalates the hint
-# when the model keeps failing the same file (stale view, ambiguous old_string).
-# Reset on a successful patch to that path.
+# when the model keeps failing the same file. Reset on a successful patch.
 _patch_failure_lock = threading.Lock()
 _patch_failure_tracker: dict = {}  # {task_id: {resolved_path: count}}
 _PATCH_FAILURE_PATHS_CAP = 64
@@ -48,24 +37,17 @@ _NOT_FOUND_CAP = 500
 _NOT_FOUND_TTL_SECONDS = 60.0  # a path that didn't exist may be created soon
 
 
-def _new_task_data() -> dict:
-    return {
-        "last_key": None, "consecutive": 0,
-        "read_history": set(), "dedup": {},
-        "dedup_hits": {}, "read_timestamps": {},
-    }
-
-
 def _task_data(task_id: str) -> dict:
     """Get-or-create the tracker entry for *task_id*, back-filling any missing keys.
 
     Must be called with ``_read_tracker_lock`` held. Entries created by older
     code paths (or injected by tests) may lack the newer containers.
     """
-    task_data = _read_tracker.setdefault(task_id, _new_task_data())
-    for key, factory in (("dedup", dict), ("dedup_hits", dict), ("read_timestamps", dict)):
-        if key not in task_data:
-            task_data[key] = factory()
+    task_data = _read_tracker.setdefault(task_id, {
+        "last_key": None, "consecutive": 0, "read_history": set(),
+    })
+    for key in ("dedup", "dedup_hits", "read_timestamps"):
+        task_data.setdefault(key, {})
     return task_data
 
 
@@ -74,11 +56,8 @@ def _record_patch_failure(task_id: str, resolved_path: str) -> int:
     with _patch_failure_lock:
         task_failures = _patch_failure_tracker.setdefault(task_id, {})
         # Evict the oldest entry once a task has failed on many distinct files.
-        if len(task_failures) >= _PATCH_FAILURE_PATHS_CAP and resolved_path not in task_failures:
-            try:
-                del task_failures[next(iter(task_failures))]
-            except StopIteration:
-                pass
+        if resolved_path not in task_failures:
+            _evict_oldest(task_failures, _PATCH_FAILURE_PATHS_CAP - 1)
         task_failures[resolved_path] = task_failures.get(resolved_path, 0) + 1
         return task_failures[resolved_path]
 
@@ -89,29 +68,10 @@ def _reset_patch_failures(task_id: str, resolved_paths: list) -> None:
         return
     with _patch_failure_lock:
         task_failures = _patch_failure_tracker.get(task_id)
-        if not task_failures:
-            return
-        for rp in resolved_paths:
+        for rp in resolved_paths if task_failures else ():
             task_failures.pop(rp, None)
 
 
-def _evict_oldest(container, cap: int) -> None:
-    """Pop entries until *container* is within *cap*.
-
-    Sets pop arbitrary entries (they only feed diagnostic summaries); dicts pop
-    oldest by insertion order. An evicted entry costs one redundant re-send
-    (dedup) or one non-mtime staleness check — graceful degradation, not a bug.
-    """
-    for _ in range(len(container) - cap):
-        try:
-            if isinstance(container, set):
-                container.pop()
-            else:
-                container.pop(next(iter(container)))
-        except (StopIteration, KeyError):
-            break
-
-
 def _cap_read_tracker_data(task_data: dict) -> None:
     """Enforce size caps on the per-task sub-containers. Call with ``_read_tracker_lock`` held."""
     # Caps are read at call time so tests can monkeypatch the module constants.
@@ -127,6 +87,14 @@ def _cap_read_tracker_data(task_data: dict) -> None:
             _evict_oldest(container, cap)
 
 
+def _pop_not_found(op: str, resolved_str: str, task_id: str) -> None:
+    """Drop the negative-cache entry for *(op, resolved_str)*. Lock must be held."""
+    task_data = _read_tracker.get(task_id)
+    nf = task_data.get("not_found") if task_data else None
+    if nf:
+        nf.pop((op, resolved_str), None)
+
+
 def _check_not_found_cache(op: str, resolved_str: str, task_id: str) -> str | None:
     """Return cached not-found JSON for *(op, resolved_str)* if still fresh.
 
@@ -136,28 +104,20 @@ def _check_not_found_cache(op: str, resolved_str: str, task_id: str) -> str | No
     """
     with _read_tracker_lock:
         task_data = _read_tracker.get(task_id)
-        if not task_data:
-            return None
-        nf = task_data.get("not_found")
-        if not nf:
-            return None
-        entry = nf.get((op, resolved_str))
+        entry = (task_data.get("not_found") or {}).get((op, resolved_str)) if task_data else None
         if entry is None:
             return None
         ts, cached_json = entry
         if time.monotonic() - ts > _NOT_FOUND_TTL_SECONDS:
-            nf.pop((op, resolved_str), None)
+            _pop_not_found(op, resolved_str, task_id)
             return None
     # The path may have been created since the miss was cached (terminal,
-    # another agent, ...) — the "check → create → read" pattern is common, so
-    # serving a stale miss breaks it. The stat runs OUTSIDE the global tracker
-    # lock: a hung stat on a dead network mount must not stall every task.
+    # another agent, ...) — "check → create → read" is common, so serving a
+    # stale miss breaks it. The stat runs OUTSIDE the global tracker lock: a
+    # hung stat on a dead network mount must not stall every task.
     if os.path.exists(resolved_str):
         with _read_tracker_lock:
-            task_data = _read_tracker.get(task_id)
-            nf = task_data.get("not_found") if task_data else None
-            if nf:
-                nf.pop((op, resolved_str), None)
+            _pop_not_found(op, resolved_str, task_id)
         return None
     return cached_json
 
@@ -166,8 +126,7 @@ def _record_not_found(op: str, resolved_str: str, task_id: str, error_json: str)
     """Cache a not-found error so the next *op* call for *resolved_str* skips I/O."""
     with _read_tracker_lock:
         task_data = _task_data(task_id)
-        nf = task_data.setdefault("not_found", {})
-        nf[(op, resolved_str)] = (time.monotonic(), error_json)
+        task_data.setdefault("not_found", {})[(op, resolved_str)] = (time.monotonic(), error_json)
         _cap_read_tracker_data(task_data)
 
 
@@ -212,11 +171,9 @@ def notify_other_tool_call(task_id: str = "default"):
         if task_data:
             task_data["last_key"] = None
             task_data["consecutive"] = 0
-            if "dedup_hits" in task_data:
-                task_data["dedup_hits"].clear()
-            nf = task_data.get("not_found")
-            if nf:
-                nf.clear()
+            for key in ("dedup_hits", "not_found"):
+                if task_data.get(key):
+                    task_data[key].clear()
 
 
 def _invalidate_dedup_for_path(filepath: str, task_id: str) -> None:
@@ -237,10 +194,8 @@ def _invalidate_dedup_for_path(filepath: str, task_id: str) -> None:
         if dedup:
             for k in [k for k in dedup if k[0] == resolved]:
                 del dedup[k]
-        nf = task_data.get("not_found")
-        if nf:
-            nf.pop(("read", resolved), None)
-            nf.pop(("search", resolved), None)
+        _pop_not_found("read", resolved, task_id)
+        _pop_not_found("search", resolved, task_id)
 
 
 def _update_read_timestamp(filepath: str, task_id: str) -> None:
@@ -271,9 +226,7 @@ def _check_file_staleness(filepath: str, task_id: str) -> str | None:
         return None
     with _read_tracker_lock:
         task_data = _read_tracker.get(task_id)
-        if not task_data:
-            return None
-        read_mtime = task_data.get("read_timestamps", {}).get(resolved)
+        read_mtime = task_data.get("read_timestamps", {}).get(resolved) if task_data else None
     if read_mtime is None:
         return None
     try:
@@ -289,11 +242,8 @@ def _check_file_staleness(filepath: str, task_id: str) -> str | None:
     return None
 
 
-def _mark_verification_stale(
-    task_id: str,
-    resolved_paths: list[str],
-    session_id: str | None = None,
-) -> None:
+def _mark_verification_stale(task_id: str, resolved_paths: list[str],
+                             session_id: str | None = None) -> None:
     """Best-effort note that successful edits made prior verification stale.
 
     The workspace cwd is the first edited path's project root when one is
@@ -308,22 +258,9 @@ def _mark_verification_stale(
         from agent.coding_context import project_facts_for
         from agent.verification_evidence import mark_workspace_edited
 
-        cwd = None
-        for path in paths:
-            try:
-                candidate = str(Path(path).parent)
-            except Exception:
-                continue
-            if project_facts_for(candidate):
-                cwd = candidate
-                break
-        if cwd is None:
-            cwd = _authoritative_workspace_root(task_id)
-        if cwd is None:
-            try:
-                cwd = str(Path(paths[0]).parent)
-            except Exception:
-                cwd = None
+        parents = [str(Path(p).parent) for p in paths]
+        cwd = (next((c for c in parents if project_facts_for(c)), None)
+               or _authoritative_workspace_root(task_id) or parents[0])
         mark_workspace_edited(session_id=session_id or task_id, cwd=cwd, paths=paths)
     except Exception:
         logger.debug("verification stale marker failed", exc_info=True)
diff --git a/tools/file_tools_write_guards.py b/tools/file_tools_write_guards.py
index 294ce52715..f17db4ee23 100644
--- a/tools/file_tools_write_guards.py
+++ b/tools/file_tools_write_guards.py
@@ -2,14 +2,10 @@
 
 Every guard returns ``None`` when the write may proceed, else an error string
 the tool returns verbatim. ``tools.file_tools`` re-imports every name here.
-
-Guards (in the order the tools apply them):
-  * ``_check_sensitive_path``            — system paths + the Hermes config.yaml (hard deny).
-  * ``_check_binary_document_write``     — text write would corrupt an Office/PDF container.
-  * ``_check_protected_instruction_write`` — AGENTS.md-style files: ALWAYS ask, no yolo bypass.
-  * ``_check_approval_required_write``   — ~/.ssh/config-style files: normal approval gate.
-  * ``_check_cross_profile_path``        — sandbox-mirror writes the host never reads (lost-work guard).
-  * ``_is_internal_file_tool_content``   — refuse to persist read_file display text as a file.
+Guards, in the order the tools apply them: ``_check_sensitive_path`` (hard
+deny), ``_check_binary_document_write``, ``_check_protected_instruction_write``
+(ALWAYS ask), ``_check_approval_required_write`` (normal gate),
+``_check_cross_profile_path`` (sandbox-mirror lost-work), ``_is_internal_file_tool_content``.
 """
 
 import fnmatch
@@ -31,25 +27,42 @@ _SENSITIVE_EXACT_PATHS = {"/var/run/docker.sock", "/run/docker.sock"}
 
 _hermes_config_resolved: str | None = None
 _hermes_config_resolved_loaded = False
+_real_hermes_home_cached: str | None = None
+_real_hermes_home_loaded = False
 
 
 def _get_hermes_config_resolved() -> str | None:
     """Return the resolved absolute path of the Hermes config file (cached)."""
     global _hermes_config_resolved, _hermes_config_resolved_loaded
-    if _hermes_config_resolved_loaded:
-        return _hermes_config_resolved
-    _hermes_config_resolved_loaded = True
-    try:
-        from hermes_cli.config import get_config_path
-        _hermes_config_resolved = str(get_config_path().resolve())
-    except Exception:
+    if not _hermes_config_resolved_loaded:
+        _hermes_config_resolved_loaded = True
         try:
-            _hermes_config_resolved = str(Path(_expand_tilde("~/.hermes/config.yaml")).resolve())
+            from hermes_cli.config import get_config_path
+            _hermes_config_resolved = str(get_config_path().resolve())
         except Exception:
-            _hermes_config_resolved = None
+            try:
+                _hermes_config_resolved = str(Path(_expand_tilde("~/.hermes/config.yaml")).resolve())
+            except Exception:
+                _hermes_config_resolved = None
     return _hermes_config_resolved
 
 
+def _get_real_hermes_home() -> str | None:
+    """Return the realpath of the authoritative Hermes home (cached)."""
+    global _real_hermes_home_cached, _real_hermes_home_loaded
+    if not _real_hermes_home_loaded:
+        _real_hermes_home_loaded = True
+        try:
+            from hermes_constants import get_hermes_home
+            _real_hermes_home_cached = os.path.realpath(str(get_hermes_home()))
+        except Exception:
+            try:
+                _real_hermes_home_cached = os.path.realpath(_expand_tilde("~/.hermes"))
+            except Exception:
+                _real_hermes_home_cached = None
+    return _real_hermes_home_cached
+
+
 def _resolved_or_raw(filepath: str, task_id: str) -> str:
     """Task-resolved path string, falling back to the raw input on resolution failure."""
     try:
@@ -60,21 +73,16 @@ def _resolved_or_raw(filepath: str, task_id: str) -> str:
 
 def _check_sensitive_path(filepath: str, task_id: str = "default") -> str | None:
     """Return an error message if the path targets a sensitive system location."""
-    resolved = _resolved_or_raw(filepath, task_id)
-    normalized = os.path.normpath(_expand_tilde(filepath))
-    _err = (
-        f"Refusing to write to sensitive system path: {filepath}\n"
-        "Use the terminal tool with sudo if you need to modify system files."
-    )
-    for prefix in _SENSITIVE_PATH_PREFIXES:
-        if resolved.startswith(prefix) or normalized.startswith(prefix):
-            return _err
-    if resolved in _SENSITIVE_EXACT_PATHS or normalized in _SENSITIVE_EXACT_PATHS:
-        return _err
+    candidates = (_resolved_or_raw(filepath, task_id), os.path.normpath(_expand_tilde(filepath)))
+    if any(c.startswith(_SENSITIVE_PATH_PREFIXES) or c in _SENSITIVE_EXACT_PATHS for c in candidates):
+        return (
+            f"Refusing to write to sensitive system path: {filepath}\n"
+            "Use the terminal tool with sudo if you need to modify system files."
+        )
     # approvals.mode and other security settings live in config.yaml; a
     # prompt-injected agent could silently disable exec approval by editing it.
     hermes_config = _get_hermes_config_resolved()
-    if hermes_config and (resolved == hermes_config or normalized == hermes_config):
+    if hermes_config and hermes_config in candidates:
         return (
             f"Refusing to write to Hermes config file: {filepath}\n"
             "Agent cannot modify security-sensitive configuration. "
@@ -88,37 +96,13 @@ def _check_sensitive_path(filepath: str, task_id: str = "default") -> str | None
 # ---------------------------------------------------------------------------
 # Files that steer FUTURE agent behavior are a prompt-injection persistence
 # vector: an injected edit to AGENTS.md / CLAUDE.md / SOUL.md / .cursorrules
-# (or a project-local .hermes tree) outlives the turn and poisons every later
-# session. Writes ALWAYS require human approval — even under --yolo — and fail
-# closed when no human channel exists. (Ported from Roo-Code's
-# RooProtectedController; the terminal-tool vector is gated separately.)
-#
-# Basenames match in ANY directory (instruction files load from cwd trees) and
-# case-insensitively (case-insensitive filesystems; loaders probe variants).
+# (or a project-local .hermes tree) outlives the turn. Writes ALWAYS require
+# human approval — even under --yolo — and fail closed without a human channel.
+# Basenames match in ANY directory and case-insensitively (loaders probe variants).
 _PROTECTED_INSTRUCTION_BASENAMES = frozenset({
     "agents.md", "claude.md", "soul.md", ".cursorrules",
 })
 
-_real_hermes_home_cached: str | None = None
-_real_hermes_home_loaded = False
-
-
-def _get_real_hermes_home() -> str | None:
-    """Return the realpath of the authoritative Hermes home (cached)."""
-    global _real_hermes_home_cached, _real_hermes_home_loaded
-    if _real_hermes_home_loaded:
-        return _real_hermes_home_cached
-    _real_hermes_home_loaded = True
-    try:
-        from hermes_constants import get_hermes_home
-        _real_hermes_home_cached = os.path.realpath(str(get_hermes_home()))
-    except Exception:
-        try:
-            _real_hermes_home_cached = os.path.realpath(_expand_tilde("~/.hermes"))
-        except Exception:
-            _real_hermes_home_cached = None
-    return _real_hermes_home_cached
-
 
 def _protected_instruction_config() -> tuple[bool, list[str]]:
     """Return ``(enabled, extra_patterns)`` from ``security.protected_instruction_files`` /
@@ -129,10 +113,8 @@ def _protected_instruction_config() -> tuple[bool, list[str]]:
     try:
         from hermes_cli.config import load_config, cfg_get
         cfg = load_config()
-        enabled = cfg_get(cfg, "security", "protected_instruction_files",
-                          default=True)
-        extra = cfg_get(cfg, "security", "protected_instruction_extra_patterns",
-                        default=[])
+        enabled = cfg_get(cfg, "security", "protected_instruction_files", default=True)
+        extra = cfg_get(cfg, "security", "protected_instruction_extra_patterns", default=[])
     except Exception:
         return True, []
     if not isinstance(enabled, bool):
@@ -166,18 +148,15 @@ def _protected_instruction_reason(filepath: str, task_id: str = "default",
     # mirror guard, write_approval); this gate targets PROJECT-LOCAL files only.
     # Must run before the ``.hermes`` component rule, which would match the home.
     real_home = _get_real_hermes_home()
-    if real_home and (resolved == real_home
-                      or resolved.startswith(real_home + os.sep)):
+    if real_home and (resolved == real_home or resolved.startswith(real_home + os.sep)):
         return None
 
     for candidate in (normalized, resolved):
         base = os.path.basename(candidate)
         base_lower = base.lower()
-        if base_lower in _PROTECTED_INSTRUCTION_BASENAMES:
+        if base_lower in _PROTECTED_INSTRUCTION_BASENAMES or any(
+                fnmatch.fnmatch(base_lower, pattern.lower()) for pattern in extra_patterns):
             return base
-        for pattern in extra_patterns:
-            if fnmatch.fnmatch(base_lower, pattern.lower()):
-                return base
         # Project-local .hermes config dirs (<repo>/.hermes/config.yaml) steer
         # behavior too. Only the IMMEDIATE parent counts — matching any ancestor
         # would gate every write inside a checkout living under ~/.hermes.
@@ -187,8 +166,11 @@ def _protected_instruction_reason(filepath: str, task_id: str = "default",
     return None
 
 
-def _request_protected_instruction_approval(
-        reasons: list[str], task_id: str = "default") -> str | None:
+_APPROVAL_UNAVAILABLE = "requires approval but the approval subsystem is unavailable."
+_NO_HUMAN = "requires approval but no interactive user or gateway is present to approve it."
+
+
+def _request_protected_instruction_approval(reasons: list[str], task_id: str = "default") -> str | None:
     """Ask the human to approve a write to protected instruction file(s).
 
     Returns ``None`` when approved, else a BLOCKED error string. Deliberately
@@ -209,21 +191,17 @@ def _request_protected_instruction_approval(
         "attempt the same edit via another path (terminal, execute_code, "
         "etc.)."
     )
-    timed_out = blocked.format(
-        why="approval prompt timed out without a user response. "
-            "Silence is not consent.")
+    timed_out = blocked.format(why="approval prompt timed out without a user response. Silence is not consent.")
     denied = blocked.format(why="was denied by the user.")
 
     try:
         import tools.approval as _approval
     except Exception:
-        return blocked.format(why="requires approval but the approval "
-                                  "subsystem is unavailable.")
+        return blocked.format(why=_APPROVAL_UNAVAILABLE)
 
     # Gateway surface: block on the button round-trip when a notify callback
     # is registered for this session. One-operation only — no scope buttons.
     session_key = _approval.get_current_session_key()
-    notify_cb = None
     try:
         with _approval._lock:
             notify_cb = _approval._gateway_notify_cbs.get(session_key)
@@ -239,23 +217,15 @@ def _request_protected_instruction_approval(
             "allow_permanent": False,
             "allow_session": False,
         }
-        decision = _approval._await_gateway_decision(
-            session_key, notify_cb, approval_data, surface="gateway",
-        )
+        decision = _approval._await_gateway_decision(session_key, notify_cb, approval_data, surface="gateway")
         if decision.get("notify_failed"):
-            return blocked.format(
-                why="requires approval but the approval request could not "
-                    "be delivered.")
-        choice = decision.get("choice")
+            return blocked.format(why="requires approval but the approval request could not be delivered.")
         # Any tapped scope is a one-operation grant; nothing is persisted.
-        if decision.get("resolved") and choice in {"once", "session", "always"}:
+        if decision.get("resolved") and decision.get("choice") in {"once", "session", "always"}:
             return None
-        if not decision.get("resolved"):
-            return timed_out
-        return denied
+        return timed_out if not decision.get("resolved") else denied
 
     # CLI surface: per-thread approval callback (prompt_toolkit panel).
-    callback = None
     try:
         from tools.terminal_tool import _get_approval_callback
         callback = _get_approval_callback()
@@ -264,26 +234,17 @@ def _request_protected_instruction_approval(
 
     if callback is not None:
         choice = _approval.prompt_dangerous_approval(
-            display, description,
-            allow_permanent=False,
-            allow_session=False,
-            approval_callback=callback,
-        )
+            display, description, allow_permanent=False, allow_session=False, approval_callback=callback)
         if choice in {"once", "session", "always"}:
             return None
-        if choice == "timeout":
-            return timed_out
-        return denied
+        return timed_out if choice == "timeout" else denied
 
     # No human channel (script, cron, background thread): fail closed —
     # auto-approving here would recreate the persistence vector.
-    return blocked.format(
-        why="requires approval but no interactive user or gateway is "
-            "present to approve it.")
+    return blocked.format(why=_NO_HUMAN)
 
 
-def _check_protected_instruction_write(paths: list[str],
-                                       task_id: str = "default") -> str | None:
+def _check_protected_instruction_write(paths: list[str], task_id: str = "default") -> str | None:
     """Gate a write/patch touching protected instruction files.
 
     ONE protected file gates the ENTIRE multi-file patch: a single prompt lists
@@ -293,19 +254,14 @@ def _check_protected_instruction_write(paths: list[str],
     enabled, extra = _protected_instruction_config()
     if not enabled:
         return None
-    reasons = [
-        r for r in (
-            _protected_instruction_reason(p, task_id, enabled=enabled, extra_patterns=extra)
-            for p in paths
-        ) if r
-    ]
+    reasons = [r for r in (_protected_instruction_reason(p, task_id, enabled=enabled, extra_patterns=extra)
+                           for p in paths) if r]
     if not reasons:
         return None
     return _request_protected_instruction_approval(reasons, task_id)
 
 
-def _check_approval_required_write(paths: list[str],
-                                   task_id: str = "default") -> str | None:
+def _check_approval_required_write(paths: list[str], task_id: str = "default") -> str | None:
     """Gate a write/patch touching an approval-required path (``~/.ssh/config``).
 
     Not credentials and not hard-denied, but they can steer process execution
@@ -337,15 +293,13 @@ def _check_approval_required_write(paths: list[str],
     try:
         import tools.approval as _approval
     except Exception:
-        return blocked.format(why="requires approval but the approval "
-                                  "subsystem is unavailable.")
+        return blocked.format(why=_APPROVAL_UNAVAILABLE)
 
     result = _approval._run_approval_gate(
         pattern_key="ssh_config_write",
         description=description,
         display_target=f"<write to {display_targets}>",
-        cron_deny_message=blocked.format(
-            why="requires approval but this cron session denies it."),
+        cron_deny_message=blocked.format(why="requires approval but this cron session denies it."),
         single_query_deny_message=blocked.format(
             why="requires approval but single-query (-q) sessions run "
                 "without a user present to approve it. To allow flagged "
@@ -353,9 +307,7 @@ def _check_approval_required_write(paths: list[str],
                 "approve in config.yaml."),
         autoapprove_log_prefix="ssh_config_write",
         fail_closed_when_no_human=True,
-        no_human_block_message=blocked.format(
-            why="requires approval but no interactive user or gateway is "
-                "present to approve it."),
+        no_human_block_message=blocked.format(why=_NO_HUMAN),
     )
     if result.get("approved"):
         return None
@@ -366,31 +318,18 @@ def _get_container_mirror_prefix_for_task(task_id: str = "default") -> str | Non
     """Return the container-side Hermes mirror prefix for persistent Docker file tools."""
     try:
         from tools.terminal_tool import (
-            _active_environments,
-            _env_lock,
-            _get_env_config,
-            _resolve_container_task_id,
+            _active_environments, _env_lock, _get_env_config, _resolve_container_task_id,
         )
-
         container_key = _resolve_container_task_id(task_id)
-    except Exception:
-        return None
-
-    try:
         with _env_lock:
             env = _active_environments.get(container_key) or _active_environments.get(task_id)
-
         if env is not None:
-            if env.__class__.__name__ == "DockerEnvironment" and bool(
-                getattr(env, "_persistent", False)
-            ):
-                return "/root/.hermes"
-            return None
-
+            persistent_docker = (env.__class__.__name__ == "DockerEnvironment"
+                                 and bool(getattr(env, "_persistent", False)))
+            return "/root/.hermes" if persistent_docker else None
         config = _get_env_config()
     except Exception:
         return None
-
     if config.get("env_type") == "docker" and config.get("container_persistent", True):
         return "/root/.hermes"
     return None
@@ -406,23 +345,14 @@ def _check_cross_profile_path(filepath: str, task_id: str = "default") -> str |
     Fails open on import error — the sensitive-path guard and denylist still apply.
     """
     try:
-        from agent.file_safety import (
-            get_container_mirror_warning,
-            get_sandbox_mirror_warning,
-        )
+        from agent.file_safety import get_container_mirror_warning, get_sandbox_mirror_warning
     except Exception:
         return None
-
     resolved = _resolved_or_raw(filepath, task_id)
-
     warning = get_sandbox_mirror_warning(resolved)
     if warning is not None:
         return warning
-
-    return get_container_mirror_warning(
-        resolved,
-        mirror_prefix=_get_container_mirror_prefix_for_task(task_id),
-    )
+    return get_container_mirror_warning(resolved, mirror_prefix=_get_container_mirror_prefix_for_task(task_id))
 
 
 def _check_binary_document_write(filepath: str, task_id: str = "default") -> str | None:
@@ -484,12 +414,8 @@ def _is_internal_file_status_text(content: str) -> bool:
     if not isinstance(content, str):
         return False
     stripped = content.strip()
-    if not stripped:
-        return False
-    if stripped == _READ_DEDUP_STATUS_MESSAGE:
-        return True
-    return (_READ_DEDUP_STATUS_MESSAGE in stripped
-            and len(stripped) <= 2 * len(_READ_DEDUP_STATUS_MESSAGE))
+    return bool(stripped) and _READ_DEDUP_STATUS_MESSAGE in stripped and (
+        len(stripped) <= 2 * len(_READ_DEDUP_STATUS_MESSAGE))
 
 
 def _looks_like_read_file_line_numbered_content(content: str) -> bool:
@@ -501,30 +427,20 @@ def _looks_like_read_file_line_numbered_content(content: str) -> bool:
     """
     if not isinstance(content, str):
         return False
-
     lines = [line for line in content.splitlines() if line.strip()]
     if len(lines) < 2:
         return False
-
     numbered: list[int] = []
     for line in lines:
         prefix, sep, _rest = line.lstrip().partition("|")
         if sep and prefix.isdigit():
             numbered.append(int(prefix))
-
     if len(numbered) < 2 or len(numbered) / len(lines) < 0.6:
         return False
-
-    consecutive_pairs = sum(
-        1 for prev, current in zip(numbered, numbered[1:])
-        if current == prev + 1
-    )
+    consecutive_pairs = sum(1 for prev, current in zip(numbered, numbered[1:]) if current == prev + 1)
     return consecutive_pairs >= len(numbered) - 1
 
 
 def _is_internal_file_tool_content(content: str) -> bool:
     """Return True when content is file-tool display text, not intended file bytes."""
-    return (
-        _is_internal_file_status_text(content)
-        or _looks_like_read_file_line_numbered_content(content)
-    )
+    return _is_internal_file_status_text(content) or _looks_like_read_file_line_numbered_content(content)

From 40f2c7951b70ec540f6ade5bc65181ebd869e3c4 Mon Sep 17 00:00:00 2001
From: Teknium <127238744+teknium1@users.noreply.github.com>
Date: Wed, 2 Sep 2026 22:51:55 -0700
Subject: [PATCH 2/4] refactor(tools): AST-neutral layout compaction of
 file_tools group

---
 tools/file_state.py               |  18 ++----
 tools/file_tools.py               | 102 ++++++++++--------------------
 tools/file_tools_paths.py         |  11 +---
 tools/file_tools_read_tracking.py |   9 +--
 tools/file_tools_write_guards.py  |  42 ++++--------
 tools/fuzzy_match.py              |  39 ++++--------
 6 files changed, 72 insertions(+), 149 deletions(-)

diff --git a/tools/file_state.py b/tools/file_state.py
index 99f3e2d069..36dd1da640 100644
--- a/tools/file_state.py
+++ b/tools/file_state.py
@@ -139,16 +139,14 @@ class FileStateRegistry:
                         f"{resolved} was modified by sibling subagent "
                         f"{writer_tid!r} but this agent never read it. "
                         "Read the file before writing to avoid overwriting "
-                        "the sibling's changes."
-                    )
+                        "the sibling's changes.")
                 read_ts = stamp[1]
                 if writer_ts > read_ts:
                     return (
                         f"{resolved} was modified by sibling subagent "
                         f"{writer_tid!r} at {_fmt_ts(writer_ts)} — after "
                         f"this agent's last read at {_fmt_ts(read_ts)}. "
-                        "Re-read the file before writing."
-                    )
+                        "Re-read the file before writing.")
 
         if stamp is not None:
             read_mtime, _read_ts, partial = stamp
@@ -156,20 +154,17 @@ class FileStateRegistry:
                 return (
                     f"{resolved} was modified since you last read it "
                     "on disk (external edit or unrecorded writer). "
-                    "Re-read the file before writing."
-                )
+                    "Re-read the file before writing.")
             if partial:
                 return (
                     f"{resolved} was last read with offset/limit pagination "
                     "(partial view). Re-read the whole file before "
-                    "overwriting it."
-                )
+                    "overwriting it.")
             return None
 
         return (
             f"{resolved} was not read by this agent. "
-            "Read the file first so you can write an informed edit."
-        )
+            "Read the file first so you can write an informed edit.")
 
     def writes_since(self, exclude_task_id: str, since_ts: float,
                      paths: Iterable[str]) -> Dict[str, List[str]]:
@@ -242,5 +237,4 @@ __all__ = [
     "check_stale",
     "lock_path",
     "writes_since",
-    "known_reads",
-]
+    "known_reads"]
diff --git a/tools/file_tools.py b/tools/file_tools.py
index 7c24dae8a4..29bf67fb47 100644
--- a/tools/file_tools.py
+++ b/tools/file_tools.py
@@ -24,10 +24,7 @@ from pathlib import Path
 from agent.file_safety import get_read_block_error
 from tools.binary_extensions import has_binary_extension
 from tools.file_operations import (
-    ShellFileOperations,
-    normalize_read_pagination,
-    normalize_search_pagination,
-)
+    ShellFileOperations, normalize_read_pagination, normalize_search_pagination)
 from tools import file_state
 from agent.redact import redact_sensitive_text
 from tools.file_tools_paths import (  # noqa: F401  (re-exported)
@@ -44,8 +41,7 @@ from tools.file_tools_paths import (  # noqa: F401  (re-exported)
     _resolve_path_for_task,
     _sentinel_free_abs_cwd,
     _terminal_env_type_for_task,
-    _uses_container_paths,
-)
+    _uses_container_paths)
 from tools.file_tools_write_guards import (  # noqa: F401  (re-exported)
     _PROTECTED_INSTRUCTION_BASENAMES,
     _READ_DEDUP_STATUS_MESSAGE,
@@ -64,8 +60,7 @@ from tools.file_tools_write_guards import (  # noqa: F401  (re-exported)
     _looks_like_read_file_line_numbered_content,
     _protected_instruction_config,
     _protected_instruction_reason,
-    _request_protected_instruction_approval,
-)
+    _request_protected_instruction_approval)
 from tools.file_tools_read_tracking import (  # noqa: F401  (re-exported)
     _DEDUP_CAP,
     _NOT_FOUND_CAP,
@@ -88,8 +83,7 @@ from tools.file_tools_read_tracking import (  # noqa: F401  (re-exported)
     _task_data,
     _update_read_timestamp,
     notify_other_tool_call,
-    reset_file_dedup,
-)
+    reset_file_dedup)
 
 logger = logging.getLogger(__name__)
 
@@ -155,14 +149,12 @@ def _apply_char_budget(result_dict: dict, content: str, offset: int, total_lines
     result_dict["hint"] = (
         f"Output truncated at the {max_chars:,}-char read budget after "
         f"{lines_kept} line(s) (showing lines {offset}-{next_offset - 1} of "
-        f"{total_lines}). Use offset={next_offset} to continue."
-    )
+        f"{total_lines}). Use offset={next_offset} to continue.")
     if len(trimmed.split("\n", 1)[0]) >= max_chars:
         result_dict["hint"] += (
             " Note: the first line alone exceeded the budget and was "
             "clamped mid-line; its remainder is not retrievable via "
-            "offset."
-        )
+            "offset.")
     return trimmed
 
 
@@ -182,8 +174,7 @@ _BLOCKED_DEVICE_PATHS = frozenset({
 _BLOCKED_PROC_SUFFIXES = (
     "/fd/0", "/fd/1", "/fd/2",  # stdio aliases
     "/environ", "/cmdline", "/maps", "/smaps", "/smaps_rollup", "/numa_maps",
-    "/mem", "/auxv", "/pagemap",
-)
+    "/mem", "/auxv", "/pagemap")
 
 
 def _file_ops_uses_host_paths(file_ops) -> bool:
@@ -330,8 +321,7 @@ def _create_terminal_env_for_file_ops(raw_task_id: str, task_id: str):
         _resolve_task_host_cwd,
         _select_image,
         get_session_cwd,
-        resolve_task_overrides,
-    )
+        resolve_task_overrides)
 
     config = _get_env_config()
     env_type = config["env_type"]
@@ -351,8 +341,7 @@ def _create_terminal_env_for_file_ops(raw_task_id: str, task_id: str):
             logger.info(
                 "Ignoring host/relative cwd override %r for %s backend "
                 "(won't exist in sandbox). Using %r instead.",
-                cwd, env_type, config["cwd"],
-            )
+                cwd, env_type, config["cwd"])
         cwd = config["cwd"]
     logger.info("Creating new %s environment for task %s...", env_type, task_id[:8])
     terminal_env = _create_configured_env(
@@ -376,8 +365,7 @@ def _get_file_ops(task_id: str = "default") -> ShellFileOperations:
     from tools.terminal_tool import (
         _active_environments, _env_lock, _last_activity, _start_cleanup_thread,
         _creation_locks, _creation_locks_lock, _resolve_container_task_id,
-        get_session_cwd, record_session_cwd,
-    )
+        get_session_cwd, record_session_cwd)
 
     raw_task_id = task_id or "default"
     task_id = _resolve_container_task_id(raw_task_id)
@@ -441,8 +429,7 @@ _SPECIAL_FILE_KINDS = (
     (stat.S_ISFIFO, "a FIFO (named pipe)"),
     (stat.S_ISSOCK, "a socket"),
     (stat.S_ISCHR, "a character device"),
-    (stat.S_ISBLK, "a block device"),
-)
+    (stat.S_ISBLK, "a block device"))
 
 
 def _special_file_kind(path) -> str | None:
@@ -478,8 +465,7 @@ def _read_extracted_document(path: str, _resolved, offset: int, limit: int, task
         MAX_DOCUMENT_BYTES,
         ExtractionError,
         extract_document_bytes,
-        is_extractable_document,
-    )
+        is_extractable_document)
 
     if not is_extractable_document(str(_resolved)):
         return None
@@ -504,8 +490,7 @@ def _read_extracted_document(path: str, _resolved, offset: int, limit: int, task
             return tool_error(
                 f"Cannot read '{path}' ({_doc_ext}): document "
                 f"extraction failed — {exc}. Use terminal utilities "
-                "to inspect or convert the file."
-            )
+                "to inspect or convert the file.")
         return None
 
     lines = extracted_text.splitlines()
@@ -517,13 +502,11 @@ def _read_extracted_document(path: str, _resolved, offset: int, limit: int, task
         "total_lines": total_lines,
         "file_size": binary.file_size,
         "truncated": total_lines > end_line,
-        "extracted_document": True,
-    }
+        "extracted_document": True}
     if result_dict["truncated"]:
         result_dict["hint"] = (
             f"Use offset={end_line + 1} to continue reading "
-            f"(showing {offset}-{min(end_line, total_lines)} of {total_lines} lines)"
-        )
+            f"(showing {offset}-{min(end_line, total_lines)} of {total_lines} lines)")
     max_chars = _get_max_read_chars()
     if len(result_dict["content"]) > max_chars:
         _apply_char_budget(result_dict, result_dict["content"], offset, total_lines, max_chars)
@@ -550,8 +533,7 @@ def _dedup_stub_or_block(task_data: dict, dedup_key: tuple, path: str) -> str:
             "still current. Proceed with your task using "
             "the information you already have.",
             path=path,
-            already_read=hits + 1,
-        )
+            already_read=hits + 1)
 
     return json.dumps({
         "status": "unchanged",
@@ -614,8 +596,7 @@ def read_file_tool(path: str, offset: int = 1, limit: int = 2000, task_id: str =
         if _is_blocked_device(path, base_dir=device_base):
             return tool_error(
                 f"Cannot read '{path}': this is a device file that would "
-                "block or produce infinite output."
-            )
+                "block or produce infinite output.")
 
         _resolved = _resolve_path_for_task(path, task_id)
 
@@ -629,9 +610,7 @@ def read_file_tool(path: str, offset: int = 1, limit: int = 2000, task_id: str =
                         f"'{path}' is {kind}, not a regular file — reading "
                         "it would block indefinitely, so no read was "
                         "attempted. Use terminal utilities if you need to "
-                        "interact with it."
-                    ),
-                })
+                        "interact with it.")})
 
         extracted = _read_extracted_document(path, _resolved, offset, limit, task_id)
         if extracted is not None:
@@ -642,8 +621,7 @@ def read_file_tool(path: str, offset: int = 1, limit: int = 2000, task_id: str =
         if has_binary_extension(str(_resolved)):
             return tool_error(
                 f"Cannot read binary file '{path}' ({_resolved.suffix.lower()}). "
-                "Use vision_analyze for images, or terminal to inspect binary files."
-            )
+                "Use vision_analyze for images, or terminal to inspect binary files.")
 
         # Hermes internal denylist: blocks prompt injection via catalog/hub
         # metadata and credential stores under HERMES_HOME. Pass the resolved
@@ -692,8 +670,7 @@ def read_file_tool(path: str, offset: int = 1, limit: int = 2000, task_id: str =
         if len(result.content or "") > max_chars:
             result.content = _apply_char_budget(
                 result_dict, result.content or "", offset,
-                result_dict.get("total_lines", "unknown"), max_chars,
-            )
+                result_dict.get("total_lines", "unknown"), max_chars)
         if result.content:
             result.content = redact_sensitive_text(result.content, file_read=True)
             result_dict["content"] = result.content
@@ -703,8 +680,7 @@ def read_file_tool(path: str, offset: int = 1, limit: int = 2000, task_id: str =
             result_dict.setdefault("_hint", (
                 f"This file is large ({file_size:,} bytes). "
                 "Consider reading only the section you need with offset and limit "
-                "to keep context usage efficient."
-            ))
+                "to keep context usage efficient."))
 
         count = _record_successful_read(task_data, task_id, path, resolved_str, offset, limit,
                                         dedup_key, partial=(offset > 1) or bool(result_dict.get("truncated")))
@@ -714,14 +690,12 @@ def read_file_tool(path: str, offset: int = 1, limit: int = 2000, task_id: str =
                 "The content has NOT changed. You already have this information. "
                 "STOP re-reading and proceed with your task.",
                 path=path,
-                already_read=count,
-            )
+                already_read=count)
         if count >= 3:
             result_dict["_warning"] = (
                 f"You have read this exact file region {count} times consecutively. "
                 "The content has not changed since your last read. Use the information you already have. "
-                "If you are stuck in a loop, stop reading and proceed with writing or responding."
-            )
+                "If you are stuck in a loop, stop reading and proceed with writing or responding.")
         return json.dumps(result_dict, ensure_ascii=False)
     except Exception as e:
         return tool_error(str(e))
@@ -764,8 +738,7 @@ def _write_precheck_error(paths: list[str], content_paths: list[str], task_id: s
     return _first_error(
         *(lambda p=p: _check_binary_document_write(p, task_id) for p in content_paths),
         lambda: _check_protected_instruction_write(paths, task_id),
-        lambda: _check_approval_required_write(paths, task_id),
-    )
+        lambda: _check_approval_required_write(paths, task_id))
 
 
 def _edit_warnings(paths: list[str], path_to_resolved: dict, task_id: str) -> list[str]:
@@ -813,8 +786,7 @@ def write_file_tool(path: str, content: str, task_id: str = "default",
             "Refusing to write internal read_file display text as file content. "
             "Strip read_file line-number prefixes or reconstruct the intended "
             "file contents before writing."
-        ) if _is_internal_file_tool_content(content) else None,
-    )
+        ) if _is_internal_file_tool_content(content) else None)
     if err:
         return tool_error(err)
     try:
@@ -873,8 +845,7 @@ def _collect_v4a_header_paths(patch: str) -> tuple[list[str], list[str]] | str:
                 f"V4A patch header contains '..' traversal: {v4a_path!r}. "
                 "Use the agent's cwd-relative path (no '..') or an absolute "
                 "path in '*** Update File:' / '*** Add File:' / "
-                "'*** Delete File:' / '*** Move File:' headers."
-            )
+                "'*** Delete File:' / '*** Move File:' headers.")
         paths.append(v4a_path)
         if writes_text:
             content_paths.append(v4a_path)
@@ -958,13 +929,11 @@ def patch_tool(mode: str = "replace", path: str = None, old_string: str = None,
                     "content, (2) use a longer / more unique old_string with "
                     "surrounding context lines, or (3) use write_file to "
                     "replace the entire file if the targeted region is hard "
-                    "to anchor."
-                )
+                    "to anchor.")
             elif "Did you mean one of these sections?" not in str(result_dict["error"]):
                 result_dict["_hint"] = (
                     "old_string not found. Use read_file to verify the current "
-                    "content, or search_files to locate the text."
-                )
+                    "content, or search_files to locate the text.")
         return json.dumps(result_dict, ensure_ascii=False)
     except Exception as e:
         return tool_error(str(e))
@@ -983,8 +952,7 @@ def search_tool(pattern: str, target: str = "content", path: str = ".",
         search_key = ("search", pattern, target, str(path), file_glob or "", limit, offset)
         with _read_tracker_lock:
             task_data = _read_tracker.setdefault(task_id, {
-                "last_key": None, "consecutive": 0, "read_history": set(),
-            })
+                "last_key": None, "consecutive": 0, "read_history": set()})
             count = _bump_consecutive(task_data, search_key)
 
         if count >= 4:
@@ -993,8 +961,7 @@ def search_tool(pattern: str, target: str = "content", path: str = ".",
                 "The results have NOT changed. You already have this information. "
                 "STOP re-searching and proceed with your task.",
                 pattern=pattern,
-                already_searched=count,
-            )
+                already_searched=count)
 
         try:
             resolved_search_path = str(_resolve_path_for_task(path, task_id))
@@ -1016,8 +983,7 @@ def search_tool(pattern: str, target: str = "content", path: str = ".",
 
         result = _get_file_ops(task_id).search(
             pattern=pattern, path=path, target=target, file_glob=file_glob,
-            limit=limit, offset=offset, output_mode=output_mode, context=context
-        )
+            limit=limit, offset=offset, output_mode=output_mode, context=context)
         omitted = _filter_read_blocked_search_results(result, task_id)
         for m in getattr(result, "matches", None) or ():
             if getattr(m, "content", None):
@@ -1027,8 +993,7 @@ def search_tool(pattern: str, target: str = "content", path: str = ".",
         if omitted:
             result_dict["_omitted"] = (
                 f"{omitted} result(s) omitted because they target credential, "
-                "token, cache, or secret-bearing environment files."
-            )
+                "token, cache, or secret-bearing environment files.")
 
         # No early return on a cached miss — same rationale as the read path.
         _search_err = result_dict.get("error") or ""
@@ -1038,8 +1003,7 @@ def search_tool(pattern: str, target: str = "content", path: str = ".",
         if count >= 3:
             result_dict["_warning"] = (
                 f"You have run this exact search {count} times consecutively. "
-                "The results have not changed. Use the information you already have."
-            )
+                "The results have not changed. Use the information you already have.")
 
         result_json = json.dumps(result_dict, ensure_ascii=False)
         if result_dict.get("truncated"):
diff --git a/tools/file_tools_paths.py b/tools/file_tools_paths.py
index d25b0f6247..dda9a90573 100644
--- a/tools/file_tools_paths.py
+++ b/tools/file_tools_paths.py
@@ -47,8 +47,7 @@ def _terminal_env_type_for_task(task_id: str = "default") -> str:
     """Best-effort terminal backend type for path-resolution decisions."""
     try:
         from tools.terminal_tool import (
-            _active_environments, _env_lock, _get_env_config, _resolve_container_task_id,
-        )
+            _active_environments, _env_lock, _get_env_config, _resolve_container_task_id)
 
         try:
             container_key = _resolve_container_task_id(task_id)
@@ -147,10 +146,7 @@ def _authoritative_workspace_root(task_id: str = "default") -> str | None:
 
 
 def _resolve_base_dir(
-    task_id: str = "default",
-    *,
-    container_paths: bool | None = None,
-) -> Path | PurePosixPath:
+    task_id: str = "default", *, container_paths: bool | None = None) -> Path | PurePosixPath:
     """Return the ABSOLUTE base directory for resolving relative paths.
 
     Uses ``_authoritative_workspace_root`` (live cwd → registered override →
@@ -249,7 +245,6 @@ def _path_resolution_warning(filepath: str, resolved: Path, task_id: str = "defa
             f"OUTSIDE the active workspace ({str(root)!r}). The edit will land in "
             f"a different directory than the terminal's cwd. If this is not "
             f"intended (e.g. a git-worktree session writing into the main "
-            f"checkout), pass an absolute path under the workspace instead."
-        )
+            f"checkout), pass an absolute path under the workspace instead.")
     except Exception:
         return None
diff --git a/tools/file_tools_read_tracking.py b/tools/file_tools_read_tracking.py
index de50abe7af..00435e4caf 100644
--- a/tools/file_tools_read_tracking.py
+++ b/tools/file_tools_read_tracking.py
@@ -44,8 +44,7 @@ def _task_data(task_id: str) -> dict:
     code paths (or injected by tests) may lack the newer containers.
     """
     task_data = _read_tracker.setdefault(task_id, {
-        "last_key": None, "consecutive": 0, "read_history": set(),
-    })
+        "last_key": None, "consecutive": 0, "read_history": set()})
     for key in ("dedup", "dedup_hits", "read_timestamps"):
         task_data.setdefault(key, {})
     return task_data
@@ -80,8 +79,7 @@ def _cap_read_tracker_data(task_data: dict) -> None:
         ("dedup", _DEDUP_CAP),
         ("dedup_hits", _DEDUP_CAP),
         ("read_timestamps", _READ_TIMESTAMPS_CAP),
-        ("not_found", _NOT_FOUND_CAP),
-    ):
+        ("not_found", _NOT_FOUND_CAP)):
         container = task_data.get(key)
         if container is not None and len(container) > cap:
             _evict_oldest(container, cap)
@@ -237,8 +235,7 @@ def _check_file_staleness(filepath: str, task_id: str) -> str | None:
         return (
             f"Warning: {filepath} was modified since you last read it "
             "(external edit or concurrent agent). The content you read may be "
-            "stale. Consider re-reading the file to verify before writing."
-        )
+            "stale. Consider re-reading the file to verify before writing.")
     return None
 
 
diff --git a/tools/file_tools_write_guards.py b/tools/file_tools_write_guards.py
index f17db4ee23..9ce02b4a08 100644
--- a/tools/file_tools_write_guards.py
+++ b/tools/file_tools_write_guards.py
@@ -21,8 +21,7 @@ from tools.file_tools_paths import _expand_tilde, _resolve_path_for_task
 _SENSITIVE_PATH_PREFIXES = (
     "/etc/", "/boot/", "/usr/lib/systemd/",
     "/private/etc/",
-    "/private/var/db/", "/private/var/root/",
-)
+    "/private/var/db/", "/private/var/root/")
 _SENSITIVE_EXACT_PATHS = {"/var/run/docker.sock", "/run/docker.sock"}
 
 _hermes_config_resolved: str | None = None
@@ -77,8 +76,7 @@ def _check_sensitive_path(filepath: str, task_id: str = "default") -> str | None
     if any(c.startswith(_SENSITIVE_PATH_PREFIXES) or c in _SENSITIVE_EXACT_PATHS for c in candidates):
         return (
             f"Refusing to write to sensitive system path: {filepath}\n"
-            "Use the terminal tool with sudo if you need to modify system files."
-        )
+            "Use the terminal tool with sudo if you need to modify system files.")
     # approvals.mode and other security settings live in config.yaml; a
     # prompt-injected agent could silently disable exec approval by editing it.
     hermes_config = _get_hermes_config_resolved()
@@ -86,8 +84,7 @@ def _check_sensitive_path(filepath: str, task_id: str = "default") -> str | None
         return (
             f"Refusing to write to Hermes config file: {filepath}\n"
             "Agent cannot modify security-sensitive configuration. "
-            "Edit ~/.hermes/config.yaml directly or use 'hermes config' instead."
-        )
+            "Edit ~/.hermes/config.yaml directly or use 'hermes config' instead.")
     return None
 
 
@@ -100,8 +97,7 @@ def _check_sensitive_path(filepath: str, task_id: str = "default") -> str | None
 # human approval — even under --yolo — and fail closed without a human channel.
 # Basenames match in ANY directory and case-insensitively (loaders probe variants).
 _PROTECTED_INSTRUCTION_BASENAMES = frozenset({
-    "agents.md", "claude.md", "soul.md", ".cursorrules",
-})
+    "agents.md", "claude.md", "soul.md", ".cursorrules"})
 
 
 def _protected_instruction_config() -> tuple[bool, list[str]]:
@@ -182,15 +178,13 @@ def _request_protected_instruction_approval(reasons: list[str], task_id: str = "
     description = (
         f"Write to protected agent-instruction file(s): {targets}. "
         "These files steer future agent behavior; approval is always "
-        "required (not bypassed by auto-approve)."
-    )
+        "required (not bypassed by auto-approve).")
     display = f"<write to {targets}>"
     blocked = (
         f"BLOCKED: write to protected agent-instruction file(s) ({targets}) "
         "{why} The user has NOT consented to this write. Do NOT retry it or "
         "attempt the same edit via another path (terminal, execute_code, "
-        "etc.)."
-    )
+        "etc.).")
     timed_out = blocked.format(why="approval prompt timed out without a user response. Silence is not consent.")
     denied = blocked.format(why="was denied by the user.")
 
@@ -215,8 +209,7 @@ def _request_protected_instruction_approval(reasons: list[str], task_id: str = "
             "pattern_keys": ["protected_instruction_file"],
             "description": description,
             "allow_permanent": False,
-            "allow_session": False,
-        }
+            "allow_session": False}
         decision = _approval._await_gateway_decision(session_key, notify_cb, approval_data, surface="gateway")
         if decision.get("notify_failed"):
             return blocked.format(why="requires approval but the approval request could not be delivered.")
@@ -282,13 +275,11 @@ def _check_approval_required_write(paths: list[str], task_id: str = "default") -
     description = (
         f"Write to SSH client config file(s): {display_targets}. "
         "The SSH config can carry ProxyCommand / Match exec directives that "
-        "run commands, so writes require your approval."
-    )
+        "run commands, so writes require your approval.")
     blocked = (
         f"BLOCKED: write to SSH config file(s) ({display_targets}) "
         "{why} Do NOT retry it via another path (terminal, execute_code) "
-        "without the user's explicit consent."
-    )
+        "without the user's explicit consent.")
 
     try:
         import tools.approval as _approval
@@ -307,8 +298,7 @@ def _check_approval_required_write(paths: list[str], task_id: str = "default") -
                 "approve in config.yaml."),
         autoapprove_log_prefix="ssh_config_write",
         fail_closed_when_no_human=True,
-        no_human_block_message=blocked.format(why=_NO_HUMAN),
-    )
+        no_human_block_message=blocked.format(why=_NO_HUMAN))
     if result.get("approved"):
         return None
     return result.get("message") or blocked.format(why="was denied.")
@@ -318,8 +308,7 @@ def _get_container_mirror_prefix_for_task(task_id: str = "default") -> str | Non
     """Return the container-side Hermes mirror prefix for persistent Docker file tools."""
     try:
         from tools.terminal_tool import (
-            _active_environments, _env_lock, _get_env_config, _resolve_container_task_id,
-        )
+            _active_environments, _env_lock, _get_env_config, _resolve_container_task_id)
         container_key = _resolve_container_task_id(task_id)
         with _env_lock:
             env = _active_environments.get(container_key) or _active_environments.get(task_id)
@@ -372,8 +361,7 @@ def _check_binary_document_write(filepath: str, task_id: str = "default") -> str
             "corrupt the file (read_file showed you EXTRACTED text, not the real "
             "bytes). Use the docx/xlsx/powerpoint skills or a library like "
             "python-docx/openpyxl/python-pptx via the terminal to create or edit "
-            "this document."
-        )
+            "this document.")
     if is_pdf_path(filepath):
         try:
             resolved = Path(_resolve_path_for_task(filepath, task_id))
@@ -386,8 +374,7 @@ def _check_binary_document_write(filepath: str, task_id: str = "default") -> str
                     "read_file showed you EXTRACTED text, not the real bytes — writing "
                     "text back would destroy the document. Use the pdf skill or a PDF "
                     "library via the terminal to modify it. (Creating a NEW .pdf file "
-                    "is allowed.)"
-                )
+                    "is allowed.)")
         except OSError:
             pass
     return None
@@ -399,8 +386,7 @@ def _check_binary_document_write(filepath: str, task_id: str = "default") -> str
 _READ_DEDUP_STATUS_MESSAGE = (
     "File unchanged since last read. The content from "
     "the earlier read_file result in this conversation is "
-    "still current — refer to that instead of re-reading."
-)
+    "still current — refer to that instead of re-reading.")
 
 
 def _is_internal_file_status_text(content: str) -> bool:
diff --git a/tools/fuzzy_match.py b/tools/fuzzy_match.py
index e0c4e98f07..5ab92b8cb8 100644
--- a/tools/fuzzy_match.py
+++ b/tools/fuzzy_match.py
@@ -18,8 +18,7 @@ Span = tuple[int, int]
 IDENTICAL_STRINGS_ERROR = (
     "No edit was applied because old_string and new_string are identical. "
     "Provide the existing text to replace in old_string and the changed "
-    "replacement text in new_string."
-)
+    "replacement text in new_string.")
 
 UNICODE_MAP = {
     "\u201c": '"', "\u201d": '"',  # smart double quotes
@@ -32,8 +31,7 @@ UNICODE_MAP = {
     "\u2000": " ", "\u2001": " ", "\u2002": " ", "\u2003": " ",
     "\u2004": " ", "\u2005": " ", "\u2006": " ", "\u2007": " ",
     "\u2008": " ", "\u2009": " ", "\u200a": " ", "\u202f": " ",
-    "\u205f": " ", "\u3000": " ",
-}
+    "\u205f": " ", "\u3000": " "}
 
 
 def _unicode_normalize(text: str) -> str:
@@ -58,8 +56,7 @@ def _window_spans(content: str, content_lines: list[str], n: int,
     """Spans of every ``n``-line window starting at ``i`` for which ``accept(i)``."""
     return [
         _calculate_line_positions(content_lines, i, i + n, len(content))
-        for i in range(len(content_lines) - n + 1) if accept(i)
-    ]
+        for i in range(len(content_lines) - n + 1) if accept(i)]
 
 
 def _match_transformed_lines(content: str, pattern: str,
@@ -257,8 +254,7 @@ def _strategy_block_anchor(content: str, pattern: str) -> list[Span]:
     potential_matches = {
         i for i in range(len(norm_content_lines) - n + 1)
         if norm_content_lines[i].strip() == first_line
-        and norm_content_lines[i + n - 1].strip() == last_line
-    }
+        and norm_content_lines[i + n - 1].strip() == last_line}
     # Looser thresholds (0.10/0.30) matched unrelated blocks; these are the safe floor.
     threshold = 0.50 if len(potential_matches) == 1 else 0.70
     pattern_middle = '\n'.join(pattern_lines[1:-1])
@@ -300,8 +296,7 @@ def _strategy_context_aware(content: str, pattern: str) -> list[Span]:
             return False
         return all(
             not p_line.strip() or _sim(p_line.strip(), c_line.strip()) >= 0.80
-            for p_line, c_line in zip(pattern_lines, block_lines)
-        )
+            for p_line, c_line in zip(pattern_lines, block_lines))
 
     return _window_spans(content, content_lines, n, accept)
 
@@ -316,8 +311,7 @@ STRATEGIES: list[tuple[str, Callable[[str, str], list[Span]]]] = [
     ("trimmed_boundary", _strategy_trimmed_boundary),
     ("unicode_normalized", _strategy_unicode_normalized),
     ("block_anchor", _strategy_block_anchor),
-    ("context_aware", _strategy_context_aware),
-]
+    ("context_aware", _strategy_context_aware)]
 
 # Matches from these only *approximately* resemble old_string — fine for one
 # unique replacement, never safe under replace_all.
@@ -386,15 +380,13 @@ def fuzzy_find_and_replace(content: str, old_string: str, new_string: str,
             return content, 0, None, (
                 f"Found {len(matches)} matches for old_string. "
                 f"Provide more context to make it unique, or use replace_all=True. "
-                f"Matches:\n{locations}"
-            )
+                f"Matches:\n{locations}")
         if replace_all and len(matches) > 1 and strategy_name in SIMILARITY_STRATEGIES:
             return content, 0, None, (
                 f"Found {len(matches)} approximate matches via the "
                 f"'{strategy_name}' strategy; replace_all only applies to exact "
                 f"matches. Provide the precise text (whitespace included) so an "
-                f"exact/line-trimmed match can be made."
-            )
+                f"exact/line-trimmed match can be made.")
 
         # Non-exact matches came through some normalization, so new_string may
         # carry serialization drift the file doesn't have.
@@ -408,8 +400,7 @@ def fuzzy_find_and_replace(content: str, old_string: str, new_string: str,
             effective_new = _preserve_unicode_in_replacement(content, matches, old_string, effective_new)
         new_content = _apply_replacements(
             content, matches, effective_new,
-            old_string=old_string if strategy_name != "exact" else None,
-        )
+            old_string=old_string if strategy_name != "exact" else None)
         return new_content, len(matches), strategy_name, None
 
     return content, 0, None, "Could not find a match for old_string in the file"
@@ -441,8 +432,7 @@ def _detect_escape_drift(content: str, matches: list[Span],
                     f"serialization artifact where an apostrophe or quote got "
                     f"prefixed with a spurious backslash. Re-read the file with "
                     f"read_file and pass old_string/new_string without "
-                    f"backslash-escaping {plain!r} characters."
-                )
+                    f"backslash-escaping {plain!r} characters.")
     return _detect_backslash_doubling(matched_regions, old_string, new_string)
 
 
@@ -476,8 +466,7 @@ def _detect_backslash_doubling(matched_regions: str, old_string: str,
         "were JSON-escaped one extra time; applying new_string verbatim would "
         "double every backslash in the file. Re-read the file with read_file "
         "and resend old_string/new_string with the backslash counts exactly "
-        "as they appear in the file."
-    )
+        "as they appear in the file.")
 
 
 def _maybe_unescape_new_string(new_string: str, content: str, matches: list[Span]) -> str:
@@ -626,8 +615,7 @@ def find_closest_lines(old_string: str, content: str, context_lines: int = 2, ma
             continue
         seen_ranges.add((start, end))
         parts.append("\n".join(
-            f"{start + j + 1:4d}| {content_lines[start + j]}" for j in range(end - start)
-        ))
+            f"{start + j + 1:4d}| {content_lines[start + j]}" for j in range(end - start)))
     if not parts:
         return ""
     result = "\n---\n".join(parts)
@@ -640,8 +628,7 @@ def find_closest_lines(old_string: str, content: str, context_lines: int = 2, ma
             "\n\nWhitespace difference detected (→ = tab, · = space):\n"
             f"  file has: {_visualize_whitespace(best_line)}\n"
             f"  you sent: {_visualize_whitespace(old_lines[0])}\n"
-            "Use the exact whitespace shown in 'file has'."
-        )
+            "Use the exact whitespace shown in 'file has'.")
     return result
 
 

From af95ff9f9fc485f935768e4ce9f56f6989d2e22e Mon Sep 17 00:00:00 2001
From: Teknium <127238744+teknium1@users.noreply.github.com>
Date: Wed, 2 Sep 2026 23:01:16 -0700
Subject: [PATCH 3/4] refactor(tools): pack re-export import blocks in
 file_tools

---
 tools/file_tools.py | 88 +++++++++++----------------------------------
 1 file changed, 21 insertions(+), 67 deletions(-)

diff --git a/tools/file_tools.py b/tools/file_tools.py
index 29bf67fb47..6f1feb8692 100644
--- a/tools/file_tools.py
+++ b/tools/file_tools.py
@@ -28,62 +28,26 @@ from tools.file_operations import (
 from tools import file_state
 from agent.redact import redact_sensitive_text
 from tools.file_tools_paths import (  # noqa: F401  (re-exported)
-    _CONTAINER_PATH_BACKENDS_FALLBACK,
-    _TERMINAL_CWD_SENTINELS,
-    _authoritative_workspace_root,
-    _configured_terminal_cwd,
-    _expand_tilde,
-    _normalize_without_host_deref,
-    _path_resolution_warning,
-    _registered_task_cwd_override,
-    _resolve_base_dir,
-    _resolve_path,
-    _resolve_path_for_task,
-    _sentinel_free_abs_cwd,
-    _terminal_env_type_for_task,
+    _CONTAINER_PATH_BACKENDS_FALLBACK, _TERMINAL_CWD_SENTINELS, _authoritative_workspace_root,
+    _configured_terminal_cwd, _expand_tilde, _normalize_without_host_deref,
+    _path_resolution_warning, _registered_task_cwd_override, _resolve_base_dir, _resolve_path,
+    _resolve_path_for_task, _sentinel_free_abs_cwd, _terminal_env_type_for_task,
     _uses_container_paths)
 from tools.file_tools_write_guards import (  # noqa: F401  (re-exported)
-    _PROTECTED_INSTRUCTION_BASENAMES,
-    _READ_DEDUP_STATUS_MESSAGE,
-    _SENSITIVE_EXACT_PATHS,
-    _SENSITIVE_PATH_PREFIXES,
-    _check_approval_required_write,
-    _check_binary_document_write,
-    _check_cross_profile_path,
-    _check_protected_instruction_write,
-    _check_sensitive_path,
-    _get_container_mirror_prefix_for_task,
-    _get_hermes_config_resolved,
-    _get_real_hermes_home,
-    _is_internal_file_status_text,
-    _is_internal_file_tool_content,
-    _looks_like_read_file_line_numbered_content,
-    _protected_instruction_config,
-    _protected_instruction_reason,
-    _request_protected_instruction_approval)
+    _PROTECTED_INSTRUCTION_BASENAMES, _READ_DEDUP_STATUS_MESSAGE, _SENSITIVE_EXACT_PATHS,
+    _SENSITIVE_PATH_PREFIXES, _check_approval_required_write, _check_binary_document_write,
+    _check_cross_profile_path, _check_protected_instruction_write, _check_sensitive_path,
+    _get_container_mirror_prefix_for_task, _get_hermes_config_resolved, _get_real_hermes_home,
+    _is_internal_file_status_text, _is_internal_file_tool_content,
+    _looks_like_read_file_line_numbered_content, _protected_instruction_config,
+    _protected_instruction_reason, _request_protected_instruction_approval)
 from tools.file_tools_read_tracking import (  # noqa: F401  (re-exported)
-    _DEDUP_CAP,
-    _NOT_FOUND_CAP,
-    _NOT_FOUND_TTL_SECONDS,
-    _READ_HISTORY_CAP,
-    _READ_TIMESTAMPS_CAP,
-    _bump_consecutive,
-    _cap_read_tracker_data,
-    _check_file_staleness,
-    _check_not_found_cache,
-    _invalidate_dedup_for_path,
-    _mark_verification_stale,
-    _patch_failure_lock,
-    _patch_failure_tracker,
-    _read_tracker,
-    _read_tracker_lock,
-    _record_not_found,
-    _record_patch_failure,
-    _reset_patch_failures,
-    _task_data,
-    _update_read_timestamp,
-    notify_other_tool_call,
-    reset_file_dedup)
+    _DEDUP_CAP, _NOT_FOUND_CAP, _NOT_FOUND_TTL_SECONDS, _READ_HISTORY_CAP, _READ_TIMESTAMPS_CAP,
+    _bump_consecutive, _cap_read_tracker_data, _check_file_staleness, _check_not_found_cache,
+    _invalidate_dedup_for_path, _mark_verification_stale, _patch_failure_lock,
+    _patch_failure_tracker, _read_tracker, _read_tracker_lock, _record_not_found,
+    _record_patch_failure, _reset_patch_failures, _task_data, _update_read_timestamp,
+    notify_other_tool_call, reset_file_dedup)
 
 logger = logging.getLogger(__name__)
 
@@ -314,14 +278,8 @@ def _create_terminal_env_for_file_ops(raw_task_id: str, task_id: str):
     command uses the configured backend (docker, modal, ...) rather than local.
     """
     from tools.terminal_tool import (
-        _CONTAINER_BACKENDS,
-        _create_configured_env,
-        _get_env_config,
-        _is_unusable_container_cwd,
-        _resolve_task_host_cwd,
-        _select_image,
-        get_session_cwd,
-        resolve_task_overrides)
+        _CONTAINER_BACKENDS, _create_configured_env, _get_env_config, _is_unusable_container_cwd,
+        _resolve_task_host_cwd, _select_image, get_session_cwd, resolve_task_overrides)
 
     config = _get_env_config()
     env_type = config["env_type"]
@@ -460,12 +418,8 @@ def _read_extracted_document(path: str, _resolved, offset: int, limit: int, task
     BEFORE the binary-extension guard so Office files can render as text.
     """
     from tools.read_extract import (
-        ANYDOC_EXTENSIONS,
-        EXTRACTABLE_EXTENSIONS,
-        MAX_DOCUMENT_BYTES,
-        ExtractionError,
-        extract_document_bytes,
-        is_extractable_document)
+        ANYDOC_EXTENSIONS, EXTRACTABLE_EXTENSIONS, MAX_DOCUMENT_BYTES, ExtractionError,
+        extract_document_bytes, is_extractable_document)
 
     if not is_extractable_document(str(_resolved)):
         return None

From 48de33b77f48446ba692236a16071fa38f0e1406 Mon Sep 17 00:00:00 2001
From: Teknium <127238744+teknium1@users.noreply.github.com>
Date: Wed, 2 Sep 2026 23:10:54 -0700
Subject: [PATCH 4/4] refactor(tools): compact file_tools group docstrings;
 regex exact strategy; one approval tail

---
 tools/file_tools.py              | 191 +++++++++++--------------------
 tools/file_tools_write_guards.py |  52 ++++-----
 tools/fuzzy_match.py             |  24 +---
 3 files changed, 98 insertions(+), 169 deletions(-)

diff --git a/tools/file_tools.py b/tools/file_tools.py
index 6f1feb8692..2e0e2d7e47 100644
--- a/tools/file_tools.py
+++ b/tools/file_tools.py
@@ -1,12 +1,10 @@
 #!/usr/bin/env python3
 """File Tools Module - LLM agent file manipulation tools.
 
-Companion modules (every name is re-imported here so ``tools.file_tools.X``
-keeps working for callers and test patches):
-  * ``file_tools_paths``         — task-aware path resolution / ``~`` expansion.
-  * ``file_tools_write_guards``  — sensitive-path, protected-instruction,
-                                   approval, mirror and binary-document guards.
-  * ``file_tools_read_tracking`` — per-task dedup / loop-detection / staleness state.
+Companions (every name re-imported here so ``tools.file_tools.X`` keeps working
+for callers and test patches): ``file_tools_paths`` (task-aware resolution),
+``file_tools_write_guards`` (write-side guards), ``file_tools_read_tracking``
+(per-task dedup / loop-detection / staleness state).
 """
 
 import base64
@@ -78,11 +76,8 @@ def _truncate_to_char_budget(content: str, max_chars: int) -> tuple[str, int, bo
     """Trim line-numbered ``read_file`` content to the last COMPLETE line within *max_chars*.
 
     Returns ``(kept_text, lines_kept, truncated)`` so the caller can offer a
-    ``next_offset`` instead of rejecting the read. Lines are already clamped to
-    ``get_max_line_length()`` upstream; the overflow handled here is many lines
-    under the line-count limit (logs, wide CSV). If not even the first line
-    fits it is clamped mid-line (slicing never splits a code point) so the read
-    is never empty and the cursor advances.
+    ``next_offset`` instead of rejecting the read. If not even the first line
+    fits it is clamped mid-line so the read is never empty and the cursor advances.
     """
     if len(content) <= max_chars:
         return content, (content.count("\n") + 1 if content else 0), False
@@ -125,8 +120,7 @@ def _apply_char_budget(result_dict: dict, content: str, offset: int, total_lines
 # Above this size, a wide read (limit > 200) gets a hint toward targeted reads.
 _LARGE_FILE_HINT_BYTES = 512_000
 
-# Device/fd paths whose reads hang the process (infinite output or blocking on
-# input). Checked by path only — no I/O.
+# Device/fd paths whose reads hang the process. Checked by path only — no I/O.
 _BLOCKED_DEVICE_PATHS = frozenset({
     "/dev/zero", "/dev/random", "/dev/urandom", "/dev/full",     # never reach EOF
     "/dev/stdin", "/dev/tty", "/dev/console",                    # block on input
@@ -142,11 +136,8 @@ _BLOCKED_PROC_SUFFIXES = (
 
 
 def _file_ops_uses_host_paths(file_ops) -> bool:
-    """True when *file_ops* targets the same host filesystem as Hermes.
-
-    Only then may we stat paths or rewrite V4A headers to host-absolute paths;
-    a container/remote backend has its own filesystem namespace.
-    """
+    """True when *file_ops* targets the host filesystem (only then may we stat paths
+    or rewrite V4A headers to host-absolute paths; sandboxes have their own namespace)."""
     env = getattr(file_ops, "env", None)
     if env is None:
         return True
@@ -165,12 +156,11 @@ _V4A_MOVE_HEADER_RE = re.compile(r'^(\*\*\*\s*Move\s+File:\s*)(.+?)\s*->\s*(.+)$
 
 
 def _rewrite_v4a_patch_paths_for_host(patch: str, path_to_resolved: dict, file_ops) -> str:
-    """Rewrite V4A ``*** Update/Add/Delete/Move File:`` headers to the resolved host paths.
+    """Rewrite V4A file headers to the resolved host paths (host backends only).
 
-    ``patch_tool`` resolves every header against the task's workspace for
-    locking/staleness/reporting; the shell layer must patch those SAME files
-    rather than re-resolving a relative header against its own cwd (which can
-    differ — the git-worktree cwd bug). Only applied for host-filesystem backends.
+    The shell layer must patch the SAME files ``patch_tool`` resolved for
+    locking/staleness, not re-resolve a relative header against its own cwd
+    (which can differ — the git-worktree cwd bug).
     """
     if not _file_ops_uses_host_paths(file_ops):
         return patch
@@ -194,9 +184,8 @@ def _is_blocked_device_path(path: str) -> bool:
 def _is_blocked_device(filepath: str, base_dir: str | Path | None = None) -> bool:
     """True if the path (literal, any symlink hop, or final realpath) is a blocked device.
 
-    The literal path is checked first so aliases like /dev/stdin are caught
-    before they resolve to terminal-specific paths; each symlink hop is checked
-    so an alias to a device cannot bypass the guard.
+    Literal first so /dev/stdin is caught before resolving to a terminal path;
+    every symlink hop is checked so an alias cannot bypass the guard.
     """
     expanded = _expand_tilde(filepath)
     if base_dir is not None and not os.path.isabs(expanded):
@@ -263,20 +252,14 @@ def _is_expected_write_exception(exc: Exception) -> bool:
         isinstance(exc, OSError) and exc.errno in _EXPECTED_WRITE_ERRNOS)
 
 
-# ---------------------------------------------------------------------------
-# ShellFileOperations per terminal environment
-# ---------------------------------------------------------------------------
+# ── ShellFileOperations per terminal environment ─────────────────────────
 _file_ops_lock = threading.Lock()
 _file_ops_cache: dict = {}
 
 
 def _create_terminal_env_for_file_ops(raw_task_id: str, task_id: str):
-    """Build the terminal environment for *task_id* from config + per-task overrides.
-
-    Mirrors terminal_tool's own environment construction (shared
-    ``_create_configured_env``) so a file tool that runs before any terminal
-    command uses the configured backend (docker, modal, ...) rather than local.
-    """
+    """Build the terminal environment for *task_id* via the shared ``_create_configured_env``,
+    so a file tool that runs before any terminal command still gets the configured backend."""
     from tools.terminal_tool import (
         _CONTAINER_BACKENDS, _create_configured_env, _get_env_config, _is_unusable_container_cwd,
         _resolve_task_host_cwd, _select_image, get_session_cwd, resolve_task_overrides)
@@ -289,11 +272,9 @@ def _create_terminal_env_for_file_ops(raw_task_id: str, task_id: str):
     except Exception:
         recorded_cwd = None
     cwd = overrides.get("cwd") or recorded_cwd or config["cwd"]
-    # Re-apply the container cwd guard _get_env_config() already ran on
-    # config["cwd"]: a gateway/TUI/ACP cwd override is a raw HOST path, and
-    # ``docker run -w <host-path>`` starts the container in a directory that
-    # doesn't exist there, so search_files & co silently return nothing.
-    # Valid in-container overrides (/workspace, /root, ...) pass untouched.
+    # Re-apply the container cwd guard: a gateway/TUI/ACP override is a raw HOST
+    # path and ``docker run -w <host-path>`` makes search_files & co silently
+    # return nothing. Valid in-container overrides (/workspace, /root) pass.
     if env_type in _CONTAINER_BACKENDS and _is_unusable_container_cwd(cwd):
         if cwd != config["cwd"]:
             logger.info(
@@ -314,11 +295,10 @@ def _create_terminal_env_for_file_ops(raw_task_id: str, task_id: str):
 def _get_file_ops(task_id: str = "default") -> ShellFileOperations:
     """Get or create ShellFileOperations for the task's terminal environment.
 
-    Thread-safe via terminal_tool's per-task creation locks, so concurrent tool
-    calls never build duplicate sandboxes. Subagent task_ids collapse to
-    "default" (``_resolve_container_task_id``) so delegate_task children share
-    the parent's container and cached file_ops; RL/benchmark task_ids with a
-    registered env override keep their isolation.
+    Uses terminal_tool's per-task creation locks (no duplicate sandboxes).
+    Subagent task_ids collapse to "default" (``_resolve_container_task_id``) so
+    delegate_task children share the parent's container; RL/benchmark task_ids
+    with a registered env override keep their isolation.
     """
     from tools.terminal_tool import (
         _active_environments, _env_lock, _last_activity, _start_cleanup_thread,
@@ -328,8 +308,7 @@ def _get_file_ops(task_id: str = "default") -> ShellFileOperations:
     raw_task_id = task_id or "default"
     task_id = _resolve_container_task_id(raw_task_id)
 
-    # Fast path: cached AND the environment is still alive (the cleanup thread
-    # may have killed it).
+    # Fast path: cached AND the environment is still alive (cleanup thread may have killed it).
     with _file_ops_lock:
         cached = _file_ops_cache.get(task_id)
     if cached is not None:
@@ -337,10 +316,8 @@ def _get_file_ops(task_id: str = "default") -> ShellFileOperations:
             if task_id in _active_environments:
                 _last_activity[task_id] = time.time()
                 return cached
-            # Environment was cleaned up: rescue the old cwd into the session
-            # record before dropping the stale entry, FILL-ONLY — ``cached.cwd``
-            # is a snapshot of the SHARED env's cwd, not attributable to this
-            # session, so never overwrite a record the session wrote for itself.
+            # Env was cleaned up: rescue its cwd into the session record FILL-ONLY
+            # (``cached.cwd`` is the SHARED env's cwd, not this session's own).
             old_cwd = getattr(cached, "cwd", None)
             if old_cwd:
                 try:
@@ -393,11 +370,9 @@ _SPECIAL_FILE_KINDS = (
 def _special_file_kind(path) -> str | None:
     """Human name for a non-regular file type that would hang a read, else None.
 
-    Stat-based sibling of the name-based ``_is_blocked_device`` guard: a FIFO
-    or socket inside a workspace hangs ``read_file`` as hard as ``/dev/zero``
-    but carries no recognizable name. Host filesystems only (see
-    ``_file_ops_uses_host_paths``). Missing/unstattable paths return None and
-    flow to the normal read path's own error handling.
+    Stat-based sibling of ``_is_blocked_device``: a FIFO/socket in a workspace
+    hangs like ``/dev/zero`` but has no recognizable name. Host filesystems
+    only; unstattable paths return None and flow to the normal read path.
     """
     try:
         mode = os.stat(os.fspath(path)).st_mode  # follows symlinks, matching a real read
@@ -412,10 +387,9 @@ def _special_file_kind(path) -> str | None:
 def _read_extracted_document(path: str, _resolved, offset: int, limit: int, task_id: str) -> str | None:
     """Render an extractable document (.docx/.xlsx/.pdf/...) as paginated text.
 
-    Returns the JSON result, a tool_error for a binary document whose
-    extraction failed for an actionable reason (size cap, encrypted,
-    malformed), or ``None`` to fall through to the normal read path. Runs
-    BEFORE the binary-extension guard so Office files can render as text.
+    Returns the JSON result, a tool_error for an actionable extraction failure
+    (size cap, encrypted, malformed), or ``None`` to fall through to the normal
+    read path. Runs BEFORE the binary-extension guard.
     """
     from tools.read_extract import (
         ANYDOC_EXTENSIONS, EXTRACTABLE_EXTENSIONS, MAX_DOCUMENT_BYTES, ExtractionError,
@@ -432,10 +406,8 @@ def _read_extracted_document(path: str, _resolved, offset: int, limit: int, task
         extracted_text = extract_document_bytes(document_bytes, str(_resolved))
     except (ExtractionError, ValueError, base64.binascii.Error) as exc:
         logger.debug("document extraction failed for %s", path, exc_info=True)
-        # Binary document formats surface the specific failure: the fallthrough
-        # can only yield a generic binary-file error or garbage bytes. .ipynb
-        # (plain JSON) and byte-transport errors (ValueError/binascii) fall
-        # through — only a specific ExtractionError carries an actionable reason.
+        # Binary formats surface the specific failure (fallthrough would only
+        # give a generic binary error); .ipynb and byte-transport errors fall through.
         _doc_ext = _resolved.suffix.lower()
         _binary_doc = _doc_ext in ANYDOC_EXTENSIONS or (
             _doc_ext in EXTRACTABLE_EXTENSIONS and _doc_ext != ".ipynb")
@@ -502,13 +474,10 @@ def _record_successful_read(task_data: dict, task_id: str, path: str, resolved_s
                             offset: int, limit: int, dedup_key: tuple, *, partial: bool) -> int:
     """Bookkeeping after a real (non-stub) read; returns the consecutive-read count.
 
-    Per-task tracker (under the lock): clear this key's stub-loop counter, add to
-    history, bump the consecutive counter, and store the mtime (feeds dedup and
-    the write/patch staleness warning). Then, OUTSIDE our lock so the registry's
-    own locking isn't nested under it: the cross-agent registry (lets write/patch
-    detect sibling-subagent writes) and the background-review read-mark (a FULL
-    read of a skill file counts like skill_view so a follow-up
-    skill_manage(patch) is accepted; no-op outside review forks).
+    Per-task tracker under the lock (stub counter, history, consecutive count,
+    mtime for dedup + staleness). Then OUTSIDE our lock (no nested locking): the
+    cross-agent registry, and the background-review read-mark (a FULL read of a
+    skill file counts like skill_view so a follow-up skill_manage(patch) is accepted).
     """
     with _read_tracker_lock:
         task_data["dedup_hits"].pop(dedup_key, None)
@@ -577,11 +546,9 @@ def read_file_tool(path: str, offset: int = 1, limit: int = 2000, task_id: str =
                 f"Cannot read binary file '{path}' ({_resolved.suffix.lower()}). "
                 "Use vision_analyze for images, or terminal to inspect binary files.")
 
-        # Hermes internal denylist: blocks prompt injection via catalog/hub
-        # metadata and credential stores under HERMES_HOME. Pass the resolved
-        # path: get_read_block_error's own resolve() runs against the process
-        # cwd, so a relative "auth.json" read with TERMINAL_CWD == HERMES_HOME
-        # would otherwise miss the denylist.
+        # Hermes internal denylist (prompt injection via catalog metadata,
+        # credential stores). Pass the RESOLVED path: the denylist's own
+        # resolve() uses the process cwd and would miss a relative "auth.json".
         block_error = get_read_block_error(str(_resolved))
         if block_error:
             return tool_error(block_error)
@@ -608,17 +575,14 @@ def read_file_tool(path: str, offset: int = 1, limit: int = 2000, task_id: str =
         result_dict = result.to_dict()
 
         # Cache a not-found result for retries. Deliberately NO early return:
-        # error results still flow through the tracking block below
-        # (consecutive-loop detection, dedup bookkeeping) — serving from the
-        # cache is the optimization, recording must stay side-effect-identical.
+        # error results still flow through the tracking below unchanged.
         _err = result_dict.get("error") or ""
         if isinstance(_err, str) and _err.startswith("File not found:"):
             _record_not_found("read", resolved_str, task_id, json.dumps(result_dict, ensure_ascii=False))
 
-        # Char budget is checked on the FORMATTED content (that is what enters
-        # context) and BEFORE redaction, to skip the regex pass on huge content.
-        # Graceful truncation instead of rejection: the model gets what fits plus
-        # a next_offset, rather than guessing a smaller limit and burning a turn.
+        # Char budget on the FORMATTED content (what enters context), BEFORE
+        # redaction (skip the regex pass on huge content); truncate gracefully
+        # with a next_offset instead of rejecting.
         file_size = result_dict.get("file_size", 0)
         max_chars = _get_max_read_chars()
         if len(result.content or "") > max_chars:
@@ -655,9 +619,7 @@ def read_file_tool(path: str, offset: int = 1, limit: int = 2000, task_id: str =
         return tool_error(str(e))
 
 
-# ---------------------------------------------------------------------------
-# Shared write/patch plumbing
-# ---------------------------------------------------------------------------
+# ── Shared write/patch plumbing ──────────────────────────────────────────
 
 def _resolve_or_none(filepath: str, task_id: str) -> str | None:
     """Task-resolved path string, or None when resolution fails for any reason."""
@@ -667,15 +629,6 @@ def _resolve_or_none(filepath: str, task_id: str) -> str | None:
         return None
 
 
-def _first_error(*checks) -> str | None:
-    """Run zero-arg guard thunks in order; return the first non-empty error string.
-
-    Lazy on purpose: later guards may prompt the user, and a hard deny earlier
-    in the chain must short-circuit them.
-    """
-    return next(filter(None, (check() for check in checks)), None)
-
-
 def _write_precheck_error(paths: list[str], content_paths: list[str], task_id: str,
                           cross_profile: bool) -> str | None:
     """Run the shared write/patch guards in order; return the first error string.
@@ -689,10 +642,12 @@ def _write_precheck_error(paths: list[str], content_paths: list[str], task_id: s
             None if cross_profile else _check_cross_profile_path(p, task_id))
         if err:
             return err
-    return _first_error(
-        *(lambda p=p: _check_binary_document_write(p, task_id) for p in content_paths),
-        lambda: _check_protected_instruction_write(paths, task_id),
-        lambda: _check_approval_required_write(paths, task_id))
+    for p in content_paths:
+        err = _check_binary_document_write(p, task_id)
+        if err:
+            return err
+    return (_check_protected_instruction_write(paths, task_id)
+            or _check_approval_required_write(paths, task_id))
 
 
 def _edit_warnings(paths: list[str], path_to_resolved: dict, task_id: str) -> list[str]:
@@ -730,17 +685,15 @@ def write_file_tool(path: str, content: str, task_id: str = "default",
     cross-PROFILE guard it was named for no longer exists).
     """
     # write_file checks the binary-document guard before the mirror guard.
-    err = _first_error(
-        lambda: _check_sensitive_path(path, task_id),
-        lambda: _check_binary_document_write(path, task_id),
-        lambda: _check_protected_instruction_write([path], task_id),
-        lambda: _check_approval_required_write([path], task_id),
-        lambda: None if cross_profile else _check_cross_profile_path(path, task_id),
-        lambda: (
-            "Refusing to write internal read_file display text as file content. "
-            "Strip read_file line-number prefixes or reconstruct the intended "
-            "file contents before writing."
-        ) if _is_internal_file_tool_content(content) else None)
+    err = (_check_sensitive_path(path, task_id)
+           or _check_binary_document_write(path, task_id)
+           or _check_protected_instruction_write([path], task_id)
+           or _check_approval_required_write([path], task_id)
+           or (None if cross_profile else _check_cross_profile_path(path, task_id)))
+    if not err and _is_internal_file_tool_content(content):
+        err = ("Refusing to write internal read_file display text as file content. "
+               "Strip read_file line-number prefixes or reconstruct the intended "
+               "file contents before writing.")
     if err:
         return tool_error(err)
     try:
@@ -779,12 +732,10 @@ def write_file_tool(path: str, content: str, task_id: str = "default",
 def _collect_v4a_header_paths(patch: str) -> tuple[list[str], list[str]] | str:
     """Extract every path named in V4A headers, rejecting ``..`` traversal.
 
-    Returns ``(all_paths, content_write_paths)`` or a tool_error string.
-    Header paths come from patch CONTENT (skill text, web extract, prompt
-    injection) so they are more attacker-influenceable than the explicit
-    ``path=`` arg, which keeps its legitimate ``..`` use. Move headers check
-    BOTH endpoints (a Move onto /etc/crontab must hit the sensitive-path check).
-    Delete/Move don't write text, so only Update/Add feed the binary-document guard.
+    Returns ``(all_paths, content_write_paths)`` or a tool_error string. Header
+    paths come from patch CONTENT (more attacker-influenceable than ``path=``,
+    which keeps its legitimate ``..`` use). Move headers check BOTH endpoints;
+    only Update/Add write text and feed the binary-document guard.
     """
     from tools.path_security import has_traversal_component
 
@@ -866,11 +817,9 @@ def patch_tool(mode: str = "replace", path: str = None, old_string: str = None,
                 _note_edited(task_id, _paths_to_check, _path_to_resolved, session_id)
                 # Clear failure counters so a future miss starts a fresh count.
                 _reset_patch_failures(task_id, [_r for _r in _path_to_resolved.values() if _r])
-        # old_string-not-found hint. Per-file failure escalation is tracked for
-        # replace mode only (V4A failures are rare; the generic hint suffices).
-        # The generic hint is suppressed when patch_replace already attached a
-        # richer "Did you mean?" snippet. The escalating hint after 3 failures
-        # exists so the model recognises the loop and changes approach.
+        # old_string-not-found hint. Failure escalation is tracked for replace
+        # mode only (V4A misses are rare); the generic hint is suppressed when
+        # patch_replace already attached a richer "Did you mean?" snippet.
         if result_dict.get("error") and "Could not find" in str(result_dict["error"]):
             failure_count = 0
             if mode == "replace" and path:
diff --git a/tools/file_tools_write_guards.py b/tools/file_tools_write_guards.py
index 9ce02b4a08..5451f32e30 100644
--- a/tools/file_tools_write_guards.py
+++ b/tools/file_tools_write_guards.py
@@ -88,14 +88,11 @@ def _check_sensitive_path(filepath: str, task_id: str = "default") -> str | None
     return None
 
 
-# ---------------------------------------------------------------------------
-# Protected agent-instruction files (always-ask approval gate)
-# ---------------------------------------------------------------------------
+# ── Protected agent-instruction files (always-ask approval gate) ─────────
 # Files that steer FUTURE agent behavior are a prompt-injection persistence
-# vector: an injected edit to AGENTS.md / CLAUDE.md / SOUL.md / .cursorrules
-# (or a project-local .hermes tree) outlives the turn. Writes ALWAYS require
-# human approval — even under --yolo — and fail closed without a human channel.
-# Basenames match in ANY directory and case-insensitively (loaders probe variants).
+# vector (AGENTS.md / CLAUDE.md / SOUL.md / .cursorrules / project .hermes tree).
+# Writes ALWAYS require human approval — even under --yolo — and fail closed
+# without a human channel. Basenames match in ANY directory, case-insensitively.
 _PROTECTED_INSTRUCTION_BASENAMES = frozenset({
     "agents.md", "claude.md", "soul.md", ".cursorrules"})
 
@@ -213,28 +210,25 @@ def _request_protected_instruction_approval(reasons: list[str], task_id: str = "
         decision = _approval._await_gateway_decision(session_key, notify_cb, approval_data, surface="gateway")
         if decision.get("notify_failed"):
             return blocked.format(why="requires approval but the approval request could not be delivered.")
-        # Any tapped scope is a one-operation grant; nothing is persisted.
-        if decision.get("resolved") and decision.get("choice") in {"once", "session", "always"}:
-            return None
-        return timed_out if not decision.get("resolved") else denied
-
-    # CLI surface: per-thread approval callback (prompt_toolkit panel).
-    try:
-        from tools.terminal_tool import _get_approval_callback
-        callback = _get_approval_callback()
-    except Exception:
-        callback = None
-
-    if callback is not None:
+        choice, timed = decision.get("choice"), not decision.get("resolved")
+    else:
+        # CLI surface: per-thread approval callback (prompt_toolkit panel).
+        try:
+            from tools.terminal_tool import _get_approval_callback
+            callback = _get_approval_callback()
+        except Exception:
+            callback = None
+        if callback is None:
+            # No human channel (script, cron, background thread): fail closed —
+            # auto-approving here would recreate the persistence vector.
+            return blocked.format(why=_NO_HUMAN)
         choice = _approval.prompt_dangerous_approval(
             display, description, allow_permanent=False, allow_session=False, approval_callback=callback)
-        if choice in {"once", "session", "always"}:
-            return None
-        return timed_out if choice == "timeout" else denied
-
-    # No human channel (script, cron, background thread): fail closed —
-    # auto-approving here would recreate the persistence vector.
-    return blocked.format(why=_NO_HUMAN)
+        timed = choice == "timeout"
+    # Any tapped scope is a one-operation grant; nothing is persisted.
+    if not timed and choice in {"once", "session", "always"}:
+        return None
+    return timed_out if timed else denied
 
 
 def _check_protected_instruction_write(paths: list[str], task_id: str = "default") -> str | None:
@@ -380,9 +374,7 @@ def _check_binary_document_write(filepath: str, task_id: str = "default") -> str
     return None
 
 
-# ---------------------------------------------------------------------------
-# Internal display text must never be persisted as file content
-# ---------------------------------------------------------------------------
+# ── Internal display text must never be persisted as file content ────────
 _READ_DEDUP_STATUS_MESSAGE = (
     "File unchanged since last read. The content from "
     "the earlier read_file result in this conversation is "
diff --git a/tools/fuzzy_match.py b/tools/fuzzy_match.py
index 5ab92b8cb8..f3b6151e0a 100644
--- a/tools/fuzzy_match.py
+++ b/tools/fuzzy_match.py
@@ -79,12 +79,9 @@ def _strip_boundary(lines: list[str]) -> list[str]:
 
 
 def _build_orig_to_norm_map(original: str) -> list[int]:
-    """Map each original index to its index in ``_unicode_normalize(original)``.
-
-    UNICODE_MAP replacements can expand one char into several, so the map is
-    needed to translate normalised spans back. Length is ``len(original)+1``;
-    the last entry is a sentinel one past the final character.
-    """
+    """Map each original index to its index in ``_unicode_normalize(original)``
+    (replacements can expand one char into several). Length ``len(original)+1``;
+    the last entry is a sentinel one past the final character."""
     result: list[int] = []
     norm_pos = 0
     for char in original:
@@ -177,18 +174,9 @@ def _map_normalized_positions(original: str, normalized: str,
 # ORIGINAL content.
 
 def _strategy_exact(content: str, pattern: str) -> list[Span]:
-    """Strategy 1: exact, non-overlapping occurrences (str.replace semantics)."""
-    matches = []
-    start = 0
-    while True:
-        pos = content.find(pattern, start)
-        if pos == -1:
-            break
-        matches.append((pos, pos + len(pattern)))
-        # Advance past the whole match: overlapping spans would corrupt the
-        # file under replace_all (reverse-order apply on stale offsets).
-        start = pos + len(pattern)
-    return matches
+    """Strategy 1: exact, non-overlapping occurrences (str.replace semantics —
+    overlapping spans would corrupt the file under replace_all)."""
+    return [m.span() for m in re.finditer(re.escape(pattern), content)]
 
 
 def _strategy_line_trimmed(content: str, pattern: str) -> list[Span]: