From e4436ec9e50ec994f6cf7e0af592f7a9590b0ef0 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 23:17:52 -0700 Subject: [PATCH] refactor(tools): hand-compact docstrings across file_tools group (keep every WHY) --- tools/file_state.py | 21 ++------ tools/file_tools.py | 3 +- tools/file_tools_paths.py | 86 ++++++++++--------------------- tools/file_tools_read_tracking.py | 53 ++++++------------- tools/file_tools_write_guards.py | 78 ++++++++-------------------- tools/fuzzy_match.py | 76 +++++++-------------------- 6 files changed, 91 insertions(+), 226 deletions(-) diff --git a/tools/file_state.py b/tools/file_state.py index 36dd1da640..b063ec5229 100644 --- a/tools/file_state.py +++ b/tools/file_state.py @@ -45,12 +45,8 @@ def _fmt_ts(ts: float) -> str: 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. - """ + """Pop entries until *container* is within *cap* (sets: arbitrary; dicts: oldest + by insertion order). An eviction only costs one redundant re-send or staleness check.""" for _ in range(len(container) - cap): try: if isinstance(container, set): @@ -110,22 +106,15 @@ class FileStateRegistry: 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``. - - Checked in severity order: (1) a sibling subagent wrote after this - agent's last read; (2) mtime drifted since our read (external edit) or - the read was partial; (3) this agent never read the file. Never raises - — callers decide whether to block or warn. - """ + """Model-facing warning if this write would be stale, else ``None``. Severity + order: sibling wrote after our read > mtime drift / partial read > never read.""" if _disabled(): return None with self._state_lock: stamp = self._reads.get(task_id, {}).get(resolved) last_writer = self._last_writer.get(resolved) - # Never read and no write record: net-new file or first touch — - # existing sensitive-path / file-exists logic handles it. - if stamp is None and last_writer is None: + if stamp is None and last_writer is None: # net-new file / first touch return None current_mtime = _mtime_or_none(resolved) if current_mtime is None: diff --git a/tools/file_tools.py b/tools/file_tools.py index 2e0e2d7e47..a6e74da744 100644 --- a/tools/file_tools.py +++ b/tools/file_tools.py @@ -112,8 +112,7 @@ def _apply_char_budget(result_dict: dict, content: str, offset: int, total_lines 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.") + "clamped mid-line; its remainder is not retrievable via offset.") return trimmed diff --git a/tools/file_tools_paths.py b/tools/file_tools_paths.py index dda9a90573..2ff8b38896 100644 --- a/tools/file_tools_paths.py +++ b/tools/file_tools_paths.py @@ -1,11 +1,9 @@ """Path resolution for the file tools: task-aware base dir, ``~`` expansion, workspace-divergence warning. -Companion to ``tools.file_tools`` (which re-imports every name here). The core -invariant: the base directory used to anchor relative paths is ALWAYS absolute +Core invariant: the base directory anchoring relative paths is ALWAYS absolute and derived from the task's terminal cwd, never from the process cwd unless no -other anchor exists. A relative or sentinel ``TERMINAL_CWD`` would otherwise -silently anchor edits to the agent process cwd (e.g. the main repo while a -worktree session is active). +other anchor exists (a relative/sentinel ``TERMINAL_CWD`` would silently anchor +edits to the agent process cwd, e.g. the main repo during a worktree session). """ import os @@ -13,9 +11,8 @@ import posixpath import sys from pathlib import Path, PurePosixPath -# ``TERMINAL_CWD`` values that mean "not configured", not a directory to resolve -# against ("." from a stale config; "auto"/"cwd" are setup-wizard placeholders). -# The gateway sanitizes the same set at import time (gateway/run.py). +# ``TERMINAL_CWD`` values that mean "not configured" ("." from a stale config; +# "auto"/"cwd" are wizard placeholders). gateway/run.py sanitizes the same set. _TERMINAL_CWD_SENTINELS = frozenset({"", ".", "./", "auto", "cwd"}) _CONTAINER_PATH_BACKENDS_FALLBACK = frozenset({"docker", "singularity", "modal", "daytona", "vercel_sandbox"}) # Backend name inferred from the live environment's class name (first match wins). @@ -23,13 +20,8 @@ _ENV_CLASS_NAME_HINTS = ("local", "ssh", "docker", "singularity", "modal", "dayt def _expand_tilde(path: str) -> str: - """Expand ``~`` using the effective profile home when available. - - In-process file tools share the gateway process's HOME, which may differ - from the profile-specific HOME interactive CLI sessions use; mirroring - ``hermes_constants.get_subprocess_home()`` keeps ``~`` consistent across - interactive and gateway-driven (cron) runs. - """ + """Expand ``~`` using the effective profile home (``get_subprocess_home``) so + gateway/cron runs, whose process HOME may differ, agree with interactive CLI sessions.""" if not path or "~" not in path: return path try: @@ -77,21 +69,14 @@ def _uses_container_paths(task_id: str = "default") -> bool: def _normalize_without_host_deref(path: str | Path | PurePosixPath) -> PurePosixPath: - """Normalize path syntax without following host symlinks. - - Container paths are meaningful inside the sandbox; ``Path.resolve()`` on the - host could dereference a host-side symlink (e.g. ``/workspace``) and rewrite - the path before Docker sees it. - """ + """Normalize path syntax without following host symlinks: container paths are + meaningful inside the sandbox, and a host-side ``/workspace`` symlink must not rewrite them.""" return PurePosixPath(posixpath.normpath(str(path))) def _sentinel_free_abs_cwd(raw: str | None) -> str | None: - """Return *raw* expanded when it is a non-sentinel ABSOLUTE anchor, else ``None``. - - A relative anchor is meaningless without knowing which cwd it is relative - to — exactly the ambiguity that misroutes worktree edits. - """ + """Return *raw* expanded when it is a non-sentinel ABSOLUTE anchor, else ``None`` + (a relative anchor is exactly the ambiguity that misroutes worktree edits).""" raw = str(raw or "").strip() if raw.lower() in _TERMINAL_CWD_SENTINELS: return None @@ -101,10 +86,7 @@ def _sentinel_free_abs_cwd(raw: str | None) -> str | None: def _configured_terminal_cwd() -> str | None: """Return ``$TERMINAL_CWD`` only when it names a real (absolute, non-sentinel) anchor. - - Scope-aware: under gateway multiplexing the routed profile's cwd lives in - the per-turn terminal scope, not the process env. - """ + Scope-aware: under gateway multiplexing the routed profile's cwd lives in the per-turn scope.""" from agent.runtime_cwd import scope_terminal_cwd return _sentinel_free_abs_cwd(scope_terminal_cwd() or None) @@ -113,10 +95,8 @@ def _configured_terminal_cwd() -> str | None: def _registered_task_cwd_override(task_id: str = "default") -> str | None: """Return a registered cwd override keyed by the RAW task id, when available. - ``terminal_tool`` collapses CWD-only task overrides to the shared - ``"default"`` environment (TUI/dashboard/ACP sessions share one sandbox), - but the cwd value itself stays keyed by the raw session id — so read the - raw override before falling back to the collapsed container key. + ``terminal_tool`` collapses CWD-only overrides to the shared ``"default"`` + env, but the cwd value stays keyed by the raw session id. """ try: from tools.terminal_tool import resolve_task_overrides @@ -131,10 +111,9 @@ def _registered_task_cwd_override(task_id: str = "default") -> str | None: def _authoritative_workspace_root(task_id: str = "default") -> str | None: """Best-effort absolute workspace root, or ``None`` when no reliable anchor exists. - Order: (1) the session's own cwd record (written on every completed terminal - command; per-session, so one session's ``cd`` never leaks into another); - (2) a registered raw-keyed task/session cwd override (TUI/Desktop/ACP); - (3) a sentinel-free absolute ``$TERMINAL_CWD`` (``-w`` sessions). + Order: (1) the session's own cwd record (per-session, so one session's + ``cd`` never leaks into another); (2) a registered raw-keyed cwd override + (TUI/Desktop/ACP); (3) a sentinel-free absolute ``$TERMINAL_CWD``. """ try: from tools.terminal_tool import get_session_cwd @@ -147,14 +126,8 @@ 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: - """Return the ABSOLUTE base directory for resolving relative paths. - - Uses ``_authoritative_workspace_root`` (live cwd → registered override → - ``$TERMINAL_CWD``), falling back to the process cwd only as a last resort. - Sentinel/relative ``TERMINAL_CWD`` values are rejected outright rather than - anchored to the process cwd, so the result never depends on where the - agent process happens to run. - """ + """Return the ABSOLUTE base directory for resolving relative paths: + ``_authoritative_workspace_root``, else the process cwd as a last resort.""" root = _authoritative_workspace_root(task_id) if container_paths is None: container_paths = _uses_container_paths(task_id) @@ -163,8 +136,7 @@ def _resolve_base_dir( 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 so - # relative file-tool paths don't anchor under a nonexistent ``\\c\\Users``. + # 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) @@ -176,8 +148,7 @@ def _resolve_base_dir( return Path(ntpath.normpath(base_text)) base = Path(base_text) if not base.is_absolute(): - # A backend reporting a relative cwd is anchored to the process cwd - # once, here, so the result no longer depends on cwd at resolve(). + # Anchor a backend's relative cwd once, here, not at resolve() time. base = Path(os.getcwd()) / base return base.resolve() @@ -186,9 +157,8 @@ def _resolve_path_for_task(filepath: str, task_id: str = "default") -> Path | Pu """Resolve *filepath* against the task's absolute base directory. Absolute inputs are returned resolved-but-unanchored. On native Windows, - Git Bash / MSYS drive paths (``/c/Users/...``) are translated first so - they aren't treated as relative ``\\c\\Users\\...`` under the process cwd; - container/WSL Linux paths are never rewritten. + MSYS drive paths (``/c/Users/...``) are translated first; container/WSL + Linux paths are never rewritten. """ container_paths = _uses_container_paths(task_id) if container_paths: @@ -221,13 +191,9 @@ _resolve_path = _resolve_path_for_task def _path_resolution_warning(filepath: str, resolved: Path, task_id: str = "default") -> str | None: - """Warn when a RELATIVE path resolved OUTSIDE the task's workspace root. - - Surfaces the worktree-cwd divergence the moment it matters — the edit is - about to land in a different checkout than the terminal's cwd. ``None`` for - absolute paths, an unknown root, or a path correctly under the root. Fires - on the very first write even before any ``cd`` populated the cwd registry. - """ + """Warn when a RELATIVE path resolved OUTSIDE the task's workspace root (the + edit is about to land in a different checkout than the terminal's cwd). + ``None`` for absolute paths, an unknown root, or a path under the root.""" try: if Path(_expand_tilde(filepath)).is_absolute(): return None diff --git a/tools/file_tools_read_tracking.py b/tools/file_tools_read_tracking.py index 00435e4caf..286dfcf267 100644 --- a/tools/file_tools_read_tracking.py +++ b/tools/file_tools_read_tracking.py @@ -38,11 +38,8 @@ _NOT_FOUND_TTL_SECONDS = 60.0 # a path that didn't exist may be created soon 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. - """ + """Get-or-create the tracker entry for *task_id*, back-filling missing containers + (search_tool / tests create partial entries). Lock must be held.""" task_data = _read_tracker.setdefault(task_id, { "last_key": None, "consecutive": 0, "read_history": set()}) for key in ("dedup", "dedup_hits", "read_timestamps"): @@ -96,9 +93,8 @@ def _pop_not_found(op: str, resolved_str: str, task_id: 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. - Skips the subprocess + similar-name walk when the model retries the same - missing path. *op* is "read" or "search" (different error JSON shapes). - Evicted by TTL, by write_file/patch on the path, or by any other tool call. + *op* is "read" or "search" (different error JSON shapes). Evicted by TTL, + by write_file/patch on the path, or by any other tool call. """ with _read_tracker_lock: task_data = _read_tracker.get(task_id) @@ -109,10 +105,9 @@ def _check_not_found_cache(op: str, resolved_str: str, task_id: str) -> str | No if time.monotonic() - ts > _NOT_FOUND_TTL_SECONDS: _pop_not_found(op, resolved_str, task_id) return None - # The path may have been created since the miss was cached (terminal, - # 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. + # "check → create → read" is common, so never serve a stale miss for a path + # that now exists. The stat runs OUTSIDE the 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: _pop_not_found(op, resolved_str, task_id) @@ -139,11 +134,8 @@ def _bump_consecutive(task_data: dict, key: tuple) -> int: def reset_file_dedup(task_id: str = None): - """Clear the read-dedup cache (one task, or all when ``task_id`` is None). - - Called after context compression: the original read content was summarised - away, so a "file unchanged" stub would point at content no longer in context. - """ + """Clear the read-dedup cache (one task, or all when ``task_id`` is None). Called + after context compression: a "file unchanged" stub would point at summarised-away content.""" with _read_tracker_lock: if task_id: targets = [_read_tracker[task_id]] if _read_tracker.get(task_id) else [] @@ -158,11 +150,9 @@ def reset_file_dedup(task_id: str = None): def notify_other_tool_call(task_id: str = "default"): """Reset the consecutive read/search counter for a task. - Called by the dispatcher for every tool OTHER than read_file/search_files, - so loop detection only fires on truly consecutive repeats. Also clears the - stub-hit counters and the not-found cache: any other tool may have created - a previously-missing path (the serve-side stat covers most cases; clearing - covers the rest, e.g. permission flips). + Called by the dispatcher for every tool OTHER than read_file/search_files. + Also clears stub-hit counters and the not-found cache: any other tool may + have created a previously-missing path (or flipped its permissions). """ with _read_tracker_lock: task_data = _read_tracker.get(task_id) @@ -175,11 +165,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*. - - Called after write_file/patch so the next read returns fresh content - instead of a stale "unchanged" stub. Acquires ``_read_tracker_lock`` itself. - """ + """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): @@ -214,10 +201,7 @@ 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 to report). - """ + ``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): @@ -241,11 +225,8 @@ def _check_file_staleness(filepath: str, task_id: str) -> str | 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 - recognised, else the task's workspace root, else the first path's parent. - """ + """Best-effort note that successful edits made prior verification stale. cwd: the + first edited path's recognised project root, else the workspace root, else the first parent.""" from pathlib import Path paths = [p for p in resolved_paths if p] diff --git a/tools/file_tools_write_guards.py b/tools/file_tools_write_guards.py index 5451f32e30..bfe958f806 100644 --- a/tools/file_tools_write_guards.py +++ b/tools/file_tools_write_guards.py @@ -99,10 +99,8 @@ _PROTECTED_INSTRUCTION_BASENAMES = frozenset({ def _protected_instruction_config() -> tuple[bool, list[str]]: """Return ``(enabled, extra_patterns)`` from ``security.protected_instruction_files`` / - ``security.protected_instruction_extra_patterns`` (fnmatch on basename). - - Config read failures keep the gate ON — fail-safe for a security boundary. - """ + ``security.protected_instruction_extra_patterns`` (fnmatch on basename). Config read + failures keep the gate ON — fail-safe for a security boundary.""" try: from hermes_cli.config import load_config, cfg_get cfg = load_config() @@ -121,11 +119,7 @@ def _protected_instruction_reason(filepath: str, task_id: str = "default", *, enabled: bool | None = None, extra_patterns: list[str] | None = None) -> str | None: """Return a short label when ``filepath`` targets a protected instruction file, else ``None``. - - Matches BOTH the normalized input and its realpath so neither a symlink - pointing AT a protected file nor a protected name that is itself a symlink - escapes; ``..`` traversal is neutralized by normpath/realpath first. - """ + Matches BOTH the normalized input and its realpath so no symlink direction escapes.""" if enabled is None or extra_patterns is None: enabled, extra_patterns = _protected_instruction_config() if not enabled: @@ -164,12 +158,11 @@ _NO_HUMAN = "requires approval but no interactive user or gateway is present to 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). + """Ask the human to approve a write to protected instruction file(s); ``None`` when approved. - Returns ``None`` when approved, else a BLOCKED error string. Deliberately - NOT routed through ``_run_approval_gate``: that honors --yolo and - session/permanent allowlists, and this gate is one-operation approval EVERY - time with no persisted scope. Fail-closed when no human channel exists. + Deliberately NOT routed through ``_run_approval_gate`` (honors --yolo and + allowlists): this gate is one-operation approval EVERY time, no persisted + scope, fail-closed without a human channel. """ targets = ", ".join(dict.fromkeys(reasons)) description = ( @@ -232,12 +225,8 @@ def _request_protected_instruction_approval(reasons: list[str], task_id: str = " 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 - every protected target and a deny applies nothing (atomic all-or-nothing - beats a partially-applied patch). - """ + """Gate a write/patch touching protected instruction files. ONE protected file gates + the ENTIRE multi-file patch (one prompt, all-or-nothing).""" enabled, extra = _protected_instruction_config() if not enabled: return None @@ -249,13 +238,9 @@ def _check_protected_instruction_write(paths: list[str], task_id: str = "default 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 - (SSH ``ProxyCommand`` / ``Match exec``). Unlike the protected-instruction - gate this is a routine user edit: the prompt offers once/session/always and - honors --yolo. Fail-closed when no interactive/gateway channel exists. - """ + """Gate a write/patch touching an approval-required path (``~/.ssh/config`` can steer + execution via ``ProxyCommand``). Routine gate: once/session/always, honors --yolo, + fail-closed without an interactive/gateway channel.""" try: from agent.file_safety import is_write_approval_required except Exception: @@ -319,14 +304,9 @@ def _get_container_mirror_prefix_for_task(task_id: str = "default") -> str | Non def _check_cross_profile_path(filepath: str, task_id: str = "default") -> str | None: - """Soft-guard: warn when ``filepath`` lands on a host-side or Docker sandbox - MIRROR of Hermes state — a write the host process never reads (lost work). - - Not profile isolation: the former cross-PROFILE guard was removed by - maintainer decision (profiles were never isolated). ``cross_profile=True`` - on the tools still bypasses these mirror guards (name kept for replay compat). - Fails open on import error — the sensitive-path guard and denylist still apply. - """ + """Soft-guard: warn when ``filepath`` lands on a host-side or Docker sandbox MIRROR of + Hermes state (a write the host never reads). Not profile isolation — that guard was + removed; ``cross_profile=True`` keeps bypassing this one for replay compat. Fails open.""" try: from agent.file_safety import get_container_mirror_warning, get_sandbox_mirror_warning except Exception: @@ -339,14 +319,9 @@ def _check_cross_profile_path(filepath: str, task_id: str = "default") -> str | def _check_binary_document_write(filepath: str, task_id: str = "default") -> str | None: - """Reject text-tool writes that would corrupt a binary document. - - ``read_file`` auto-extracts Office/PDF to text, so the model plausibly - believes it holds the file's bytes and writes edited text back — which can - never form a valid container. Opaque formats (.docx/.xlsx/.pptx/.odt/...) - are always rejected; .pdf only when OVERWRITING an existing regular file - (raw PDF syntax is text-authorable, so new-file creation stays allowed). - """ + """Reject text-tool writes that would corrupt a binary document (read_file showed + EXTRACTED text, so the model may write it back). Opaque formats are always rejected; + .pdf only when OVERWRITING an existing file (raw PDF syntax is text-authorable).""" if has_opaque_document_extension(filepath): ext = filepath[filepath.rfind("."):].lower() return ( @@ -382,13 +357,8 @@ _READ_DEDUP_STATUS_MESSAGE = ( def _is_internal_file_status_text(content: str) -> bool: - """True when content is the read_file dedup status message (verbatim or lightly framed). - - Models echo the message verbatim OR wrap it with short framing ("Note:", - a trailing comment). Any write whose stripped body contains the full - message and is <=2x its length is status-dominated — a real file quoting - this message would be dramatically longer. - """ + """True when content is the read_file dedup status message, verbatim or lightly framed + (contains the full message and is <=2x its length — a real file quoting it would be longer).""" if not isinstance(content, str): return False stripped = content.strip() @@ -397,12 +367,8 @@ def _is_internal_file_status_text(content: str) -> bool: def _looks_like_read_file_line_numbered_content(content: str) -> bool: - """True for content dominated by read_file's ``LINE_NUM|CONTENT`` display. - - Rejects writes whose non-empty lines are mostly (>=60%) consecutive - numbered lines, while allowing sparse literal pipe content such as a - single ``1|value`` line. - """ + """True for content dominated by read_file's ``LINE_NUM|CONTENT`` display (>=60% of + non-empty lines are consecutive numbered lines; a lone ``1|value`` is allowed).""" if not isinstance(content, str): return False lines = [line for line in content.splitlines() if line.strip()] diff --git a/tools/fuzzy_match.py b/tools/fuzzy_match.py index f3b6151e0a..50283fb589 100644 --- a/tools/fuzzy_match.py +++ b/tools/fuzzy_match.py @@ -260,11 +260,7 @@ def _strategy_block_anchor(content: str, pattern: str) -> list[Span]: def _strategy_context_aware(content: str, pattern: str) -> list[Span]: """Strategy 9 (last resort): anchored per-line similarity, every non-blank line >= 0.80. - - The first/last-line anchor pre-filter keeps a miss from being an - O(file x pattern) scan; the all-lines requirement stops one coincidental - line match from replacing an unrelated block. - """ + The anchor pre-filter bounds the scan; the all-lines rule stops coincidental matches.""" pattern_lines = pattern.split('\n') content_lines = content.split('\n') n = len(pattern_lines) @@ -309,11 +305,8 @@ SIMILARITY_STRATEGIES = frozenset({"block_anchor", "context_aware"}) # ── Orchestrator ───────────────────────────────────────────────────────── def is_already_applied(content: str, old_string: str, new_string: str) -> bool: - """True when the requested edit is already present (re-sent edit -> success-shaped no-op). - - Conservative: new_string must be non-trivial (>= 8 chars stripped) and - appear EXACTLY; when it differs from old_string, old_string must be gone. - """ + """True when the edit is already present (re-sent edit -> success-shaped no-op). + Conservative: new_string non-trivial (>= 8 chars) and present EXACTLY; old_string gone.""" if not new_string or len(new_string.strip()) < 8 or new_string not in content: return False return old_string == new_string or old_string not in content @@ -398,12 +391,8 @@ def fuzzy_find_and_replace(content: str, old_string: str, new_string: str, def _detect_escape_drift(content: str, matches: list[Span], old_string: str, new_string: str) -> Optional[str]: - """Error string when new_string carries tool-call escape artifacts, else None. - - Fires on ``\\'``/``\\"`` present in both old_string and new_string but - absent from the matched region (spurious shell-style escaping), and on - JSON double-escaped backslash runs (see ``_detect_backslash_doubling``). - """ + """Error string when new_string carries tool-call escape artifacts, else None: + ``\\'``/``\\"`` in both strings but not the matched region, or doubled backslash runs.""" has_quote_suspects = "\\'" in new_string or '\\"' in new_string if not has_quote_suspects and "\\" not in old_string: return None @@ -431,14 +420,9 @@ def _backslash_runs(s: str) -> list[int]: def _detect_backslash_doubling(matched_regions: str, old_string: str, new_string: str) -> Optional[str]: - """Detect old_string whose every backslash run is exactly 2x the file's. - - That pattern means the arguments were JSON-escaped one extra time; a - similarity strategy still matches, and writing new_string verbatim would - double every backslash in the file. Requires the same run count, a - non-trivial signal (a run >= 2 or 2+ runs), and new_string not already - matching the file's counts. - """ + """Detect old_string whose every backslash run is exactly 2x the file's (arguments + JSON-escaped one extra time). Requires the same run count, a non-trivial signal + (a run >= 2 or 2+ runs), and new_string not already matching the file's counts.""" old_runs = _backslash_runs(old_string) file_runs = _backslash_runs(matched_regions) if (not old_runs or not file_runs or len(old_runs) != len(file_runs) @@ -458,14 +442,9 @@ def _detect_backslash_doubling(matched_regions: str, old_string: str, def _maybe_unescape_new_string(new_string: str, content: str, matches: list[Span]) -> str: - """Convert literal ``\\t``/``\\r`` in new_string to control chars, per sequence, - only when the matched file region already contains the real control char. - - Files that legitimately contain the two-char string (e.g. ``sep = "\\t"``) - have a backslash+t in the region, not a tab, so they're left alone. - ``\\n`` is deliberately excluded: newlines serialize correctly through - JSON and rewriting them would mangle escape sequences in source literals. - """ + """Convert literal ``\\t``/``\\r`` in new_string to control chars, per sequence, only + when the matched region already contains the real control char (so ``sep = "\\t"`` files + are left alone). ``\\n`` is excluded: rewriting it would mangle source escape literals.""" if "\\t" not in new_string and "\\r" not in new_string: return new_string matched_regions = _matched_regions(content, matches) @@ -486,13 +465,9 @@ def _first_meaningful_line(text: str) -> Optional[str]: def _reindent_replacement(file_region: str, old_string: str, new_string: str) -> str: - """Re-anchor ``new_string``'s indentation onto the file's actual base indent. - - After a non-exact match the LLM's base indent (first non-blank line of - old_string) may differ from the file's. Each non-blank new_string line - swaps the LLM base prefix for the file's, preserving relative nesting; - lines shallower than the LLM base are anchored to the file base. - """ + """Re-anchor ``new_string``'s indentation onto the file's actual base indent after a + non-exact match: swap the LLM base prefix (first non-blank old_string line) for the + file's, preserving relative nesting; shallower lines anchor to the file base.""" if not new_string: return new_string old_first = _first_meaningful_line(old_string) @@ -517,13 +492,8 @@ def _reindent_replacement(file_region: str, old_string: str, new_string: str) -> def _preserve_unicode_in_replacement(content: str, matches: list[Span], old_string: str, new_string: str) -> str: - """Apply only the old->new edits onto the file's original (Unicode) text. - - After a unicode_normalized match, writing the LLM's ASCII new_string - verbatim would flatten the file's em-dashes/smart quotes. Diff the - normalized old_string against new_string and keep the file's original - characters for every ``equal`` span. - """ + """Apply only the old->new edits onto the file's original (Unicode) text, so a + unicode_normalized match doesn't flatten the file's em-dashes/smart quotes.""" file_region = _matched_regions(content, matches) norm_old = _unicode_normalize(old_string) if norm_old != _unicode_normalize(file_region): @@ -545,11 +515,8 @@ def _preserve_unicode_in_replacement(content: str, matches: list[Span], def _apply_replacements(content: str, matches: list[Span], new_string: str, old_string: Optional[str] = None) -> str: - """Splice ``new_string`` over each span (end-to-start so offsets stay valid). - - ``old_string`` non-None signals a non-exact match: new_string is - re-indented per region to the file's actual indentation. - """ + """Splice ``new_string`` over each span (end-to-start so offsets stay valid); + ``old_string`` non-None (non-exact match) re-indents it per region.""" result = content for start, end in sorted(matches, key=lambda x: x[0], reverse=True): adjusted = new_string @@ -622,11 +589,8 @@ def find_closest_lines(old_string: str, content: str, context_lines: int = 2, ma def format_no_match_hint(error: Optional[str], match_count: int, old_string: str, content: str) -> str: - """'\\n\\nDid you mean...' snippet for plain no-match errors only, else ''. - - Ambiguous-match, escape-drift and identical-strings errors also have - ``match_count == 0`` but a hint would mislead there. - """ + """'\\n\\nDid you mean...' snippet for plain no-match errors only, else '' (ambiguous / + escape-drift / identical errors also have ``match_count == 0`` but a hint would mislead).""" if match_count != 0 or not error or not error.startswith("Could not find"): return "" hint = find_closest_lines(old_string, content)