refactor(tools): unify path anchoring, cached lookups and resolve-or-none clones

tools/file_tools_paths.py

@@ -124,6 +124,35 @@ def _authoritative_workspace_root(task_id: str = "default") -> str | None:
     return recorded or _registered_task_cwd_override(task_id) or _configured_terminal_cwd()
 
 
+def _host_text(text: str, container_paths: bool) -> str:
+    """Expand ``~``; on host backends also translate Git Bash ``/c/Users/...`` drive
+    paths before Path sees them. Container/WSL Linux paths are never rewritten."""
+    if not container_paths:
+        from tools.environments.local import _msys_to_windows_path
+
+        text = _msys_to_windows_path(text)
+    return _expand_tilde(text)
+
+
+def _anchor(text: str, base, container_paths: bool) -> Path | PurePosixPath:
+    """Return *text* as an absolute, normalized path, joining it onto ``base()`` when
+    relative. Container: pure-posix, no host deref. Host: resolve() (win32: ntpath normpath)."""
+    if container_paths:
+        if not posixpath.isabs(text):
+            text = posixpath.join(str(base()), text)
+        return _normalize_without_host_deref(text)
+    if sys.platform == "win32":
+        import ntpath
+
+        if not ntpath.isabs(text):
+            text = ntpath.join(str(base()), text)
+        return Path(ntpath.normpath(text))
+    p = Path(text)
+    if not p.is_absolute():
+        p = Path(base()) / p
+    return p.resolve()
+
+
 def _resolve_base_dir(
     task_id: str = "default", *, container_paths: bool | None = None) -> Path | PurePosixPath:
     """Return the ABSOLUTE base directory for resolving relative paths:
@@ -131,59 +160,16 @@ def _resolve_base_dir(
     root = _authoritative_workspace_root(task_id)
     if container_paths is None:
         container_paths = _uses_container_paths(task_id)
-    base_text = _expand_tilde(root) if root else os.getcwd()
-    if container_paths:
-        if not posixpath.isabs(base_text):
-            base_text = posixpath.join(os.getcwd(), base_text)
-        return _normalize_without_host_deref(base_text)
-    # Git Bash ``pwd -P`` reports ``/c/Users/...``; translate before Path.
-    from tools.environments.local import _msys_to_windows_path
-
-    base_text = _msys_to_windows_path(base_text)
-    if sys.platform == "win32":
-        import ntpath
-
-        if not ntpath.isabs(base_text):
-            base_text = ntpath.join(os.getcwd(), base_text)
-        return Path(ntpath.normpath(base_text))
-    base = Path(base_text)
-    if not base.is_absolute():
-        # Anchor a backend's relative cwd once, here, not at resolve() time.
-        base = Path(os.getcwd()) / base
-    return base.resolve()
+    # A backend's relative cwd is anchored to the process cwd once, here.
+    return _anchor(_host_text(root or os.getcwd(), container_paths), os.getcwd, container_paths)
 
 
 def _resolve_path_for_task(filepath: str, task_id: str = "default") -> Path | PurePosixPath:
-    """Resolve *filepath* against the task's absolute base directory.
-
-    Absolute inputs are returned resolved-but-unanchored. On native Windows,
-    MSYS drive paths (``/c/Users/...``) are translated first; container/WSL
-    Linux paths are never rewritten.
-    """
+    """Resolve *filepath* against the task's absolute base directory
+    (absolute inputs are returned resolved-but-unanchored)."""
     container_paths = _uses_container_paths(task_id)
-    if container_paths:
-        expanded = _expand_tilde(filepath)
-        if posixpath.isabs(expanded):
-            return _normalize_without_host_deref(expanded)
-        resolved = _resolve_base_dir(task_id, container_paths=True) / expanded
-        return _normalize_without_host_deref(resolved)
-
-    from tools.environments.local import _msys_to_windows_path
-
-    expanded = _expand_tilde(_msys_to_windows_path(filepath))
-    if sys.platform == "win32":
-        import ntpath
-
-        if ntpath.isabs(expanded):
-            return Path(ntpath.normpath(expanded))
-        joined = ntpath.join(str(_resolve_base_dir(task_id, container_paths=False)), expanded)
-        return Path(ntpath.normpath(joined))
-
-    p = Path(expanded)
-    if p.is_absolute():
-        return p.resolve()
-    resolved = _resolve_base_dir(task_id, container_paths=False) / p
-    return resolved.resolve()
+    return _anchor(_host_text(filepath, container_paths),
+                   lambda: _resolve_base_dir(task_id, container_paths=container_paths), container_paths)
 
 
 # Back-compat alias (imported by agent.context_references and tests).

tools/file_tools_read_tracking.py

@@ -82,6 +82,13 @@ def _cap_read_tracker_data(task_data: dict) -> None:
             _evict_oldest(container, cap)
 
 
+def _resolved_or_none(filepath: str, task_id: str) -> str | None:
+    try:
+        return str(_resolve_path_for_task(filepath, task_id))
+    except (OSError, ValueError):
+        return None
+
+
 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)
@@ -167,9 +174,8 @@ def notify_other_tool_call(task_id: str = "default"):
 def _invalidate_dedup_for_path(filepath: str, task_id: str) -> None:
     """Evict every dedup entry (all offset/limit ranges) and not-found entry for *filepath*
     after a write, so the next read returns fresh content. Acquires the lock itself."""
-    try:
-        resolved = str(_resolve_path_for_task(filepath, task_id))
-    except (OSError, ValueError):
+    resolved = _resolved_or_none(filepath, task_id)
+    if resolved is None:
         return
     with _read_tracker_lock:
         task_data = _read_tracker.get(task_id)
@@ -187,10 +193,12 @@ def _update_read_timestamp(filepath: str, task_id: str) -> None:
     """After a successful write: invalidate dedup and refresh the stored mtime so
     consecutive edits by the same task don't trigger false staleness warnings."""
     _invalidate_dedup_for_path(filepath, task_id)
