refactor(tools): compact file_operations comments/docstrings, keep every WHY
This commit is contained in:
Teknium
+44
-81
@@ -35,9 +35,7 @@ from tools.file_operations_search import ( # noqa: F401 (re-exported)
|
||||
# Controller home; SearchMixin reads it (tests monkeypatch it here).
|
||||
_HOME = str(Path.home())
|
||||
|
||||
# =============================================================================
|
||||
# Binary-content identification
|
||||
# =============================================================================
|
||||
# --- Binary-content identification -------------------------------------------
|
||||
|
||||
_MAGIC_SIGNATURES: tuple = (
|
||||
# (prefix bytes, human name) — ordered, first match wins. Longest
|
||||
@@ -135,9 +133,7 @@ class FileOperations(ABC):
|
||||
"""Search for content or files."""
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# Shell-based Implementation
|
||||
# =============================================================================
|
||||
# --- Shell-based implementation ----------------------------------------------
|
||||
|
||||
# Image extensions (subset of binary that we can return as base64)
|
||||
IMAGE_EXTENSIONS = {'.png', '.jpg', '.jpeg', '.gif', '.webp', '.bmp', '.ico'}
|
||||
@@ -159,8 +155,7 @@ class ShellFileOperations(LintMixin, SearchMixin, FileOperations):
|
||||
|
||||
def __init__(self, terminal_env, cwd: str = None):
|
||||
self.env = terminal_env
|
||||
# Never fall back to os.getcwd(): that is the HOST path, which doesn't
|
||||
# exist inside container/cloud backends. "/" is the universal default.
|
||||
# Never os.getcwd(): that is the HOST path, absent inside container backends.
|
||||
self.cwd = cwd or getattr(terminal_env, 'cwd', None) or \
|
||||
getattr(getattr(terminal_env, 'config', None), 'cwd', None) or "/"
|
||||
self._command_cache: Dict[str, bool] = {}
|
||||
@@ -178,8 +173,7 @@ class ShellFileOperations(LintMixin, SearchMixin, FileOperations):
|
||||
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 for stdin callers other than
|
||||
# write_file, which rejects unencodable content up front).
|
||||
# child never received the input.
|
||||
if result.get("stdin_error") and exit_code == 0:
|
||||
exit_code = 1
|
||||
return ExecuteResult(stdout=result.get("output", ""), exit_code=exit_code)
|
||||
@@ -255,10 +249,8 @@ class ShellFileOperations(LintMixin, SearchMixin, FileOperations):
|
||||
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 as binary (read-only) so a read→edit→write round-trip
|
||||
# can't overwrite the original bytes with mojibake.
|
||||
# Undecodable bytes arrive as U+FFFD ("printable", so the ratio misses
|
||||
# them); treat as binary so a round-trip can't write back 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')
|
||||
@@ -295,9 +287,8 @@ class ShellFileOperations(LintMixin, SearchMixin, FileOperations):
|
||||
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.
|
||||
# ~username: validate 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
|
||||
@@ -376,9 +367,7 @@ class ShellFileOperations(LintMixin, SearchMixin, FileOperations):
|
||||
old_content.splitlines(keepends=True), new_content.splitlines(keepends=True),
|
||||
fromfile=f"a/{filename}", tofile=f"b/{filename}"))
|
||||
|
||||
# =========================================================================
|
||||
# READ Implementation
|
||||
# =========================================================================
|
||||
# --- READ ---------------------------------------------------------------
|
||||
|
||||
@staticmethod
|
||||
def _not_regular_error(path: str) -> ReadResult:
|
||||
@@ -508,9 +497,8 @@ class ShellFileOperations(LintMixin, SearchMixin, FileOperations):
|
||||
offset, limit = normalize_read_pagination(offset, limit)
|
||||
file_size, status = self._probe_regular_file(path)
|
||||
if status == "missing":
|
||||
# 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.
|
||||
# Unicode-equivalent spellings render identically, so the model can never
|
||||
# discover the byte mismatch by retyping — 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)
|
||||
@@ -532,8 +520,7 @@ class ShellFileOperations(LintMixin, SearchMixin, FileOperations):
|
||||
"Use vision_analyze with this file path to inspect the image contents."))
|
||||
is_binary, sample_bytes = self._detect_binary(path)
|
||||
if is_binary:
|
||||
# A UTF-16 text file (Notepad .txt, PowerShell `>`) arrives mangled with
|
||||
# U+FFFD and trips the binary guard; probe the raw bytes and transcode.
|
||||
# UTF-16 text (Notepad, PowerShell `>`) trips the binary guard; transcode.
|
||||
utf16_result = self._try_read_utf16(path, offset, limit, file_size)
|
||||
if utf16_result is not None:
|
||||
return utf16_result
|
||||
@@ -541,15 +528,12 @@ class ShellFileOperations(LintMixin, SearchMixin, FileOperations):
|
||||
is_binary=True, file_size=file_size,
|
||||
error=describe_binary_file(sample_bytes, file_size))
|
||||
|
||||
# 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).
|
||||
# Clamp each line to a byte budget IN THE SHELL so a 400MB single-line file
|
||||
# never crosses the exec transport. 4*max+1 BYTES (not max+1): ``cut -b`` can
|
||||
# split a multibyte codepoint, and a tighter byte clamp would yield fewer
|
||||
# CHARS than max so the Python clamp in _add_line_numbers would never fire
|
||||
# (silent truncation). UTF-8 codepoints are ≤4 bytes, so every over-long
|
||||
# line still trips the char clamp, which also drops a boundary-split U+FFFD.
|
||||
from tools.tool_output_limits import get_max_line_length
|
||||
line_clamp_bytes = 4 * get_max_line_length() + 1
|
||||
end_line = offset + limit - 1
|
||||
@@ -559,9 +543,7 @@ class ShellFileOperations(LintMixin, SearchMixin, FileOperations):
|
||||
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)
|
||||
# Only the first chunk can carry a BOM (byte 0); strip it so the model
|
||||
# never sees a phantom U+FEFF.
|
||||
if offset == 1:
|
||||
if offset == 1: # only the first chunk can carry a BOM (byte 0)
|
||||
read_output, _ = _strip_bom(read_output)
|
||||
|
||||
wc_result = self._exec(f"wc -l < {self._escape_shell_arg(path)}")
|
||||
@@ -574,16 +556,14 @@ class ShellFileOperations(LintMixin, SearchMixin, FileOperations):
|
||||
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 EOF; probe the last byte.
|
||||
# ``cut`` always newline-terminates, so a file without a trailing newline
|
||||
# would grow a phantom empty last line; when this page reaches EOF, probe.
|
||||
if not truncated and read_output.endswith('\n'):
|
||||
tail_result = self._exec(f"tail -c 1 {self._escape_shell_arg(path)} | wc -l")
|
||||
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. Name the dead end and its recovery.
|
||||
# Empty content is indistinguishable from a broken tool: name the dead end.
|
||||
if file_size == 0:
|
||||
return ReadResult(content="", total_lines=0, file_size=0, hint="File is empty (0 bytes).")
|
||||
if offset > total_lines > 0:
|
||||
@@ -661,8 +641,7 @@ 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): 1-2 edit typos the
|
||||
# substring checks miss.
|
||||
# Near-miss spelling (AGENT.md -> AGENTS.md) the substring checks miss.
|
||||
if score == 0 and difflib.SequenceMatcher(None, lower_name, lf).ratio() >= 0.8:
|
||||
score = 50
|
||||
if score > 0:
|
||||
@@ -686,9 +665,8 @@ class ShellFileOperations(LintMixin, SearchMixin, FileOperations):
|
||||
cat_result = self._exec(f"cat {self._escape_shell_arg(path)}")
|
||||
if cat_result.exit_code != 0:
|
||||
return ReadResult(error=f"Failed to read file: {cat_result.stdout}")
|
||||
# Strip a leading 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.
|
||||
# Strip a leading BOM (a phantom U+FEFF defeats an exact first-line match);
|
||||
# write_file re-probes disk and restores it.
|
||||
raw_content, _ = _strip_bom(_strip_terminal_fence_leaks(cat_result.stdout))
|
||||
return ReadResult(content=raw_content, file_size=file_size)
|
||||
|
||||
@@ -723,8 +701,8 @@ class ShellFileOperations(LintMixin, SearchMixin, FileOperations):
|
||||
denied = get_write_denied_error(path, verb="Delete")
|
||||
if denied:
|
||||
return WriteResult(error=denied)
|
||||
# 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.
|
||||
# Path baked in via repr() for shell-independent quoting; no
|
||||
# ``unlink(missing_ok=True)`` (a 3.7 remote interpreter lacks it).
|
||||
snippet = (
|
||||
"import shutil, pathlib, sys\n"
|
||||
f"p = pathlib.Path({path!r})\n"
|
||||
@@ -758,9 +736,7 @@ class ShellFileOperations(LintMixin, SearchMixin, FileOperations):
|
||||
return WriteResult(error=f"Failed to move {src} -> {dst}: {result.stdout}")
|
||||
return WriteResult()
|
||||
|
||||
# =========================================================================
|
||||
# WRITE Implementation
|
||||
# =========================================================================
|
||||
# --- WRITE --------------------------------------------------------------
|
||||
|
||||
# Lone surrogates OUTSIDE the surrogateescape range (U+DC80-U+DCFF round-trips
|
||||
# through the pipe; anything else can't be encoded at all).
|
||||
@@ -866,17 +842,14 @@ 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)
|
||||
# Best-effort snapshot so the post-write LSP layer returns only
|
||||
# diagnostics introduced by this edit.
|
||||
# Best-effort snapshot so the LSP tier reports only this edit's diagnostics.
|
||||
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`` means "parent dirs ensured" (mkdir -p is folded into
|
||||
# _atomic_write; its 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.
|
||||
# surrogateescape is the exact inverse of the decode that may have produced
|
||||
# this content, so these are 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:
|
||||
@@ -886,9 +859,8 @@ class ShellFileOperations(LintMixin, SearchMixin, FileOperations):
|
||||
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 are a separate channel, fired only when the syntax tier is
|
||||
# clean (no point asking an LSP about a file that won't parse).
|
||||
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
|
||||
@@ -896,9 +868,7 @@ class ShellFileOperations(LintMixin, SearchMixin, FileOperations):
|
||||
bytes_written=len(content_bytes), dirs_created=dirs_created, verified=content_verified,
|
||||
lint=lint_result.to_dict() if lint_result else None, lsp_diagnostics=lsp_diagnostics)
|
||||
|
||||
# =========================================================================
|
||||
# PATCH Implementation (Replace Mode)
|
||||
# =========================================================================
|
||||
# --- PATCH (replace mode) -----------------------------------------------
|
||||
|
||||
def _no_match_result(self, path: str, content: str, old_string: str,
|
||||
new_string: str, match_count: int, error: Optional[str]) -> PatchResult:
|
||||
@@ -953,9 +923,8 @@ class ShellFileOperations(LintMixin, SearchMixin, FileOperations):
|
||||
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).
|
||||
# Match and diff on BOM-stripped content (a phantom U+FEFF defeats an exact
|
||||
# first-line match); the raw read becomes write_file's pre_content.
|
||||
raw_content = read_result.stdout
|
||||
content, _ = _strip_bom(raw_content)
|
||||
|
||||
@@ -964,13 +933,11 @@ class ShellFileOperations(LintMixin, SearchMixin, FileOperations):
|
||||
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 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.
|
||||
# Models send bare-LF old/new strings; normalize the substituted region to
|
||||
# the file's ending so CRLF files stay consistent.
|
||||
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.
|
||||
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}")
|
||||
@@ -981,8 +948,7 @@ class ShellFileOperations(LintMixin, SearchMixin, FileOperations):
|
||||
return PatchResult(
|
||||
success=True, diff=self._unified_diff(content, new_content, path), files_modified=[path],
|
||||
lint=lint_result.to_dict() if lint_result else None,
|
||||
# Captured by the internal write_file call against the pre-patch
|
||||
# baseline, so the delta is correct for the whole patch.
|
||||
# From the internal write_file call, whose baseline was the pre-patch content.
|
||||
lsp_diagnostics=write_result.lsp_diagnostics)
|
||||
|
||||
def patch_v4a(self, patch_content: str) -> PatchResult:
|
||||
@@ -994,9 +960,7 @@ class ShellFileOperations(LintMixin, SearchMixin, FileOperations):
|
||||
return PatchResult(error=f"Failed to parse patch: {parse_error}")
|
||||
return apply_v4a_operations(operations, self)
|
||||
|
||||
# =========================================================================
|
||||
# SEARCH Implementation
|
||||
# =========================================================================
|
||||
# --- SEARCH -------------------------------------------------------------
|
||||
|
||||
def search(self, pattern: str, path: str = ".", target: str = "content",
|
||||
file_glob: Optional[str] = None, limit: int = 50, offset: int = 0,
|
||||
@@ -1007,8 +971,7 @@ class ShellFileOperations(LintMixin, SearchMixin, FileOperations):
|
||||
offset, limit = normalize_search_pagination(offset, limit)
|
||||
path = self._expand_path(path)
|
||||
if "not_found" in self._path_exists_probe(path):
|
||||
# Models frequently pass several paths in one string ("dir1 dir2" or
|
||||
# comma-separated): search every part that exists and report the rest.
|
||||
# Models often pass several paths in one string: search the parts that exist.
|
||||
multi = self._try_multi_path_search(
|
||||
pattern, path, target, file_glob, limit, offset, output_mode, context)
|
||||
if multi is not None:
|
||||
|
||||
@@ -12,9 +12,8 @@ from typing import Callable, Dict, Optional
|
||||
|
||||
from tools.file_operations_common import LintResult
|
||||
|
||||
# Shell linters by extension, run via _exec() for languages whose check needs an
|
||||
# external toolchain. ``.tsx`` is deliberately absent: it has never had a shell
|
||||
# linter (it hits the "No linter" skip) and LSP covers it when enabled.
|
||||
# Shell linters by extension (external toolchain). ``.tsx`` is deliberately absent:
|
||||
# it hits the "No linter" skip and LSP covers it when enabled.
|
||||
LINTERS = {
|
||||
'.py': 'python -m py_compile {file} 2>&1',
|
||||
'.js': 'node --check {file} 2>&1',
|
||||
@@ -23,16 +22,14 @@ LINTERS = {
|
||||
'.rs': 'rustfmt --check {file} 2>&1',
|
||||
}
|
||||
|
||||
# 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, ``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.
|
||||
# Per-file shell linters that flood phantom errors on real projects (single-file
|
||||
# ``tsc`` ignores tsconfig, ``go vet`` fails outside a module, ``rustfmt --check``
|
||||
# is style-only): skipped when an LSP server claims the file. 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) → ``skipped`` so the write isn't flagged and
|
||||
# the LSP tier (which gates on ok/skipped) still runs. Matched case-insensitively.
|
||||
# Output substrings (case-insensitive) meaning the linter binary exists but could
|
||||
# not run → ``skipped`` so the write isn't flagged and the LSP tier still runs.
|
||||
_LINTER_UNUSABLE_PATTERNS = {
|
||||
'npx': (
|
||||
'this is not the tsc command you are looking for', # tsc not installed locally
|
||||
@@ -71,13 +68,9 @@ def _lint_json_inproc(content: str) -> tuple[bool, str]:
|
||||
|
||||
|
||||
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 and app tags like
|
||||
CloudFormation ``!Sub`` / Ansible ``!vault``. This verdict is a fail-closed
|
||||
WRITE gate, so a false positive refuses a legitimate write.
|
||||
"""
|
||||
"""In-process YAML syntax check; ``__SKIP__`` when PyYAML is missing. Syntax-only
|
||||
(``yaml.parse``), NOT ``safe_load``: loading rejects valid multi-doc streams and
|
||||
app tags (``!Sub``, ``!vault``), and this is a fail-closed WRITE gate."""
|
||||
try:
|
||||
import yaml as _yaml
|
||||
except ImportError:
|
||||
@@ -113,9 +106,8 @@ def _lint_python_inproc(content: str) -> tuple[bool, str]:
|
||||
return False, f"{type(e).__name__}: {e}"
|
||||
|
||||
|
||||
# In-process linters, preferred over shell linters (microseconds, no subprocess).
|
||||
# Each takes content and returns (ok, error); error ``"__SKIP__"`` means the
|
||||
# linter is unavailable (missing dependency) and counts as "no linter".
|
||||
# In-process linters, preferred over shell linters (no subprocess). Each returns
|
||||
# (ok, error); error ``"__SKIP__"`` = unavailable dependency, counts as "no linter".
|
||||
LINTERS_INPROC: Dict[str, Callable[[str], tuple[bool, str]]] = {
|
||||
'.py': _lint_python_inproc,
|
||||
'.json': _lint_json_inproc,
|
||||
@@ -124,10 +116,9 @@ LINTERS_INPROC: Dict[str, Callable[[str], tuple[bool, str]]] = {
|
||||
'.toml': _lint_toml_inproc,
|
||||
}
|
||||
|
||||
# Extensions where write_file REFUSES on a parse failure (fail-closed gate) rather
|
||||
# than merely reporting. ``.py`` is excluded on purpose: test fixtures use ``*.py``
|
||||
# paths as a generic stand-in for arbitrary text, so hard-refusing invalid Python
|
||||
# would break exercised patterns. Python keeps the non-blocking lint-delta report.
|
||||
# Extensions where write_file REFUSES on a parse failure. ``.py`` is excluded on
|
||||
# purpose: test fixtures use ``*.py`` paths as a stand-in for arbitrary text, so
|
||||
# Python keeps the non-blocking lint-delta report.
|
||||
_FAIL_CLOSED_INPROC_EXTS = frozenset({'.json', '.yaml', '.yml', '.toml'})
|
||||
|
||||
|
||||
@@ -153,9 +144,8 @@ class LintMixin:
|
||||
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.
|
||||
# Single-file tsc can't read tsconfig.json and floods phantom TS2307/TS2339
|
||||
# errors the delta filter misreports as "pre-existing"; 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 "
|
||||
@@ -169,8 +159,7 @@ class LintMixin:
|
||||
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).
|
||||
# Native Windows binaries need C:/... not MSYS /c/... (→ phantom ENOENT).
|
||||
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
|
||||
@@ -191,16 +180,13 @@ class LintMixin:
|
||||
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.
|
||||
# Single-error parsers stop at the first error, so if every post error already
|
||||
# existed we can't prove the edit is clean — 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,
|
||||
message="Pre-existing lint errors — this edit didn't introduce new ones but the file is still broken.",
|
||||
)
|
||||
return LintResult(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)
|
||||
@@ -244,11 +230,8 @@ class LintMixin:
|
||||
|
||||
def _has_ancestor_tsconfig(self, path: str) -> bool:
|
||||
"""True iff a tsconfig.json exists in ``path``'s directory or any ancestor.
|
||||
|
||||
Host-side walk, local backend only: on a remote backend the tree isn't
|
||||
here, so this answers False and the shell linter runs as before — never
|
||||
suppress lint based on a probe that couldn't answer.
|
||||
"""
|
||||
Host-side walk, local backend only: on a remote backend this answers False so
|
||||
the shell linter still runs — never suppress lint on a probe that couldn't answer."""
|
||||
if not self._lsp_local_only():
|
||||
return False
|
||||
try:
|
||||
@@ -287,12 +270,9 @@ class LintMixin:
|
||||
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.
|
||||
|
||||
With both pre and post content a line-shift map remaps baseline
|
||||
diagnostics into post-edit coordinates; otherwise every pre-existing
|
||||
diagnostic below an inserted/deleted line would look newly introduced.
|
||||
"""
|
||||
unavailable/disabled/clean. With both pre and post content a line-shift map
|
||||
remaps baseline diagnostics into post-edit coordinates; otherwise every
|
||||
pre-existing diagnostic below an inserted line would look new."""
|
||||
svc = self._lsp_service()
|
||||
if svc is None or not svc.enabled_for(path):
|
||||
return ""
|
||||
|
||||
@@ -55,28 +55,22 @@ def _search_stdout_and_limit(result: ExecuteResult) -> tuple[str, Optional[str]]
|
||||
|
||||
|
||||
# A real rg/grep output line is a whitespace-free path token followed by ``:``
|
||||
# (match/count), ``-`` (context), or nothing (files_only). Tool diagnostics
|
||||
# ("rg: ...", "error: ...", indented carets) never match: the leading token
|
||||
# forbids whitespace and a tool prefix is followed by ": " (space).
|
||||
# (match/count), ``-`` (context), or nothing (files_only); tool diagnostics
|
||||
# ("rg: ...", indented carets) never match.
|
||||
_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 → ``(diagnostics, payload)``.
|
||||
|
||||
``_exec`` merges stderr into stdout, so tool errors interleave with matches.
|
||||
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.
|
||||
"""
|
||||
``_exec`` merges stderr into stdout; classifying by SHAPE lets the exit-2 guard
|
||||
tell a pure failure (no payload) from a partial one (one unreadable file, others
|
||||
matched) and guarantees error text is never parsed as a match."""
|
||||
diagnostics: list[str] = []
|
||||
payload: list[str] = []
|
||||
for line in output.split('\n'):
|
||||
if not line.strip():
|
||||
continue
|
||||
# Prefix check first: a real match path can contain "-<digit>" (e.g.
|
||||
# ".../pytest-686/..."), which the shape regex would accept as a match.
|
||||
# Prefix check first: a match path can contain "-<digit>" (".../pytest-686/...").
|
||||
if line.lstrip().startswith(("rg: ", "grep: ")):
|
||||
diagnostics.append(line)
|
||||
elif line == "--" or _SEARCH_OUTPUT_RE.match(line):
|
||||
@@ -141,12 +135,9 @@ _OUTPUT_MODE_FLAGS = {"files_only": "-l", "count": "-c"}
|
||||
def _parse_search_output(result, output_mode: str, limit: int, offset: int,
|
||||
context: int, warning: Optional[str] = None) -> SearchResult:
|
||||
"""Parse rg/grep ``| head`` output into a SearchResult (shared by both engines).
|
||||
|
||||
Exit codes: 0=matches, 1=none, 2=error — but both tools return 2 on PARTIAL
|
||||
errors (one unreadable file in a tree that otherwise matched), so an error is
|
||||
surfaced only when exit==2 AND no usable payload remains.
|
||||
``warning`` is attached to files_only/content results (rg's multiline note).
|
||||
"""
|
||||
errors (one unreadable file), so an error is surfaced only when exit==2 AND no
|
||||
usable payload remains. ``warning`` is attached to files_only/content results."""
|
||||
stdout, limit_reason = _search_stdout_and_limit(result)
|
||||
diagnostics, payload = _split_tool_diagnostics(stdout)
|
||||
if result.exit_code == 2 and not payload.strip():
|
||||
@@ -179,8 +170,7 @@ def _parse_search_output(result, output_mode: str, limit: int, offset: int,
|
||||
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,
|
||||
# to avoid false positives on dash-heavy paths.
|
||||
# Context lines only when requested, to avoid false positives on dashy paths.
|
||||
if context > 0:
|
||||
parsed = _parse_search_context_line(line)
|
||||
if parsed:
|
||||
@@ -202,14 +192,10 @@ class SearchMixin:
|
||||
``_escape_native_tool_arg``, ``env`` and ``cwd`` from the host class."""
|
||||
|
||||
def _macos_search_exclusions(self, path: str) -> List[str]:
|
||||
"""Protected descendants to prune for this search root, if any.
|
||||
|
||||
Gated on ``env.is_local``: ``sys.platform``/``_HOME`` describe the
|
||||
CONTROLLER, but the search runs on ``env``'s host — a macOS controller
|
||||
driving a Linux container must not prune the remote's Downloads. Envs
|
||||
without the flag (fakes, plugins) default to local semantics; pruning is
|
||||
a warning-carrying skip, never data loss.
|
||||
"""
|
||||
"""Protected descendants to prune for this search root, if any. Gated on
|
||||
``env.is_local``: ``sys.platform``/``_HOME`` describe the CONTROLLER, but the
|
||||
search runs on ``env``'s host. Envs without the flag default to local
|
||||
semantics; pruning is a warning-carrying skip, never data loss."""
|
||||
env = getattr(self, "env", None)
|
||||
if env is not None and getattr(env, "is_local", True) is False:
|
||||
return []
|
||||
@@ -312,12 +298,9 @@ class SearchMixin:
|
||||
)
|
||||
|
||||
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
|
||||
probes (case-insensitive, hidden/ignored, fixed-string) and report the
|
||||
first that hits. Bounded to three rg invocations.
|
||||
"""
|
||||
"""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 probes (case-insensitive,
|
||||
hidden/ignored, fixed-string) and report the first that hits."""
|
||||
if not self._has_command('rg'):
|
||||
return None
|
||||
has_meta = bool(re.search(r"[.\[\](){}?*+^$\\|]", pattern))
|
||||
@@ -354,8 +337,8 @@ class SearchMixin:
|
||||
error="File search requires 'rg' (ripgrep) or 'find'. "
|
||||
"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).
|
||||
# Hidden roots: find's path filter would exclude everything, so filter
|
||||
# descendants (and paginate) in Python.
|
||||
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 '*/.*'"
|
||||
@@ -388,7 +371,8 @@ class SearchMixin:
|
||||
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."""
|
||||
@@ -440,13 +424,10 @@ class SearchMixin:
|
||||
|
||||
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 <fetch_limit>`` 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).
|
||||
"""
|
||||
"""Run ``cmd_parts | head -n <fetch_limit>`` under pipefail and parse. Extra
|
||||
rows report the true total (context mode also emits "--" separators, so
|
||||
grab 200 more). pipefail keeps the engine's exit 2 alive across ``| head``
|
||||
(a truncating head makes rg exit 0 / grep 141, which the ==2 guard ignores)."""
|
||||
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)
|
||||
@@ -456,8 +437,7 @@ class SearchMixin:
|
||||
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.
|
||||
# A regex \n hard-errors in line-oriented mode; enable -U up front and say so.
|
||||
multiline = _pattern_has_regex_newline(pattern)
|
||||
if multiline:
|
||||
cmd_parts.append("--multiline")
|
||||
@@ -493,9 +473,8 @@ class SearchMixin:
|
||||
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."""
|
||||
# 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.
|
||||
# grep's --exclude-dir matches BASENAMES anywhere, so it can't express "only
|
||||
# the home-level Downloads"; route pruning through find's path-scoped -prune.
|
||||
protected_paths = self._protected_prune_paths(path)
|
||||
if protected_paths:
|
||||
return self._search_with_grep_pruned(
|
||||
@@ -503,9 +482,8 @@ class SearchMixin:
|
||||
# -H forces filenames; -E matches rg regex behavior; --exclude-dir='.*'
|
||||
# mirrors rg's hidden-dir default (.git/, .hub/index-cache/, ...).
|
||||
cmd_parts = self._grep_cmd(["grep", "-rnHE", "--exclude-dir='.*'"], pattern, output_mode, context, file_glob)
|
||||
# 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).
|
||||
# --exclude-dir applies to the root too, so "." would be excluded by '.*';
|
||||
# anchor relative paths at the shell's live $PWD.
|
||||
is_absolute = path.startswith(("/", "\\\\")) or bool(re.match(r"^[A-Za-z]:[\\/]", path))
|
||||
if is_absolute:
|
||||
search_root = self._escape_shell_arg(path)
|
||||
@@ -520,15 +498,11 @@ class SearchMixin:
|
||||
def _search_with_grep_pruned(self, pattern: str, path: str, file_glob: Optional[str],
|
||||
limit: int, offset: int, output_mode: str, context: int,
|
||||
protected_paths: List[str]) -> SearchResult:
|
||||
"""grep fallback with PATH-scoped protected-dir pruning.
|
||||
|
||||
``find ... -prune`` enumerates files (traversal never enters protected
|
||||
dirs, so macOS never sees an access attempt) and hands them to grep via
|
||||
``-exec {} +``; hidden dirs are pruned to mirror ``--exclude-dir='.*'``.
|
||||
Trade-off: find folds grep's exit code into its own generic non-zero, so
|
||||
a hard grep error surfaces as an empty result rather than exit 2 —
|
||||
acceptable for this darwin-local-broad-search-only branch.
|
||||
"""
|
||||
"""grep fallback with PATH-scoped protected-dir pruning: ``find ... -prune``
|
||||
enumerates files (traversal never enters protected dirs) and hands them to
|
||||
grep via ``-exec {} +``; hidden dirs pruned to mirror ``--exclude-dir='.*'``.
|
||||
Trade-off: find folds grep's exit code, so a hard grep error surfaces as an
|
||||
empty result — acceptable for this darwin-local-broad-search-only branch."""
|
||||
grep_parts = self._grep_cmd(["grep", "-nHE"], pattern, output_mode, context)
|
||||
find_parts = [
|
||||
"find", self._escape_shell_arg(path or "."),
|
||||
|
||||
@@ -1,12 +1,11 @@
|
||||
"""Spill oversized hook-injected context to disk with a preview placeholder.
|
||||
|
||||
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
|
||||
``<HERMES_HOME>/hook_outputs/<session>``) 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.
|
||||
Hook ``{"context": ...}`` output rides 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 text is written under
|
||||
``hooks.output_spill.directory`` (default ``<HERMES_HOME>/hook_outputs/<session>``)
|
||||
and the payload becomes a ``preview_head``/``preview_tail`` excerpt plus the path.
|
||||
``enabled: false`` disables. Never raises: an I/O failure still returns a preview.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
@@ -114,11 +113,5 @@ def spill_if_oversized(
|
||||
return "\n".join(parts)
|
||||
|
||||
|
||||
__all__ = [
|
||||
"DEFAULT_MAX_CHARS",
|
||||
"DEFAULT_PREVIEW_HEAD",
|
||||
"DEFAULT_PREVIEW_TAIL",
|
||||
"DEFAULT_ENABLED",
|
||||
"get_spill_config",
|
||||
"spill_if_oversized",
|
||||
]
|
||||
__all__ = ["DEFAULT_MAX_CHARS", "DEFAULT_PREVIEW_HEAD", "DEFAULT_PREVIEW_TAIL", "DEFAULT_ENABLED",
|
||||
"get_spill_config", "spill_if_oversized"]
|
||||
|
||||
Reference in New Issue
Block a user