diff --git a/tools/file_tools_paths.py b/tools/file_tools_paths.py index 2ff8b38896..a4246049be 100644 --- a/tools/file_tools_paths.py +++ b/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). diff --git a/tools/file_tools_read_tracking.py b/tools/file_tools_read_tracking.py index 286dfcf267..86c639a876 100644 --- a/tools/file_tools_read_tracking.py +++ b/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) diff --git a/tools/file_tools_write_guards.py b/tools/file_tools_write_guards.py index bfe958f806..147a1a2f35 100644 --- a/tools/file_tools_write_guards.py +++ b/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: diff --git a/tools/fuzzy_match.py b/tools/fuzzy_match.py index 50283fb589..ab9491e36f 100644 --- a/tools/fuzzy_match.py +++ b/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