+    resolved = _resolved_or_none(filepath, task_id)
+    if resolved is None:
+        return
     try:
-        resolved = str(_resolve_path_for_task(filepath, task_id))
         current_mtime = os.path.getmtime(resolved)
-    except (OSError, ValueError):
+    except OSError:
         return
     with _read_tracker_lock:
         task_data = _read_tracker.get(task_id)
@@ -202,9 +210,8 @@ def _update_read_timestamp(filepath: str, task_id: str) -> None:
 def _check_file_staleness(filepath: str, task_id: str) -> str | None:
     """Warn (don't block) when the file's mtime changed since this task last read it.
     ``None`` when never read, fresh, or unstattable (a deleted file is the write's problem)."""
-    try:
-        resolved = str(_resolve_path_for_task(filepath, task_id))
-    except (OSError, ValueError):
+    resolved = _resolved_or_none(filepath, task_id)
+    if resolved is None:
         return None
     with _read_tracker_lock:
         task_data = _read_tracker.get(task_id)

tools/file_tools_write_guards.py

@@ -30,36 +30,42 @@ _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 not _hermes_config_resolved_loaded:
-        _hermes_config_resolved_loaded = True
+def _cached_lookup(slot: str, flag: str, primary, fallback) -> str | None:
+    """Fill module global *slot* once (guarded by *flag*) from ``primary()``, else
+    ``fallback()``, else None. Module globals so tests can monkeypatch the slots."""
+    g = globals()
+    if not g[flag]:
+        g[flag] = True
         try:
-            from hermes_cli.config import get_config_path
-            _hermes_config_resolved = str(get_config_path().resolve())
+            g[slot] = primary()
         except Exception:
             try:
-                _hermes_config_resolved = str(Path(_expand_tilde("~/.hermes/config.yaml")).resolve())
+                g[slot] = fallback()
             except Exception:
-                _hermes_config_resolved = None
-    return _hermes_config_resolved
+                g[slot] = None
+    return g[slot]
+
+
+def _config_path_resolved() -> str:
+    from hermes_cli.config import get_config_path
+    return str(get_config_path().resolve())
+
+
+def _hermes_home_real() -> str:
+    from hermes_constants import get_hermes_home
+    return os.path.realpath(str(get_hermes_home()))
+
+
+def _get_hermes_config_resolved() -> str | None:
+    """Return the resolved absolute path of the Hermes config file (cached)."""
+    return _cached_lookup("_hermes_config_resolved", "_hermes_config_resolved_loaded", _config_path_resolved,
+                          lambda: str(Path(_expand_tilde("~/.hermes/config.yaml")).resolve()))
 
 
 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
+    return _cached_lookup("_real_hermes_home_cached", "_real_hermes_home_loaded", _hermes_home_real,
+                          lambda: os.path.realpath(_expand_tilde("~/.hermes")))
 
 
 def _resolved_or_raw(filepath: str, task_id: str) -> str:

tools/fuzzy_match.py

@@ -26,8 +26,7 @@ UNICODE_MAP = {
     "\u2014": "--", "\u2013": "-",  # em/en dashes
     "\u2026": "...", "\u00a0": " ",  # ellipsis and non-breaking space
     "\u2212": "-",  # typographic minus (math/scientific docs)
-    # Space-separator family (Zs) beyond NBSP: otherwise such files miss every
-    # precise strategy and fall to the similarity fallback (wrong-region risk).
+    # Space-separator family (Zs): otherwise such files fall to the similarity fallback.
     "\u2000": " ", "\u2001": " ", "\u2002": " ", "\u2003": " ",
     "\u2004": " ", "\u2005": " ", "\u2006": " ", "\u2007": " ",
     "\u2008": " ", "\u2009": " ", "\u200a": " ", "\u202f": " ",
@@ -548,18 +547,11 @@ def find_closest_lines(old_string: str, content: str, context_lines: int = 2, ma
     if not anchor:
         return ""
 
-    scored = []
-    for i, line in enumerate(content_lines):
-        stripped = line.strip()
-        if not stripped:
-            continue
-        ratio = SequenceMatcher(None, anchor, stripped).ratio()
-        if ratio > 0.3:
-            scored.append((ratio, i))
-    if not scored:
+    scored = sorted(((SequenceMatcher(None, anchor, line.strip()).ratio(), i)
+                     for i, line in enumerate(content_lines) if line.strip()), key=lambda x: -x[0])
+    top = [s for s in scored if s[0] > 0.3][:max_results]
+    if not top:
         return ""
-    scored.sort(key=lambda x: -x[0])
-    top = scored[:max_results]
 
     parts = []
     seen_ranges = set()
@@ -571,8 +563,6 @@ def find_closest_lines(old_string: str, content: str, context_lines: int = 2, ma
         seen_ranges.add((start, end))
         parts.append("\n".join(
             f"{start + j + 1:4d}| {content_lines[start + j]}" for j in range(end - start)))
-    if not parts:
-        return ""
     result = "\n---\n".join(parts)
 
     # Whitespace-shaped miss: best line equals the anchor once stripped. Show