From e86180b446e37a37699bec3156a0146341a0684c Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 22:19:23 -0700 Subject: [PATCH] =?UTF-8?q?refactor(tools):=20compact=20file=5Foperations?= =?UTF-8?q?=20tiers=20=E2=80=94=20shared=20cat/head/python-snippet=20helpe?= =?UTF-8?q?rs,=20one=20search=20pipeline=20runner,=20drop=20dead=20re-expo?= =?UTF-8?q?rts?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- tools/file_operations.py | 907 ++++++++++---------------------- tools/file_operations_common.py | 42 +- tools/file_operations_lint.py | 109 ++-- tools/file_operations_search.py | 266 ++++------ tools/hook_output_spill.py | 108 +--- 5 files changed, 446 insertions(+), 986 deletions(-) diff --git a/tools/file_operations.py b/tools/file_operations.py index 26b67bcab7..e733c7c091 100644 --- a/tools/file_operations.py +++ b/tools/file_operations.py @@ -1,17 +1,11 @@ #!/usr/bin/env python3 """File operations (read, write, patch, search) over any terminal backend. -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). - - file_ops = ShellFileOperations(terminal_env) - 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") +Every operation is a shell command run through the backend's ``execute()``, so one +implementation serves local, docker, ssh, singularity, modal, daytona and +vercel_sandbox. Companions (names re-exported here): ``file_operations_common`` +(result dataclasses, line-ending/BOM helpers), ``file_operations_lint`` +(LintMixin), ``file_operations_search`` (SearchMixin). """ import base64 @@ -24,80 +18,25 @@ import hashlib import json import unicodedata from abc import ABC, abstractmethod - from typing import Optional, Dict from pathlib import Path + from tools.binary_extensions import BINARY_EXTENSIONS - -from agent.file_safety import ( - get_write_denied_error, - is_write_denied as _shared_is_write_denied, -) +from agent.file_safety import get_write_denied_error, is_write_denied as _is_write_denied # noqa: F401 from tools.file_operations_common import ( # noqa: F401 (re-exported) - DEFAULT_READ_LIMIT, - DEFAULT_READ_OFFSET, - DEFAULT_SEARCH_LIMIT, - DEFAULT_SEARCH_OFFSET, - ExecuteResult, - LintResult, - PatchResult, - ReadResult, - SearchMatch, - SearchResult, - WriteResult, - _FENCE_MARKER_RE, - _OSC_SEQUENCE_RE, - _UTF8_BOM, - _coerce_int, - _detect_line_ending, - _has_bom, - _normalize_line_endings, - _strip_bom, - _strip_terminal_fence_leaks, - normalize_read_pagination, - normalize_search_pagination, -) -from tools.file_operations_lint import ( # noqa: F401 (re-exported) - LINTERS, - LINTERS_INPROC, - LintMixin, - _FAIL_CLOSED_INPROC_EXTS, - _LINTER_UNUSABLE_PATTERNS, - _SHELL_LINTER_LSP_REDUNDANT, - _lint_json_inproc, - _lint_python_inproc, - _lint_toml_inproc, - _lint_yaml_inproc, - _looks_like_linter_unusable, + ExecuteResult, LintResult, PatchResult, ReadResult, SearchMatch, SearchResult, WriteResult, + _UTF8_BOM, _detect_line_ending, _has_bom, _normalize_line_endings, _strip_bom, + _strip_terminal_fence_leaks, normalize_read_pagination, normalize_search_pagination, ) +from tools.file_operations_lint import LINTERS_INPROC, LintMixin, _FAIL_CLOSED_INPROC_EXTS from tools.file_operations_search import ( # noqa: F401 (re-exported) - SearchMixin, - _MACOS_TCC_PROTECTED_HOME_DIRS, - _REGEX_NEWLINE_ESCAPE_RE, - _SEARCH_OUTPUT_RE, - _SEARCH_TIMEOUT_MARKER_RE, - _is_line_oriented_newline_error, - _macos_protected_search_exclusions, - _maybe_warn_line_oriented_newline_pattern, - _parse_search_context_line, - _pattern_has_regex_newline, - _search_stdout_and_limit, - _split_tool_diagnostics, + SearchMixin, _macos_protected_search_exclusions, _parse_search_context_line, + _pattern_has_regex_newline, _search_stdout_and_limit, _split_tool_diagnostics, ) - -# --------------------------------------------------------------------------- -# Write-path deny list — blocks writes to sensitive system/credential files -# --------------------------------------------------------------------------- - +# Controller home; SearchMixin reads it (tests monkeypatch it here). _HOME = str(Path.home()) - -def _is_write_denied(path: str) -> bool: - """Return True if path is on the write deny list.""" - return _shared_is_write_denied(path) - - # ============================================================================= # Binary-content identification # ============================================================================= @@ -133,14 +72,11 @@ _MAGIC_SIGNATURES: tuple = ( def identify_binary_bytes(sample: bytes) -> str: - """Best-effort human name for binary content from its magic bytes. + """Best-effort human name for binary content from its magic bytes; never raises. - Returns e.g. ``"PNG image data"`` or ``"unknown binary"``. Never raises. - The ISO-media entry additionally checks for ``ftyp`` at offset 4, since - the leading size field alone (three NULs) is too weak a signature. + The ISO-media entry additionally requires ``ftyp`` at offset 4 — three + leading NULs alone are too weak a signature. """ - if not sample: - return "unknown binary" for prefix, name in _MAGIC_SIGNATURES: if sample.startswith(prefix): if name.startswith("ISO media") and sample[4:8] != b"ftyp": @@ -150,12 +86,8 @@ def identify_binary_bytes(sample: bytes) -> str: def describe_binary_file(sample: Optional[bytes], file_size: int) -> str: - """One-line answer for the binary-file refusal. - - Naming the dead end: "Binary file" alone sends the model hunting for - 'appropriate tools' that may not exist in its toolset. Naming the TYPE - ("PNG image data, 4.1 KB") answers what-is-this in a single read. - """ + """One-line binary-file refusal naming the TYPE ("PNG image data, 4.1 KB"), so the + model gets what-is-this in one read instead of hunting for tools it may lack.""" kind = identify_binary_bytes(sample or b"") if file_size >= 1024 * 1024: size = f"{file_size / (1024 * 1024):.1f} MB" @@ -168,69 +100,41 @@ def describe_binary_file(sample: Optional[bytes], file_size: int) -> str: class FileOperations(ABC): """Abstract interface for file operations across terminal backends.""" - + @abstractmethod def read_file(self, path: str, offset: int = 1, limit: int = 2000) -> ReadResult: """Read a file with pagination support.""" - ... @abstractmethod 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. - Returns ReadResult with .content = full file text, .error set on - failure. Always reads to EOF regardless of file size. - """ - ... - - def read_file_bytes(self, path: str, max_bytes: Optional[int] = None) -> ReadResult: - """Read complete binary content as base64 across the backend boundary.""" - return ReadResult(error="Binary reads are not implemented for this backend") + """Whole file as a plain string: no pagination, line numbers or clamping.""" @abstractmethod - def write_file(self, path: str, content: str, - pre_content: Optional[str] = None) -> WriteResult: + def write_file(self, path: str, content: str, pre_content: Optional[str] = None) -> WriteResult: """Write content to a file, creating directories as needed.""" - ... @abstractmethod 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.""" - ... @abstractmethod def patch_v4a(self, patch_content: str) -> PatchResult: """Apply a V4A format patch.""" - ... @abstractmethod def delete_file(self, path: str) -> WriteResult: """Delete a file. Returns WriteResult with .error set on failure.""" - ... - - def delete_path(self, path: str, recursive: bool = False) -> WriteResult: - """Cross-platform delete that handles files and (with recursive=True) - directory trees. Default implementation delegates to ``delete_file`` - for the non-recursive case; backends with native recursive support - should override. - """ - if recursive: - return WriteResult(error="Recursive delete not implemented for this backend") - return self.delete_file(path) @abstractmethod def move_file(self, src: str, dst: str) -> WriteResult: - """Move/rename a file from src to dst. Returns WriteResult with .error set on failure.""" - ... + """Move/rename a file. Returns WriteResult with .error set on failure.""" @abstractmethod def search(self, pattern: str, path: str = ".", target: str = "content", file_glob: Optional[str] = None, limit: int = 50, offset: int = 0, output_mode: str = "content", context: int = 0) -> SearchResult: """Search for content or files.""" - ... # ============================================================================= @@ -249,11 +153,10 @@ 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.) + cwd rule: every ``_exec`` prefers the LIVE ``env.cwd`` so a ``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 it for every call once made + patches "succeed" with a plausible diff while landing in the wrong directory). """ def __init__(self, terminal_env, cwd: str = None): @@ -273,19 +176,15 @@ class ShellFileOperations(LintMixin, SearchMixin, FileOperations): kwargs['timeout'] = timeout if stdin_data is not None: kwargs['stdin_data'] = stdin_data - 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 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). + # child never received the input (defense 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 - ) + 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).""" @@ -293,21 +192,32 @@ class ShellFileOperations(LintMixin, SearchMixin, FileOperations): result = self._exec(f"command -v {cmd} >/dev/null 2>&1 && echo 'yes'") self._command_cache[cmd] = result.stdout.strip() == 'yes' return self._command_cache[cmd] - + + def _cat(self, path: str) -> ExecuteResult: + """``cat`` the file with stderr silenced (missing file → non-zero exit).""" + return self._exec(f"cat {self._escape_shell_arg(path)} 2>/dev/null") + + def _head(self, path: str, nbytes: int) -> ExecuteResult: + return self._exec(f"head -c {nbytes} {self._escape_shell_arg(path)} 2>/dev/null") + + def _run_python_snippet(self, snippet: str) -> ExecuteResult: + """Run ``snippet`` via the backend's ``python3``, retrying with ``python`` + when only that name exists (Windows / older systems).""" + result = self._exec(f"python3 -c {self._escape_shell_arg(snippet)}") + if result.exit_code != 0 and "python3" in (result.stdout or ""): + result = self._exec(f"python -c {self._escape_shell_arg(snippet)}") + return result + def _sample_file_bytes(self, path: str, length: int = 1000): - """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" - ) + """First ``length`` raw bytes, 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``). None when no clean base64 came back (no ``base64`` binary); + callers then fall back to the text heuristic.""" + result = self._exec(f"head -c {length} {self._escape_shell_arg(path)} 2>/dev/null | base64") if result.exit_code != 0: return None - encoded = _strip_terminal_fence_leaks(result.stdout) - encoded = "".join(encoded.split()) + encoded = "".join(_strip_terminal_fence_leaks(result.stdout).split()) if not encoded: return b"" if not re.fullmatch(r"[A-Za-z0-9+/]+={0,2}", encoded): @@ -319,15 +229,14 @@ class ShellFileOperations(LintMixin, SearchMixin, FileOperations): @staticmethod def _is_likely_binary_bytes(sample: bytes) -> bool: - """Byte-layer binary detection. + """Byte-layer binary detection: text iff valid UTF-8, allowing one incomplete + multibyte sequence at the very end (an artifact of the byte-boundary cut). - 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). + 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 @@ -349,18 +258,16 @@ class ShellFileOperations(LintMixin, SearchMixin, FileOperations): def _is_likely_binary(self, path: str, content_sample: str = None) -> bool: """Legacy text-layer binary check: extension, else >30% non-printable chars.""" - ext = os.path.splitext(path)[1].lower() - if ext in BINARY_EXTENSIONS: + if os.path.splitext(path)[1].lower() in BINARY_EXTENSIONS: return True if content_sample: # 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. + # them. Treat 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') + 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 @@ -370,125 +277,84 @@ class ShellFileOperations(LintMixin, SearchMixin, FileOperations): 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). + Compact, not fixed-width: padding cost ~16% more tokens per line for no + accuracy gain in A/B, while dropping numbers regressed line-referencing. """ from tools.tool_output_limits import get_max_line_length max_line_length = get_max_line_length() - lines = content.split('\n') numbered = [] - for i, line in enumerate(lines, start=start_line): - # Truncate long lines + for i, line in enumerate(content.split('\n'), start=start_line): if len(line) > max_line_length: line = line[:max_line_length] + "... [truncated]" numbered.append(f"{i}|{line}") return '\n'.join(numbered) - + def _expand_path(self, path: str) -> str: """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: + if not path or not path.startswith('~'): return path - - if path.startswith('~'): - result = self._exec("echo $HOME") - if result.exit_code == 0 and result.stdout.strip(): - home = result.stdout.strip() - if path == '~': - return home - elif path.startswith('~/'): - return home + path[1:] # Replace ~ with home - # ~username format - extract and validate username before - # letting shell expand it (prevent shell injection via - # paths like "~; rm -rf /"). - rest = path[1:] # strip leading ~ - slash_idx = rest.find('/') - username = rest[:slash_idx] if slash_idx >= 0 else rest - if username and re.fullmatch(r'[a-zA-Z0-9._-]+', username): - # Only expand ~username (not the full path) to avoid shell - # injection via path suffixes like "~user/$(malicious)". - expand_result = self._exec(f"echo ~{username}") - if expand_result.exit_code == 0 and expand_result.stdout.strip(): - user_home = expand_result.stdout.strip() - suffix = path[1 + len(username):] # e.g. "/rest/of/path" - return user_home + suffix - + result = self._exec("echo $HOME") + if result.exit_code == 0 and result.stdout.strip(): + home = result.stdout.strip() + if path == '~': + return home + if path.startswith('~/'): + return home + path[1:] + # ~username: validate the username before letting the shell expand + # it, and expand ONLY that token, so neither "~; rm -rf /" nor + # "~user/$(malicious)" reaches the shell. + rest = path[1:] + slash_idx = rest.find('/') + username = rest[:slash_idx] if slash_idx >= 0 else rest + if username and re.fullmatch(r'[a-zA-Z0-9._-]+', username): + expand_result = self._exec(f"echo ~{username}") + if expand_result.exit_code == 0 and expand_result.stdout.strip(): + return expand_result.stdout.strip() + path[1 + len(username):] return path - + def _escape_shell_arg(self, arg: str) -> str: - """Escape a string for safe use in shell commands. - - On Windows native drive paths (``C:\\Users\\x`` / ``C:/Users/x``) - and mixed MSYS leftovers (``/c/Users\\x``) are rewritten to the - Git Bash ``/c/Users/x`` form via ``_bash_safe_path``: bash eats - backslashes and MSYS otherwise mangles drive paths into the - ``Directory \\drivers\\etc does not exist`` failure class. Reuses - the env-layer translator so shell file ops and the terminal ``cd`` - agree on the path form. No-op off Windows and for plain POSIX paths. - """ + """Single-quote ``arg`` for the shell. On Windows, native drive paths and + mixed MSYS leftovers are first rewritten to the Git Bash ``/c/Users/x`` + form via the env-layer ``_bash_safe_path`` (bash eats backslashes; MSYS + mangles drive paths), so shell file ops and the terminal ``cd`` agree.""" from tools.environments.local import _bash_safe_path - - arg = _bash_safe_path(arg) - # Use single quotes and escape any single quotes in the string - return "'" + arg.replace("'", "'\"'\"'") + "'" + return "'" + _bash_safe_path(arg).replace("'", "'\"'\"'") + "'" def _escape_native_tool_arg(self, arg: str) -> str: - """Escape a path argument destined for a NATIVE Windows binary. + """Quote a path for a NATIVE Windows binary (rg, node, git ...). - ``_escape_shell_arg`` rewrites Windows paths to the Git Bash MSYS - form (``/c/Users/x``) so bash builtins resolve them. But native - Windows binaries invoked from that bash (ripgrep installed via - winget/cargo/choco, native git, etc.) do not understand ``/c/...`` - paths — and Hermes disables MSYS argument conversion for its bash - subprocesses (``MSYS_NO_PATHCONV=1`` / ``MSYS2_ARG_CONV_EXCL=*``, - see ``_apply_windows_msys_bash_env_defaults``), so nothing ever - translates the MSYS form back. The native tool then fails with - ``The system cannot find the path specified. (os error 3)``. - - The forward-slash native form (``C:/Users/x``) is the one spelling - every layer accepts: bash passes it through untouched (it is not an - absolute POSIX path, so no conversion applies even without the - opt-outs), and Windows APIs treat ``/`` and ``\\`` as equivalent - separators. MSYS builds of the same tools accept it too, so this is - safe regardless of which flavor of the binary is installed. - - On non-Windows hosts this is exactly ``_escape_shell_arg``. + Those don't understand the MSYS ``/c/...`` form ``_escape_shell_arg`` + produces, and Hermes disables MSYS argument conversion for its bash + subprocesses, so nothing translates it back (→ ``os error 3``). The + forward-slash native form ``C:/Users/x`` is accepted by every layer: + bash passes it untouched and Windows treats ``/`` as a separator. + Identical to ``_escape_shell_arg`` off Windows. """ from tools.environments.local import _IS_WINDOWS, _msys_to_windows_path - if _IS_WINDOWS and arg: arg = _msys_to_windows_path(arg).replace("\\", "/") return "'" + arg.replace("'", "'\"'\"'") + "'" def _atomic_write(self, path: str, content: str) -> "ExecuteResult": - """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. + """Write ``content`` atomically: stdin → temp file in the SAME directory → + ``mv -f`` over the target (same-FS rename; cross-device ``mv`` degrades to + copy+unlink and is NOT atomic). ``mkdir -p`` is folded in. Exit 0 means the + swap happened; non-zero means nothing was renamed and the original 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. + Script notes: symlink targets are resolved first (replacing the link with a + plain file would orphan the target) and the temp dir recomputed from the + RESOLVED target. ``mktemp -p`` with a hidden marked template, PID-stamped + fallback without mktemp. Existing target: mode copied via ``stat`` (GNU + ``-c%a`` / BSD ``-f%Lp``) + ``chmod`` — ``chmod --reference`` is GNU-only. + New target: ``chmod "=rw"`` AFTER cat gives umask-default perms instead of + mktemp's 0600; deliberately not ``$(umask)`` arithmetic (zsh parses + leading-zero constants as decimal) and quoted so zsh doesn't =word-expand. + ``trap ... EXIT`` removes the temp on every failure path. """ q_path = self._escape_shell_arg(path) - parent = os.path.dirname(path) or "." - q_parent = self._escape_shell_arg(parent) + q_parent = self._escape_shell_arg(os.path.dirname(path) or ".") tmpl = self._escape_shell_arg(".hermes-tmp.XXXXXX") - script = ( "set -e; " f"d={q_parent}; t={q_path}; " @@ -514,25 +380,21 @@ 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]: - """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).""" + """Dominant on-disk line ending, or None when undeterminable (new/empty/ + single-line file). Uses ``pre_content`` when given, else a 4KB ``head``.""" if pre_content: return _detect_line_ending(pre_content) - head_result = self._exec(f"head -c 4096 {self._escape_shell_arg(path)} 2>/dev/null") + head_result = self._head(path, 4096) 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 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) + """Whether the on-disk file 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 → False.""" + head_result = self._head(path, 3) + return head_result.exit_code == 0 and _has_bom(head_result.stdout) def _unified_diff(self, old_content: str, new_content: str, filename: str) -> str: return ''.join(difflib.unified_diff( @@ -543,38 +405,30 @@ class ShellFileOperations(LintMixin, SearchMixin, FileOperations): # ========================================================================= # READ Implementation # ========================================================================= - - def _size_probe_cmd(self, path: str) -> str: - """Byte size of a REGULAR file without opening one that never ends. - - ``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 ( - f"if [ -f {arg} ]; then wc -c < {arg} 2>/dev/null; " - f"elif [ -e {arg} ]; then echo {NOT_REGULAR_SENTINEL}; " - f"else exit 1; fi" - ) @staticmethod def _not_regular_error(path: str) -> ReadResult: """Error for a path that exists but would block if read.""" - return ReadResult( - error=( - f"Cannot read '{path}': not a regular file (directory, FIFO, " - "socket, or device). Reading it could block indefinitely." - ) - ) + return ReadResult(error=( + f"Cannot read '{path}': not a regular file (directory, FIFO, " + "socket, or device). Reading it could block indefinitely." + )) 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)) + """Byte size of a REGULAR file: ``(file_size, status)`` with status ``"ok"``, + ``"missing"``, ``"not_regular"`` or ``"bad_size"`` (unparseable ``wc``). + + ``wc -c <`` on a writer-less FIFO, socket or /dev/zero blocks forever (read + helpers pass no timeout) and file_tools' name-based device blocklist 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. + """ + arg = self._escape_shell_arg(path) + stat_result = self._exec( + f"if [ -f {arg} ]; then wc -c < {arg} 2>/dev/null; " + f"elif [ -e {arg} ]; then echo {NOT_REGULAR_SENTINEL}; " + f"else exit 1; fi" + ) if stat_result.exit_code != 0: return 0, "missing" stat_output = _strip_terminal_fence_leaks(stat_result.stdout).strip() @@ -592,8 +446,7 @@ class ShellFileOperations(LintMixin, SearchMixin, FileOperations): 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) + sample_output = _strip_terminal_fence_leaks(self._head(path, 1000).stdout) return self._is_likely_binary(path, sample_output), None # UTF-16 rescue: trust a BOM first, then zero-byte PARITY (not density, so @@ -609,12 +462,8 @@ class ShellFileOperations(LintMixin, SearchMixin, FileOperations): """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: + if os.path.splitext(path)[1].lower() in BINARY_EXTENSIONS or file_size > self._UTF16_MAX_BYTES: return None - if file_size > self._UTF16_MAX_BYTES: - return None - snippet = ( "import sys, json, os\n" f"p = {path!r}\n" @@ -657,11 +506,7 @@ class ShellFileOperations(LintMixin, SearchMixin, FileOperations): "except Exception:\n" " print('HERMES_UTF16:NO'); sys.exit(0)\n" ) - - result = self._exec(f"python3 -c {self._escape_shell_arg(snippet)}") - if result.exit_code != 0 and "python3" in (result.stdout or ""): - result = self._exec(f"python -c {self._escape_shell_arg(snippet)}") - + result = self._run_python_snippet(snippet) stdout = _strip_terminal_fence_leaks(result.stdout or "") marker = stdout.find("HERMES_UTF16:OK") if result.exit_code != 0 or marker < 0: @@ -674,7 +519,6 @@ class ShellFileOperations(LintMixin, SearchMixin, FileOperations): encoding = str(data.get("encoding", "utf-16")) except (ValueError, KeyError, TypeError): return None - end_line = offset + limit - 1 truncated = total_lines > end_line hint_parts = [f"Transcoded from {encoding.upper()} to UTF-8 for display. " @@ -685,11 +529,8 @@ class ShellFileOperations(LintMixin, SearchMixin, FileOperations): f"(showing {offset}-{end_line} of {total_lines} lines)" ) return ReadResult( - content=self._add_line_numbers(content, offset), - total_lines=total_lines, - file_size=file_size, - truncated=truncated, - hint=" ".join(hint_parts), + content=self._add_line_numbers(content, offset), total_lines=total_lines, + file_size=file_size, truncated=truncated, hint=" ".join(hint_parts), ) def read_file(self, path: str, offset: int = 1, limit: int = 2000) -> ReadResult: @@ -699,13 +540,11 @@ class ShellFileOperations(LintMixin, SearchMixin, FileOperations): """ 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). + # Unicode-equivalent spellings (NFC/NFD, narrow no-break space, curly + # quotes) render identically, so the model retyping a visually-correct + # path can never discover the byte mismatch — retrying is the tool's job. variant = self._unicode_variant_match(path) if variant is not None: result = self.read_file(variant, offset=offset, limit=limit) @@ -720,65 +559,47 @@ class ShellFileOperations(LintMixin, SearchMixin, FileOperations): return self._suggest_similar_files(path) if status == "not_regular": return self._not_regular_error(path) - - # Images are never inlined — redirect to the vision tool - if self._is_image(path): + if self._is_image(path): # never inlined — redirect to the vision tool return ReadResult( - is_image=True, - is_binary=True, - file_size=file_size, + is_image=True, is_binary=True, file_size=file_size, hint=( "Image file detected. Automatically redirected to vision_analyze tool. " "Use vision_analyze with this file path to inspect the image contents." ), ) - is_binary, sample_bytes = self._detect_binary(path) if is_binary: - # 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. + # A UTF-16 text file (Notepad .txt, PowerShell `>`) arrives mangled with + # U+FFFD and trips the binary guard; probe the raw bytes and transcode. utf16_result = self._try_read_utf16(path, offset, limit, file_size) if utf16_result is not None: return utf16_result return ReadResult( - is_binary=True, - file_size=file_size, + is_binary=True, 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 (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: ``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. + # Clamp each line to a byte budget IN THE SHELL so a pathological single- + # line file (one 400MB minified line) never crosses the exec transport; + # _add_line_numbers still clamps in chars afterwards. Why 4*max+1 BYTES: + # ``cut -b`` is byte-based and can split a multibyte codepoint (→ U+FFFD); + # a clamp of max+1 bytes would yield far fewer CHARS than max for + # multibyte text, so the Python clamp would never fire and truncation + # would be silent. UTF-8 codepoints are ≤4 bytes, so 4*max+1 guarantees + # every over-long line still exceeds max chars and trips the Python clamp, + # which also drops any boundary-split U+FFFD (it lands beyond char max). from tools.tool_output_limits import get_max_line_length line_clamp_bytes = 4 * get_max_line_length() + 1 end_line = offset + limit - 1 - read_cmd = ( + read_result = self._exec( f"sed -n '{offset},{end_line}p' {self._escape_shell_arg(path)}" 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. - # Only the first chunk can carry it (the marker lives at byte 0). + # Only the first chunk can carry a BOM (byte 0); strip it so the model + # never sees a phantom U+FEFF. if offset == 1: read_output, _ = _strip_bom(read_output) @@ -787,88 +608,69 @@ class ShellFileOperations(LintMixin, SearchMixin, FileOperations): total_lines = int(_strip_terminal_fence_leaks(wc_result.stdout).strip()) except ValueError: total_lines = 0 - 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 EOF; probe the last byte. if not truncated and read_output.endswith('\n'): 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": + if tail_result.exit_code == 0 and _strip_terminal_fence_leaks(tail_result.stdout).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. + # from inside the model, from a broken tool. Name the dead end and its recovery. if file_size == 0: - return ReadResult( - content="", - total_lines=0, - file_size=0, - hint="File is empty (0 bytes).", - ) + return ReadResult(content="", total_lines=0, file_size=0, hint="File is empty (0 bytes).") if offset > total_lines > 0: return ReadResult( - content="", - total_lines=total_lines, - file_size=file_size, + content="", total_lines=total_lines, file_size=file_size, hint=( f"Note: offset {offset} is beyond the end of the file " f"({total_lines} lines total). Retry with offset <= " f"{total_lines}." ), ) - return ReadResult( - content=self._add_line_numbers(read_output, offset), - total_lines=total_lines, - file_size=file_size, - truncated=truncated, - hint=hint + content=self._add_line_numbers(read_output, offset), total_lines=total_lines, + file_size=file_size, truncated=truncated, hint=hint, ) - def _unicode_variant_match(self, path: str) -> Optional[str]: - """On-disk spelling of a file whose name is unicode-equivalent to ``path``. + # Confusable characters seen in real filenames, collapsed after NFC. + _CONFUSABLES = ( + ("\u202f", " "), # narrow no-break space (macOS screenshots) + ("\u00a0", " "), # no-break space + ("\u2019", "'"), # right single quotation mark (Finder) + ("\u2018", "'"), # left single quotation mark + ) - 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. - """ + def _unicode_variant_match(self, path: str) -> Optional[str]: + """On-disk spelling of a file whose name is unicode-equivalent to ``path`` + (NFC/NFD, confusable spaces/quotes). Returns the entry only when EXACTLY one + matches — several candidates = homoglyph collision, guessing would read the + wrong file.""" dir_path = os.path.dirname(path) or "." filename = os.path.basename(path) if not filename: return None def _canon(name: str) -> str: - # NFC first so composed/decomposed collapse together, then the - # confusable space/quote characters seen in real filenames. out = unicodedata.normalize("NFC", name) - for src, dst in ( - ("\u202f", " "), # narrow no-break space - ("\u00a0", " "), # no-break space - ("\u2019", "'"), # right single quotation mark - ("\u2018", "'"), # left single quotation mark - ): + for src, dst in self._CONFUSABLES: out = out.replace(src, dst) return out target = _canon(filename) - ls_cmd = f"ls -1 {self._escape_shell_arg(dir_path)} 2>/dev/null" - ls_result = self._exec(ls_cmd) + ls_result = self._exec(f"ls -1 {self._escape_shell_arg(dir_path)} 2>/dev/null") if ls_result.exit_code != 0 or not ls_result.stdout.strip(): return None candidates = [ - entry - for entry in _strip_terminal_fence_leaks(ls_result.stdout).splitlines() + entry for entry in _strip_terminal_fence_leaks(ls_result.stdout).splitlines() if entry and entry != filename and _canon(entry) == target ] - # 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 @@ -880,9 +682,7 @@ class ShellFileOperations(LintMixin, SearchMixin, FileOperations): basename_no_ext = os.path.splitext(filename)[0].lower() ext = os.path.splitext(filename)[1].lower() lower_name = filename.lower() - 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(): for f in ls_result.stdout.strip().split('\n'): @@ -904,19 +704,15 @@ class ShellFileOperations(LintMixin, SearchMixin, FileOperations): 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): a high sequence ratio - # catches 1-2 edit typos the substring checks miss. + # Near-miss spelling (AGENT.md -> AGENTS.md): 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]) - return ReadResult( - error=f"File not found: {path}", - similar_files=[fp for _, fp in scored[:5]], - ) - + return ReadResult(error=f"File not found: {path}", similar_files=[fp for _, fp in scored[:5]]) + def read_file_raw(self, path: str) -> ReadResult: """Whole file as a plain string (no pagination/line numbers/clamping).""" path = self._expand_path(path) @@ -929,24 +725,18 @@ class ShellFileOperations(LintMixin, SearchMixin, FileOperations): return ReadResult(is_image=True, is_binary=True, file_size=file_size) is_binary, sample_bytes = self._detect_binary(path) if is_binary: - return ReadResult( - is_binary=True, file_size=file_size, - error=describe_binary_file(sample_bytes, file_size), - ) + return ReadResult(is_binary=True, file_size=file_size, error=describe_binary_file(sample_bytes, file_size)) 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 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. + # 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, - file_size=file_size, - ) + return ReadResult(content=raw_content, file_size=file_size) def read_file_bytes(self, path: str, max_bytes: Optional[int] = None) -> ReadResult: - """Read binary-safe bytes from any shell-backed environment.""" + """Read binary-safe bytes (as base64) from any shell-backed environment.""" path = self._expand_path(path) file_size, status = self._probe_regular_file(path) if status == "missing": @@ -960,7 +750,6 @@ class ShellFileOperations(LintMixin, SearchMixin, FileOperations): file_size=file_size, error=f"File is too large ({file_size:,} bytes, limit is {max_bytes:,})", ) - encoded = self._exec(f"base64 < {self._escape_shell_arg(path)}") if encoded.exit_code != 0: return ReadResult(error=f"Failed to read binary file: {encoded.stdout}") @@ -969,34 +758,21 @@ class ShellFileOperations(LintMixin, SearchMixin, FileOperations): base64.b64decode(compact, validate=True) except (ValueError, base64.binascii.Error): return ReadResult(error=f"Backend returned invalid binary data for: {path}") - return ReadResult( - base64_content=compact, - file_size=file_size, - is_binary=True, - ) + return ReadResult(base64_content=compact, file_size=file_size, is_binary=True) def delete_file(self, path: str) -> WriteResult: - """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: - """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``).""" + """Delete a single file (directories rejected) via the backend's ``python -c`` + so one code path works on local/docker/ssh AND Windows shells (no ``rm``).""" path = self._expand_path(path) denied = get_write_denied_error(path, verb="Delete") if denied: return WriteResult(error=denied) - - # 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. + # Not ``unlink(missing_ok=True)``: a 3.7 remote interpreter lacks it. snippet = ( "import shutil, pathlib, sys\n" f"p = pathlib.Path({path!r})\n" - f"recursive = {bool(recursive)!r}\n" + "recursive = False\n" "try:\n" " if p.is_dir() and not p.is_symlink():\n" " if recursive:\n" @@ -1004,24 +780,15 @@ class ShellFileOperations(LintMixin, SearchMixin, FileOperations): " else:\n" " print('is a directory: ' + str(p), file=sys.stderr); sys.exit(2)\n" " else:\n" - # 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" "except Exception as exc:\n" " print(str(exc), file=sys.stderr); sys.exit(1)\n" ) - - result = self._exec(f"python3 -c {self._escape_shell_arg(snippet)}") - - # 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)}") - + result = self._run_python_snippet(snippet) if result.exit_code != 0: return WriteResult(error=f"Failed to delete {path}: {(result.stdout or '').strip() or 'unknown error'}") - return WriteResult() def move_file(self, src: str, dst: str) -> WriteResult: @@ -1031,9 +798,7 @@ class ShellFileOperations(LintMixin, SearchMixin, FileOperations): denied = get_write_denied_error(p, verb="Move") if denied: return WriteResult(error=denied) - result = self._exec( - f"mv {self._escape_shell_arg(src)} {self._escape_shell_arg(dst)}" - ) + result = self._exec(f"mv {self._escape_shell_arg(src)} {self._escape_shell_arg(dst)}") if result.exit_code != 0: return WriteResult(error=f"Failed to move {src} -> {dst}: {result.stdout}") return WriteResult() @@ -1042,40 +807,31 @@ class ShellFileOperations(LintMixin, SearchMixin, FileOperations): # WRITE Implementation # ========================================================================= + # Lone surrogates OUTSIDE the surrogateescape range (U+DC80-U+DCFF round-trips + # through the pipe; anything else can't be encoded at all). _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. - """ + """Refuse content with a lone surrogate BEFORE any subprocess: letting it + reach the pipe spawns a child that hangs or truncates the target via + empty-stdin ``cat``. A regex scan needs no encode.""" m = self._LONE_SURROGATE_RE.search(content) if m: - return WriteResult( - error=( - f"Refusing to write '{path}': content contains a lone " - f"surrogate character ({m.group(0)!r}) that cannot be " - "encoded as UTF-8. The file was NOT created or modified." - ) - ) + return WriteResult(error=( + f"Refusing to write '{path}': content contains a lone " + f"surrogate character ({m.group(0)!r}) that cannot be " + "encoded as UTF-8. The file was NOT created or modified." + )) return None @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). - - 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. + """Fail-closed pre-write gate for ``_FAIL_CLOSED_INPROC_EXTS`` (JSON/YAML/TOML): + a structured-format write that doesn't parse is a corrupt write, so refuse + before any bytes touch disk. ``.py`` keeps its non-blocking lint-delta report. 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. + post-shim would false-positive on a legitimately BOM-marked file. """ linter = LINTERS_INPROC.get(ext) if ext in _FAIL_CLOSED_INPROC_EXTS else None if linter is None: @@ -1083,41 +839,36 @@ class ShellFileOperations(LintMixin, SearchMixin, FileOperations): 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." - ) - ) + 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]: + 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). + Read only for extensions in the UNION of in-process lint and LSP coverage + (skipping binaries/opaque formats keeps the hot path fast). A caller- + supplied ``pre_content`` is reused as-is; a failed ``cat`` 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") + read_result = self._cat(path) 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: + 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 + 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": @@ -1127,57 +878,48 @@ class ShellFileOperations(LintMixin, SearchMixin, FileOperations): return content 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). + """Compare the on-disk sha256 to the intended bytes: ``(verified, error)``. - 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. + An explicit ``verified`` flag saves the model the re-read turn it otherwise + spends confirming persistence; a mismatch is a hard error, never silent + corruption. ``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 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. + def write_file(self, path: str, content: str, pre_content: Optional[str] = None) -> WriteResult: + """Write content 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. + 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: no ARG_MAX limit, never 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. + ``pre_content``: pre-edit content the caller already has (patch_replace read + it); saves a ``cat``. BOM detection always probes disk regardless. """ 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: @@ -1185,56 +927,35 @@ class ShellFileOperations(LintMixin, SearchMixin, FileOperations): 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. + # Best-effort snapshot so the post-write LSP layer returns only + # diagnostics introduced by this edit. self._snapshot_lsp_baseline(path) - # ``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)) - # 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 - # rejection above guarantees this cannot raise; the try/except is - # defense for future callers that bypass it. - try: - content_bytes = content.encode("utf-8", "surrogateescape") - except UnicodeEncodeError as exc: - return WriteResult( - error=( - f"Refusing to write '{path}': content contains a lone " - f"surrogate character ({exc}) that cannot be encoded as " - "UTF-8. The file was NOT created or modified." - ) - ) + # 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 rejection + # above guarantees this cannot raise. + content_bytes = content.encode("utf-8", "surrogateescape") write_result = self._atomic_write(path, content) if write_result.exit_code != 0: return WriteResult(error=f"Failed to write file: {write_result.stdout}") - content_verified, verify_error = self._verify_written_hash(path, content_bytes) if verify_error is not None: return verify_error lint_result = self._check_lint_delta(path, pre_content=pre_content, post_content=content) - # 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: - lsp_diagnostics = self._maybe_lsp_diagnostics( - path, pre_content=pre_content, post_content=content - ) or None - + lsp_diagnostics = self._maybe_lsp_diagnostics(path, pre_content=pre_content, post_content=content) or None return WriteResult( - 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, + 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, ) # ========================================================================= @@ -1245,19 +966,16 @@ class ShellFileOperations(LintMixin, SearchMixin, FileOperations): 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. + 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) — a success-shaped + no-op stops the model 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, + 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 " @@ -1275,15 +993,14 @@ class ShellFileOperations(LintMixin, SearchMixin, FileOperations): """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. + another task, truncated pipe) that would otherwise return success-with- + diff while the file is unchanged. Line endings are normalized before + comparing: Windows text-mode ``open()`` writes ``\\n`` as ``\\r\\n``, so the + disk legitimately holds CRLF while ``new_content`` has LF. 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") + verify_result = self._cat(path) 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) @@ -1304,15 +1021,12 @@ class ShellFileOperations(LintMixin, SearchMixin, FileOperations): """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) - denied = get_write_denied_error(path) if denied: return PatchResult(error=denied) - - read_result = self._exec(f"cat {self._escape_shell_arg(path)} 2>/dev/null") + read_result = self._cat(path) if read_result.exit_code != 0: return PatchResult(error=f"Failed to read file: {path}") - # 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). @@ -1320,42 +1034,30 @@ class ShellFileOperations(LintMixin, SearchMixin, FileOperations): 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: return self._no_match_result(path, content, old_string, new_string, match_count, error) - - # 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. + # Models send bare-LF old/new strings, so the substituted region is LF + # while the rest keeps the file's CRLF; normalize to the file's 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) - - # pre_content must be the RAW read (before _strip_bom) for BOM detection; - # passing it also saves write_file a redundant cat. + # pre_content must be the RAW read (before _strip_bom) for BOM detection. 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}") - verify_error = self._verify_patch_persisted(path, new_content) if verify_error is not None: return verify_error - - # 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=self._unified_diff(content, new_content, path), - files_modified=[path], + success=True, diff=self._unified_diff(content, new_content, path), files_modified=[path], lint=lint_result.to_dict() if lint_result else None, - # 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. + # Captured by the internal write_file call against the pre-patch + # baseline, so the delta is correct for the whole patch. lsp_diagnostics=write_result.lsp_diagnostics, ) @@ -1363,7 +1065,6 @@ class ShellFileOperations(LintMixin, SearchMixin, FileOperations): """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}") @@ -1372,47 +1073,26 @@ class ShellFileOperations(LintMixin, SearchMixin, FileOperations): # ========================================================================= # SEARCH Implementation # ========================================================================= - + def search(self, pattern: str, path: str = ".", target: str = "content", file_glob: Optional[str] = None, limit: int = 50, offset: int = 0, output_mode: str = "content", context: int = 0) -> SearchResult: - """ - Search for content or files. - - Args: - pattern: Regex (for content) or glob pattern (for files) - path: Directory/file to search (default: cwd) - target: "content" (grep) or "files" (glob) - file_glob: File pattern filter for content search (e.g., "*.py") - limit: Max results (default 50) - offset: Skip first N results - output_mode: "content", "files_only", or "count" - context: Lines of context around matches - - Returns: - SearchResult with matches or file list - """ + """Search for content (regex, ``target="content"``) or files (glob, + ``target="files"``). ``output_mode``: "content", "files_only" or "count"; + ``context``: lines of context around matches.""" offset, limit = normalize_search_pagination(offset, limit) - - # Expand ~ and other shell paths path = self._expand_path(path) - - # Validate that the path exists before searching if "not_found" in self._path_exists_probe(path): - # Multi-path recovery: models frequently pass several paths in - # one string ("dir1 dir2 dir3" or comma-separated). Instead of - # failing the whole call, split, search every path that exists, - # merge the results, and report the skipped parts. + # Models frequently pass several paths in one string ("dir1 dir2" or + # comma-separated): search every part that exists and report the rest. multi = self._try_multi_path_search( pattern, path, target, file_glob, limit, offset, output_mode, context ) if multi is not None: return multi return self._path_not_found_result(path) - result = self._dispatch_search(pattern, path, target, file_glob, limit, offset, output_mode, context) - exclusions = self._macos_search_exclusions(path) if exclusions and not result.error: skipped = ", ".join(item.split("/")[-1] for item in exclusions) @@ -1422,12 +1102,3 @@ class ShellFileOperations(LintMixin, SearchMixin, FileOperations): "folder directly when access is intentional." ) return result - - - - - - - - - diff --git a/tools/file_operations_common.py b/tools/file_operations_common.py index 64e200a6d2..0b0fbdaae5 100644 --- a/tools/file_operations_common.py +++ b/tools/file_operations_common.py @@ -111,12 +111,9 @@ class SearchResult: _DENSIFY_MIN_MATCHES: ClassVar[int] = 5 def _densify_matches(self) -> Optional[str]: - """Render content matches as a lossless, path-grouped text block. - - Path printed once, then `` : `` rows. Relies on rg/grep - emitting a file's hits consecutively, so grouping on path change needs - no reordering. Returns None when too few matches to be worth it. - """ + """Lossless path-grouped text block: path once, then `` : `` + rows. Relies on rg/grep emitting a file's hits consecutively. None when + too few matches to be worth it.""" if len(self.matches) < self._DENSIFY_MIN_MATCHES: return None lines: list[str] = [] @@ -142,8 +139,7 @@ class SearchResult: result["matches_text"] = dense else: result["matches"] = [ - {"path": m.path, "line": m.line_number, "content": m.content} - for m in self.matches + {"path": m.path, "line": m.line_number, "content": m.content} for m in self.matches ] if self.files: result["files"] = self.files @@ -198,9 +194,7 @@ def _strip_terminal_fence_leaks(text: str) -> str: cleaned_lines: List[str] = [] for line in text.splitlines(keepends=True): had_terminal_wrapper = "__HERMES_FENCE_" in line or "\x1b]" in line - cleaned = _OSC_SEQUENCE_RE.sub("", line) - cleaned = _FENCE_MARKER_RE.sub("", cleaned) - cleaned = cleaned.replace("\x07", "") + cleaned = _FENCE_MARKER_RE.sub("", _OSC_SEQUENCE_RE.sub("", line)).replace("\x07", "") if had_terminal_wrapper and cleaned.strip("'\r\n\t ") == "": continue cleaned_lines.append(cleaned) @@ -209,15 +203,10 @@ def _strip_terminal_fence_leaks(text: str) -> str: def _detect_line_ending(sample: str) -> Optional[str]: """Dominant line ending of ``sample`` (``\\r\\n`` if any CRLF in the first 4KB, - else ``\\n``), or None for empty/single-line content. - - Used to preserve a file's endings across write_file/patch: the agent's bare-LF - tool args would otherwise silently normalize CRLF files, and patch would - produce mixed endings when only the substituted region changes. - """ - if not sample: - return None - head = sample[:4096] + else ``\\n``), or None for empty/single-line content. Preserves a file's + endings across write_file/patch: bare-LF tool args would otherwise silently + normalize CRLF files, and patch would produce mixed endings.""" + head = sample[:4096] if sample else "" if "\r\n" in head: return "\r\n" if "\n" in head: @@ -239,15 +228,14 @@ def _normalize_line_endings(text: str, target: str) -> str: # UTF-8 BOM (EF BB BF == U+FEFF), prepended by some Windows editors. Stripped on # read so the model never sees a phantom first character (and patch's first-line -# match works), restored on write when the on-disk file had one — mirroring the -# line-ending preservation above. +# match works), restored on write when the on-disk file had one. _UTF8_BOM = "\ufeff" def _strip_bom(text: str) -> tuple[str, bool]: """Return (text-without-leading-BOM, had_bom). Only a leading BOM is stripped; mid-content U+FEFF is legitimate data.""" - if text and text.startswith(_UTF8_BOM): + if _has_bom(text): return text[len(_UTF8_BOM):], True return text, False @@ -281,16 +269,12 @@ def normalize_read_pagination(offset: Any = DEFAULT_READ_OFFSET, like ``0,-1p`` (schemas declare bounds, but not every caller enforces them). The ``limit`` ceiling is ``tool_output.max_lines`` from config.yaml.""" from tools.tool_output_limits import get_max_lines - max_lines = get_max_lines() normalized_offset = max(1, _coerce_int(offset, DEFAULT_READ_OFFSET)) - normalized_limit = _coerce_int(limit, DEFAULT_READ_LIMIT) - normalized_limit = max(1, min(normalized_limit, max_lines)) + normalized_limit = max(1, min(_coerce_int(limit, DEFAULT_READ_LIMIT), get_max_lines())) return normalized_offset, normalized_limit def normalize_search_pagination(offset: Any = DEFAULT_SEARCH_OFFSET, limit: Any = DEFAULT_SEARCH_LIMIT) -> tuple[int, int]: """Return safe search pagination bounds for shell head/tail pipelines.""" - normalized_offset = max(0, _coerce_int(offset, DEFAULT_SEARCH_OFFSET)) - normalized_limit = max(1, _coerce_int(limit, DEFAULT_SEARCH_LIMIT)) - return normalized_offset, normalized_limit + return max(0, _coerce_int(offset, DEFAULT_SEARCH_OFFSET)), max(1, _coerce_int(limit, DEFAULT_SEARCH_LIMIT)) diff --git a/tools/file_operations_lint.py b/tools/file_operations_lint.py index c8da5a4bd5..67578f4a45 100644 --- a/tools/file_operations_lint.py +++ b/tools/file_operations_lint.py @@ -1,8 +1,7 @@ """Syntax-lint and LSP-diagnostics tier for ``tools.file_operations``. -Extracted from ``ShellFileOperations`` as ``LintMixin``; the class inherits it -so every ``self._check_lint(...)`` call resolves unchanged via the MRO. Module -constants are re-imported into ``tools.file_operations`` for back-compat. +``ShellFileOperations`` inherits ``LintMixin``; module constants and in-process +linters are pure and re-imported into ``tools.file_operations`` for back-compat. """ import ast @@ -26,16 +25,14 @@ LINTERS = { # Extensions whose per-file shell linter is structurally weaker than a real LSP # server and floods phantom errors on real projects: single-file ``tsc`` ignores -# tsconfig (no-lib/ES5 → every ES2015+ stdlib name "missing"), ``go vet`` fails -# outside a module, ``rustfmt --check`` is style-only and rejects non-Cargo files. +# tsconfig, ``go vet`` fails outside a module, ``rustfmt --check`` is style-only. # When an LSP server claims the file, ``_check_lint`` skips the shell linter for # these; py_compile / node --check are file-local and correct so always run. _SHELL_LINTER_LSP_REDUNDANT = frozenset({'.ts', '.go', '.rs'}) # Output substrings meaning the linter binary exists but could not actually run -# (tooling gap, not a lint failure). ``_check_lint`` then returns ``skipped`` so -# the write isn't flagged and the LSP tier (which gates on ok/skipped) still runs. -# Matched case-insensitively. +# (tooling gap, not a lint failure) → ``skipped`` so the write isn't flagged and +# the LSP tier (which gates on ok/skipped) still runs. Matched case-insensitively. _LINTER_UNUSABLE_PATTERNS = { 'npx': ( 'this is not the tsc command you are looking for', # tsc not installed locally @@ -77,10 +74,9 @@ def _lint_yaml_inproc(content: str) -> tuple[bool, str]: """In-process YAML syntax check; ``__SKIP__`` when PyYAML is missing. Syntax-only (``yaml.parse``), NOT ``safe_load``: loading rejects valid YAML - that isn't one plain document — multi-doc ``---`` streams (ComposerError) and - app tags like CloudFormation ``!Sub`` / Ansible ``!vault`` (ConstructorError). - This verdict is a fail-closed WRITE gate, so a false positive refuses a - legitimate write; ``parse`` still catches real scanner/parser errors. + that isn't one plain document — multi-doc ``---`` streams and app tags like + CloudFormation ``!Sub`` / Ansible ``!vault``. This verdict is a fail-closed + WRITE gate, so a false positive refuses a legitimate write. """ try: import yaml as _yaml @@ -144,7 +140,6 @@ class LintMixin: """Syntax-check ``path``: in-process linter when one matches the extension (``content`` avoids a re-read), else the shell linter table.""" ext = os.path.splitext(path)[1].lower() - inproc = LINTERS_INPROC.get(ext) if inproc is not None: if content is None: @@ -156,94 +151,60 @@ class LintMixin: if err == "__SKIP__": return LintResult(skipped=True, message=f"No linter available for {ext} (missing dependency)") return LintResult(success=ok, output="" if ok else err) - if ext not in LINTERS: return LintResult(skipped=True, message=f"No linter for {ext} files") - # Single-file tsc can't read the project's tsconfig.json, so for project # .ts files it floods phantom TS2307/TS2339 errors the delta filter then # misreports as "pre-existing"; skip and let the LSP tier speak. if ext == '.ts' and self._has_ancestor_tsconfig(path): - return LintResult( - skipped=True, - message=( - "Project tsconfig.json detected — per-file tsc skipped " - "(single-file tsc can't resolve project aliases/globals; " - "use the LSP tier or `tsc -p tsconfig.json` for real " - "diagnostics)." - ), - ) - + return LintResult(skipped=True, message=( + "Project tsconfig.json detected — per-file tsc skipped " + "(single-file tsc can't resolve project aliases/globals; " + "use the LSP tier or `tsc -p tsconfig.json` for real " + "diagnostics)." + )) if ext in _SHELL_LINTER_LSP_REDUNDANT and self._lsp_will_handle(path): - return LintResult( - skipped=True, - message=f"LSP server handles {ext} — shell linter skipped", - ) - + return LintResult(skipped=True, message=f"LSP server handles {ext} — shell linter skipped") linter_cmd = LINTERS[ext] base_cmd = linter_cmd.split()[0] if not self._has_command(base_cmd): return LintResult(skipped=True, message=f"{base_cmd} not available") - # Linters are native Windows binaries on Windows: they need the C:/... # form, not MSYS /c/... (node would resolve it as C:\c\Users\... → phantom ENOENT). - cmd = linter_cmd.replace("{file}", self._escape_native_tool_arg(path)) - result = self._exec(cmd, timeout=30) - + result = self._exec(linter_cmd.replace("{file}", self._escape_native_tool_arg(path)), timeout=30) if result.exit_code != 0 and _looks_like_linter_unusable(base_cmd, result.stdout): from tools.ansi_strip import strip_ansi cleaned = strip_ansi(result.stdout).strip() # Collapse to one line — the npx banner is multi-line ASCII art. - first_line = next( - (ln.strip() for ln in cleaned.splitlines() if ln.strip()), - cleaned[:120], - ) - return LintResult( - skipped=True, - message=f"{base_cmd} not usable: {first_line[:200]}", - ) - - return LintResult( - success=result.exit_code == 0, - output=result.stdout.strip() if result.stdout.strip() else "" - ) + first_line = next((ln.strip() for ln in cleaned.splitlines() if ln.strip()), cleaned[:120]) + return LintResult(skipped=True, message=f"{base_cmd} not usable: {first_line[:200]}") + return LintResult(success=result.exit_code == 0, output=result.stdout.strip()) def _check_lint_delta(self, path: str, pre_content: Optional[str], post_content: Optional[str] = None) -> LintResult: - """Post-write lint; when it fails and ``pre_content`` is known, report - only errors this edit introduced (pre-existing lines filtered out). - - Semantic (LSP) diagnostics are a separate channel — see - ``_maybe_lsp_diagnostics`` — so syntax and semantic signals stay distinct. - """ + """Post-write lint; when it fails and ``pre_content`` is known, report only + errors this edit introduced (pre-existing lines filtered out). Semantic + (LSP) diagnostics are a separate channel — see ``_maybe_lsp_diagnostics``.""" post = self._check_lint(path, content=post_content) if post.success or post.skipped or pre_content is None: return post - pre = self._check_lint(path, content=pre_content) if pre.success or pre.skipped or not pre.output: return post # pre-write was clean (or unlintable): all post errors are new - # Single-error parsers (ast.parse, json.loads) stop at the first error, so # if every post error already existed we can't prove the edit is clean — # report the file as still broken but say nothing new was introduced. pre_lines = {ln.strip() for ln in pre.output.splitlines() if ln.strip()} post_lines = [ln for ln in post.output.splitlines() if ln.strip() and ln.strip() not in pre_lines] - if not post_lines: return LintResult( - success=False, - output=post.output, + success=False, output=post.output, message="Pre-existing lint errors — this edit didn't introduce new ones but the file is still broken.", ) - - return LintResult( - success=False, - output=( - "New lint errors introduced by this edit " - "(pre-existing errors filtered out):\n" + "\n".join(post_lines) - ) - ) + return LintResult(success=False, output=( + "New lint errors introduced by this edit " + "(pre-existing errors filtered out):\n" + "\n".join(post_lines) + )) def _lsp_local_only(self) -> bool: """True iff wired to a local backend. LSP servers run on the host and @@ -259,10 +220,7 @@ class LintMixin: def _lsp_service(self): """The active LSPService, or None on a non-local backend / any failure. - - Shared best-effort probe: LSP is an enrichment layer and must never - break a write, so every failure path collapses to None. - """ + LSP is an enrichment layer and must never break a write.""" if not self._lsp_local_only(): return None try: @@ -327,13 +285,8 @@ class LintMixin: except Exception: # noqa: BLE001 pass - def _maybe_lsp_diagnostics( - self, - path: str, - *, - pre_content: Optional[str] = None, - post_content: Optional[str] = None, - ) -> str: + def _maybe_lsp_diagnostics(self, path: str, *, pre_content: Optional[str] = None, + post_content: Optional[str] = None) -> str: """Formatted LSP diagnostics introduced by this edit, or "" when LSP is unavailable/disabled/clean. Never raises past the service probe. @@ -344,7 +297,6 @@ class LintMixin: svc = self._lsp_service() if svc is None or not svc.enabled_for(path): return "" - line_shift = None if pre_content is not None and post_content is not None and pre_content != post_content: try: @@ -352,7 +304,6 @@ class LintMixin: line_shift = build_line_shift(pre_content, post_content) except Exception: # noqa: BLE001 line_shift = None - try: diagnostics = svc.get_diagnostics_sync(path, delta=True, line_shift=line_shift) except Exception: # noqa: BLE001 diff --git a/tools/file_operations_search.py b/tools/file_operations_search.py index b61603f5f8..7a5e0fb3f1 100644 --- a/tools/file_operations_search.py +++ b/tools/file_operations_search.py @@ -1,8 +1,7 @@ """Content/file search tier for ``tools.file_operations``. -Extracted from ``ShellFileOperations`` as ``SearchMixin``; the class inherits it -so every ``self._search_*`` call resolves unchanged via the MRO. Module-level -helpers are re-imported into ``tools.file_operations`` for back-compat. +``ShellFileOperations`` inherits ``SearchMixin``; module-level helpers are pure +(no I/O) and re-imported into ``tools.file_operations`` for back-compat. """ import os @@ -19,11 +18,7 @@ _MACOS_TCC_PROTECTED_HOME_DIRS = ( def _macos_protected_search_exclusions( - path: str, - *, - cwd: Optional[str] = None, - home: Optional[str] = None, - platform: Optional[str] = None, + path: str, *, cwd: Optional[str] = None, home: Optional[str] = None, platform: Optional[str] = None, ) -> List[str]: """Protected home dirs (relative to ``path``) below a broad macOS search root. @@ -33,14 +28,11 @@ def _macos_protected_search_exclusions( """ if (platform or sys.platform) != "darwin": return [] - - home_path = Path(home or Path.home()).expanduser() root = Path(path).expanduser() if not root.is_absolute(): root = Path(cwd or os.getcwd()) / root root = Path(os.path.normpath(str(root))) - home_path = Path(os.path.normpath(str(home_path))) - + home_path = Path(os.path.normpath(str(Path(home or Path.home()).expanduser()))) exclusions: List[str] = [] for dirname in _MACOS_TCC_PROTECTED_HOME_DIRS: try: @@ -70,13 +62,13 @@ _SEARCH_OUTPUT_RE = re.compile(r'^([A-Za-z]:)?[^\s:][^\n]*?[:\-]\d|^[^\s:][^\s]* def _split_tool_diagnostics(output: str) -> tuple[str, str]: - """Separate rg/grep diagnostic lines from real match output. + """Separate rg/grep diagnostic lines from real match output → ``(diagnostics, payload)``. ``_exec`` merges stderr into stdout, so tool errors interleave with matches. - Returns ``(diagnostics, payload)``. Classifying by SHAPE (not error prefix) - lets the exit-2 guard tell a pure failure (no payload → surface the error) - from a partial one (one unreadable file, others matched → keep matches), and - guarantees error text is never parsed as a match. + Classifying by SHAPE (not error prefix) lets the exit-2 guard tell a pure + failure (no payload → surface the error) from a partial one (one unreadable + file, others matched → keep matches), and guarantees error text is never + parsed as a match. """ diagnostics: list[str] = [] payload: list[str] = [] @@ -85,11 +77,9 @@ def _split_tool_diagnostics(output: str) -> tuple[str, str]: continue # Prefix check first: a real match path can contain "-" (e.g. # ".../pytest-686/..."), which the shape regex would accept as a match. - stripped = line.lstrip() - if stripped.startswith("rg: ") or stripped.startswith("grep: "): + if line.lstrip().startswith(("rg: ", "grep: ")): diagnostics.append(line) - continue - if line == "--" or _SEARCH_OUTPUT_RE.match(line): + elif line == "--" or _SEARCH_OUTPUT_RE.match(line): payload.append(line) else: diagnostics.append(line) @@ -97,22 +87,17 @@ def _split_tool_diagnostics(output: str) -> tuple[str, str]: def _parse_search_context_line(line: str) -> tuple[str, int, str] | None: - """Parse a ``path-line-content`` context line. - - Filenames may contain ``--`` segments, so use the RIGHTMOST numeric - separator: ``dir/file-12-name.py-8-context`` → (``dir/file-12-name.py``, 8). - """ + """Parse a ``path-line-content`` context line using the RIGHTMOST numeric + separator (filenames may contain ``--`` segments): + ``dir/file-12-name.py-8-context`` → (``dir/file-12-name.py``, 8, ``context``).""" if not line or line == "--": return None match = None for candidate in re.finditer(r'-(\d+)-', line): match = candidate - if match is None: + if match is None or match.start() == 0: return None - path = line[:match.start()] - if not path: - return None - return path, int(match.group(1)), line[match.end():] + return line[:match.start()], int(match.group(1)), line[match.end():] _REGEX_NEWLINE_ESCAPE_RE = re.compile(r"(? bool: def _is_line_oriented_newline_error(error: Optional[str]) -> bool: """Return True for rg's hard error when multiline mode is required.""" - if not error: - return False - return "literal \"\\n\" is not allowed" in error and "--multiline" in error + return bool(error) and "literal \"\\n\" is not allowed" in error and "--multiline" in error def _maybe_warn_line_oriented_newline_pattern(result: SearchResult, pattern: str) -> SearchResult: @@ -170,17 +153,12 @@ def _parse_search_output(result, output_mode: str, limit: int, offset: int, if result.exit_code == 2 and not payload.strip(): error_msg = diagnostics.strip() or result.stdout.strip() or "Search error" return SearchResult(error=f"Search failed: {error_msg}", total_count=0) - lines = [ln for ln in payload.strip().split('\n') if ln] if output_mode == "files_only": return SearchResult( - files=lines[offset:offset + limit], - total_count=len(lines), - truncated=bool(limit_reason), - limit_reason=limit_reason, - warning=warning, + files=lines[offset:offset + limit], total_count=len(lines), + truncated=bool(limit_reason), limit_reason=limit_reason, warning=warning, ) - if output_mode == "count": counts = {} for line in lines: @@ -191,12 +169,9 @@ def _parse_search_output(result, output_mode: str, limit: int, offset: int, except ValueError: pass return SearchResult( - counts=counts, - total_count=sum(counts.values()), - truncated=bool(limit_reason), - limit_reason=limit_reason, + counts=counts, total_count=sum(counts.values()), + truncated=bool(limit_reason), limit_reason=limit_reason, ) - matches = [] for line in lines: if line == "--": @@ -204,9 +179,7 @@ def _parse_search_output(result, output_mode: str, limit: int, offset: int, m = _MATCH_LINE_RE.match(line) if m: matches.append(SearchMatch( - path=(m.group(1) or '') + m.group(2), - line_number=int(m.group(3)), - content=m.group(4)[:500], + path=(m.group(1) or '') + m.group(2), line_number=int(m.group(3)), content=m.group(4)[:500], )) continue # Context lines ("file-line-content") only when context was requested, @@ -214,19 +187,18 @@ def _parse_search_output(result, output_mode: str, limit: int, offset: int, if context > 0: parsed = _parse_search_context_line(line) if parsed: - matches.append(SearchMatch( - path=parsed[0], line_number=parsed[1], content=parsed[2][:500], - )) + matches.append(SearchMatch(path=parsed[0], line_number=parsed[1], content=parsed[2][:500])) total = len(matches) return SearchResult( - matches=matches[offset:offset + limit], - total_count=total, - truncated=total > offset + limit or bool(limit_reason), - limit_reason=limit_reason, - warning=warning, + matches=matches[offset:offset + limit], total_count=total, + truncated=total > offset + limit or bool(limit_reason), limit_reason=limit_reason, warning=warning, ) +def _has_hidden_part(parts) -> bool: + return any(part not in {".", ".."} and part.startswith(".") for part in parts) + + class SearchMixin: """File-name and content search via rg with find/grep fallbacks. Requires ``_exec``, ``_has_command``, ``_expand_path``, ``_escape_shell_arg``, @@ -246,51 +218,49 @@ class SearchMixin: return [] from tools import file_operations as _fo # lazy: _HOME is monkeypatched there cwd = getattr(self.env, "cwd", None) or self.cwd - return _macos_protected_search_exclusions( - path, cwd=cwd, home=_fo._HOME, platform=sys.platform - ) + return _macos_protected_search_exclusions(path, cwd=cwd, home=_fo._HOME, platform=sys.platform) def _protected_prune_paths(self, path: str) -> List[str]: """Absolute-ish protected paths for find's ``-path ... -prune``.""" - return [ - os.path.normpath(os.path.join(path, item)) - for item in self._macos_search_exclusions(path) - ] + return [os.path.normpath(os.path.join(path, item)) for item in self._macos_search_exclusions(path)] + + def _prune_expr(self, protected_paths: List[str]) -> str: + """find ``\\( -path A -o -path B \\) -prune`` clause for the protected dirs.""" + terms = " -o ".join(f"-path {self._escape_shell_arg(item)}" for item in protected_paths) + return f"\\( {terms} \\) -prune" + + def _rg_exclusion_globs(self, path: str) -> List[str]: + """``--glob '!/**'`` pairs excluding protected dirs from an rg run.""" + out: List[str] = [] + for item in self._macos_search_exclusions(path): + out.extend(["--glob", self._escape_shell_arg(f"!{item}/**")]) + return out def _path_exists_probe(self, path: str) -> str: """Stdout of the existence probe: contains "exists" or "not_found".""" - return self._exec( - f"test -e {self._escape_shell_arg(path)} && echo exists || echo not_found" - ).stdout + return self._exec(f"test -e {self._escape_shell_arg(path)} && echo exists || echo not_found").stdout def _dispatch_search(self, pattern: str, path: str, target: str, file_glob: Optional[str], limit: int, offset: int, output_mode: str, context: int) -> SearchResult: if target == "files": return self._search_files(pattern, path, limit, offset) - return self._search_content(pattern, path, file_glob, limit, offset, - output_mode, context) + return self._search_content(pattern, path, file_glob, limit, offset, output_mode, context) def _path_not_found_result(self, path: str) -> SearchResult: """Error result for a missing search root, with nearby-entry suggestions.""" parent = os.path.dirname(path) or "." basename_query = os.path.basename(path) hint_parts = [f"Path not found: {path}"] - parent_check = self._exec( - f"test -d {self._escape_shell_arg(parent)} && echo yes || echo no" - ) + parent_check = self._exec(f"test -d {self._escape_shell_arg(parent)} && echo yes || echo no") if "yes" in parent_check.stdout and basename_query: - ls_result = self._exec( - f"ls -1 {self._escape_shell_arg(parent)} 2>/dev/null | head -20" - ) + ls_result = self._exec(f"ls -1 {self._escape_shell_arg(parent)} 2>/dev/null | head -20") if ls_result.exit_code == 0 and ls_result.stdout.strip(): lower_q = basename_query.lower() candidates = [] for entry in ls_result.stdout.strip().split('\n'): - if not entry: - continue le = entry.lower() - if lower_q in le or le in lower_q or le.startswith(lower_q[:3]): + if entry and (lower_q in le or le in lower_q or le.startswith(lower_q[:3])): candidates.append(os.path.join(parent, entry)) if candidates: hint_parts.append("Similar paths: " + ", ".join(candidates[:5])) @@ -311,11 +281,9 @@ class SearchMixin: (existing if "exists" in self._path_exists_probe(expanded) else missing).append(expanded) if not existing: return None - merged = SearchResult() for p in existing: - sub = self._dispatch_search(pattern, p, target, file_glob, limit, offset, - output_mode, context) + sub = self._dispatch_search(pattern, p, target, file_glob, limit, offset, output_mode, context) if sub.error: continue merged.matches.extend(sub.matches) @@ -346,8 +314,7 @@ class SearchMixin: "(or pass a simpler substring)."), ) - def _zero_match_probe(self, pattern: str, path: str, - file_glob: Optional[str]) -> Optional[str]: + def _zero_match_probe(self, pattern: str, path: str, file_glob: Optional[str]) -> Optional[str]: """Steering hint for a 0-match content search, or None. A bare zero gives the model nothing to act on, so run cheap count-only rg @@ -383,13 +350,6 @@ class SearchMixin: """Search for files by name (glob-like): rg --files, else find.""" search_pattern = pattern if (not pattern.startswith('**/') and '/' not in pattern) \ else pattern.split('/')[-1] - - search_root = Path(path) - has_hidden_path_ancestor = any( - part not in {".", ".."} and part.startswith(".") - for part in search_root.parts - ) - # rg respects .gitignore, skips hidden dirs, and walks in parallel (~200x find). if self._has_command('rg'): return self._search_files_rg(search_pattern, path, limit, offset) @@ -399,21 +359,15 @@ class SearchMixin: "Install ripgrep for best results: " "https://github.com/BurntSushi/ripgrep#installation" ) - # Hidden roots: find's path filter would exclude everything under the root, # so gather full output and filter descendants in Python (pagination too). + search_root = Path(path) + has_hidden_path_ancestor = _has_hidden_part(search_root.parts) hidden_filter_expr = "" if has_hidden_path_ancestor else " -not -path '*/.*'" pagination_expr = "" if has_hidden_path_ancestor else f" | tail -n +{offset + 1} | head -n {limit}" - # Prune protected dirs BEFORE traversal so macOS never sees an access attempt. protected_paths = self._protected_prune_paths(path) - prune_expr = "" - if protected_paths: - prune_terms = " -o ".join( - f"-path {self._escape_shell_arg(item)}" for item in protected_paths - ) - prune_expr = f" \\( {prune_terms} \\) -prune -o" - + prune_expr = f" {self._prune_expr(protected_paths)} -o" if protected_paths else "" base = (f"find {self._escape_shell_arg(path)}{prune_expr}{hidden_filter_expr} " f"-type f -name {self._escape_shell_arg(search_pattern)} ") result = self._exec(f"{base}-printf '%T@ %p\\n' 2>/dev/null | sort -rn{pagination_expr}", timeout=60) @@ -422,14 +376,12 @@ class SearchMixin: # BSD find (macOS) has no -printf. result = self._exec(f"{base}2>/dev/null | sort -rn{pagination_expr}", timeout=60) stdout, limit_reason = _search_stdout_and_limit(result) - files = [] for line in stdout.strip().split('\n'): if not line: continue parts = line.split(' ', 1) files.append(parts[1] if len(parts) == 2 and parts[0].replace('.', '').isdigit() else line) - if has_hidden_path_ancestor: normalized_root = search_root.resolve() filtered_files = [] @@ -438,65 +390,46 @@ class SearchMixin: rel_parts = Path(file_path).resolve().relative_to(normalized_root).parts except ValueError: rel_parts = Path(file_path).parts - if any(part not in {".", ".."} and part.startswith(".") for part in rel_parts): - continue - filtered_files.append(file_path) + if not _has_hidden_part(rel_parts): + filtered_files.append(file_path) files = filtered_files[offset:offset + limit] - - return SearchResult( - files=files, - total_count=len(files), - truncated=bool(limit_reason), - limit_reason=limit_reason, - ) + return SearchResult(files=files, total_count=len(files), truncated=bool(limit_reason), limit_reason=limit_reason) def _search_files_rg(self, pattern: str, path: str, limit: int, offset: int) -> SearchResult: """File-name search via ``rg --files``, mtime-sorted when rg >= 13 supports --sortr.""" # Wrap bare names so -g matches at any depth (equivalent to find -name). glob_pattern = f"*{pattern}" if ('/' not in pattern and not pattern.startswith('*')) else pattern - fetch_limit = limit + offset - exclusion_globs = " ".join( - f"--glob {self._escape_shell_arg(f'!{item}/**')}" - for item in self._macos_search_exclusions(path) - ) + exclusion_globs = " ".join(self._rg_exclusion_globs(path)) exclusion_args = f" {exclusion_globs}" if exclusion_globs else "" tail = (f"-g {self._escape_shell_arg(glob_pattern)}{exclusion_args} " f"{self._escape_native_tool_arg(path)} 2>/dev/null | head -n {fetch_limit}") result = self._exec(f"rg --files --sortr=modified {tail}", timeout=60) stdout, limit_reason = _search_stdout_and_limit(result) all_files = [f for f in stdout.strip().split('\n') if f] - if not all_files and not limit_reason: # --sortr may have failed on older rg; retry without it. result = self._exec(f"rg --files {tail}", timeout=60) stdout, limit_reason = _search_stdout_and_limit(result) all_files = [f for f in stdout.strip().split('\n') if f] - return SearchResult( - files=all_files[offset:offset + limit], - total_count=len(all_files), - truncated=len(all_files) >= fetch_limit or bool(limit_reason), - limit_reason=limit_reason, + files=all_files[offset:offset + limit], total_count=len(all_files), + truncated=len(all_files) >= fetch_limit or bool(limit_reason), limit_reason=limit_reason, ) def _search_content(self, pattern: str, path: str, file_glob: Optional[str], limit: int, offset: int, output_mode: str, context: int) -> SearchResult: """Content search: rg, else grep; attaches zero-match steering hints.""" - used_rg = False - if self._has_command('rg'): - used_rg = True - result = self._search_with_rg(pattern, path, file_glob, limit, offset, - output_mode, context) + used_rg = self._has_command('rg') + if used_rg: + result = self._search_with_rg(pattern, path, file_glob, limit, offset, output_mode, context) elif self._has_command('grep'): - result = self._search_with_grep(pattern, path, file_glob, limit, offset, - output_mode, context) + result = self._search_with_grep(pattern, path, file_glob, limit, offset, output_mode, context) else: return SearchResult( error="Content search requires ripgrep (rg) or grep. " "Install ripgrep: https://github.com/BurntSushi/ripgrep#installation" ) - if (not result.error and result.total_count == 0 and not result.matches and not result.files and not result.counts): try: @@ -505,18 +438,30 @@ class SearchMixin: hint = None if hint: result.warning = hint if not result.warning else f"{result.warning} {hint}" - # rg auto-enables --multiline for \n patterns, so the line-oriented # explanation only applies to the grep fallback. if used_rg: return result return _maybe_warn_line_oriented_newline_pattern(result, pattern) + def _run_search_pipeline(self, cmd_parts: List[str], output_mode: str, limit: int, + offset: int, context: int, warning: Optional[str] = None) -> SearchResult: + """Run ``cmd_parts | head -n `` under pipefail and parse. + + Extra rows are fetched to report the true total; context mode also emits + "--" separators, so grab 200 more and filter in Python. pipefail keeps the + engine's exit 2 alive across ``| head`` (a truncating head makes rg exit 0 + / grep exit 141 on SIGPIPE, neither of which the strict ==2 guard flags). + """ + fetch_limit = limit + offset + (200 if context > 0 else 0) + cmd = "set -o pipefail; " + " ".join(cmd_parts + ["|", "head", "-n", str(fetch_limit)]) + result = self._exec(cmd, timeout=60) + return _parse_search_output(result, output_mode, limit, offset, context, warning=warning) + def _search_with_rg(self, pattern: str, path: str, file_glob: Optional[str], limit: int, offset: int, output_mode: str, context: int) -> SearchResult: """Search using ripgrep.""" cmd_parts = ["rg", "--line-number", "--no-heading", "--with-filename"] - # A regex \n can't match in line-oriented mode (rg hard-errors); enable -U # up front when the pattern clearly wants to cross lines, and say so. multiline = _pattern_has_regex_newline(pattern) @@ -524,8 +469,7 @@ class SearchMixin: cmd_parts.append("--multiline") if context > 0: cmd_parts.extend(["-C", str(context)]) - for item in self._macos_search_exclusions(path): - cmd_parts.extend(["--glob", self._escape_shell_arg(f"!{item}/**")]) + cmd_parts.extend(self._rg_exclusion_globs(path)) if file_glob: cmd_parts.extend(["--glob", self._escape_shell_arg(file_glob)]) if output_mode in _OUTPUT_MODE_FLAGS: @@ -533,39 +477,26 @@ class SearchMixin: cmd_parts.append(self._escape_shell_arg(pattern)) # rg is a native Windows binary (winget/cargo/choco): needs C:/... not MSYS /c/... cmd_parts.append(self._escape_native_tool_arg(path)) - - # Fetch extra rows to report the true total; context mode also emits "--" - # separators, so grab generously and filter in Python. - fetch_limit = limit + offset + 200 if context > 0 else limit + offset - cmd_parts.extend(["|", "head", "-n", str(fetch_limit)]) - - # pipefail so rg's exit 2 survives `| head` (else head's 0 masks it). rg - # exits 0 on SIGPIPE from a truncating head, so no false errors. - cmd = "set -o pipefail; " + " ".join(cmd_parts) - result = self._exec(cmd, timeout=60) ml_note = ( "Pattern contains \\n — multiline mode (-U) was enabled automatically " "so the regex can match across line boundaries." ) if multiline else None - return _parse_search_output(result, output_mode, limit, offset, context, warning=ml_note) + return self._run_search_pipeline(cmd_parts, output_mode, limit, offset, context, warning=ml_note) def _search_with_grep(self, pattern: str, path: str, file_glob: Optional[str], limit: int, offset: int, output_mode: str, context: int) -> SearchResult: """Fallback search using grep.""" - # -H forces filenames; -E matches rg regex behavior; --exclude-dir='.*' - # mirrors rg's hidden-dir default (.git/, .hub/index-cache/, ...). - cmd_parts = ["grep", "-rnHE", "--exclude-dir='.*'"] - # grep's --exclude-dir matches BASENAMES anywhere in the tree, so it can't # express "only the home-level Downloads"; route protected-dir pruning # through find's path-scoped -prune instead. protected_paths = self._protected_prune_paths(path) if protected_paths: return self._search_with_grep_pruned( - pattern, path, file_glob, limit, offset, output_mode, context, - protected_paths, + pattern, path, file_glob, limit, offset, output_mode, context, protected_paths, ) - + # -H forces filenames; -E matches rg regex behavior; --exclude-dir='.*' + # mirrors rg's hidden-dir default (.git/, .hub/index-cache/, ...). + cmd_parts = ["grep", "-rnHE", "--exclude-dir='.*'"] if context > 0: cmd_parts.extend(["-C", str(context)]) if file_glob: @@ -573,13 +504,10 @@ class SearchMixin: if output_mode in _OUTPUT_MODE_FLAGS: cmd_parts.append(_OUTPUT_MODE_FLAGS[output_mode]) cmd_parts.append(self._escape_shell_arg(pattern)) - # grep applies --exclude-dir to the search root too, so a relative root # "." would be excluded by '.*'. Anchor relative paths at the shell's # live $PWD (quoted separately so user paths stay escaped). - is_absolute = path.startswith(("/", "\\\\")) or bool( - re.match(r"^[A-Za-z]:[\\/]", path) - ) + is_absolute = path.startswith(("/", "\\\\")) or bool(re.match(r"^[A-Za-z]:[\\/]", path)) if is_absolute: search_root = self._escape_shell_arg(path) else: @@ -588,15 +516,7 @@ class SearchMixin: if relative_path not in {"", "."}: search_root += f"/{self._escape_shell_arg(relative_path)}" cmd_parts.append(search_root) - - fetch_limit = limit + offset + (200 if context > 0 else 0) - cmd_parts.extend(["|", "head", "-n", str(fetch_limit)]) - - # pipefail so grep's exit 2 survives `| head`; a truncating head makes - # grep exit 141 (SIGPIPE), which the strict ==2 guard ignores. - cmd = "set -o pipefail; " + " ".join(cmd_parts) - result = self._exec(cmd, timeout=60) - return _parse_search_output(result, output_mode, limit, offset, context) + return self._run_search_pipeline(cmd_parts, output_mode, limit, offset, context) def _search_with_grep_pruned(self, pattern: str, path: str, file_glob: Optional[str], limit: int, offset: int, output_mode: str, context: int, @@ -616,23 +536,13 @@ class SearchMixin: if output_mode in _OUTPUT_MODE_FLAGS: grep_parts.append(_OUTPUT_MODE_FLAGS[output_mode]) grep_parts.append(self._escape_shell_arg(pattern)) - - prune_terms = " -o ".join( - f"-path {self._escape_shell_arg(item)}" for item in protected_paths - ) find_parts = [ "find", self._escape_shell_arg(path or "."), - f"\\( {prune_terms} \\) -prune", "-o", + self._prune_expr(protected_paths), "-o", "\\( -type d -name '.*' \\) -prune", "-o", "-type f", ] if file_glob: find_parts.extend(["-name", self._escape_shell_arg(file_glob)]) - find_parts.extend(["-exec", *grep_parts, "{}", "+"]) - fetch_limit = limit + offset + (200 if context > 0 else 0) - cmd = ( - "set -o pipefail; " + " ".join(find_parts) - + f" 2>/dev/null | head -n {fetch_limit}" - ) - result = self._exec(cmd, timeout=60) - return _parse_search_output(result, output_mode, limit, offset, context) + find_parts.extend(["-exec", *grep_parts, "{}", "+", "2>/dev/null"]) + return self._run_search_pipeline(find_parts, output_mode, limit, offset, context) diff --git a/tools/hook_output_spill.py b/tools/hook_output_spill.py index 14ca516c73..5070cc69ab 100644 --- a/tools/hook_output_spill.py +++ b/tools/hook_output_spill.py @@ -1,25 +1,12 @@ """Spill oversized hook-injected context to disk with a preview placeholder. -Shell hooks and plugin ``pre_llm_call`` hooks can return ``{"context": ...}`` -that is concatenated into the user message on EVERY subsequent API call, so a -large blob inflates every turn and breaks the prompt-cache prefix. Above a -configured budget the full text is written to a per-session directory and the -in-prompt payload becomes a head/tail preview plus the saved path. - -Config (``config.yaml``):: - - hooks: - output_spill: - enabled: true # default: true; set false to disable spilling - max_chars: 10000 # default; context above this is spilled - preview_head: 500 # chars shown at the start of the preview - preview_tail: 500 # chars shown at the end of the preview - directory: null # default: /hook_outputs - -Invariants: unchanged input when disabled or under the cap; never raises — -an I/O failure still returns a bounded preview with an in-prompt notice. Spill -files are grouped per session so ``/new`` sessions don't pile into one directory. -Ported from openai/codex PR #21069. +Hook ``{"context": ...}`` output is concatenated into the user message on EVERY +subsequent API call, so a large blob inflates every turn and breaks the +prompt-cache prefix. Above ``hooks.output_spill.max_chars`` (default 10000) the +full text is written under ``hooks.output_spill.directory`` (default +``/hook_outputs/``) and the in-prompt payload becomes a +``preview_head``/``preview_tail`` excerpt plus the saved path. ``enabled: false`` +disables. Never raises: an I/O failure still returns a bounded preview. """ from __future__ import annotations @@ -48,27 +35,19 @@ def get_spill_config() -> Dict[str, Any]: from hermes_cli.config import load_config cfg = load_config() or {} hooks = cfg.get("hooks") if isinstance(cfg, dict) else None - if isinstance(hooks, dict): - sub = hooks.get("output_spill") - if isinstance(sub, dict): - section = sub + if isinstance(hooks, dict) and isinstance(hooks.get("output_spill"), dict): + section = hooks["output_spill"] except Exception: section = {} - enabled_raw = section.get("enabled", DEFAULT_ENABLED) - enabled = bool(enabled_raw) if enabled_raw is not None else DEFAULT_ENABLED - directory = section.get("directory") - if directory is not None and not isinstance(directory, str): - directory = None - return { - "enabled": enabled, + "enabled": bool(enabled_raw) if enabled_raw is not None else DEFAULT_ENABLED, "max_chars": _coerce_positive_int(section.get("max_chars"), DEFAULT_MAX_CHARS), # head/tail allow zero (empty tail), max_chars must be positive. "preview_head": _coerce_int(section.get("preview_head"), DEFAULT_PREVIEW_HEAD, 0), "preview_tail": _coerce_int(section.get("preview_tail"), DEFAULT_PREVIEW_TAIL, 0), - "directory": directory, + "directory": directory if isinstance(directory, str) else None, } @@ -78,46 +57,13 @@ def _resolve_spill_dir(directory_override: Optional[str], session_id: Optional[s base = Path(os.path.expanduser(directory_override)) else: from hermes_constants import get_hermes_home - base = Path(get_hermes_home()) / "hook_outputs" - - session_segment = session_id or "no-session" - session_segment = session_segment.replace("/", "_").replace("\\", "_").replace("..", "_") + session_segment = (session_id or "no-session").replace("/", "_").replace("\\", "_").replace("..", "_") return base / session_segment -def _build_preview( - text: str, - head: int, - tail: int, - saved_path: Optional[str], - *, - source: str, -) -> str: - """Assemble the in-prompt preview with head/tail and saved-path footer.""" - total = len(text) - head_chunk = text[:head] if head > 0 else "" - tail_chunk = text[-tail:] if tail > 0 and total > head else "" - - parts = [ - f"[{source} output truncated — {total:,} chars; full content " - + (f"saved to {saved_path}]" if saved_path else "unavailable — spill write failed]"), - ] - if head_chunk: - parts.append("--- head ---") - parts.append(head_chunk) - if tail_chunk: - parts.append("--- tail ---") - parts.append(tail_chunk) - return "\n".join(parts) - - def spill_if_oversized( - text: str, - *, - session_id: Optional[str] = None, - source: str = "hook", - config: Optional[Dict[str, Any]] = None, + text: str, *, session_id: Optional[str] = None, source: str = "hook", config: Optional[Dict[str, Any]] = None, ) -> str: """Spill ``text`` to disk if it exceeds the configured cap. @@ -132,42 +78,40 @@ def spill_if_oversized( text = str(text) except Exception: return "" - cfg = config if config is not None else get_spill_config() if not cfg.get("enabled", True): return text - - max_chars = int(cfg.get("max_chars") or DEFAULT_MAX_CHARS) - if len(text) <= max_chars: + if len(text) <= int(cfg.get("max_chars") or DEFAULT_MAX_CHARS): return text - head = int(cfg.get("preview_head") or 0) tail = int(cfg.get("preview_tail") or 0) - directory_override = cfg.get("directory") # A disk failure must never blow up the turn — fall through to a preview # without a saved path. saved_path: Optional[str] = None try: - spill_dir = _resolve_spill_dir(directory_override, session_id) + spill_dir = _resolve_spill_dir(cfg.get("directory"), session_id) from tools.spill_safety import ensure_spill_dir, write_text_exclusive - # Hook context may embed raw secrets: private perms + exclusive, # symlink-refusing create (the per-session dir is predictable). ensure_spill_dir(spill_dir, private=True) spill_path = spill_dir / f"{uuid.uuid4().hex}.txt" # Trailing newline so tail readers don't report "missing newline". - write_text_exclusive( - spill_path, - text if text.endswith("\n") else text + "\n", - private=True, - ) + write_text_exclusive(spill_path, text if text.endswith("\n") else text + "\n", private=True) saved_path = str(spill_path) except Exception as exc: logger.warning("hook output spill failed: %s", exc) - saved_path = None - return _build_preview(text, head, tail, saved_path, source=source) + total = len(text) + parts = [ + f"[{source} output truncated — {total:,} chars; full content " + + (f"saved to {saved_path}]" if saved_path else "unavailable — spill write failed]"), + ] + if head > 0 and text[:head]: + parts.extend(["--- head ---", text[:head]]) + if tail > 0 and total > head: + parts.extend(["--- tail ---", text[-tail:]]) + return "\n".join(parts) __all__ = [