hermes-admin/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 30 Packages Projects Releases Wiki Activity

refactor(tools): shared find/rg fallback runner, collapse small helpers in file_operations tiers

Browse Source
This commit is contained in:
Teknium
2026-09-02 23:04:41 -07:00
parent 10ebc75c7d
commit 5df140f868
4 changed files with 46 additions and 63 deletions
Download Patch File Download Diff File Expand all files Collapse all files

30
tools/file_operations.py
View File

@@ -2,10 +2,9 @@
"""File operations (read, write, patch, search) over any terminal backend.
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).
implementation serves every environment (local, docker, ssh, modal, ...). Companions,
re-exported here: ``file_operations_common`` (result dataclasses, text helpers),
``file_operations_lint`` (LintMixin), ``file_operations_search`` (SearchMixin).
"""
import base64
@@ -69,10 +68,8 @@ _MAGIC_SIGNATURES: tuple = (
def identify_binary_bytes(sample: bytes) -> str:
"""Best-effort human name for binary content from its magic bytes; never raises.
The ISO-media entry additionally requires ``ftyp`` at offset 4 — three
leading NULs alone are too weak a signature.
"""
The ISO-media entry additionally requires ``ftyp`` at offset 4 (three leading
NULs alone are too weak a signature)."""
for prefix, name in _MAGIC_SIGNATURES:
if sample.startswith(prefix):
if name.startswith("ISO media") and sample[4:8] != b"ftyp":
@@ -261,19 +258,14 @@ class ShellFileOperations(LintMixin, SearchMixin, FileOperations):
return os.path.splitext(path)[1].lower() in IMAGE_EXTENSIONS
def _add_line_numbers(self, content: str, start_line: int = 1) -> str:
"""Prefix each line with a compact ``<n>|`` gutter, clamping long lines.
Compact, not fixed-width: padding cost ~16% more tokens per line for no
accuracy gain in A/B, while dropping numbers regressed line-referencing.
"""
"""Prefix each line with a compact ``<n>|`` gutter, clamping long lines. 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()
numbered = []
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)
return '\n'.join(
f"{i}|{line if len(line) <= max_line_length else line[:max_line_length] + '... [truncated]'}"
for i, line in enumerate(content.split('\n'), start=start_line))
def _expand_path(self, path: str) -> str:
"""Expand ``~`` / ``~user`` via the backend's shell (its HOME, not the

10
tools/file_operations_common.py
View File

@@ -1,10 +1,8 @@
"""Result dataclasses and pure text helpers shared by ``tools.file_operations``
and its search/lint mixins.
Leaf module (imports nothing from ``tools`` at module scope) so the mixins and
the origin module can all depend on it without an import cycle. The
``to_dict`` output of every class here IS tool behavior — key names, key
order, and omission rules are pinned by tests and read by the model.
and its search/lint mixins. Leaf module (imports nothing from ``tools`` at module
scope) so nothing cycles. The ``to_dict`` output of every class here IS tool
behavior — key names, order and omission rules are pinned by tests and read by
the model.
"""
import re

12
tools/file_operations_lint.py
View File

@@ -225,8 +225,7 @@ class LintMixin:
from agent.lsp.servers import SERVERS
except Exception: # noqa: BLE001
return False
ext_lower = ext.lower()
return any(ext_lower in srv.extensions for srv in SERVERS)
return any(ext.lower() in srv.extensions for srv in SERVERS)
def _has_ancestor_tsconfig(self, path: str) -> bool:
"""True iff a tsconfig.json exists in ``path``'s directory or any ancestor.
@@ -236,13 +235,12 @@ class LintMixin:
return False
try:
d = os.path.dirname(os.path.abspath(path))
while True:
if os.path.isfile(os.path.join(d, "tsconfig.json")):
return True
while not os.path.isfile(os.path.join(d, "tsconfig.json")):
parent = os.path.dirname(d)
if parent == d:
return False
d = parent
return True
except Exception: # noqa: BLE001
return False
@@ -292,8 +290,6 @@ class LintMixin:
try:
from agent.lsp.reporter import report_for_file, truncate
block = report_for_file(path, diagnostics)
if not block:
return ""
return truncate("LSP diagnostics introduced by this edit:\n" + block)
return truncate("LSP diagnostics introduced by this edit:\n" + block) if block else ""
except Exception: # noqa: BLE001
return ""

57
tools/file_operations_search.py
View File

@@ -239,12 +239,10 @@ class SearchMixin:
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")
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'):
le = entry.lower()
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))
lq = basename_query.lower()
candidates = [
os.path.join(parent, e) for e in ls_result.stdout.strip().split('\n')
if e and (lq in e.lower() or e.lower() in lq or e.lower().startswith(lq[:3]))]
if candidates:
hint_parts.append("Similar paths: " + ", ".join(candidates[:5]))
return SearchResult(error=". ".join(hint_parts), total_count=0)
@@ -348,32 +346,36 @@ class SearchMixin:
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)
stdout, limit_reason = _search_stdout_and_limit(result)
if not stdout.strip() and not limit_reason:
# 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)
# BSD find (macOS) has no -printf: retry without the mtime prefix.
lines, limit_reason = self._exec_lines_with_fallback(
f"{base}-printf '%T@ %p\\n' 2>/dev/null | sort -rn{pagination_expr}",
f"{base}2>/dev/null | sort -rn{pagination_expr}")
files = []
for line in stdout.strip().split('\n'):
if not line:
continue
for line in lines:
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 = []
for file_path in files:
def rel_parts(file_path):
try:
rel_parts = Path(file_path).resolve().relative_to(normalized_root).parts
return Path(file_path).resolve().relative_to(normalized_root).parts
except ValueError:
rel_parts = Path(file_path).parts
if not _has_hidden_part(rel_parts):
filtered_files.append(file_path)
files = filtered_files[offset:offset + limit]
return Path(file_path).parts
files = [f for f in files if not _has_hidden_part(rel_parts(f))][offset:offset + limit]
return SearchResult(files=files, total_count=len(files),
truncated=bool(limit_reason), limit_reason=limit_reason)
def _exec_lines_with_fallback(self, cmd: str, fallback_cmd: str) -> tuple[List[str], Optional[str]]:
"""Non-empty stdout lines of ``cmd``; when it yields nothing (and didn't time
out) run ``fallback_cmd`` instead. Returns ``(lines, limit_reason)``."""
result = self._exec(cmd, timeout=60)
stdout, limit_reason = _search_stdout_and_limit(result)
if not stdout.strip() and not limit_reason:
result = self._exec(fallback_cmd, timeout=60)
stdout, limit_reason = _search_stdout_and_limit(result)
return [f for f in stdout.strip().split('\n') if f], 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).
@@ -383,14 +385,9 @@ class SearchMixin:
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]
# --sortr may have failed on older rg; retry without it.
all_files, limit_reason = self._exec_lines_with_fallback(
f"rg --files --sortr=modified {tail}", f"rg --files {tail}")
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,