diff --git a/tests/tools/test_file_operations.py b/tests/tools/test_file_operations.py index 9af6dcb73e..b26ca1ba62 100644 --- a/tests/tools/test_file_operations.py +++ b/tests/tools/test_file_operations.py @@ -16,7 +16,6 @@ from tools.file_operations import ( SearchMatch, LintResult, ShellFileOperations, - MAX_LINE_LENGTH, normalize_read_pagination, normalize_search_pagination, ) diff --git a/tools/file_operations.py b/tools/file_operations.py index 5e5fa28677..26b67bcab7 100644 --- a/tools/file_operations.py +++ b/tools/file_operations.py @@ -1,35 +1,24 @@ #!/usr/bin/env python3 -""" -File Operations Module +"""File operations (read, write, patch, search) over any terminal backend. -Provides file manipulation capabilities (read, write, patch, search) that work -across all terminal backends (local, docker, ssh, singularity, modal, daytona, vercel_sandbox). +Every operation is expressed as a shell command run through the backend's +``execute()``, so one implementation serves local, docker, ssh, singularity, +modal, daytona and vercel_sandbox. Companion modules (names re-exported here): +``file_operations_common`` (result dataclasses, line-ending/BOM helpers), +``file_operations_lint`` (LintMixin: syntax lint + LSP), ``file_operations_search`` +(SearchMixin: rg/grep/find backends). -The key insight is that all file operations can be expressed as shell commands, -so we wrap the terminal backend's execute() interface to provide a unified file API. - -Usage: - from tools.file_operations import ShellFileOperations - from tools.terminal_tool import _active_environments - - # Get file operations for a terminal environment file_ops = ShellFileOperations(terminal_env) - - # Read a file - result = file_ops.read_file("/path/to/file.py") - - # Write a file - result = file_ops.write_file("/path/to/new.py", "print('hello')") - - # Search for content - result = file_ops.search("TODO", path=".", file_glob="*.py") + file_ops.read_file("/path/to/file.py") + file_ops.write_file("/path/to/new.py", "print('hello')") + file_ops.search("TODO", path=".", file_glob="*.py") """ import base64 import binascii import os import re -import sys +import sys # noqa: F401 (tests monkeypatch tools.file_operations.sys.platform) import difflib import hashlib import json @@ -41,8 +30,6 @@ from pathlib import Path from tools.binary_extensions import BINARY_EXTENSIONS from agent.file_safety import ( - build_write_denied_paths, - build_write_denied_prefixes, get_write_denied_error, is_write_denied as _shared_is_write_denied, ) @@ -72,6 +59,7 @@ from tools.file_operations_common import ( # noqa: F401 (re-exported) ) from tools.file_operations_lint import ( # noqa: F401 (re-exported) LINTERS, + LINTERS_INPROC, LintMixin, _FAIL_CLOSED_INPROC_EXTS, _LINTER_UNUSABLE_PATTERNS, @@ -105,23 +93,13 @@ from tools.file_operations_search import ( # noqa: F401 (re-exported) _HOME = str(Path.home()) -WRITE_DENIED_PATHS = build_write_denied_paths(_HOME) - -WRITE_DENIED_PREFIXES = build_write_denied_prefixes(_HOME) - - def _is_write_denied(path: str) -> bool: """Return True if path is on the write deny list.""" return _shared_is_write_denied(path) # ============================================================================= -# Result Data Classes -# ============================================================================= - - -# ============================================================================= -# Abstract Interface +# Binary-content identification # ============================================================================= _MAGIC_SIGNATURES: tuple = ( @@ -262,113 +240,53 @@ class FileOperations(ABC): # Image extensions (subset of binary that we can return as base64) IMAGE_EXTENSIONS = {'.png', '.jpg', '.jpeg', '.gif', '.webp', '.bmp', '.ico'} - -# In-process linters by file extension. Preferred over shell linters when -# present — no subprocess overhead, microseconds per call. Each callable -# takes file content (str) and returns (ok: bool, error: str). An error -# string of ``"__SKIP__"`` signals the linter isn't available (missing -# dependency) and should be treated as "no linter". -LINTERS_INPROC = { - '.py': _lint_python_inproc, - '.json': _lint_json_inproc, - '.yaml': _lint_yaml_inproc, - '.yml': _lint_yaml_inproc, - '.toml': _lint_toml_inproc, -} - - -# Max limits for read operations -MAX_LINES = 2000 -MAX_LINE_LENGTH = 2000 -MAX_FILE_SIZE = 50 * 1024 # 50KB - # Echoed by the size probe when the path exists but is not a regular file. # `wc -c` prints only digits, so this can never collide with a real size. NOT_REGULAR_SENTINEL = "__hermes_not_regular__" class ShellFileOperations(LintMixin, SearchMixin, FileOperations): + """File operations over any terminal backend exposing ``execute(command, cwd)`` + returning ``{"output": str, "returncode": int}``. + + cwd rule: every ``_exec`` prefers the LIVE ``env.cwd`` so ``cd`` run via the + terminal tool is picked up immediately; the init-time ``self.cwd`` is only + a fallback for envs that don't track cwd. (Using the init-time cwd for + every call once made patches "succeed" with a plausible diff while landing + in the wrong directory.) """ - File operations implemented via shell commands. - - Works with ANY terminal backend that has execute(command, cwd) method. - This includes local, docker, singularity, ssh, modal, and daytona environments. - """ - + def __init__(self, terminal_env, cwd: str = None): - """ - Initialize file operations with a terminal environment. - - Args: - terminal_env: Any object with execute(command, cwd) method. - Returns {"output": str, "returncode": int} - cwd: Optional explicit fallback cwd when the terminal env has - no cwd attribute (rare — most backends track cwd live). - - Note: - Every _exec() call prefers the LIVE ``terminal_env.cwd`` over - ``self.cwd`` so ``cd`` commands run via the terminal tool are - picked up immediately. ``self.cwd`` is only used as a fallback - when the env has no cwd at all — it is NOT the authoritative - cwd, despite being settable at init time. - - Historical bug (fixed): prior versions of this class used the - init-time cwd for every _exec() call, which caused relative - paths passed to patch/read/write to target the wrong directory - after the user ran ``cd`` in the terminal. Patches would - claim success and return a plausible diff but land in the - original directory, producing apparent silent failures. - """ self.env = terminal_env - # Determine cwd from various possible sources. - # IMPORTANT: do NOT fall back to os.getcwd() -- that's the HOST's local - # path which doesn't exist inside container/cloud backends (modal, docker). - # If nothing provides a cwd, use "/" as a safe universal default. + # Never fall back to os.getcwd(): that is the HOST path, which doesn't + # exist inside container/cloud backends. "/" is the universal default. self.cwd = cwd or getattr(terminal_env, 'cwd', None) or \ getattr(getattr(terminal_env, 'config', None), 'cwd', None) or "/" - - # Cache for command availability checks self._command_cache: Dict[str, bool] = {} - + def _exec(self, command: str, cwd: str = None, timeout: int = None, stdin_data: str = None) -> ExecuteResult: - """Execute command via terminal backend. - - Args: - stdin_data: If provided, piped to the process's stdin instead of - embedding in the command string. Bypasses ARG_MAX. - - Cwd resolution order (critical — see class docstring): - 1. Explicit ``cwd`` arg (if provided) - 2. Live ``self.env.cwd`` (tracks ``cd`` commands run via terminal) - 3. Init-time ``self.cwd`` (fallback when env has no cwd attribute) - - This ordering ensures relative paths in file operations follow the - terminal's current directory — not the directory this file_ops was - originally created in. See test_file_ops_cwd_tracking.py. - """ + """Run ``command`` on the backend. cwd: explicit arg → live ``env.cwd`` → + init-time ``self.cwd``. ``stdin_data`` is piped (bypasses ARG_MAX).""" kwargs = {} if timeout: kwargs['timeout'] = timeout if stdin_data is not None: kwargs['stdin_data'] = stdin_data - # Resolve cwd from the live env so `cd` commands are picked up. - # Fall through to init-time self.cwd only if the env doesn't track cwd. effective_cwd = cwd or getattr(self.env, 'cwd', None) or self.cwd result = self.env.execute(command, cwd=effective_cwd, **kwargs) exit_code = result.get("returncode", 0) - # A stdin write failure with an otherwise-clean child exit is still - # a failure: the child never received the intended input. write_file - # rejects such content up front (Task 3); this mapping is - # defense-in-depth for any other stdin caller. + # A stdin write failure with a clean child exit is still a failure: the + # child never received the input. Defense-in-depth for stdin callers + # other than write_file (which rejects unencodable content up front). if result.get("stdin_error") and exit_code == 0: exit_code = 1 return ExecuteResult( stdout=result.get("output", ""), exit_code=exit_code ) - + def _has_command(self, cmd: str) -> bool: """Check if a command exists in the environment (cached).""" if cmd not in self._command_cache: @@ -377,19 +295,11 @@ class ShellFileOperations(LintMixin, SearchMixin, FileOperations): return self._command_cache[cmd] def _sample_file_bytes(self, path: str, length: int = 1000): - """Fetch the first ``length`` raw bytes of a file through the terminal. - - File operations run through a terminal backend (possibly remote), so - raw bytes cannot cross the transport directly — the terminal decodes - stdout with ``errors="replace"`` and manufactures U+FFFD at every - byte it cannot decode, including a multibyte character cut in half by - ``head -c``. Wrapping the sample in base64 lets the original bytes - survive the transport, so binary detection can happen at the byte - layer where it is well-defined (#80308 and friends). - - Returns the sample bytes, or ``None`` when the transport could not - produce clean base64 (exotic shells without ``base64``); callers fall - back to the legacy text-sample heuristic in that case. + """First ``length`` raw bytes of a file, base64-wrapped so they survive the + terminal transport (which decodes stdout with ``errors="replace"`` and + manufactures U+FFFD for every undecodable byte, including a multibyte + char cut in half by ``head -c``). Returns None when the shell produced + no clean base64 (no ``base64`` binary); callers fall back to the text heuristic. """ result = self._exec( f"head -c {length} {self._escape_shell_arg(path)} 2>/dev/null | base64" @@ -409,20 +319,15 @@ class ShellFileOperations(LintMixin, SearchMixin, FileOperations): @staticmethod def _is_likely_binary_bytes(sample: bytes) -> bool: - """Byte-layer binary detection (the boundary for the #80308 class). + """Byte-layer binary detection. - Contract: a file is text when its sample is valid UTF-8, allowing one - incomplete multibyte sequence at the very end (an artifact of cutting - the sample at a byte boundary, not a property of the file). Anything - else — NUL bytes, mid-stream invalid UTF-8 such as latin-1 or true - binaries — stays read-only, preserving the anti-mojibake guarantee - the old U+FFFD check existed for: a read→edit→write round-trip must - never rewrite undecodable bytes with replacement characters. - - A file that legitimately *contains* U+FFFD (EF BF BD — e.g. logs of - lossy output) is valid UTF-8 and reads as text; the old text-layer - check misclassified it because it could not tell a stored replacement - character from a transport-manufactured one. + Text iff the sample is valid UTF-8, allowing one incomplete multibyte + sequence at the very end (an artifact of the byte-boundary cut, not of + the file). NUL bytes or mid-stream invalid UTF-8 (latin-1, true binaries) + stay read-only so a read→edit→write round-trip never rewrites + undecodable bytes as U+FFFD. A file that legitimately CONTAINS U+FFFD + is valid UTF-8 and reads as text (the text-layer check couldn't tell a + stored replacement char from a transport-manufactured one). """ if not sample: return False @@ -443,53 +348,31 @@ class ShellFileOperations(LintMixin, SearchMixin, FileOperations): return True def _is_likely_binary(self, path: str, content_sample: str = None) -> bool: - """ - Check if a file is likely binary. - - Uses extension check (fast) + content analysis (fallback). - """ + """Legacy text-layer binary check: extension, else >30% non-printable chars.""" ext = os.path.splitext(path)[1].lower() if ext in BINARY_EXTENSIONS: return True - - # Content analysis: >30% non-printable chars = binary if content_sample: - # Undecodable bytes: the terminal env decodes stdout with - # errors="replace", so any non-UTF-8 byte arrives here already - # turned into U+FFFD. That char is "printable" (ord 65533), so the - # non-printable ratio below never catches it — and returning the - # lossy text would let a read→edit→write round-trip silently - # overwrite the original bytes with mojibake. Treat a file whose - # sample carries the replacement char as binary (read-only) so the - # agent can't corrupt it. Legitimate UTF-8 text effectively never - # contains U+FFFD. + # The terminal decodes stdout with errors="replace", so undecodable + # bytes arrive as U+FFFD — "printable", so the ratio below misses + # them. Treat the sample as binary (read-only) so a read→edit→write + # round-trip can't overwrite the original bytes with mojibake. if "\ufffd" in content_sample[:1000]: return True non_printable = sum(1 for c in content_sample[:1000] if ord(c) < 32 and c not in '\n\r\t') return non_printable / min(len(content_sample), 1000) > 0.30 - return False - - def _is_image(self, path: str) -> bool: - """Check if file is an image we can return as base64.""" - ext = os.path.splitext(path)[1].lower() - return ext in IMAGE_EXTENSIONS - - def _add_line_numbers(self, content: str, start_line: int = 1) -> str: - """Add line numbers to content in ``LINE_NUM|CONTENT`` format. - The gutter uses a compact ``|`` prefix (e.g. ``34|foo``) rather - than a fixed-width zero/space-padded one (`` 34|foo``). The - padding was pure token overhead: on dense source the padded gutter - cost ~48% more tokens than the bare content and ~16% more than the - compact form, because the leading spaces + zero-padding tokenize - into extra tokens on every single line. An A/B (Sonnet 4.6, 2 - passes) showed the compact gutter matches the padded gutter on - line-reference / patch / value-lookup / structure tasks (4/4 both), - while dropping line numbers entirely regressed line-referencing - (the model hand-counted and was off-by-one, 3/4) — so we keep the - numbers, just not the padding. + def _is_image(self, path: str) -> bool: + return os.path.splitext(path)[1].lower() in IMAGE_EXTENSIONS + + def _add_line_numbers(self, content: str, start_line: int = 1) -> str: + """Prefix each line with a compact ``|`` gutter, clamping long lines. + + Compact (not fixed-width padded): padding cost ~16% more tokens per + line for no accuracy gain in A/B, while dropping numbers entirely + regressed line-referencing (models hand-count off-by-one). """ from tools.tool_output_limits import get_max_line_length max_line_length = get_max_line_length() @@ -503,18 +386,12 @@ class ShellFileOperations(LintMixin, SearchMixin, FileOperations): return '\n'.join(numbered) def _expand_path(self, path: str) -> str: - """ - Expand shell-style paths like ~ and ~user to absolute paths. - - This must be done BEFORE shell escaping, since ~ doesn't expand - inside single quotes. - """ + """Expand ``~`` / ``~user`` via the backend's shell (its HOME, not the + host's). Must run BEFORE shell escaping — ~ doesn't expand in quotes.""" if not path: return path - - # Handle ~ and ~user + if path.startswith('~'): - # Get home directory via the terminal environment result = self._exec("echo $HOME") if result.exit_code == 0 and result.stdout.strip(): home = result.stdout.strip() @@ -585,87 +462,51 @@ class ShellFileOperations(LintMixin, SearchMixin, FileOperations): return "'" + arg.replace("'", "'\"'\"'") + "'" def _atomic_write(self, path: str, content: str) -> "ExecuteResult": - """Write ``content`` to ``path`` atomically via temp-file + rename. + """Write ``content`` to ``path`` atomically: stdin → temp file in the SAME + directory → ``mv -f`` over the target (same-FS rename; a cross-device + ``mv`` degrades to copy+unlink and is NOT atomic). ``mkdir -p`` is folded + in (one subprocess). ``exit_code == 0`` means the swap happened; non-zero + means nothing was renamed and the original (if any) is intact. - Streams ``content`` over stdin into a temp file in the SAME - directory as ``path`` (so the final ``mv`` is a real rename on the - same filesystem, not a non-atomic cross-device copy), preserves the - existing file's mode if it exists, then renames over the target. - On any failure the temp file is removed so we never leak a partial - ``.hermes-tmp`` file next to the user's data, and the original file - is left untouched. Content rides stdin so there is no ARG_MAX limit. - - ``mkdir -p`` for the parent directory is folded into this script - (one fewer subprocess vs. a separate ``mkdir -p`` call). - - Returns an :class:`ExecuteResult`; ``exit_code == 0`` means the file - was swapped into place atomically. A non-zero exit means nothing was - renamed and the original (if any) is intact. + Script notes: + - Symlink targets are resolved first so we edit the file the link points + at (replacing the link with a plain file orphans the target); the + temp dir is recomputed from the RESOLVED target. Best-effort. + - ``mktemp -p`` with a hidden, marked template (an orphan is only possible + on a hard crash between cat and mv); PID-stamped fallback without mktemp. + - Existing target: copy its mode via ``stat`` (GNU ``-c%a`` / BSD + ``-f%Lp``) + explicit ``chmod`` — ``chmod --reference`` is GNU-only. + Best-effort; a failure leaves mktemp's 0600. + - New target: ``chmod "=rw"`` AFTER cat gives umask-default perms (0644 + under 022) instead of 0600. Deliberately not ``$(umask)`` arithmetic: + zsh parses leading-zero constants as decimal and computes garbage; + quoted so zsh doesn't =word-expand it. + - ``trap ... EXIT`` removes the temp on every failure path; cleared + after a successful mv. """ q_path = self._escape_shell_arg(path) parent = os.path.dirname(path) or "." q_parent = self._escape_shell_arg(parent) - # template basename: hidden so it doesn't show up in casual `ls`, - # carries a marker so an orphaned temp (only possible on a hard - # crash *between* cat and mv) is identifiable. tmpl = self._escape_shell_arg(".hermes-tmp.XXXXXX") - # One shell script, fully quoted. Notes: - # - `mkdir -p "$d"` is folded in here so the parent directory is - # created in the same subprocess that writes the temp file — - # saves one entire subprocess spawn vs. a separate mkdir call. - # - `mktemp` lands the temp in the target's own dir (-p) so `mv` is - # same-FS atomic; we fall back to a PID-stamped name if the - # backend lacks mktemp (rare; busybox/macOS/Linux all ship it). - # - `chmod --reference` is GNU-only, so we read the octal mode with - # `stat` (GNU `-c%a` or BSD `-f%Lp`) and `chmod` it explicitly; - # silent best-effort — a perms-copy failure must not abort the - # write (the file then lands at mktemp's 0600, same as pre-fix). - # - brand-new targets get `chmod "=rw"` — the POSIX who-less - # symbolic form, which sets rw minus the process umask (e.g. - # 0644 under umask 022) instead of mktemp's hardcoded 0600 - # (#70856). Deliberately NOT shell arithmetic on `$(umask)`: - # zsh (reachable via _find_bash's $SHELL fallback) parses - # leading-zero constants as decimal and silently computes a - # garbage mode, while `chmod "=rw"` is spec-identical in - # bash/dash/ash/zsh and degrades to 0600 (pre-fix behavior) - # if an exotic chmod rejects it. - # - `trap ... EXIT` guarantees the temp is removed on every error - # path (cat failure, mv failure, signal) but NOT after a - # successful mv (the temp no longer exists by then). - # - we `cat >` the temp, then `mv -f` it over the target. script = ( "set -e; " f"d={q_parent}; t={q_path}; " - # Follow a symlink target so we edit the file the link points at, - # rather than replacing the symlink itself with a plain file (which - # orphans the real target and destroys the link). Recompute the - # temp dir from the RESOLVED target so `mv` stays same-filesystem - # atomic. Best-effort: a broken link or missing readlink/realpath - # falls back to the original path (pre-fix behavior, no regression). 'if [ -L "$t" ]; then ' 'rt="$(readlink -f "$t" 2>/dev/null || realpath "$t" 2>/dev/null || true)"; ' '[ -n "$rt" ] && { t="$rt"; d="$(dirname "$t")"; }; ' "fi; " - # Create the parent dir in the SAME subprocess that writes the - # temp file (one fewer exec vs. a separate mkdir call). Runs - # AFTER symlink resolution so a resolved target's directory is - # the one created/confirmed. 'mkdir -p "$d"; ' 'tmp="$(mktemp -p "$d" ' + tmpl + ' 2>/dev/null ' '|| mktemp "$d/.hermes-tmp.$$.XXXXXX" 2>/dev/null ' '|| { tmp="$d/.hermes-tmp.$$"; : > "$tmp" && echo "$tmp"; })"; ' '[ -n "$tmp" ] || { echo "atomic write: could not create temp file" >&2; exit 1; }; ' "trap 'rm -f \\\"$tmp\\\"' EXIT; " - # preserve mode of an existing target (best-effort, never fatal) 'if [ -e "$t" ]; then ' 'm="$(stat -c%a "$t" 2>/dev/null || stat -f%Lp "$t" 2>/dev/null || true)"; ' '[ -n "$m" ] && chmod "$m" "$tmp" 2>/dev/null || true; ' "fi; " 'cat > "$tmp"; ' - # new file: umask-default perms instead of mktemp's 0600 (#70856). - # Runs AFTER cat so a write-masking umask can't EACCES the stream; - # quoted "=rw" so zsh doesn't =word-expand it. 'if [ ! -e "$t" ]; then chmod "=rw" "$tmp" 2>/dev/null || true; fi; ' 'mv -f "$tmp" "$t"; ' "trap - EXIT" @@ -673,77 +514,44 @@ class ShellFileOperations(LintMixin, SearchMixin, FileOperations): return self._exec(script, stdin_data=content) def _detect_file_line_ending(self, path: str, pre_content: Optional[str] = None) -> Optional[str]: - """Detect the dominant line ending of a file on disk. - - If ``pre_content`` is already available (we just read the file - for lint/LSP purposes), inspect that — zero extra exec calls. - Otherwise issue a tiny ``head -c 4096`` to sample the first 4KB. - - Returns ``"\\r\\n"`` for CRLF (Windows), ``"\\n"`` for LF (Unix), - or ``None`` if undetermined (new file, empty file, single-line - file with no line break in the first chunk). - """ + """Dominant line ending on disk (``"\\r\\n"``/``"\\n"``), or None when + undeterminable (new/empty/single-line file). Uses ``pre_content`` when + given, else a 4KB ``head`` sample (exits 0 with no output for a new file).""" if pre_content: return _detect_line_ending(pre_content) - # File may not exist (new write) — `head` exits 0 with empty - # stdout in that case which yields None below. Cheap probe. - head_cmd = f"head -c 4096 {self._escape_shell_arg(path)} 2>/dev/null" - head_result = self._exec(head_cmd) + head_result = self._exec(f"head -c 4096 {self._escape_shell_arg(path)} 2>/dev/null") if head_result.exit_code != 0 or not head_result.stdout: return None return _detect_line_ending(head_result.stdout) def _file_has_bom(self, path: str, pre_content: Optional[str] = None) -> bool: - """Whether the file on disk starts with a UTF-8 BOM. - - Always probes the first 3 bytes on disk — do NOT trust - ``pre_content`` for BOM detection because the most common - provider (``read_file_raw``) deliberately strips BOMs so the - agent never sees U+FEFF glyphs. Passing BOM-stripped content - through ``pre_content`` would cause a false-negative and - silently remove the marker on rewrite. - - A missing/empty file returns False (new writes get no BOM - unless the caller explicitly includes one). - """ - head_cmd = f"head -c 3 {self._escape_shell_arg(path)} 2>/dev/null" - head_result = self._exec(head_cmd) + """Whether the file on disk starts with a UTF-8 BOM. ALWAYS probes disk: + ``pre_content`` usually comes from ``read_file_raw``, which strips BOMs, + so trusting it would silently drop the marker on rewrite. Missing/empty + file → False (new writes get no BOM unless the content carries one).""" + head_result = self._exec(f"head -c 3 {self._escape_shell_arg(path)} 2>/dev/null") if head_result.exit_code != 0 or not head_result.stdout: return False return _has_bom(head_result.stdout) - def _unified_diff(self, old_content: str, new_content: str, filename: str) -> str: - """Generate unified diff between old and new content.""" - old_lines = old_content.splitlines(keepends=True) - new_lines = new_content.splitlines(keepends=True) - diff = difflib.unified_diff( - old_lines, new_lines, - fromfile=f"a/{filename}", - tofile=f"b/{filename}" - ) - return ''.join(diff) - + return ''.join(difflib.unified_diff( + old_content.splitlines(keepends=True), new_content.splitlines(keepends=True), + fromfile=f"a/{filename}", tofile=f"b/{filename}", + )) + # ========================================================================= # READ Implementation # ========================================================================= def _size_probe_cmd(self, path: str) -> str: - """Byte size of a regular file, without opening one that never ends. + """Byte size of a REGULAR file without opening one that never ends. - ``wc -c < path`` opens the path. On a FIFO with no writer, a socket, - or a character device like /dev/zero that never reaches EOF, that - read blocks forever — and the read helpers pass no timeout to - :meth:`_exec`, so the turn wedges until the process is killed. The - device blocklist in ``tools/file_tools.py`` cannot cover this: it - matches literal ``/dev/*`` names, while a FIFO is a file *type* and - can sit at any path. - - ``[ -f ]`` is a stat, not an open — it answers exactly the question - the size probe needs (regular file, symlinks followed) without - touching the contents. Non-regular paths that exist report the - sentinel so callers can say so instead of claiming the file is - missing; a genuinely absent path still exits non-zero. + ``wc -c <`` on a writer-less FIFO, socket or /dev/zero blocks forever + (read helpers pass no timeout). The name-based device blocklist in + file_tools can't cover a FIFO — it's a file TYPE at any path. ``[ -f ]`` + is a stat (symlinks followed), so it answers without touching content; + existing non-regular paths echo the sentinel, absent paths exit non-zero. """ arg = self._escape_shell_arg(path) return ( @@ -762,29 +570,45 @@ class ShellFileOperations(LintMixin, SearchMixin, FileOperations): ) ) - # UTF-16 rescue constants (ported from MoonshotAI/kimi-code#2647, - # detection derived from VS Code's encoding sniffer): sample the leading - # bytes; trust a BOM first, then a zero-byte parity heuristic — zeros - # clustering at odd indices mean UTF-16 LE (`0xAA 0x00`), at even indices - # UTF-16 BE (`0x00 0xAA`). Only the *placement* of zeros is checked, not - # density, so mixed Latin/CJK content (whose CJK units carry no zero - # byte) still detects. Zeros at both parities, or a single isolated - # zero, mean real binary. Legacy 8-bit encodings (GBK, Big5, ...) are - # never guessed — a wrong silent guess is worse than a clear refusal. + def _probe_regular_file(self, path: str) -> tuple[int, str]: + """Run the size probe. Returns ``(file_size, status)`` with status one of + ``"ok"``, ``"missing"`` (path absent), ``"not_regular"`` (FIFO/socket/ + device/directory) or ``"bad_size"`` (unparseable ``wc`` output; size 0).""" + stat_result = self._exec(self._size_probe_cmd(path)) + if stat_result.exit_code != 0: + return 0, "missing" + stat_output = _strip_terminal_fence_leaks(stat_result.stdout).strip() + if stat_output == NOT_REGULAR_SENTINEL: + return 0, "not_regular" + try: + return int(stat_output), "ok" + except ValueError: + return 0, "bad_size" + + def _detect_binary(self, path: str) -> tuple[bool, Optional[bytes]]: + """``(is_binary, sample_bytes)`` — byte-layer detection when the transport + allows (base64 sample), else the legacy text heuristic (sample is None).""" + sample_bytes = self._sample_file_bytes(path) + if sample_bytes is not None: + ext_binary = os.path.splitext(path)[1].lower() in BINARY_EXTENSIONS + return ext_binary or self._is_likely_binary_bytes(sample_bytes), sample_bytes + sample_result = self._exec(f"head -c 1000 {self._escape_shell_arg(path)} 2>/dev/null") + sample_output = _strip_terminal_fence_leaks(sample_result.stdout) + return self._is_likely_binary(path, sample_output), None + + # UTF-16 rescue: trust a BOM first, then zero-byte PARITY (not density, so + # mixed Latin/CJK still detects): zeros at odd indices → UTF-16 LE, at even + # → BE; both parities or a single zero → real binary. Legacy 8-bit + # encodings (GBK, Big5) are never guessed — a wrong silent guess is worse + # than a clear refusal. _UTF16_MAX_BYTES = 10 * 1024 * 1024 _UTF16_SAMPLE_BYTES = 512 def _try_read_utf16(self, path: str, offset: int, limit: int, file_size: int) -> "Optional[ReadResult]": - """Attempt to read ``path`` as UTF-16 text, transcoded to UTF-8. - - Returns a populated ``ReadResult`` when the file is UTF-16 (BOM or - zero-byte parity heuristic), else ``None`` so the caller falls back - to the binary-file error. Files over 10 MiB are not rescued. - ``path`` must already be expanded (caller ran ``_expand_path``). - """ - # Extensions that are definitively binary (images, archives, ...) - # never contain UTF-16 text worth rescuing — skip the subprocess. + """Read ``path`` as UTF-16 transcoded to UTF-8, or None (caller falls back + to the binary-file error). Skips known-binary extensions and files over + 10 MiB. ``path`` must already be expanded.""" ext = os.path.splitext(path)[1].lower() if ext in BINARY_EXTENSIONS: return None @@ -869,31 +693,19 @@ class ShellFileOperations(LintMixin, SearchMixin, FileOperations): ) def read_file(self, path: str, offset: int = 1, limit: int = 2000) -> ReadResult: - """ - Read a file with pagination, binary detection, and line numbers. - - Args: - path: File path (absolute or relative to cwd) - offset: Line number to start from (1-indexed, default 1) - limit: Maximum lines to return (default 500, max 2000) - - Returns: - ReadResult with content, metadata, or error info - """ - # Expand ~ and other shell paths - path = self._expand_path(path) - - offset, limit = normalize_read_pagination(offset, limit) - - # Check if file exists and get size (POSIX, works on Linux + macOS) - stat_result = self._exec(self._size_probe_cmd(path)) + """Read a file with pagination, binary detection, and line numbers. - if stat_result.exit_code != 0: - # File not found. Before failing, try unicode-equivalent - # spellings — NFC/NFD, narrow no-break space, curly quotes - # render identically in a terminal, so the model retyping a - # visually-correct path can never discover the byte mismatch - # on its own (retrying is the tool's job, not the model's). + ``offset`` is 1-indexed; ``limit`` is clamped by ``normalize_read_pagination``. + """ + path = self._expand_path(path) # before shell escaping: ~ doesn't expand in quotes + offset, limit = normalize_read_pagination(offset, limit) + + file_size, status = self._probe_regular_file(path) + if status == "missing": + # Before failing, try unicode-equivalent spellings — NFC/NFD, narrow + # no-break space, curly quotes render identically in a terminal, so + # the model retyping a visually-correct path can never discover the + # byte mismatch on its own (retrying is the tool's job, not the model's). variant = self._unicode_variant_match(path) if variant is not None: result = self.read_file(variant, offset=offset, limit=limit) @@ -905,22 +717,10 @@ class ShellFileOperations(LintMixin, SearchMixin, FileOperations): ) result.hint = f"{note} {result.hint}" if result.hint else note return result - # No equivalent spelling — suggest similar files return self._suggest_similar_files(path) - - stat_output = _strip_terminal_fence_leaks(stat_result.stdout) - if stat_output.strip() == NOT_REGULAR_SENTINEL: + if status == "not_regular": return self._not_regular_error(path) - try: - file_size = int(stat_output.strip()) - except ValueError: - file_size = 0 - - # Check if file is too large - if file_size > MAX_FILE_SIZE: - # Still try to read, but warn - pass - + # Images are never inlined — redirect to the vision tool if self._is_image(path): return ReadResult( @@ -932,27 +732,14 @@ class ShellFileOperations(LintMixin, SearchMixin, FileOperations): "Use vision_analyze with this file path to inspect the image contents." ), ) - - # Read a sample to check for binary content — at the byte layer when - # the transport allows, falling back to the legacy text heuristic. - sample_bytes = self._sample_file_bytes(path) - if sample_bytes is not None: - ext_binary = os.path.splitext(path)[1].lower() in BINARY_EXTENSIONS - is_binary = ext_binary or self._is_likely_binary_bytes(sample_bytes) - else: - sample_cmd = f"head -c 1000 {self._escape_shell_arg(path)} 2>/dev/null" - sample_result = self._exec(sample_cmd) - sample_output = _strip_terminal_fence_leaks(sample_result.stdout) - is_binary = self._is_likely_binary(path, sample_output) + is_binary, sample_bytes = self._detect_binary(path) if is_binary: - # UTF-16 rescue (ported from MoonshotAI/kimi-code#2647): the - # terminal env decodes stdout as UTF-8 with errors="replace", so - # a UTF-16 text file (Windows Notepad .txt, PowerShell `>` - # redirects) arrives mangled with U+FFFD and trips the binary - # guard. Probe the raw bytes via the backend's Python and - # transcode to UTF-8 when a BOM or the zero-byte parity - # heuristic identifies UTF-16. + # UTF-16 rescue: the terminal env decodes stdout as UTF-8 with + # errors="replace", so a UTF-16 text file (Windows Notepad .txt, + # PowerShell `>` redirects) arrives mangled with U+FFFD and trips + # the binary guard. Probe the raw bytes via the backend's Python + # and transcode when a BOM or zero-byte parity identifies UTF-16. utf16_result = self._try_read_utf16(path, offset, limit, file_size) if utf16_result is not None: return utf16_result @@ -961,31 +748,23 @@ class ShellFileOperations(LintMixin, SearchMixin, FileOperations): file_size=file_size, error=describe_binary_file(sample_bytes, file_size), ) - - # Read with pagination using sed, clamping each line to a byte - # budget IN THE SHELL so a pathological single-line file (e.g. one - # 400MB minified line) never crosses the exec transport. The Python - # clamp in _add_line_numbers still runs afterwards; the shell clamp - # only bounds what reaches it. + + # Read with pagination using sed, clamping each line to a byte budget IN + # THE SHELL so a pathological single-line file (one 400MB minified line) + # never crosses the exec transport; the Python clamp in + # _add_line_numbers still runs afterwards. # - # Why 4*max_line_length + 1 bytes (not max_line_length + 1): - # ``cut -c`` on GNU coreutils is byte-based despite its name, and a - # byte clamp can split a multibyte UTF-8 codepoint at the boundary. - # The transport decodes with errors="replace", so a split codepoint - # becomes U+FFFD rather than an exception — but a clamp of - # max_line_length+1 BYTES yields far fewer CHARS than - # max_line_length for multibyte text, so the Python clamp would - # never fire and truncation would be silent (no "... [truncated]" - # suffix). UTF-8 codepoints are at most 4 bytes, so any line whose - # first max_line_length chars survive occupies at most - # 4*max_line_length bytes; keeping one byte more guarantees that - # every line longer than max_line_length chars still decodes to - # more than max_line_length chars, which triggers the existing - # Python-side clamp (len(line) > max_line_length) and its - # "... [truncated]" suffix. Any U+FFFD from a boundary split lands - # beyond char max_line_length and is always removed by that clamp, - # so mojibake is never visible. ``cut -b`` is used explicitly to - # document the byte semantics. + # Why 4*max_line_length + 1 bytes: ``cut -c`` is byte-based on GNU + # coreutils, and a byte clamp can split a multibyte UTF-8 codepoint (the + # transport decodes with errors="replace", so that becomes U+FFFD). A + # clamp of max_line_length+1 BYTES yields far fewer CHARS than + # max_line_length for multibyte text, so the Python clamp would never + # fire and truncation would be silent (no "... [truncated]" suffix). + # UTF-8 codepoints are at most 4 bytes, so keeping 4*max+1 bytes + # guarantees every over-long line still decodes to more than + # max_line_length chars and trips the Python clamp, which also removes + # any boundary-split U+FFFD (it lands beyond char max_line_length). + # ``cut -b`` documents the byte semantics explicitly. from tools.tool_output_limits import get_max_line_length line_clamp_bytes = 4 * get_max_line_length() + 1 end_line = offset + limit - 1 @@ -994,46 +773,39 @@ class ShellFileOperations(LintMixin, SearchMixin, FileOperations): f" | cut -b1-{line_clamp_bytes}" ) read_result = self._exec(read_cmd) - + if read_result.exit_code != 0: return ReadResult(error=f"Failed to read file: {read_result.stdout}") read_output = _strip_terminal_fence_leaks(read_result.stdout) - # Strip a leading UTF-8 BOM so the model never sees a phantom U+FEFF - # before the first real character. Only meaningful on the first - # chunk (the marker lives at byte 0); later pages can't carry it. + # Strip a leading UTF-8 BOM so the model never sees a phantom U+FEFF. + # Only the first chunk can carry it (the marker lives at byte 0). if offset == 1: read_output, _ = _strip_bom(read_output) - - # Get total line count - wc_cmd = f"wc -l < {self._escape_shell_arg(path)}" - wc_result = self._exec(wc_cmd) - wc_output = _strip_terminal_fence_leaks(wc_result.stdout) + + wc_result = self._exec(f"wc -l < {self._escape_shell_arg(path)}") try: - total_lines = int(wc_output.strip()) + total_lines = int(_strip_terminal_fence_leaks(wc_result.stdout).strip()) except ValueError: total_lines = 0 - - # Check if truncated + truncated = total_lines > end_line hint = None if truncated: hint = f"Use offset={end_line + 1} to continue reading (showing {offset}-{end_line} of {total_lines} lines)" - # ``cut`` (unlike sed -n p) always newline-terminates its output, - # so a file whose final line has no trailing newline would grow a - # phantom empty last line. Only possible when this page reaches the - # file's final line; probe the last byte and strip the artifact. + # ``cut`` (unlike sed -n p) always newline-terminates its output, so a + # file whose final line has no trailing newline would grow a phantom + # empty last line. Only possible when this page reaches the file's + # final line; probe the last byte and strip the artifact. if not truncated and read_output.endswith('\n'): - tail_cmd = f"tail -c 1 {self._escape_shell_arg(path)} | wc -l" - tail_result = self._exec(tail_cmd) + tail_result = self._exec(f"tail -c 1 {self._escape_shell_arg(path)} | wc -l") tail_output = _strip_terminal_fence_leaks(tail_result.stdout) if tail_result.exit_code == 0 and tail_output.strip() == "0": read_output = read_output[:-1] - # Ambiguous-silence guards: an empty content string is - # indistinguishable, from inside the model, from a broken tool — - # it re-reads, widens the window, tries another path. Name the - # dead end and its recovery instead. + # Ambiguous-silence guards: an empty content string is indistinguishable, + # from inside the model, from a broken tool — it re-reads, widens the + # window, tries another path. Name the dead end and its recovery instead. if file_size == 0: return ReadResult( content="", @@ -1060,15 +832,13 @@ class ShellFileOperations(LintMixin, SearchMixin, FileOperations): truncated=truncated, hint=hint ) - - def _unicode_variant_match(self, path: str) -> Optional[str]: - """Find an existing file whose name is unicode-equivalent to ``path``. - macOS names screenshots with a NARROW NO-BREAK SPACE (U+202F) before - AM/PM, stores names NFD-decomposed, and Finder renames turn ' into - \u2019 — all invisible in rendered text. Compare directory entries - under a normalization that erases exactly those differences and - return the on-disk spelling when exactly one entry matches. + def _unicode_variant_match(self, path: str) -> Optional[str]: + """On-disk spelling of a file whose name is unicode-equivalent to ``path``. + + macOS puts a NARROW NO-BREAK SPACE (U+202F) in screenshot names, stores + NFD, and Finder turns ' into \u2019 — all invisible when rendered. + Returns the entry only when EXACTLY one matches under normalization. """ dir_path = os.path.dirname(path) or "." filename = os.path.basename(path) @@ -1098,24 +868,20 @@ class ShellFileOperations(LintMixin, SearchMixin, FileOperations): for entry in _strip_terminal_fence_leaks(ls_result.stdout).splitlines() if entry and entry != filename and _canon(entry) == target ] - # Exactly one equivalent spelling = unambiguous repair. Zero or - # several = fall through to suggestions; guessing among homoglyph - # collisions would silently read the wrong file. + # Several candidates = homoglyph collision; guessing would read the wrong file. if len(candidates) == 1: return os.path.join(dir_path, candidates[0]) if dir_path != "." or "/" in path else candidates[0] return None def _suggest_similar_files(self, path: str) -> ReadResult: - """Suggest similar files when the requested file is not found.""" + """"File not found" result listing up to 5 similar names from the same directory.""" dir_path = os.path.dirname(path) or "." filename = os.path.basename(path) - basename_no_ext = os.path.splitext(filename)[0] + basename_no_ext = os.path.splitext(filename)[0].lower() ext = os.path.splitext(filename)[1].lower() lower_name = filename.lower() - # List files in the target directory - ls_cmd = f"ls -1 {self._escape_shell_arg(dir_path)} 2>/dev/null | head -50" - ls_result = self._exec(ls_cmd) + ls_result = self._exec(f"ls -1 {self._escape_shell_arg(dir_path)} 2>/dev/null | head -50") scored: list = [] # (score, filepath) — higher is better if ls_result.exit_code == 0 and ls_result.stdout.strip(): @@ -1124,73 +890,44 @@ class ShellFileOperations(LintMixin, SearchMixin, FileOperations): continue lf = f.lower() score = 0 - - # Exact match (shouldn't happen, but guard) if lf == lower_name: score = 100 - # Same base name, different extension (e.g. config.yml vs config.yaml) - elif os.path.splitext(f)[0].lower() == basename_no_ext.lower(): + elif os.path.splitext(f)[0].lower() == basename_no_ext: # config.yml vs config.yaml score = 90 - # Target is prefix of candidate or vice-versa elif lf.startswith(lower_name) or lower_name.startswith(lf): score = 70 - # Substring match (candidate contains query) elif lower_name in lf: score = 60 - # Reverse substring (query contains candidate name) elif lf in lower_name and len(lf) > 2: score = 40 - # Same extension with some overlap elif ext and os.path.splitext(f)[1].lower() == ext: common = set(lower_name) & set(lf) if len(common) >= max(len(lower_name), len(lf)) * 0.4: score = 30 - # Near-miss spelling (AGENT.md -> AGENTS.md): substring - # checks above find nothing, but a high sequence ratio - # catches 1-2 edit typos without a homegrown levenshtein. - if score == 0 and difflib.SequenceMatcher( - None, lower_name, lf - ).ratio() >= 0.8: + # Near-miss spelling (AGENT.md -> AGENTS.md): a high sequence ratio + # catches 1-2 edit typos the substring checks miss. + if score == 0 and difflib.SequenceMatcher(None, lower_name, lf).ratio() >= 0.8: score = 50 - if score > 0: scored.append((score, os.path.join(dir_path, f))) scored.sort(key=lambda x: -x[0]) - similar = [fp for _, fp in scored[:5]] - return ReadResult( error=f"File not found: {path}", - similar_files=similar + similar_files=[fp for _, fp in scored[:5]], ) def read_file_raw(self, path: str) -> ReadResult: - """Read the complete file content as a plain string. - - No pagination, no line-number prefixes, no per-line truncation. - Uses cat so the full file is returned regardless of size. - """ + """Whole file as a plain string (no pagination/line numbers/clamping).""" path = self._expand_path(path) - stat_result = self._exec(self._size_probe_cmd(path)) - if stat_result.exit_code != 0: + file_size, status = self._probe_regular_file(path) + if status == "missing": return self._suggest_similar_files(path) - stat_output = _strip_terminal_fence_leaks(stat_result.stdout) - if stat_output.strip() == NOT_REGULAR_SENTINEL: + if status == "not_regular": return self._not_regular_error(path) - try: - file_size = int(stat_output.strip()) - except ValueError: - file_size = 0 if self._is_image(path): return ReadResult(is_image=True, is_binary=True, file_size=file_size) - sample_bytes = self._sample_file_bytes(path) - if sample_bytes is not None: - ext_binary = os.path.splitext(path)[1].lower() in BINARY_EXTENSIONS - is_binary = ext_binary or self._is_likely_binary_bytes(sample_bytes) - else: - sample_result = self._exec(f"head -c 1000 {self._escape_shell_arg(path)} 2>/dev/null") - sample_output = _strip_terminal_fence_leaks(sample_result.stdout) - is_binary = self._is_likely_binary(path, sample_output) + is_binary, sample_bytes = self._detect_binary(path) if is_binary: return ReadResult( is_binary=True, file_size=file_size, @@ -1199,11 +936,9 @@ class ShellFileOperations(LintMixin, SearchMixin, FileOperations): cat_result = self._exec(f"cat {self._escape_shell_arg(path)}") if cat_result.exit_code != 0: return ReadResult(error=f"Failed to read file: {cat_result.stdout}") - # Strip a leading UTF-8 BOM so patch's fuzzy matcher operates on - # clean content (a phantom U+FEFF before line 1 would defeat an - # exact first-line match). write_file restores the BOM on the way - # back out — it re-probes the on-disk file, which still has the - # marker — so the round-trip preserves it. + # Strip a leading BOM so patch's fuzzy matcher sees clean content (a + # phantom U+FEFF defeats an exact first-line match); write_file + # re-probes disk and restores it, so the round-trip preserves it. raw_content, _ = _strip_bom(_strip_terminal_fence_leaks(cat_result.stdout)) return ReadResult( content=raw_content, @@ -1213,15 +948,12 @@ class ShellFileOperations(LintMixin, SearchMixin, FileOperations): def read_file_bytes(self, path: str, max_bytes: Optional[int] = None) -> ReadResult: """Read binary-safe bytes from any shell-backed environment.""" path = self._expand_path(path) - stat_result = self._exec(self._size_probe_cmd(path)) - if stat_result.exit_code != 0: + file_size, status = self._probe_regular_file(path) + if status == "missing": return ReadResult(error=f"File not found: {path}") - stat_output = _strip_terminal_fence_leaks(stat_result.stdout) - if stat_output.strip() == NOT_REGULAR_SENTINEL: + if status == "not_regular": return self._not_regular_error(path) - try: - file_size = int(stat_output.strip()) - except ValueError: + if status == "bad_size": return ReadResult(error=f"Could not determine file size: {path}") if max_bytes is not None and file_size > max_bytes: return ReadResult( @@ -1244,33 +976,23 @@ class ShellFileOperations(LintMixin, SearchMixin, FileOperations): ) def delete_file(self, path: str) -> WriteResult: - """Delete a single file. - - Cross-platform: runs via ``python -c`` against the terminal env's - Python so it works on Windows shells (``cmd.exe``/PowerShell) that - don't ship ``rm``. Directories are rejected here — use - ``delete_path(recursive=True)`` for trees. - """ + """Delete a single file (directories rejected; see ``delete_path``).""" return self._python_delete(path, recursive=False) def delete_path(self, path: str, recursive: bool = False) -> WriteResult: - """Cross-platform delete that handles files and (with recursive=True) - directory trees. Always preferred over emitting ``rm -rf`` / - ``Remove-Item -Recurse`` directly so the same tool call works on - every backend (local / docker / ssh / Windows). - """ + """Delete a file or (``recursive=True``) a directory tree.""" return self._python_delete(path, recursive=recursive) def _python_delete(self, path: str, recursive: bool) -> WriteResult: + """Delete via the backend's ``python -c`` so one code path works on + local/docker/ssh AND Windows shells (no ``rm`` / ``Remove-Item``).""" path = self._expand_path(path) denied = get_write_denied_error(path, verb="Delete") if denied: return WriteResult(error=denied) - # We can't shell out to ``rm`` here — it doesn't exist on Windows - # ``cmd.exe`` or PowerShell, so this code path is what's left when - # the backend's terminal is a Windows shell. Path is baked into the - # snippet via ``repr()`` so quoting is correct on every shell. + # No ``rm``: it doesn't exist on Windows cmd.exe/PowerShell backends. + # Path is baked in via ``repr()`` so quoting is correct on every shell. snippet = ( "import shutil, pathlib, sys\n" f"p = pathlib.Path({path!r})\n" @@ -1282,10 +1004,8 @@ class ShellFileOperations(LintMixin, SearchMixin, FileOperations): " else:\n" " print('is a directory: ' + str(p), file=sys.stderr); sys.exit(2)\n" " else:\n" - # NOTE: avoid ``unlink(missing_ok=True)`` — that kwarg lands in - # Python 3.8 and the remote interpreter (docker/ssh) may still - # be 3.7 on older distros. The FileNotFoundError handler below - # covers the same case and works back to 3.4. + # Not ``unlink(missing_ok=True)``: a 3.7 remote interpreter lacks it; + # the FileNotFoundError handler covers the same case. " p.unlink()\n" "except FileNotFoundError:\n" " pass\n" @@ -1295,8 +1015,7 @@ class ShellFileOperations(LintMixin, SearchMixin, FileOperations): result = self._exec(f"python3 -c {self._escape_shell_arg(snippet)}") - # Fall back to ``python`` (Windows / older systems where there's no - # ``python3`` symlink but a ``python`` binary is on PATH). + # Windows / older systems: no ``python3`` symlink, only ``python``. if result.exit_code != 0 and "python3" in (result.stdout or ""): result = self._exec(f"python -c {self._escape_shell_arg(snippet)}") @@ -1306,7 +1025,6 @@ class ShellFileOperations(LintMixin, SearchMixin, FileOperations): return WriteResult() def move_file(self, src: str, dst: str) -> WriteResult: - """Move a file via mv.""" src = self._expand_path(src) dst = self._expand_path(dst) for p in (src, dst): @@ -1324,62 +1042,17 @@ class ShellFileOperations(LintMixin, SearchMixin, FileOperations): # WRITE Implementation # ========================================================================= - def write_file(self, path: str, content: str, - pre_content: Optional[str] = None) -> WriteResult: + _LONE_SURROGATE_RE = re.compile(r"[\ud800-\udc7f\udd00-\udfff]") + + def _reject_unencodable(self, path: str, content: str) -> Optional[WriteResult]: + """Refuse content with a lone surrogate BEFORE any subprocess. + + surrogateescape-decoded content (U+DC80-U+DCFF) round-trips through the + pipe; surrogates outside that range cannot be encoded at all, and + letting them reach the pipe spawns a child that hangs or truncates the + target via empty-stdin ``cat``. A regex scan needs no encode. """ - Write content to a file, creating parent directories as needed. - - Pipes content through stdin to avoid OS ARG_MAX limits on large - files. The content never appears in the shell command string — - only the file path does. - - Before anything touches disk, a fail-closed syntax gate runs - against the CANDIDATE content: if ``path``'s extension is in - ``_FAIL_CLOSED_INPROC_EXTS`` (JSON/YAML/TOML — structured data - formats where a parse failure always means corruption) and the - candidate content doesn't parse, the write is refused outright. - No temp file, no rename, nothing on disk changes. - - After a write that clears the gate, runs a post-first / pre-lazy - lint check via ``_check_lint_delta()``. If the new content is - clean, the lint call is O(one parse). If the new content has - errors the gate didn't already catch (i.e. errors from a linter - outside ``_FAIL_CLOSED_INPROC_EXTS``, such as Python), the - pre-write content is linted too and only errors newly introduced - by this write are surfaced — pre-existing problems are filtered - out so the agent isn't distracted chasing them. - - Args: - path: File path to write - content: Content to write - pre_content: Pre-edit file content if the caller already has it - (e.g. patch_replace read the file for fuzzy matching). - When provided, skips a redundant ``cat`` subprocess to - re-read the file for lint baseline / line-ending - detection. BOM detection always probes disk (the most - common provider — ``read_file_raw`` — strips BOMs, so - trusting ``pre_content`` for BOM would cause false - negatives and silent marker loss on rewrite). When - None, reads from disk as before. - - Returns: - WriteResult with bytes written, lint summary, or error. - """ - # Expand ~ and other shell paths - path = self._expand_path(path) - - # Block writes to sensitive paths - denied = get_write_denied_error(path) - if denied: - return WriteResult(error=denied) - - # Reject lone surrogates up front with a regex scan (no encode, no - # subprocess). surrogateescape-decoded content (U+DC80–U+DCFF) - # round-trips through the pipe fine, but surrogates outside that - # range cannot be encoded at all — and letting them reach the pipe - # would spawn a child that then hangs, or truncates the target via - # empty-stdin `cat`. Refuse synchronously before any subprocess. - m = re.search(r"[\ud800-\udc7f\udd00-\udfff]", content) + m = self._LONE_SURROGATE_RE.search(content) if m: return WriteResult( error=( @@ -1388,131 +1061,140 @@ class ShellFileOperations(LintMixin, SearchMixin, FileOperations): "encoded as UTF-8. The file was NOT created or modified." ) ) + return None - # ── Fail-closed pre-write syntax gate ─────────────────────────── - # Validate the CANDIDATE content BEFORE any bytes touch disk — - # previously this only ran as a post-write lint *report* that the - # caller could ignore (or that ``files_modified`` gating wouldn't - # catch, since a lint failure never set the top-level ``error`` - # key). A structured-format write that doesn't even parse (mashed - # quotes, truncated generation, wrong indentation dialect) is a - # corrupt write, not a style nit — refuse it outright instead of - # writing first and reporting the damage afterward. - # - # Scope: only extensions in ``_FAIL_CLOSED_INPROC_EXTS`` (JSON/ - # YAML/TOML). ``.py`` deliberately keeps its pre-existing, - # non-blocking lint-delta *report* instead of a hard refusal — see - # ``_FAIL_CLOSED_INPROC_EXTS``'s docstring above for why. Extensions - # with no in-process linter at all (including ones only covered by - # a shell linter) are completely unaffected — this gate never runs - # for them, so behavior there is unchanged. - # - # Checked against the raw ``content`` argument, before the - # BOM/CRLF preservation shims below run. Those shims exist purely - # to match the on-disk file's existing conventions; linting - # post-shim would false-positive a JSONDecodeError on a - # legitimately BOM-marked JSON file purely because this method - # re-adds the marker the read layer strips — see - # ``_file_has_bom``/``_UTF8_BOM`` below. - ext = os.path.splitext(path)[1].lower() - inproc_linter = LINTERS_INPROC.get(ext) if ext in _FAIL_CLOSED_INPROC_EXTS else None - if inproc_linter is not None: - _ok, _lint_err = inproc_linter(content) - if not _ok and _lint_err != "__SKIP__": - return WriteResult( - error=( - f"Refusing to write '{path}': candidate content fails " - f"{ext} syntax validation ({_lint_err}). The file was " - "NOT created or modified. Fix the content and retry." - ) - ) + @staticmethod + def _fail_closed_syntax_error(path: str, ext: str, content: str) -> Optional[WriteResult]: + """Fail-closed pre-write gate for ``_FAIL_CLOSED_INPROC_EXTS`` (JSON/YAML/TOML). - # Capture pre-write content. Two consumers want it: - # - # 1. The lint-delta layer (for in-process linters like ast.parse - # and json.loads) needs the previous content to compute the - # set of NEW lint errors introduced by this write. - # 2. The LSP layer needs pre/post content to build a line-shift - # map — pre-existing diagnostics below the edit point shift - # when lines are added/removed, and the shift map remaps - # baseline diagnostics into post-edit coordinates so the - # strict (range-aware) delta key matches. - # - # The set of extensions we capture pre_content for is therefore - # the UNION of in-process lint coverage and LSP coverage. For - # extensions outside both sets (binaries, opaque formats), - # skipping the read keeps the hot path fast. - want_pre = ext in LINTERS_INPROC or self._lsp_handles_extension(ext) - if want_pre: - if pre_content is not None: - # Caller already has file content (e.g. patch_replace read it - # for fuzzy matching) — reuse directly, skip redundant cat. - pass - else: - # Best-effort read; failure (file missing, permission) leaves - # pre_content as None which makes both downstream consumers - # degrade gracefully (lint reports all errors; LSP skips the - # shift map). - read_cmd = f"cat {self._escape_shell_arg(path)} 2>/dev/null" - read_result = self._exec(read_cmd) - if read_result.exit_code == 0 and read_result.stdout: - pre_content = read_result.stdout + A structured-format write that doesn't parse (mashed quotes, truncated + generation) is a corrupt write, not a style nit: refuse it before any + bytes touch disk instead of reporting damage afterwards. ``.py`` keeps + its non-blocking lint-delta report (see ``_FAIL_CLOSED_INPROC_EXTS``); + extensions without an in-process linter are untouched. - # ── Line-ending preservation (Roo Code pattern) ────────────── - # If the file existed with CRLF endings and the agent's content - # has bare LFs, convert to CRLF before writing. Otherwise the - # write silently normalizes a Windows-line-ending file (and patch - # produces mixed endings when only a substituted region changes). - # Detect from a small head sample to avoid reading the full file - # for line-ending purposes alone. - original_ending = self._detect_file_line_ending(path, pre_content) - if original_ending == "\r\n": + Checked against the RAW content, before the BOM/CRLF shims: linting + post-shim would false-positive a JSONDecodeError on a legitimately + BOM-marked file purely because write_file re-adds the marker. + """ + linter = LINTERS_INPROC.get(ext) if ext in _FAIL_CLOSED_INPROC_EXTS else None + if linter is None: + return None + ok, err = linter(content) + if ok or err == "__SKIP__": + return None + return WriteResult( + error=( + f"Refusing to write '{path}': candidate content fails " + f"{ext} syntax validation ({err}). The file was " + "NOT created or modified. Fix the content and retry." + ) + ) + + def _capture_pre_content(self, path: str, ext: str, + pre_content: Optional[str]) -> Optional[str]: + """Pre-write content for the lint-delta and LSP line-shift consumers. + + Captured only for extensions in the UNION of in-process lint coverage + and LSP coverage — for anything else (binaries, opaque formats) skipping + the read keeps the hot path fast. A caller-supplied ``pre_content`` is + reused as-is; otherwise a best-effort ``cat`` whose failure (missing + file, permissions) leaves None so both consumers degrade gracefully + (lint reports all errors; LSP skips the shift map). + """ + if pre_content is not None: + return pre_content + if ext in LINTERS_INPROC or self._lsp_handles_extension(ext): + read_result = self._exec(f"cat {self._escape_shell_arg(path)} 2>/dev/null") + if read_result.exit_code == 0 and read_result.stdout: + return read_result.stdout + return None + + def _match_on_disk_conventions(self, path: str, content: str, + pre_content: Optional[str]) -> str: + """Re-apply the on-disk file's CRLF endings and UTF-8 BOM to ``content``. + + read_file strips the BOM and models send bare-LF text, so a round-trip + would otherwise silently normalize a CRLF file (patch would leave mixed + endings) and drop a byte signature some Windows toolchains key on. The + BOM is only prepended when the original had one and ``content`` doesn't + already (guards double-BOM from callers passing raw bytes). + """ + if self._detect_file_line_ending(path, pre_content) == "\r\n": content = _normalize_line_endings(content, "\r\n") - - # ── BOM preservation ────────────────────────────────────────── - # If the file on disk started with a UTF-8 BOM, keep it. read_file - # strips the BOM so the agent never sees it, which means the - # content it hands back to write_file / patch has no BOM either — - # without restoring it here a round-trip would silently strip the - # marker and change the file's byte signature (some Windows - # toolchains key on it). Only prepend when the original had a BOM - # and the new content doesn't already carry one (guards against - # double-BOM if a caller passed raw bytes). if self._file_has_bom(path, pre_content) and not _has_bom(content): content = _UTF8_BOM + content + return content - # Snapshot LSP diagnostics for this file (best-effort) so the - # post-write LSP layer can return only diagnostics introduced - # by this specific edit. Mirrors claude-code's - # ``beforeFileEdited`` pattern but wired to the local LSP - # rather than an external IDE. + def _verify_written_hash(self, path: str, content_bytes: bytes) -> tuple[Optional[bool], Optional[WriteResult]]: + """Compare the on-disk sha256 to the intended bytes (one shell call). + + Production mining shows models re-reading files right after writing to + confirm persistence; an explicit ``verified`` flag makes that turn + unnecessary, and a mismatch is a hard error instead of silent + corruption. Returns ``(verified, error_result)``; ``verified`` is None + when the hash could not be taken. + """ + try: + hash_result = self._exec(f"sha256sum {self._escape_shell_arg(path)} 2>/dev/null") + if hash_result.exit_code == 0 and hash_result.stdout.strip(): + disk_sha = hash_result.stdout.strip().split()[0] + if disk_sha != hashlib.sha256(content_bytes).hexdigest(): + return False, WriteResult( + error=( + f"Post-write verification failed for {path}: on-disk " + "content hash differs from the intended write. The " + "write did not persist correctly — re-read the file " + "and retry." + ) + ) + return True, None + except Exception: + pass + return None, None + + def write_file(self, path: str, content: str, + pre_content: Optional[str] = None) -> WriteResult: + """Write content to a file atomically, creating parent directories as needed. + + Order: deny list → lone-surrogate refusal → fail-closed syntax gate on + the CANDIDATE content (JSON/YAML/TOML; nothing touches disk on failure) + → pre-content capture → CRLF/BOM preservation → LSP baseline snapshot → + atomic write (content rides stdin, so no ARG_MAX limit and the content + never appears in the command string) → sha256 verification → lint delta + (only errors THIS write introduced) → LSP diagnostics when syntax is clean. + + ``pre_content``: pre-edit content the caller already has (patch_replace + read it for fuzzy matching); saves a ``cat``. BOM detection always + probes disk regardless — ``read_file_raw`` strips BOMs, so trusting + ``pre_content`` would silently drop the marker on rewrite. + """ + path = self._expand_path(path) + + denied = get_write_denied_error(path) + if denied: + return WriteResult(error=denied) + refused = self._reject_unencodable(path, content) + if refused is not None: + return refused + + ext = os.path.splitext(path)[1].lower() + refused = self._fail_closed_syntax_error(path, ext, content) + if refused is not None: + return refused + + pre_content = self._capture_pre_content(path, ext, pre_content) + content = self._match_on_disk_conventions(path, content, pre_content) + + # Snapshot LSP diagnostics (best-effort) so the post-write LSP layer + # returns only diagnostics introduced by this edit. self._snapshot_lsp_baseline(path) - # Write atomically. ``mkdir -p`` is folded into _atomic_write - # (one fewer subprocess vs. a separate mkdir call). - # ``dirs_created`` has always meant "parent dirs ensured" — - # ``mkdir -p`` exits 0 even when the dirs pre-exist, so the old - # separate-mkdir code reported True in exactly the same cases. - # A mkdir failure now surfaces as the atomic-write error return - # below, before this field is ever emitted. - parent = os.path.dirname(path) - dirs_created = bool(parent) + # ``dirs_created`` has always meant "parent dirs ensured": mkdir -p is + # folded into _atomic_write and exits 0 even when they pre-exist; a + # mkdir failure surfaces as the atomic-write error below. + dirs_created = bool(os.path.dirname(path)) - # Write atomically: stream into a temp file in the SAME directory, - # then ``mv`` it over the target. The rename is atomic on POSIX - # (and on every backend FS we run on), so a crash / power loss / - # truncated pipe mid-write leaves the original file intact instead - # of a half-written corrupt file. Same-directory is load-bearing — - # ``mv`` across filesystems degrades to copy+unlink, which is NOT - # atomic; keeping the temp beside the target guarantees a real - # rename. Content still rides stdin so there's no ARG_MAX limit. - # - # The temp file is created with ``mktemp`` (collision-safe) when the - # backend has it, falling back to a PID-stamped name otherwise. We - # then chmod the temp to match the existing file's mode (if any) so - # the atomic swap doesn't silently widen or narrow permissions, and - # clean the temp up on any failure so we never leak a ``.hermes-tmp`` - # turd next to the user's file. # Encode once for byte count + sha256. surrogateescape is the exact # inverse of the decode that may have produced this content, so these # are the bytes the pipe transmits and the bytes on disk. The early @@ -1529,254 +1211,164 @@ class ShellFileOperations(LintMixin, SearchMixin, FileOperations): ) ) write_result = self._atomic_write(path, content) - if write_result.exit_code != 0: return WriteResult(error=f"Failed to write file: {write_result.stdout}") - # Bytes written — computed from the exact bytes we just wrote (len - # matches wc -c) instead of spawning a ``wc -c`` subprocess. The - # encode happened up front with surrogateescape — the inverse of the - # decode that produces surrogate content — so content_bytes == what - # rode stdin == what is on disk, and the sha256 below compares like - # with like. - bytes_written = len(content_bytes) + content_verified, verify_error = self._verify_written_hash(path, content_bytes) + if verify_error is not None: + return verify_error - # Post-write content verification (cheap, one shell call): compare - # the on-disk sha256 to the intended content's hash. Production - # mining shows models re-reading files right after writing them to - # confirm persistence (154 verify-reads in a 400k-msg window) — - # an explicit verified flag makes that turn unnecessary, and a - # mismatch is surfaced as a hard error instead of silent corruption - # (mirrors patch_replace's post-write verification). - content_verified: Optional[bool] = None - try: - hash_cmd = f"sha256sum {self._escape_shell_arg(path)} 2>/dev/null" - hash_result = self._exec(hash_cmd) - if hash_result.exit_code == 0 and hash_result.stdout.strip(): - disk_sha = hash_result.stdout.strip().split()[0] - expected_sha = hashlib.sha256(content_bytes).hexdigest() - content_verified = disk_sha == expected_sha - if not content_verified: - return WriteResult( - error=( - f"Post-write verification failed for {path}: on-disk " - "content hash differs from the intended write. The " - "write did not persist correctly — re-read the file " - "and retry." - ) - ) - except Exception: - content_verified = None - - # Post-write lint with delta refinement. lint_result = self._check_lint_delta(path, pre_content=pre_content, post_content=content) - # Semantic diagnostics from the LSP layer — separate channel. - # Only fired when the syntax tier reported clean (no point asking - # an LSP for a file that won't even parse). Pass pre/post - # content so the LSP layer can build a line-shift map and - # remap baseline diagnostics into post-edit coordinates. - # Best-effort: ``""`` is returned for any failure path. + # Semantic (LSP) diagnostics are a separate channel, fired only when the + # syntax tier is clean (no point asking an LSP about a file that won't + # parse). Best-effort: "" on any failure path. lsp_diagnostics: Optional[str] = None if lint_result.success or lint_result.skipped: - block = self._maybe_lsp_diagnostics( + lsp_diagnostics = self._maybe_lsp_diagnostics( path, pre_content=pre_content, post_content=content - ) - if block: - lsp_diagnostics = block + ) or None return WriteResult( - bytes_written=bytes_written, + bytes_written=len(content_bytes), dirs_created=dirs_created, verified=content_verified, lint=lint_result.to_dict() if lint_result else None, lsp_diagnostics=lsp_diagnostics, ) - + # ========================================================================= # PATCH Implementation (Replace Mode) # ========================================================================= - + + def _no_match_result(self, path: str, content: str, old_string: str, + new_string: str, match_count: int, error: Optional[str]) -> PatchResult: + """PatchResult for a failed fuzzy match. + + Already-applied detection first: the most common production patch + failure is a re-send of an edit that already landed (identical + old/new, or old_string gone while new_string is present verbatim). + That becomes a success-shaped no-op so the model moves on instead of + burning turns on re-reads. Otherwise attach a best-effort + "Did you mean?" snippet to the error. + """ + from tools.fuzzy_match import format_no_match_hint, is_already_applied + + if is_already_applied(content, old_string, new_string): + return PatchResult( + success=True, + no_change=True, + note=( + f"File already contains the target text — the edit " + f"appears to be already applied to {path}. No write " + "performed; do not re-send this patch." + ), + ) + err_msg = error or f"Could not find match for old_string in {path}" + try: + err_msg += format_no_match_hint(err_msg, match_count, old_string, content) + except Exception: + pass + return PatchResult(error=err_msg) + + def _verify_patch_persisted(self, path: str, new_content: str) -> Optional[PatchResult]: + """Re-read ``path`` and confirm the intended bytes landed; error result or None. + + Catches silent persistence failures (backend FS oddities, a race with + another task, truncated pipe) that would otherwise return + success-with-diff while the file is unchanged. Line endings are + normalized before comparing: on Windows text-mode ``open()`` writes + ``\\n`` as ``\\r\\n``, so the disk legitimately holds CRLF while + ``new_content`` has LF (POSIX is a no-op). The re-read's leading BOM is + stripped too — write_file restored it on disk but ``new_content`` is + the BOM-less string we matched against. + """ + verify_result = self._exec(f"cat {self._escape_shell_arg(path)} 2>/dev/null") + if verify_result.exit_code != 0: + return PatchResult(error=f"Post-write verification failed: could not re-read {path}") + bomless, _ = _strip_bom(verify_result.stdout) + on_disk = bomless.replace("\r\n", "\n").replace("\r", "\n") + intended = new_content.replace("\r\n", "\n").replace("\r", "\n") + if on_disk != intended: + return PatchResult(error=( + f"Post-write verification failed for {path}: on-disk content " + f"differs from intended write " + f"(wrote {len(intended)} chars, read back " + f"{len(on_disk)} chars after normalizing line endings). " + "The patch did not persist. Re-read the file and try again." + )) + return None + def patch_replace(self, path: str, old_string: str, new_string: str, replace_all: bool = False) -> PatchResult: - """ - Replace text in a file using fuzzy matching. - - Args: - path: File path to modify - old_string: Text to find (must be unique unless replace_all=True) - new_string: Replacement text - replace_all: If True, replace all occurrences - - Returns: - PatchResult with diff and lint results - """ - # Expand ~ and other shell paths + """Replace text in a file using fuzzy matching (``old_string`` must be + unique unless ``replace_all``). Returns a PatchResult with diff + lint.""" path = self._expand_path(path) - # Block writes to sensitive paths denied = get_write_denied_error(path) if denied: return PatchResult(error=denied) - # Read current content - read_cmd = f"cat {self._escape_shell_arg(path)} 2>/dev/null" - read_result = self._exec(read_cmd) - + read_result = self._exec(f"cat {self._escape_shell_arg(path)} 2>/dev/null") if read_result.exit_code != 0: return PatchResult(error=f"Failed to read file: {path}") - - content = read_result.stdout - # Preserve raw content (including BOM) for write_file's pre_content - # so write_file can detect/restore BOM correctly. - raw_content = content - # Strip a leading UTF-8 BOM before matching so the fuzzy matcher and - # the diff operate on clean content (a phantom U+FEFF before line 1 - # defeats an exact first-line match). write_file restores the BOM on - # the way back out by re-probing the on-disk file, so the round-trip - # preserves the marker. - content, _ = _strip_bom(content) - # Import and use fuzzy matching + # Keep the raw read (with BOM) as write_file's pre_content so it can + # detect/restore the BOM; match and diff on BOM-stripped content (a + # phantom U+FEFF before line 1 defeats an exact first-line match). + raw_content = read_result.stdout + content, _ = _strip_bom(raw_content) + from tools.fuzzy_match import fuzzy_find_and_replace - + new_content, match_count, _strategy, error = fuzzy_find_and_replace( content, old_string, new_string, replace_all ) - if error or match_count == 0: - # Already-applied detection: the most common patch failure in - # production is a re-send of an edit that has already landed - # (identical old/new strings, or old_string gone while - # new_string is present verbatim). Surface that as an explicit - # success-shaped no-op so the model moves on instead of - # burning turns on re-reads and re-patches. - from tools.fuzzy_match import is_already_applied - if is_already_applied(content, old_string, new_string): - return PatchResult( - success=True, - no_change=True, - note=( - f"File already contains the target text — the edit " - f"appears to be already applied to {path}. No write " - "performed; do not re-send this patch." - ), - ) - err_msg = error or f"Could not find match for old_string in {path}" - try: - from tools.fuzzy_match import format_no_match_hint - err_msg += format_no_match_hint(err_msg, match_count, old_string, content) - except Exception: - pass - return PatchResult(error=err_msg) + return self._no_match_result(path, content, old_string, new_string, match_count, error) - # ── Line-ending preservation ────────────────────────────────── - # Models nearly always send old_string/new_string with bare LF - # in tool args (JSON-encoded), but the file may have CRLF on - # disk. After fuzzy_find_and_replace, ``new_content`` is a - # mixed-ending string: the substituted region is LF, surrounding - # text keeps the file's CRLF. Normalize the whole thing to the - # file's detected line ending so the on-disk file is consistent - # and the unified diff below reflects the actual change. + # Models send bare-LF old/new strings, so after replacement the + # substituted region is LF while the rest keeps the file's CRLF. + # Normalize to the file's detected ending so the file stays consistent + # and the diff reflects the real change. file_ending = _detect_line_ending(content) if file_ending: new_content = _normalize_line_endings(new_content, file_ending) - # Write back — pass pre_content (original read, with BOM) to avoid - # a redundant cat subprocess inside write_file. Must be the raw - # content (before _strip_bom) so write_file can detect/restore BOM. - write_result = self.write_file(path, new_content, - pre_content=raw_content) + # pre_content must be the RAW read (before _strip_bom) for BOM detection; + # passing it also saves write_file a redundant cat. + write_result = self.write_file(path, new_content, pre_content=raw_content) if write_result.error: return PatchResult(error=f"Failed to write changes: {write_result.error}") - # Post-write verification — re-read the file and confirm the bytes we - # intended to write actually landed. Catches silent persistence - # failures (backend FS oddities, race with another task, truncated - # pipe, etc.) that would otherwise return success-with-diff while the - # file is unchanged on disk. - verify_cmd = f"cat {self._escape_shell_arg(path)} 2>/dev/null" - verify_result = self._exec(verify_cmd) - if verify_result.exit_code != 0: - return PatchResult(error=f"Post-write verification failed: could not re-read {path}") - # Normalize line endings before comparing. On Windows, Python's - # default text-mode ``open()`` translates ``\n`` → ``\r\n`` on - # write, so the file on disk legitimately holds CRLFs while our - # ``new_content`` string has bare LFs. Without this normalization - # every patch on Windows returns a bogus "wrote 39, read 42" - # false-negative even though the edit landed correctly. POSIX - # backends don't translate, so this is a no-op there. We also - # strip a leading BOM from the re-read: write_file restored the - # marker on disk but ``new_content`` is the BOM-less string we - # matched against, so the comparison must drop it to stay - # apples-to-apples. - _verify_bomless, _ = _strip_bom(verify_result.stdout) - _verify_stdout_normalized = _verify_bomless.replace("\r\n", "\n").replace("\r", "\n") - _new_content_normalized = new_content.replace("\r\n", "\n").replace("\r", "\n") - if _verify_stdout_normalized != _new_content_normalized: - return PatchResult(error=( - f"Post-write verification failed for {path}: on-disk content " - f"differs from intended write " - f"(wrote {len(_new_content_normalized)} chars, read back " - f"{len(_verify_stdout_normalized)} chars after normalizing line endings). " - "The patch did not persist. Re-read the file and try again." - )) + verify_error = self._verify_patch_persisted(path, new_content) + if verify_error is not None: + return verify_error - # Generate diff - diff = self._unified_diff(content, new_content, path) - - # Auto-lint with delta refinement: only surface errors introduced - # by this patch, filtering out pre-existing lint failures so the - # agent isn't distracted by problems that were already there. + # Lint delta: only surface errors introduced by this patch. lint_result = self._check_lint_delta(path, pre_content=content, post_content=new_content) return PatchResult( success=True, - diff=diff, + diff=self._unified_diff(content, new_content, path), files_modified=[path], lint=lint_result.to_dict() if lint_result else None, - # Propagate the LSP diagnostics already captured by the - # internal ``write_file`` call. Its baseline was the - # pre-patch content (taken at the start of write_file via - # ``_snapshot_lsp_baseline``) so the delta is correct for - # the patch as a whole. Keep the field separate from the - # syntax-check ``lint`` so the agent can read both signals. + # LSP diagnostics already captured by the internal write_file call; + # its baseline was the pre-patch content, so the delta is correct for + # the whole patch. Kept separate from ``lint`` so both signals are readable. lsp_diagnostics=write_result.lsp_diagnostics, ) - + def patch_v4a(self, patch_content: str) -> PatchResult: - """ - Apply a V4A format patch. - - V4A format: - *** Begin Patch - *** Update File: path/to/file.py - @@ context hint @@ - context line - -removed line - +added line - *** End Patch - - Args: - patch_content: V4A format patch string - - Returns: - PatchResult with changes made - """ - # Import patch parser + """Apply a V4A format patch (``*** Begin Patch`` / ``*** Update File:`` / + ``@@ hint @@`` hunks / ``*** End Patch``).""" from tools.patch_parser import parse_v4a_patch, apply_v4a_operations - + operations, parse_error = parse_v4a_patch(patch_content) if parse_error: return PatchResult(error=f"Failed to parse patch: {parse_error}") - - # Apply operations - result = apply_v4a_operations(operations, self) - return result - + return apply_v4a_operations(operations, self) - - # ========================================================================= # SEARCH Implementation # ========================================================================= diff --git a/tools/file_tools.py b/tools/file_tools.py index c34801f8fd..a76ec2e8af 100644 --- a/tools/file_tools.py +++ b/tools/file_tools.py @@ -660,6 +660,46 @@ def _dedup_stub_or_block(task_data: dict, dedup_key: tuple, path: str) -> str: }, ensure_ascii=False) +def _record_successful_read(task_data: dict, task_id: str, path: str, resolved_str: str, + 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 — it feeds both + 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 after our read) 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). + """ + with _read_tracker_lock: + task_data["dedup_hits"].pop(dedup_key, None) + task_data["read_history"].add((path, offset, limit)) + count = _bump_consecutive(task_data, ("read", path, offset, limit)) + try: + _mtime_now = os.path.getmtime(resolved_str) + task_data["dedup"][dedup_key] = _mtime_now + task_data.setdefault("read_timestamps", {})[resolved_str] = _mtime_now + except OSError: + pass + _cap_read_tracker_data(task_data) + + try: + file_state.record_read(task_id, resolved_str, partial=partial) + except Exception: + logger.debug("file_state.record_read failed", exc_info=True) + + if not partial: + try: + from tools.skill_manager_tool import mark_background_review_skill_read + + mark_background_review_skill_read(Path(resolved_str)) + except Exception: + logger.debug("background-review read-mark failed", exc_info=True) + return count + + def read_file_tool(path: str, offset: int = 1, limit: int = 2000, task_id: str = "default") -> str: """Read a file with pagination and line numbers. @@ -706,9 +746,11 @@ def read_file_tool(path: str, offset: int = 1, limit: int = 2000, task_id: str = "Use vision_analyze for images, or terminal to inspect binary files." ) - # Pass the already-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: 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. block_error = get_read_block_error(str(_resolved)) if block_error: return tool_error(block_error) @@ -737,14 +779,18 @@ 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, exactly - # as before the cache existed. + # error results still flow through the tracking block below + # (consecutive-loop detection, dedup bookkeeping) exactly as before the + # cache existed — serving from the cache is the optimization, recording + # must stay side-effect-identical. _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. file_size = result_dict.get("file_size", 0) max_chars = _get_max_read_chars() if len(result.content or "") > max_chars: @@ -766,43 +812,8 @@ def read_file_tool(path: str, offset: int = 1, limit: int = 2000, task_id: str = "to keep context usage efficient." )) - read_key = ("read", path, offset, limit) - with _read_tracker_lock: - # A real read succeeded, so this key is no longer in a stub-loop. - task_data["dedup_hits"].pop(dedup_key, None) - task_data["read_history"].add((path, offset, limit)) - count = _bump_consecutive(task_data, read_key) - # mtime at read time feeds dedup AND the write/patch staleness warning. - try: - _mtime_now = os.path.getmtime(resolved_str) - task_data["dedup"][dedup_key] = _mtime_now - task_data.setdefault("read_timestamps", {})[resolved_str] = _mtime_now - except OSError: - pass - _cap_read_tracker_data(task_data) - - # Cross-agent registry (separate from the per-task tracker): lets - # write/patch detect sibling-subagent writes after our read. Partial - # when offset>1 or truncated. Outside our lock so the registry's own - # locking isn't nested under it. - _partial = (offset > 1) or bool(result_dict.get("truncated")) - try: - file_state.record_read(task_id, resolved_str, partial=_partial) - except Exception: - logger.debug("file_state.record_read failed", exc_info=True) - - # Background-review read-before-write guard: 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. - if not _partial: - try: - from tools.skill_manager_tool import mark_background_review_skill_read - - mark_background_review_skill_read(Path(resolved_str)) - except Exception: - logger.debug( - "background-review read-mark failed", exc_info=True - ) + count = _record_successful_read(task_data, task_id, path, resolved_str, offset, limit, + dedup_key, partial=(offset > 1) or bool(result_dict.get("truncated"))) if count >= 4: return tool_error( @@ -824,6 +835,14 @@ def read_file_tool(path: str, offset: int = 1, limit: int = 2000, task_id: str = return tool_error(str(e)) +def _resolve_or_none(filepath: str, task_id: str) -> str | None: + """Task-resolved path string, or None when resolution fails for any reason.""" + try: + return str(_resolve_path_for_task(filepath, task_id)) + except Exception: + return 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. @@ -883,10 +902,7 @@ def write_file_tool(path: str, content: str, task_id: str = "default", try: # Resolution failure falls back to the legacy unlocked path (the write # still proceeds; the per-task staleness check still runs). - try: - _resolved = str(_resolve_path_for_task(path, task_id)) - except Exception: - _resolved = None + _resolved = _resolve_or_none(path, task_id) if _resolved is None: stale_warning = _check_file_staleness(path, task_id) @@ -900,10 +916,12 @@ def write_file_tool(path: str, content: str, task_id: str = "default", _update_read_timestamp(path, task_id) return json.dumps(result_dict, ensure_ascii=False) - # Per-path lock serializes read→modify→write across concurrent subagents. + # Per-path lock serializes read→modify→write across concurrent + # subagents; different paths stay fully parallel. with file_state.lock_path(_resolved): # Warning priority: cross-agent (names the sibling subagent) > - # per-task staleness > workspace divergence. + # per-task staleness > workspace divergence (relative path resolving + # outside the terminal's cwd — the worktree-cwd bug). cross_warning = file_state.check_stale(task_id, _resolved) stale_warning = _check_file_staleness(path, task_id) cwd_warning = _path_resolution_warning(path, Path(_resolved), task_id) @@ -913,11 +931,14 @@ def write_file_tool(path: str, content: str, task_id: str = "default", effective_warning = cross_warning or stale_warning or cwd_warning if effective_warning: result_dict["_warning"] = effective_warning - # Always report the ABSOLUTE path written so a wrong-cwd mismatch is visible. + # Always report the ABSOLUTE path written so a wrong-cwd mismatch is + # visible in the response instead of silently landing elsewhere. result_dict["resolved_path"] = _resolved if not result_dict.get("error"): result_dict["files_modified"] = [_resolved] _mark_verification_stale(task_id, [_resolved], session_id=session_id) + # Refresh stamps after the write so consecutive edits by the same + # task don't trigger false staleness warnings. _update_read_timestamp(path, task_id) if not result_dict.get("error"): file_state.note_write(task_id, _resolved) @@ -972,13 +993,6 @@ def _collect_v4a_header_paths(patch: str) -> tuple[list[str], list[str]] | str: return paths, content_paths -def _resolve_or_none(filepath: str, task_id: str) -> str | None: - try: - return str(_resolve_path_for_task(filepath, task_id)) - except Exception: - return None - - def patch_tool(mode: str = "replace", path: str = None, old_string: str = None, new_string: str = None, replace_all: bool = False, patch: str = None, task_id: str = "default", cross_profile: bool = False, @@ -1001,13 +1015,15 @@ def patch_tool(mode: str = "replace", path: str = None, old_string: str = None, return tool_error(precheck_err) try: # Lock paths in sorted, deduplicated order so concurrent callers with - # overlapping multi-file patches can't deadlock. An unresolvable path - # is simply not locked. + # overlapping multi-file patches can't deadlock (every caller locks in + # the same order). An unresolvable path is simply not locked. _path_to_resolved: dict[str, str] = { _p: _resolve_or_none(_p, task_id) for _p in _paths_to_check } _resolved_paths = sorted({_r for _r in _path_to_resolved.values() if _r}) + # ExitStack: one lock per path; degenerates to a single lock (or none + # for an unresolvable path) without special-casing. from contextlib import ExitStack with ExitStack() as _locks: for _r in _resolved_paths: @@ -1049,6 +1065,8 @@ def patch_tool(mode: str = "replace", path: str = None, old_string: str = None, result_dict = result.to_dict() if stale_warnings: result_dict["_warning"] = stale_warnings[0] if len(stale_warnings) == 1 else " | ".join(stale_warnings) + # Report the ABSOLUTE path(s) actually patched so a wrong-cwd + # mismatch is visible instead of silently landing elsewhere. _resolved_modified = [ _path_to_resolved.get(_p) or _p for _p in _paths_to_check ] @@ -1057,6 +1075,9 @@ def patch_tool(mode: str = "replace", path: str = None, old_string: str = None, if len(_resolved_modified) == 1: result_dict["resolved_path"] = _resolved_modified[0] _mark_verification_stale(task_id, _resolved_modified, session_id=session_id) + # Refresh stamps for every patched path (no false staleness on + # the next edit) and clear failure counters so a future miss + # starts a fresh count. for _p in _paths_to_check: _update_read_timestamp(_p, task_id) _r = _path_to_resolved.get(_p) @@ -1068,7 +1089,8 @@ def patch_tool(mode: str = "replace", path: str = None, old_string: str = None, # 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. + # richer "Did you mean?" snippet. The escalating hint after 3 failures + # exists so the model recognises the loop and changes approach. if result_dict.get("error") and "Could not find" in str(result_dict["error"]): failure_count = 0 if mode == "replace" and path: @@ -1137,8 +1159,8 @@ def search_tool(pattern: str, target: str = "content", path: str = ".", if block_error: return tool_error(block_error) - # A missing search root costs two shells (search + parent listing); - # cache the miss. + # A missing search root costs two shells (search + parent listing for + # "Similar paths"); cache the miss so a retry skips both. try: resolved_search_path = str(_resolve_path_for_task(path, task_id)) except (OSError, ValueError):