1a72042115
A volume that already owns /workspace skipped the configured working directory, so tools treated that host path as unmounted. Bind it at a second mount, or point tools at the volume that already has it, for any drive path.
1704 lines
91 KiB
Python
1704 lines
91 KiB
Python
#!/usr/bin/env python3
|
|
"""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 every environment (local, docker, ssh, modal, ...). Companions:
|
|
``file_operations_common`` (result dataclasses, text helpers), ``file_operations_lint``
|
|
(LintMixin), ``file_operations_search`` (SearchMixin).
|
|
"""
|
|
|
|
import base64
|
|
import binascii
|
|
import os
|
|
import re
|
|
import sys
|
|
import difflib
|
|
import hashlib
|
|
import json
|
|
import logging
|
|
import secrets
|
|
import unicodedata
|
|
from abc import ABC, abstractmethod
|
|
from typing import Optional, Dict
|
|
from pathlib import Path
|
|
|
|
from tools.binary_extensions import has_binary_extension
|
|
from agent.file_safety import get_write_denied_error
|
|
from tools.file_operations_common import (
|
|
ExecuteResult, PatchResult, ReadResult, 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 SearchMixin
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Controller home; SearchMixin reads it (tests monkeypatch it here).
|
|
_HOME = str(Path.home())
|
|
|
|
# --- Binary-content identification -------------------------------------------
|
|
|
|
_MAGIC_SIGNATURES: tuple = (
|
|
# (prefix bytes, human name) — ordered, first match wins. Longest
|
|
# prefixes for a shared first byte come first.
|
|
(b"\x89PNG\r\n\x1a\n", "PNG image data"),
|
|
(b"\xff\xd8\xff", "JPEG image data"),
|
|
(b"GIF87a", "GIF image data"),
|
|
(b"GIF89a", "GIF image data"),
|
|
(b"RIFF", "RIFF container (WAV/AVI/WebP family)"),
|
|
(b"%PDF-", "PDF document"),
|
|
(b"PK\x03\x04", "ZIP archive (also docx/xlsx/jar/apk)"),
|
|
(b"PK\x05\x06", "ZIP archive (empty)"),
|
|
(b"\x1f\x8b", "gzip compressed data"),
|
|
(b"BZh", "bzip2 compressed data"),
|
|
(b"\xfd7zXZ\x00", "xz compressed data"),
|
|
(b"7z\xbc\xaf\x27\x1c", "7-Zip archive"),
|
|
(b"\x7fELF", "ELF executable"),
|
|
(b"MZ", "Windows PE executable"),
|
|
(b"\xcf\xfa\xed\xfe", "Mach-O executable (64-bit)"),
|
|
(b"\xca\xfe\xba\xbe", "Mach-O universal binary / Java class"),
|
|
(b"SQLite format 3\x00", "SQLite database"),
|
|
(b"OggS", "Ogg container"),
|
|
(b"fLaC", "FLAC audio"),
|
|
(b"ID3", "MP3 audio (ID3 tag)"),
|
|
(b"\x00\x00\x00", "ISO media container (MP4/MOV family)"), # ftyp at +4
|
|
(b"BM", "BMP image data"),
|
|
(b"II*\x00", "TIFF image data (little-endian)"),
|
|
(b"MM\x00*", "TIFF image data (big-endian)"),
|
|
)
|
|
|
|
|
|
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)."""
|
|
for prefix, name in _MAGIC_SIGNATURES:
|
|
if sample.startswith(prefix):
|
|
if name.startswith("ISO media") and sample[4:8] != b"ftyp":
|
|
continue
|
|
return name
|
|
return "unknown binary"
|
|
|
|
|
|
def describe_binary_file(sample: Optional[bytes], file_size: int) -> str:
|
|
"""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"
|
|
elif file_size >= 1024:
|
|
size = f"{file_size / 1024:.1f} KB"
|
|
else:
|
|
size = f"{file_size} bytes"
|
|
return f"Binary file ({kind}, {size}) — cannot display as text."
|
|
|
|
|
|
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:
|
|
"""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:
|
|
"""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."""
|
|
|
|
@abstractmethod
|
|
def move_file(self, src: str, dst: str) -> WriteResult:
|
|
"""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,
|
|
order: str = "discovery") -> SearchResult:
|
|
"""Search for content or files."""
|
|
|
|
|
|
# --- Shell-based implementation ----------------------------------------------
|
|
|
|
# Image extensions (subset of binary that we can return as base64)
|
|
IMAGE_EXTENSIONS = {'.png', '.jpg', '.jpeg', '.gif', '.webp', '.bmp', '.ico'}
|
|
|
|
# Echoed by the size probe when the path exists but is not a regular file.
|
|
# `wc -c` prints only digits, so this can never collide with a real size.
|
|
NOT_REGULAR_SENTINEL = "__hermes_not_regular__"
|
|
|
|
# Echoed by the compound read/write probes when the path does not exist. A
|
|
# compound command only reports its *last* exit status, so the missing-file
|
|
# signal that ``_probe_regular_file`` carries in ``exit 1`` travels in-band.
|
|
MISSING_SENTINEL = "__hermes_missing__"
|
|
|
|
_READ_SENTINEL_PREFIX = "__HERMES_RF_"
|
|
_WRITE_SENTINEL_PREFIX = "__HERMES_WF_"
|
|
_BYTES_SENTINEL_PREFIX = "__HERMES_RB_"
|
|
|
|
|
|
def _new_sentinel(prefix: str) -> str:
|
|
"""Per-call separator line for a compound shell probe. 128 random bits make a
|
|
collision with file content negligible; the underscores keep the token outside
|
|
the base64 alphabet, so a sentinel leaking into a sample segment fails base64
|
|
validation instead of decoding into bytes."""
|
|
return f"{prefix}{secrets.token_hex(16)}__"
|
|
|
|
|
|
def _split_segments(output: str, sentinel: str) -> list[str]:
|
|
"""Split compound-probe stdout on its sentinel lines. Every producer (``wc``,
|
|
``base64``, ``cut``) newline-terminates or prints nothing, so the separator is
|
|
always ``sentinel + "\n"`` on its own line; the text after the final sentinel
|
|
is the status segment."""
|
|
return output.split(sentinel + "\n")
|
|
|
|
|
|
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 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):
|
|
self.env = terminal_env
|
|
# 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 "/"
|
|
# Ordinary executables: bool cache (hits AND misses). rg is special — it has
|
|
# an off-PATH resolver and may be installed mid-session — so only successful
|
|
# rg resolutions are cached (see SearchMixin._resolve_command).
|
|
self._command_cache: Dict[str, bool] = {}
|
|
self._rg_resolution_cache: Dict[str, str] = {}
|
|
self._rg_modified_capability: Dict[str, Optional[str]] = {}
|
|
|
|
def _exec(self, command: str, cwd: str = None, timeout: int = None,
|
|
stdin_data: str = None) -> ExecuteResult:
|
|
"""Run ``command`` on the backend. cwd: explicit arg → live ``env.cwd`` →
|
|
init-time ``self.cwd``. ``stdin_data`` is piped (bypasses ARG_MAX)."""
|
|
kwargs = {}
|
|
if timeout:
|
|
kwargs['timeout'] = timeout
|
|
if stdin_data is not None:
|
|
kwargs['stdin_data'] = stdin_data
|
|
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)
|
|
output = result.get("output", "")
|
|
# The command wrapper's own ``builtin cd -- <cwd> || exit 126`` failed: the
|
|
# working directory does not exist on this backend (typically ``terminal.cwd``
|
|
# is a host path and the backend is a container). Name that, or the raw
|
|
# ``cd:`` line reads like a sandbox/mount fault at the requested path.
|
|
cwd_error = ""
|
|
if exit_code == 126 and "cd: " in output:
|
|
from tools.terminal_tool_config import _is_container_backend
|
|
env_type = getattr(self.env, "env_type", None)
|
|
hint = ("; for container backends use a path inside the container, e.g. /workspace"
|
|
if env_type and _is_container_backend(env_type) else "")
|
|
cwd_error = output = (
|
|
f"working directory {effective_cwd!r} does not exist on the active terminal "
|
|
f"backend (check terminal.cwd or the session cwd{hint}). {output.strip()}")
|
|
# A stdin write failure with a clean child exit is still a failure: the
|
|
# child never received the input.
|
|
if result.get("stdin_error") and exit_code == 0:
|
|
exit_code = 1
|
|
return ExecuteResult(stdout=output, exit_code=exit_code, cwd_error=cwd_error)
|
|
|
|
def _has_command(self, cmd: str) -> bool:
|
|
"""Check if a command exists in the environment (cached); rg goes through
|
|
the resolver so a mid-session install becomes visible."""
|
|
if cmd == "rg":
|
|
return self._resolve_command(cmd) is not None
|
|
if cmd not in self._command_cache:
|
|
result = self._exec(f"command -v {cmd} >/dev/null 2>&1 && echo 'yes'")
|
|
if result.cwd_error: # the probe never ran: no verdict to cache
|
|
return False
|
|
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 _fenced_read(self, body: str, *more: str) -> "tuple[Optional[list[str]], Optional[int], ExecuteResult]":
|
|
"""Run BODY, then each of MORE, each in its own sentinel-delimited segment; return (those
|
|
segments, BODY's exit status, reply).
|
|
|
|
The transport merges the backend's own stdout with the command's, and every caller here
|
|
decodes a segment into file bytes, so the payload has to be delimited rather than taken to
|
|
be the whole reply: a remote shell announcing ``TERM`` is four base64 characters that would
|
|
otherwise join the payload and decode to ``b"LDL"`` at the head of it. The fence drops noise
|
|
OUTSIDE it only; output emitted while BODY runs (a ``BASH_ENV`` DEBUG hook) lands inside the
|
|
payload, so a caller that writes the bytes back must verify them independently (MORE). The
|
|
status rides in its own trailing segment so a failed BODY is still told apart from an empty
|
|
file. ``(None, None, reply)`` when no fenced reply came back — the command never ran as
|
|
written.
|
|
"""
|
|
sentinel = _new_sentinel(_BYTES_SENTINEL_PREFIX)
|
|
mark = f"echo {sentinel}"
|
|
rest = "".join(f"{mark}; {cmd}; " for cmd in more)
|
|
# xtrace off first: a traced ``+ echo <sentinel>`` line is an extra separator, and the
|
|
# traces of the transport commands would land inside the payload segments.
|
|
result = self._exec(f"{{ set +x; }} 2>/dev/null; {mark}; {body}; __hb=$?; {rest}{mark}; echo $__hb")
|
|
segments = _split_segments(result.stdout or "", sentinel)
|
|
if len(segments) != len(more) + 3:
|
|
return None, None, result
|
|
try:
|
|
return segments[1:-1], int(_strip_terminal_fence_leaks(segments[-1]).split()[0]), result
|
|
except (IndexError, ValueError):
|
|
return segments[1:-1], None, result
|
|
|
|
@staticmethod
|
|
def _matches_size(data: bytes, size_segment: str) -> bool:
|
|
"""Whether DATA is exactly as long as the file's own ``wc -c``. Noise inside the payload
|
|
only ever ADDS text, and any addition that still decodes adds bytes, so equal length is
|
|
the check; noise in the size segment breaks its single-integer shape instead."""
|
|
tokens = _strip_terminal_fence_leaks(size_segment).split()
|
|
return len(tokens) == 1 and tokens[0].isdigit() and int(tokens[0]) == len(data)
|
|
|
|
def _sample_file_bytes(self, path: str, length: int = 1000):
|
|
"""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.
|
|
|
|
Wrapping the sample in base64 lets the original bytes survive the transport, so binary detection can
|
|
happen at the byte layer where it is well-defined (#80308 and friends). Fenced like the
|
|
byte-exact read below: this sample is the binary-admission gate in FRONT of that read, so
|
|
backend noise decoded into it decides whether a file is editable at all.
|
|
"""
|
|
segments, read_rc, _ = self._fenced_read(
|
|
f"head -c {length} {self._escape_shell_arg(path)} 2>/dev/null | base64")
|
|
if segments is None or read_rc != 0:
|
|
return None
|
|
return self._decode_base64_sample(segments[0])
|
|
|
|
def _read_exact_bytes(self, path: str) -> "tuple[Optional[bytes], Optional[ExecuteResult]]":
|
|
"""The file's bytes exactly, for the edit paths that write back every line they did not touch.
|
|
|
|
The text transport cannot carry them: it decodes with errors="replace", so a byte UTF-8 cannot
|
|
decode comes back as U+FFFD and the edit then persists it. A native read on the local POSIX host,
|
|
else base64 over the transport; ``(None, result)`` hands back the failed shell read for the
|
|
caller's message. Only a regular file gets a native open (a FIFO would block this thread); the
|
|
rest take the shell path and its timeout, as before."""
|
|
if self._native_read_enabled():
|
|
import stat as _stat
|
|
full = path if os.path.isabs(path) else os.path.join(
|
|
getattr(self.env, "cwd", None) or self.cwd, path)
|
|
try:
|
|
# One lookup, not two: a stat-then-open pair can have the path swapped for a FIFO in
|
|
# between, and that open blocks this thread forever (no backend timeout covers it).
|
|
# O_NONBLOCK returns a descriptor for a FIFO instead of waiting, and fstat judges THAT
|
|
# descriptor, so a non-regular file is rejected rather than read.
|
|
fd = os.open(full, os.O_RDONLY | getattr(os, "O_NONBLOCK", 0))
|
|
try:
|
|
if _stat.S_ISREG(os.fstat(fd).st_mode):
|
|
with open(fd, "rb", closefd=False) as fh:
|
|
return fh.read(), None
|
|
finally:
|
|
os.close(fd)
|
|
except OSError:
|
|
pass # missing/unreadable/would-block: the shell read below reports it the usual way
|
|
# Fenced like the compound read probe, and for the same reason: a backend whose merged
|
|
# stdout carries login-shell noise (a remote shell announcing TERM, a banner) would
|
|
# otherwise have it whitespace-joined onto the payload and decoded INTO the file's bytes,
|
|
# which the edit paths then write back. The file's own byte count travels beside it: output
|
|
# INSIDE the fence decodes too, so only a read that matches it is ever handed to a writer.
|
|
arg = self._escape_shell_arg(path)
|
|
segments, read_rc, result = self._fenced_read(f"base64 < {arg}", f"wc -c < {arg}")
|
|
garbled = ExecuteResult(stdout=f"{path}: the backend returned a garbled byte-exact read", exit_code=1)
|
|
if segments is None:
|
|
# No fenced reply: the command never ran as written (a wrapper ``cd`` failed, the backend
|
|
# refused it). Hand the backend's own text back so the caller reports what it said.
|
|
return None, result if result.exit_code != 0 else garbled
|
|
if read_rc is None:
|
|
return None, garbled
|
|
if read_rc == 127: # no base64 on this backend (busybox, distroless): try the hex transport
|
|
return self._read_exact_bytes_hex(path)
|
|
payload, size = segments
|
|
if read_rc != 0:
|
|
# stderr is merged into the fenced segment, so that segment holds base64's own diagnostic
|
|
# ("No such file or directory", "Permission denied"): keep it for the caller's message.
|
|
return None, self._failed_read(path, payload, read_rc)
|
|
data = self._decode_base64_sample(payload)
|
|
if data is None or not self._matches_size(data, size): # stray output: refuse, never echo it back
|
|
return None, garbled
|
|
return data, None
|
|
|
|
@staticmethod
|
|
def _failed_read(path: str, payload: str, read_rc: int) -> ExecuteResult:
|
|
"""The backend's own diagnostic for a read that ran and failed, else a bare exit status."""
|
|
return ExecuteResult(stdout=_strip_terminal_fence_leaks(payload).strip() or f"{path}: exit {read_rc}",
|
|
exit_code=read_rc)
|
|
|
|
def _read_exact_bytes_hex(self, path: str) -> "tuple[Optional[bytes], Optional[ExecuteResult]]":
|
|
"""``od`` fallback for a backend without ``base64``, fenced the same way.
|
|
|
|
``read_file_raw`` is the edit paths' source read AND, through ``_apply_add``, their
|
|
existence check, so a transport that simply is not installed must not read as "no such
|
|
file" — that clobbers the file the Add was refusing to overwrite. ``od`` is POSIX and
|
|
present in busybox; when it is missing too the caller gets a transport error, never a
|
|
not-found."""
|
|
arg = self._escape_shell_arg(path)
|
|
segments, read_rc, result = self._fenced_read(f"od -An -v -tx1 < {arg}", f"wc -c < {arg}")
|
|
unavailable = ExecuteResult(
|
|
stdout=f"{path}: this backend has neither base64 nor od, so a byte-exact read is unavailable",
|
|
exit_code=1)
|
|
if segments is None:
|
|
return None, result if result.exit_code != 0 else unavailable
|
|
if read_rc is None or read_rc == 127:
|
|
return None, unavailable
|
|
payload, size = segments
|
|
if read_rc != 0:
|
|
return None, self._failed_read(path, payload, read_rc)
|
|
try:
|
|
data = bytes.fromhex("".join(_strip_terminal_fence_leaks(payload).split()))
|
|
except ValueError:
|
|
data = None
|
|
if data is None or not self._matches_size(data, size):
|
|
return None, ExecuteResult(stdout=f"{path}: the backend returned a garbled byte-exact read",
|
|
exit_code=1)
|
|
return data, None
|
|
|
|
@staticmethod
|
|
def _decode_base64_sample(text: str) -> Optional[bytes]:
|
|
"""Decode one ``base64`` transport reply (a ``head -c N`` sample or a whole file). Whitespace-joins
|
|
the whole text first (``base64`` wraps at 76 columns), so callers hand over exactly one
|
|
segment; anything else fails validation → None."""
|
|
encoded = "".join(_strip_terminal_fence_leaks(text).split())
|
|
if not encoded:
|
|
return b""
|
|
if not re.fullmatch(r"[A-Za-z0-9+/]+={0,2}", encoded):
|
|
return None
|
|
try:
|
|
return base64.b64decode(encoded, validate=True)
|
|
except (binascii.Error, ValueError):
|
|
return None
|
|
|
|
@staticmethod
|
|
def _is_likely_binary_bytes(sample: bytes) -> bool:
|
|
"""Byte-layer binary detection: text iff valid UTF-8, allowing one incomplete
|
|
multibyte sequence at the very end (artifact of the byte-boundary cut).
|
|
NUL bytes or mid-stream invalid UTF-8 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.
|
|
|
|
See #80308.
|
|
"""
|
|
if not sample:
|
|
return False
|
|
if b"\x00" in sample:
|
|
return True
|
|
try:
|
|
sample.decode("utf-8")
|
|
return False
|
|
except UnicodeDecodeError as exc:
|
|
# UTF-8 sequences are at most 4 bytes: an error starting in the
|
|
# last 3 bytes with a clean prefix is a boundary cut, not binary.
|
|
if exc.start >= len(sample) - 3:
|
|
try:
|
|
sample[: exc.start].decode("utf-8")
|
|
return False
|
|
except UnicodeDecodeError:
|
|
pass
|
|
return True
|
|
|
|
def _is_likely_binary(self, path: str, content_sample: str = None) -> bool:
|
|
"""Legacy text-layer binary check: extension, else >30% non-printable chars."""
|
|
if has_binary_extension(path):
|
|
return True
|
|
if content_sample:
|
|
# 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')
|
|
return non_printable / min(len(content_sample), 1000) > 0.30
|
|
return False
|
|
|
|
def _is_image(self, path: str) -> bool:
|
|
return os.path.splitext(path)[1].lower() in IMAGE_EXTENSIONS
|
|
|
|
def _add_line_numbers(self, content: str, start_line: int = 1) -> str:
|
|
"""Prefix each line with a compact ``<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()
|
|
# A trailing newline terminates the final line — it does not start a new,
|
|
# empty one. Splitting without dropping it rendered a phantom "<N+1>|"
|
|
# gutter line on every newline-terminated file (`cat -n` semantics).
|
|
# Exactly ONE terminator is dropped, so a genuinely selected trailing
|
|
# blank line in a page keeps its own number.
|
|
if content.endswith('\n'):
|
|
content = content[:-1]
|
|
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
|
|
host's). A host path under the configured workspace mount is rewritten
|
|
to that container path first, so a Windows drive path is readable
|
|
inside Docker. Must run BEFORE shell escaping — ~ doesn't expand in quotes."""
|
|
from tools.terminal_tool_config import translate_mounted_host_path
|
|
host_root = getattr(self.env, "host_cwd", None)
|
|
container_root = getattr(self.env, "host_cwd_mount", None) or "/workspace"
|
|
translated = translate_mounted_host_path(path, host_root or "", container_root)
|
|
if translated:
|
|
return translated
|
|
if not path or not path.startswith('~'):
|
|
return path
|
|
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 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, translate_path: bool = True) -> 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.
|
|
|
|
``translate_path=False`` skips that translation for non-path values
|
|
such as regex patterns. Backslash compensation applies only to the
|
|
local Windows argv transport. Serialized shell text stays literal.
|
|
"""
|
|
from tools.environments.local import _IS_WINDOWS, _bash_safe_path
|
|
|
|
if translate_path:
|
|
arg = _bash_safe_path(arg)
|
|
elif _IS_WINDOWS and getattr(self.env, "is_local", False):
|
|
arg = arg.replace("\\", "\\\\")
|
|
# Use single quotes and escape any single quotes in the string
|
|
return "'" + arg.replace("'", "'\"'\"'") + "'"
|
|
|
|
def _escape_native_tool_arg(self, arg: str) -> str:
|
|
"""Quote a path for a NATIVE Windows binary (rg, node, git ...): those don't
|
|
understand the MSYS ``/c/...`` form and Hermes disables MSYS argument
|
|
conversion, so nothing translates it back (→ ``os error 3``). ``C:/Users/x``
|
|
is accepted by every layer. 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`` atomically: stdin → temp file in the SAME directory →
|
|
``mv -f`` (same-FS rename; cross-device ``mv`` is copy+unlink, NOT atomic).
|
|
``mkdir -p`` folded in. Exit 0 = swap happened; non-zero = original intact.
|
|
|
|
Symlink targets are resolved first (replacing the link would orphan the
|
|
target) and the temp dir recomputed from the RESOLVED target. 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 — not ``$(umask)``
|
|
arithmetic (zsh parses leading-zero constants as decimal), quoted so zsh
|
|
doesn't =word-expand. ``trap ... EXIT`` removes the temp on every failure.
|
|
"""
|
|
q_path = self._escape_shell_arg(path)
|
|
q_parent = self._escape_shell_arg(os.path.dirname(path) or ".")
|
|
tmpl = self._escape_shell_arg(".hermes-tmp.XXXXXX")
|
|
script = (
|
|
"set -e; "
|
|
# One shell script, fully quoted. Notes: - `mkdir -p "$d"` is folded in here so the parent
|
|
# directory is created in the same subprocess that writes the temp file — saves one entire
|
|
# subprocess spawn vs. a separate mkdir call. - `mktemp` lands the temp in the target's own dir
|
|
# (-p) so `mv` is same-FS atomic; we fall back to a PID-stamped name if the backend lacks mktemp
|
|
# (rare; busybox/macOS/Linux all ship it). - `chmod --reference` is GNU-only, so we read the
|
|
# octal mode with `stat` (GNU `-c%a` or BSD `-f%Lp`) and `chmod` it explicitly; silent
|
|
# best-effort — a perms-copy failure must not abort the write (the file then lands at mktemp's
|
|
# 0600, same as pre-fix). - brand-new targets get `chmod "=rw"` — the POSIX who-less symbolic
|
|
# form, which sets rw minus the process umask (e.g. 0644 under umask 022) instead of mktemp's
|
|
# hardcoded 0600 (#70856). Deliberately NOT shell arithmetic on `$(umask)`: zsh (reachable via
|
|
# _find_bash's $SHELL fallback) parses leading-zero constants as decimal and silently computes a
|
|
# garbage mode, while `chmod "=rw"` is spec-identical in bash/dash/ash/zsh and degrades to 0600
|
|
# (pre-fix behavior) if an exotic chmod rejects it. - `trap ... EXIT` guarantees the temp is
|
|
# removed on every error path (cat failure, mv failure, signal) but NOT after a successful mv
|
|
# (the temp no longer exists by then). - we `cat >` the temp, then `mv -f` it over the target.
|
|
f"d={q_parent}; t={q_path}; "
|
|
'if [ -L "$t" ]; then '
|
|
'rt="$(readlink -f "$t" 2>/dev/null || realpath "$t" 2>/dev/null || true)"; '
|
|
'[ -n "$rt" ] && { t="$rt"; d="$(dirname "$t")"; }; '
|
|
"fi; "
|
|
'mkdir -p "$d"; '
|
|
'tmp="$(mktemp -p "$d" ' + tmpl + ' 2>/dev/null '
|
|
'|| mktemp "$d/.hermes-tmp.$$.XXXXXX" 2>/dev/null '
|
|
'|| { tmp="$d/.hermes-tmp.$$"; : > "$tmp" && echo "$tmp"; })"; '
|
|
'[ -n "$tmp" ] || { echo "atomic write: could not create temp file" >&2; exit 1; }; '
|
|
"trap 'rm -f \\\"$tmp\\\"' EXIT; "
|
|
'if [ -e "$t" ]; then '
|
|
'm="$(stat -c%a "$t" 2>/dev/null || stat -f%Lp "$t" 2>/dev/null || true)"; '
|
|
'[ -n "$m" ] && chmod "$m" "$tmp" 2>/dev/null || true; '
|
|
"fi; "
|
|
'cat > "$tmp"; '
|
|
# new file: umask-default perms instead of mktemp's 0600 (#70856). Runs AFTER cat so a
|
|
# write-masking umask can't EACCES the stream; quoted "=rw" so zsh doesn't =word-expand it.
|
|
'if [ ! -e "$t" ]; then chmod "=rw" "$tmp" 2>/dev/null || true; fi; '
|
|
'mv -f "$tmp" "$t"; '
|
|
"trap - EXIT")
|
|
return self._exec(script, stdin_data=content)
|
|
|
|
def _file_has_bom(self, path: str, pre_content: Optional[str] = None) -> bool:
|
|
"""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(
|
|
old_content.splitlines(keepends=True), new_content.splitlines(keepends=True),
|
|
fromfile=f"a/{filename}", tofile=f"b/{filename}"))
|
|
|
|
# --- READ ---------------------------------------------------------------
|
|
|
|
@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, dangling symlink, "
|
|
"FIFO, socket, or device). Reading it could block indefinitely."))
|
|
|
|
def _probe_regular_file(self, path: str) -> tuple[int, str]:
|
|
"""Byte size of a REGULAR file: ``(file_size, status)`` with status ``"ok"``,
|
|
``"missing"``, ``"not_regular"``, ``"bad_size"`` (unparseable ``wc``),
|
|
``"env_unavailable"``, or the named working-directory error when the exec
|
|
wrapper itself failed (``_env_unavailable_error`` surfaces it verbatim).
|
|
``wc -c <`` on a writer-less FIFO/socket//dev/zero blocks forever and a
|
|
name-based blocklist can't cover a FIFO (a file TYPE at any path); ``[ -f ]``
|
|
is a stat (symlinks followed) so it answers without touching content. A dangling
|
|
symlink is ``not_regular``, never ``missing``: the entry exists, and a writer
|
|
told the path is free would follow the link and create its target."""
|
|
arg = self._escape_shell_arg(path)
|
|
# A missing path ECHOES its sentinel: a non-zero exit with no sentinel means the shell itself did
|
|
# not run (container still starting, removed out-of-band, transport down) — not a missing file.
|
|
# Reporting that as "File not found" made the model trust a false negative for the whole session.
|
|
stat_result = self._exec(
|
|
f"if [ -f {arg} ]; then wc -c < {arg} 2>/dev/null; "
|
|
f"elif [ -e {arg} ] || [ -L {arg} ]; then echo {NOT_REGULAR_SENTINEL}; "
|
|
f"else echo {MISSING_SENTINEL}; fi")
|
|
stat_output = _strip_terminal_fence_leaks(stat_result.stdout).strip()
|
|
if stat_output == MISSING_SENTINEL:
|
|
return 0, "missing"
|
|
if stat_output == NOT_REGULAR_SENTINEL:
|
|
return 0, "not_regular"
|
|
if stat_result.exit_code != 0:
|
|
return 0, stat_result.cwd_error or "env_unavailable"
|
|
try:
|
|
return int(stat_output), "ok"
|
|
except ValueError:
|
|
return 0, "bad_size"
|
|
|
|
def _env_unavailable_error(self, path: str, status: str = "env_unavailable") -> ReadResult:
|
|
if status != "env_unavailable":
|
|
return ReadResult(error=status)
|
|
return ReadResult(error=(f"Terminal environment unavailable: could not stat {path} "
|
|
"(the sandbox may still be starting or was removed). Retry shortly."))
|
|
|
|
def _detect_binary(self, path: str) -> tuple[bool, Optional[bytes]]:
|
|
"""``(is_binary, sample_bytes)`` — byte-layer detection when the transport
|
|
allows (base64 sample), else the legacy text heuristic (sample is None)."""
|
|
sample_bytes = self._sample_file_bytes(path)
|
|
if sample_bytes is not None:
|
|
ext_binary = has_binary_extension(path)
|
|
return ext_binary or self._is_likely_binary_bytes(sample_bytes), sample_bytes
|
|
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
|
|
# mixed Latin/CJK still detects): zeros at odd indices → UTF-16 LE, at even
|
|
# → BE; both parities or a single zero → real binary. Legacy 8-bit
|
|
# encodings (GBK, Big5) are never guessed — a wrong silent guess is worse
|
|
# than a clear refusal.
|
|
# UTF-16 rescue constants (ported from MoonshotAI/kimi-code#2647, detection derived from VS Code's
|
|
# encoding sniffer): sample the leading bytes; trust a BOM first, then a zero-byte parity heuristic —
|
|
# zeros clustering at odd indices mean UTF-16 LE (`0xAA 0x00`), at even indices UTF-16 BE (`0x00 0xAA`).
|
|
_UTF16_MAX_BYTES = 10 * 1024 * 1024
|
|
_UTF16_SAMPLE_BYTES = 512
|
|
|
|
def _python_interpreter_cmd(self) -> str:
|
|
"""Return a shell-safe Python interpreter for the terminal backend.
|
|
|
|
On the local backend ``sys.executable`` is always a working
|
|
interpreter — and on Windows it dodges the Microsoft Store
|
|
``python``/``python3`` alias stub (exit 49, "Python was not
|
|
found") that otherwise breaks inline ``-c`` snippets. Remote
|
|
backends (docker/ssh/...) don't have the agent's interpreter, so
|
|
they fall back to ``python3`` on their own PATH (the ``python``
|
|
fallback is handled at the call site).
|
|
"""
|
|
if self._lsp_local_only():
|
|
return self._escape_shell_arg(sys.executable)
|
|
return "python3"
|
|
|
|
def _exec_python_snippet(self, snippet: str, py: str = None) -> ExecuteResult:
|
|
"""Run a Python ``snippet`` in the terminal backend's interpreter.
|
|
|
|
Base64-encodes the snippet so it survives every shell/quoting layer
|
|
as pure ASCII: Windows ``subprocess`` list-arg quoting and ``bash``
|
|
double-quote processing both eat backslashes, which otherwise
|
|
corrupts Windows paths (``C:\\Users\\x``) and byte literals
|
|
(``b'\\xfe\\xff'``) embedded in a ``-c`` program. ``exec`` decodes
|
|
and runs it unchanged.
|
|
"""
|
|
encoded = base64.b64encode(snippet.encode("utf-8")).decode("ascii")
|
|
if py is None:
|
|
py = self._python_interpreter_cmd()
|
|
return self._exec(
|
|
f"{py} -c \"import base64; exec(base64.b64decode('{encoded}').decode())\""
|
|
)
|
|
|
|
def _try_read_utf16(self, path: str, offset: int, limit: int,
|
|
file_size: int) -> "Optional[ReadResult]":
|
|
"""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."""
|
|
if has_binary_extension(path) or file_size > self._UTF16_MAX_BYTES:
|
|
return None
|
|
snippet = (
|
|
"import sys, json, os\n"
|
|
f"p = {json.dumps(path)}\n"
|
|
f"offset = {int(offset)}\n"
|
|
f"limit = {int(limit)}\n"
|
|
f"MAX = {self._UTF16_MAX_BYTES}\n"
|
|
f"SAMPLE = {self._UTF16_SAMPLE_BYTES}\n"
|
|
"try:\n"
|
|
" size = os.path.getsize(p)\n"
|
|
" if size > MAX:\n"
|
|
" print('HERMES_UTF16:NO'); sys.exit(0)\n"
|
|
" with open(p, 'rb') as f:\n"
|
|
" data = f.read()\n"
|
|
" sample = data[:SAMPLE]\n"
|
|
" enc = None\n"
|
|
" if sample[:2] == b'\\xfe\\xff':\n"
|
|
" enc = 'utf-16-be'\n"
|
|
" elif sample[:2] == b'\\xff\\xfe':\n"
|
|
" enc = 'utf-16-le'\n"
|
|
" else:\n"
|
|
" odd = sum(1 for i in range(1, len(sample), 2) if sample[i] == 0)\n"
|
|
" even = sum(1 for i in range(0, len(sample), 2) if sample[i] == 0)\n"
|
|
" if even == 0 and odd >= 2:\n"
|
|
" enc = 'utf-16-le'\n"
|
|
" elif odd == 0 and even >= 2:\n"
|
|
" enc = 'utf-16-be'\n"
|
|
" if enc is None:\n"
|
|
" print('HERMES_UTF16:NO'); sys.exit(0)\n"
|
|
" text = data.decode(enc, 'replace')\n"
|
|
" if text[:1] == '\\ufeff':\n"
|
|
" text = text[1:]\n"
|
|
" text = text.replace('\\r\\n', '\\n')\n"
|
|
" lines = text.split('\\n')\n"
|
|
" total = len(lines)\n"
|
|
" sel = lines[offset - 1: offset - 1 + limit]\n"
|
|
" out = {'total_lines': total, 'encoding': enc,\n"
|
|
" 'content': '\\n'.join(sel)}\n"
|
|
" print('HERMES_UTF16:OK')\n"
|
|
" print(json.dumps(out, ensure_ascii=True))\n"
|
|
"except Exception:\n"
|
|
" print('HERMES_UTF16:NO'); sys.exit(0)\n"
|
|
)
|
|
|
|
result = self._exec_python_snippet(snippet)
|
|
if result.exit_code != 0 and "python3" in (result.stdout or ""):
|
|
result = self._exec_python_snippet(snippet, py="python")
|
|
|
|
stdout = _strip_terminal_fence_leaks(result.stdout or "")
|
|
marker = stdout.find("HERMES_UTF16:OK")
|
|
if result.exit_code != 0 or marker < 0:
|
|
return None
|
|
payload = stdout[marker + len("HERMES_UTF16:OK"):].strip()
|
|
try:
|
|
data = json.loads(payload.split("\n", 1)[0] if "\n" in payload else payload)
|
|
content = data["content"]
|
|
total_lines = int(data["total_lines"])
|
|
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. "
|
|
"Text edits via patch/write_file would re-encode as UTF-8."]
|
|
if truncated:
|
|
hint_parts.append(
|
|
f"Use offset={end_line + 1} to continue reading "
|
|
f"(showing {offset}-{end_line} of {total_lines} lines)")
|
|
from tools.tool_output_limits import get_max_line_length
|
|
max_line_length = get_max_line_length()
|
|
truncated_lines = any(len(line) > max_line_length for line in content.split('\n'))
|
|
return ReadResult(
|
|
content=self._add_line_numbers(content, offset), total_lines=total_lines,
|
|
file_size=file_size, truncated=truncated, hint=" ".join(hint_parts),
|
|
truncated_lines=True if truncated_lines else None)
|
|
|
|
def read_file(self, path: str, offset: int = 1, limit: int = 2000) -> ReadResult:
|
|
"""Read a file with pagination, binary detection, and line numbers.
|
|
|
|
``offset`` is 1-indexed; ``limit`` is clamped by ``normalize_read_pagination``.
|
|
One shell round-trip answers every question the read needs (existence, size,
|
|
binary sample, page, line count, trailing newline; see ``_read_probe_cmd``).
|
|
An unparseable reply falls back to ``_read_file_sequential`` (one probe per
|
|
question), so an exotic shell can never do worse than before. On a local
|
|
POSIX environment the read never touches the shell (``_read_file_native``).
|
|
"""
|
|
path = self._expand_path(path) # before shell escaping: ~ doesn't expand in quotes
|
|
offset, limit = normalize_read_pagination(offset, limit)
|
|
|
|
if self._native_read_enabled():
|
|
return self._read_file_native(path, offset, limit)
|
|
|
|
# Images / known-binary extensions never inline content; the sequential
|
|
# path stops at the probes for them, so don't stream their bytes.
|
|
if self._is_image(path) or has_binary_extension(path):
|
|
return self._read_file_sequential(path, offset, limit)
|
|
|
|
from tools.tool_output_limits import get_max_line_length
|
|
line_clamp_bytes = 4 * get_max_line_length() + 1
|
|
end_line = offset + limit - 1
|
|
sentinel = _new_sentinel(_READ_SENTINEL_PREFIX)
|
|
probe = self._exec(self._read_probe_cmd(path, offset, end_line, line_clamp_bytes, sentinel))
|
|
output = probe.stdout or ""
|
|
|
|
if sentinel not in output:
|
|
# Single-line replies: the path is missing or not a regular file.
|
|
marker = _strip_terminal_fence_leaks(output).strip()
|
|
if marker == MISSING_SENTINEL:
|
|
return self._read_file_missing(path, offset, limit)
|
|
if marker == NOT_REGULAR_SENTINEL:
|
|
return self._not_regular_error(path)
|
|
logger.debug(
|
|
"read_file: compound probe reply for %s has no sentinel "
|
|
"(exit %s, %d chars); falling back to sequential probes",
|
|
path, probe.exit_code, len(output))
|
|
return self._read_file_sequential(path, offset, limit)
|
|
|
|
segments = _split_segments(output, sentinel)
|
|
if probe.exit_code != 0 or len(segments) != 6:
|
|
logger.debug(
|
|
"read_file: compound probe for %s returned exit %s with %d "
|
|
"segments (want 6); falling back to sequential probes",
|
|
path, probe.exit_code, len(segments))
|
|
return self._read_file_sequential(path, offset, limit)
|
|
size_seg, sample_seg, page_seg, wc_seg, tail_seg, status_seg = segments
|
|
|
|
status = _strip_terminal_fence_leaks(status_seg).split()
|
|
try:
|
|
sample_rc, read_rc = int(status[0]), int(status[1])
|
|
except (IndexError, ValueError):
|
|
logger.debug(
|
|
"read_file: compound probe for %s has unparseable status %r; "
|
|
"falling back to sequential probes", path, status_seg[-40:])
|
|
return self._read_file_sequential(path, offset, limit)
|
|
|
|
try:
|
|
file_size = int(_strip_terminal_fence_leaks(size_seg).strip())
|
|
except ValueError:
|
|
file_size = 0
|
|
|
|
# Byte-layer binary detection when base64 was available, else the legacy
|
|
# text heuristic over a plain sample (one extra round-trip, shells without base64).
|
|
sample_bytes = self._decode_base64_sample(sample_seg) if sample_rc == 0 else None
|
|
if sample_bytes is not None:
|
|
is_binary = self._is_likely_binary_bytes(sample_bytes)
|
|
else:
|
|
logger.debug(
|
|
"read_file: no usable base64 sample for %s (base64 exit %s); "
|
|
"paying one extra round-trip for the text heuristic", path, sample_rc)
|
|
sample_output = _strip_terminal_fence_leaks(self._head(path, 1000).stdout)
|
|
is_binary = self._is_likely_binary(path, sample_output)
|
|
if is_binary:
|
|
return self._read_binary_file(path, offset, limit, file_size, sample_bytes)
|
|
|
|
if read_rc != 0:
|
|
return ReadResult(error=f"Failed to read file: {_strip_terminal_fence_leaks(page_seg)}")
|
|
read_output = _strip_terminal_fence_leaks(page_seg)
|
|
try:
|
|
total_lines = int(_strip_terminal_fence_leaks(wc_seg).strip())
|
|
except ValueError:
|
|
total_lines = 0
|
|
tail_flag = _strip_terminal_fence_leaks(tail_seg).strip()
|
|
file_ends_with_newline = tail_flag == "1" if tail_flag in ("0", "1") else None
|
|
return self._assemble_read_result(
|
|
read_output, offset=offset, end_line=end_line, total_lines=total_lines,
|
|
file_size=file_size, file_ends_with_newline=file_ends_with_newline)
|
|
|
|
def _native_read_enabled(self) -> bool:
|
|
"""Whether ``read_file`` and ``search_files`` may bypass the shell: only POSIX + ``LocalEnvironment``
|
|
(file is on this host, path already native; Windows keeps the shell path since
|
|
file_operations holds Git-Bash-style paths there). ``HERMES_NATIVE_FILE_READ=0``
|
|
turns the fast path off."""
|
|
flag = os.environ.get("HERMES_NATIVE_FILE_READ", "1").strip().lower()
|
|
if flag in ("0", "false", "no", "off"):
|
|
return False
|
|
# Same "is this env the local host" test the LSP path uses; isinstance is
|
|
# microseconds and self.env is never rebound, so nothing to memoize.
|
|
return sys.platform != "win32" and self._lsp_local_only()
|
|
|
|
def _read_file_native(self, path: str, offset: int, limit: int) -> ReadResult:
|
|
"""``read_file`` without a shell — same contract as the shell path, byte for
|
|
byte. ``os.stat`` is the ``[ -f ]`` guard (a stat, never an open, so FIFOs and
|
|
devices are refused before anything touches them); the first 1000 bytes drive
|
|
the byte-layer binary check; the page is produced exactly as
|
|
``sed -n 'a,bp' | cut -b1-N`` prints it (each line clamped to N bytes and
|
|
newline-terminated) then decoded with errors="replace" like the transport.
|
|
One chunked pass counts lines and collects the page, so neither the file nor
|
|
a pathological line is ever held whole. ``path`` is already expanded; any
|
|
unexpected OSError hands over to the shell path."""
|
|
import stat as _stat
|
|
|
|
full = path if os.path.isabs(path) else os.path.join(
|
|
getattr(self.env, "cwd", None) or self.cwd, path)
|
|
try:
|
|
st = os.stat(full)
|
|
except (FileNotFoundError, NotADirectoryError):
|
|
if os.path.islink(full): # dangling: an entry, not an absent path (``_probe_regular_file``)
|
|
return self._not_regular_error(path)
|
|
return self._read_file_missing(path, offset, limit)
|
|
except OSError:
|
|
return self._read_file_sequential(path, offset, limit)
|
|
if not _stat.S_ISREG(st.st_mode):
|
|
return self._not_regular_error(path)
|
|
file_size = st.st_size
|
|
if self._is_image(path):
|
|
return self._image_redirect_result(file_size)
|
|
|
|
from tools.tool_output_limits import get_max_line_length
|
|
clamp = 4 * get_max_line_length() + 1
|
|
end_line = offset + limit - 1
|
|
|
|
page: list[bytes] = []
|
|
total_lines = 0
|
|
lineno = 1 # the line currently being scanned
|
|
kept = bytearray() # first ``clamp`` bytes of that line
|
|
have_partial = False # that line has bytes but no newline yet
|
|
last_byte = b""
|
|
digest = hashlib.sha256()
|
|
try:
|
|
with open(full, "rb") as fh:
|
|
sample = fh.read(1000)
|
|
ext_binary = has_binary_extension(path)
|
|
if ext_binary or self._is_likely_binary_bytes(sample):
|
|
return self._read_binary_file(path, offset, limit, file_size, sample)
|
|
fh.seek(0)
|
|
while True:
|
|
chunk = fh.read(1 << 20)
|
|
if not chunk:
|
|
break
|
|
digest.update(chunk)
|
|
last_byte = chunk[-1:]
|
|
if lineno > end_line:
|
|
# Past the window: only the line count and trailing byte
|
|
# are needed, so let memchr do it instead of per-line work.
|
|
total_lines += chunk.count(b"\n")
|
|
have_partial = chunk[-1:] != b"\n"
|
|
continue
|
|
pos, n = 0, len(chunk)
|
|
while pos < n:
|
|
nl = chunk.find(b"\n", pos)
|
|
in_page = offset <= lineno <= end_line
|
|
if nl < 0:
|
|
if in_page and len(kept) < clamp:
|
|
kept += chunk[pos:pos + (clamp - len(kept))]
|
|
have_partial = True
|
|
break
|
|
if in_page:
|
|
if len(kept) < clamp:
|
|
kept += chunk[pos:min(nl, pos + (clamp - len(kept)))]
|
|
page.append(bytes(kept) + b"\n")
|
|
kept = bytearray()
|
|
have_partial = False
|
|
total_lines += 1
|
|
lineno += 1
|
|
pos = nl + 1
|
|
except OSError:
|
|
return self._read_file_sequential(path, offset, limit)
|
|
if have_partial and offset <= lineno <= end_line:
|
|
# ``sed`` prints a final line that lacks a newline; ``cut`` adds one.
|
|
page.append(bytes(kept) + b"\n")
|
|
|
|
read_output = _strip_terminal_fence_leaks(b"".join(page).decode("utf-8", errors="replace"))
|
|
result = self._assemble_read_result(
|
|
read_output, offset=offset, end_line=end_line, total_lines=total_lines,
|
|
file_size=file_size,
|
|
file_ends_with_newline=(last_byte == b"\n") if file_size else None)
|
|
result._snapshot = (st.st_dev, st.st_ino, st.st_size, st.st_mtime_ns, st.st_ctime_ns, digest.digest())
|
|
return result
|
|
|
|
@staticmethod
|
|
def _image_redirect_result(file_size: int) -> ReadResult:
|
|
return ReadResult(
|
|
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."))
|
|
|
|
def _read_probe_cmd(self, path: str, offset: int, end_line: int,
|
|
line_clamp_bytes: int, sentinel: str) -> str:
|
|
"""One shell command answering every question ``read_file`` asks: six
|
|
segments each closed by a ``sentinel`` line — byte size, base64 of the first
|
|
1000 bytes, the ``sed | cut`` page, ``wc -l``, whether the last byte is a
|
|
newline, then the base64 and page pipeline statuses. Probes run only inside
|
|
``[ -f ]`` (stat-not-open, like ``_probe_regular_file``) so a FIFO/device never
|
|
reaches ``head``/``sed``. A missing path echoes ``MISSING_SENTINEL`` (a compound
|
|
command only reports its last status). Every stage silences stderr: the local
|
|
backend merges stderr into stdout and a stray diagnostic would land inside a
|
|
segment. The byte clamp is ``4 * max_line_length + 1``; see ``_read_file_sequential``."""
|
|
arg = self._escape_shell_arg(path)
|
|
mark = f"echo {sentinel}"
|
|
return (
|
|
f"if [ -f {arg} ]; then "
|
|
f"wc -c < {arg} 2>/dev/null; {mark}; "
|
|
f"head -c 1000 {arg} 2>/dev/null | base64 2>/dev/null; __hs=$?; {mark}; "
|
|
f"sed -n '{offset},{end_line}p' {arg} 2>/dev/null"
|
|
f" | cut -b1-{line_clamp_bytes} 2>/dev/null; __hr=$?; {mark}; "
|
|
f"wc -l < {arg} 2>/dev/null; {mark}; "
|
|
f"tail -c 1 {arg} 2>/dev/null | wc -l; {mark}; "
|
|
f'echo "$__hs $__hr"; '
|
|
f"elif [ -e {arg} ] || [ -L {arg} ]; then echo {NOT_REGULAR_SENTINEL}; "
|
|
f"else echo {MISSING_SENTINEL}; fi")
|
|
|
|
def _read_file_missing(self, path: str, offset: int, limit: int) -> ReadResult:
|
|
"""Not-found recovery shared by every read path. Unicode-equivalent spellings
|
|
(NFC/NFD, confusable spaces/quotes) render identically, so the model can never
|
|
discover the byte mismatch by retyping — retrying is the tool's job. No
|
|
equivalent spelling → suggest similar files."""
|
|
variant = self._unicode_variant_match(path)
|
|
if variant is not None:
|
|
result = self.read_file(variant, offset=offset, limit=limit)
|
|
note = (
|
|
f"Note: '{path}' not found byte-for-byte; resolved to "
|
|
f"the unicode-equivalent file '{variant}' (invisible "
|
|
"encoding difference: NFC/NFD or special space/quote "
|
|
"characters).")
|
|
result.hint = f"{note} {result.hint}" if result.hint else note
|
|
return result
|
|
return self._suggest_similar_files(path)
|
|
|
|
def _read_binary_file(self, path: str, offset: int, limit: int,
|
|
file_size: int, sample_bytes: Optional[bytes]) -> ReadResult:
|
|
"""Binary branch shared by every read path: UTF-16 text (Notepad, PowerShell
|
|
``>``) trips the binary guard; transcode it, else refuse with the type name.
|
|
|
|
UTF-16 rescue (ported from MoonshotAI/kimi-code#2647): the terminal env decodes stdout as UTF-8 with
|
|
errors="replace", so a UTF-16 text file (Windows Notepad .txt, PowerShell `>` redirects) arrives
|
|
mangled with U+FFFD and trips the binary guard. Probe the raw bytes via the backend's Python and
|
|
transcode to UTF-8 when a BOM or the zero-byte parity heuristic identifies UTF-16.
|
|
"""
|
|
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,
|
|
error=describe_binary_file(sample_bytes, file_size))
|
|
|
|
def _read_file_sequential(self, path: str, offset: int, limit: int) -> ReadResult:
|
|
"""One-probe-per-call read: the pre-compound form, kept as fallback for
|
|
image / known-binary extensions and unparseable compound replies. ``path`` is
|
|
already expanded and ``offset``/``limit`` normalized."""
|
|
file_size, status = self._probe_regular_file(path)
|
|
if status == "missing":
|
|
return self._read_file_missing(path, offset, limit)
|
|
if status == "not_regular":
|
|
return self._not_regular_error(path)
|
|
if status not in ("ok", "bad_size"):
|
|
return self._env_unavailable_error(path, status)
|
|
if self._is_image(path): # never inlined — redirect to the vision tool
|
|
return self._image_redirect_result(file_size)
|
|
is_binary, sample_bytes = self._detect_binary(path)
|
|
if is_binary:
|
|
return self._read_binary_file(path, offset, limit, file_size, sample_bytes)
|
|
|
|
# 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
|
|
read_result = self._exec(
|
|
f"sed -n '{offset},{end_line}p' {self._escape_shell_arg(path)}"
|
|
f" | cut -b1-{line_clamp_bytes}")
|
|
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)
|
|
|
|
wc_result = self._exec(f"wc -l < {self._escape_shell_arg(path)}")
|
|
try:
|
|
total_lines = int(_strip_terminal_fence_leaks(wc_result.stdout).strip())
|
|
except ValueError:
|
|
total_lines = 0
|
|
|
|
# Only the page reaching the file's final line can carry the ``cut`` newline
|
|
# artifact (see _assemble_read_result); probe the last byte just for that case.
|
|
file_ends_with_newline: Optional[bool] = None
|
|
if not total_lines > end_line 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:
|
|
file_ends_with_newline = _strip_terminal_fence_leaks(tail_result.stdout).strip() != "0"
|
|
return self._assemble_read_result(
|
|
read_output, offset=offset, end_line=end_line, total_lines=total_lines,
|
|
file_size=file_size, file_ends_with_newline=file_ends_with_newline)
|
|
|
|
def _assemble_read_result(self, read_output: str, *, offset: int, end_line: int,
|
|
total_lines: int, file_size: int,
|
|
file_ends_with_newline: Optional[bool]) -> ReadResult:
|
|
"""Turn a raw ``sed | cut`` page into the final ``ReadResult``. Shared by every
|
|
read path so the BOM strip, pagination hint, ``cut`` newline-artifact fix and
|
|
the ambiguous-silence guards never drift apart. ``file_ends_with_newline`` is
|
|
None when the caller could not tell (artifact left alone, as before)."""
|
|
# ``wc -l`` counts newlines, not lines: a nonempty file whose last byte
|
|
# is not a newline holds one more line than the count (#3907). Adjust
|
|
# here — the single choke point — so total_lines, truncation, and the
|
|
# past-EOF guard agree on every read path (compound, sequential, native).
|
|
if file_size > 0 and file_ends_with_newline is False:
|
|
total_lines += 1
|
|
if offset == 1: # only the first chunk can carry a BOM (byte 0)
|
|
read_output, _ = _strip_bom(read_output)
|
|
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`` always newline-terminates, so a file without a trailing newline
|
|
# would grow a phantom empty last line; strip it when the last byte says so.
|
|
if not truncated and read_output.endswith('\n') and file_ends_with_newline is False:
|
|
read_output = read_output[:-1]
|
|
|
|
# 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:
|
|
return ReadResult(
|
|
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}."))
|
|
from tools.tool_output_limits import get_max_line_length
|
|
max_line_length = get_max_line_length()
|
|
truncated_lines = any(len(line) > max_line_length for line in read_output.split('\n'))
|
|
return ReadResult(
|
|
content=self._add_line_numbers(read_output, offset), total_lines=total_lines,
|
|
file_size=file_size, truncated=truncated, hint=hint,
|
|
truncated_lines=True if truncated_lines else None)
|
|
|
|
# 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
|
|
)
|
|
|
|
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:
|
|
out = unicodedata.normalize("NFC", name)
|
|
for src, dst in self._CONFUSABLES:
|
|
out = out.replace(src, dst)
|
|
return out
|
|
|
|
target = _canon(filename)
|
|
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()
|
|
if entry and entry != filename and _canon(entry) == target]
|
|
if len(candidates) == 1:
|
|
return os.path.join(dir_path, candidates[0]) if dir_path != "." or "/" in path else candidates[0]
|
|
return None
|
|
|
|
def _suggest_similar_files(self, path: str) -> ReadResult:
|
|
""""File not found" result listing up to 5 similar names from the same directory."""
|
|
dir_path = os.path.dirname(path) or "."
|
|
filename = os.path.basename(path)
|
|
basename_no_ext = os.path.splitext(filename)[0].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'):
|
|
if not f:
|
|
continue
|
|
lf = f.lower()
|
|
score = 0
|
|
if lf == lower_name:
|
|
score = 100
|
|
elif os.path.splitext(f)[0].lower() == basename_no_ext: # config.yml vs config.yaml
|
|
score = 90
|
|
elif lf.startswith(lower_name) or lower_name.startswith(lf):
|
|
score = 70
|
|
elif lower_name in lf:
|
|
score = 60
|
|
elif lf in lower_name and len(lf) > 2:
|
|
score = 40
|
|
elif ext and os.path.splitext(f)[1].lower() == ext:
|
|
common = set(lower_name) & set(lf)
|
|
if len(common) >= max(len(lower_name), len(lf)) * 0.4:
|
|
score = 30
|
|
# Near-miss spelling (AGENT.md -> AGENTS.md) 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}", not_found=True,
|
|
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)
|
|
file_size, status = self._probe_regular_file(path)
|
|
if status == "missing":
|
|
return self._suggest_similar_files(path)
|
|
if status == "not_regular":
|
|
return self._not_regular_error(path)
|
|
if status not in ("ok", "bad_size"):
|
|
return self._env_unavailable_error(path, status)
|
|
if self._is_image(path):
|
|
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))
|
|
data, failed = self._read_exact_bytes(path)
|
|
if data is None:
|
|
return ReadResult(error=f"Failed to read file: {failed.stdout}")
|
|
# V4A writes this back, so no display cleanup (nothing has emitted the __HERMES_FENCE_ wrapper it
|
|
# targets since d684d7ee7e; it can only eat the file's own escape bytes), and surrogateescape
|
|
# so write_file's encode restores any byte past the sample that UTF-8 cannot decode (#79178).
|
|
# 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(data.decode("utf-8", "surrogateescape"))
|
|
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 (as base64) from any shell-backed environment."""
|
|
path = self._expand_path(path)
|
|
file_size, status = self._probe_regular_file(path)
|
|
if status == "missing":
|
|
return ReadResult(error=f"File not found: {path}", not_found=True)
|
|
if status == "not_regular":
|
|
return self._not_regular_error(path)
|
|
if status not in ("ok", "bad_size"):
|
|
return self._env_unavailable_error(path, status)
|
|
if status == "bad_size":
|
|
return ReadResult(error=f"Could not determine file size: {path}")
|
|
if max_bytes is not None and file_size > max_bytes:
|
|
return ReadResult(
|
|
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}")
|
|
compact = "".join(_strip_terminal_fence_leaks(encoded.stdout).split())
|
|
try:
|
|
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)
|
|
|
|
def delete_file(self, path: str) -> WriteResult:
|
|
"""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)
|
|
# Delete removes the directory entry (a symlink itself, not its target), so
|
|
# the guards vet the entry as well as the target it resolves to.
|
|
denied = get_write_denied_error(path, verb="Delete", entry=True)
|
|
if denied:
|
|
return WriteResult(error=denied)
|
|
# 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({json.dumps(path)})\n"
|
|
"recursive = False\n"
|
|
"try:\n"
|
|
" if p.is_dir() and not p.is_symlink():\n"
|
|
" if recursive:\n"
|
|
" shutil.rmtree(p)\n"
|
|
" else:\n"
|
|
" print('is a directory: ' + str(p), file=sys.stderr); sys.exit(2)\n"
|
|
" else:\n"
|
|
" 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_python_snippet(snippet)
|
|
|
|
# Fall back to ``python`` (remote backends / older systems where there's no
|
|
# ``python3`` symlink but a ``python`` binary is on PATH).
|
|
if result.exit_code != 0 and "python3" in (result.stdout or ""):
|
|
result = self._exec_python_snippet(snippet, py="python")
|
|
|
|
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:
|
|
src = self._expand_path(src)
|
|
dst = self._expand_path(dst)
|
|
# Entry-level op like delete_file: vet both entries, not just their targets.
|
|
for p in (src, dst):
|
|
denied = get_write_denied_error(p, verb="Move", entry=True)
|
|
if denied:
|
|
return WriteResult(error=denied)
|
|
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()
|
|
|
|
# --- WRITE --------------------------------------------------------------
|
|
|
|
# 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: 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 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 is a corrupt write, so refuse
|
|
before any bytes touch disk. Checked against the RAW content, before the
|
|
BOM/CRLF shims (post-shim linting would false-positive on a BOM-marked file)."""
|
|
linter = LINTERS_INPROC.get(ext) if ext in _FAIL_CLOSED_INPROC_EXTS else None
|
|
if linter is None:
|
|
return None
|
|
ok, err = linter(content)
|
|
if ok or err == "__SKIP__":
|
|
return None
|
|
return WriteResult(error=(
|
|
f"Refusing to write '{path}': candidate content fails "
|
|
f"{ext} syntax validation ({err}). The file was "
|
|
"NOT created or modified. Fix the content and retry."))
|
|
|
|
def _write_probe_cmd(self, path: str, sentinel: str, body: Optional[str]) -> str:
|
|
"""One shell command for the on-disk questions ``write_file`` asks. Two
|
|
segments closed by a ``sentinel`` line: base64 of the first three bytes (BOM
|
|
detection at the byte layer, same on-disk truth as ``_file_has_bom``), then
|
|
``body``: ``"cat"`` for the full text when pre-content is wanted, ``"sample"``
|
|
for the 4 KB line-ending sample, or None for nothing. Gated on ``[ -f ]`` so a
|
|
FIFO/device never reaches ``head``/``cat``; a missing path echoes ``MISSING_SENTINEL``."""
|
|
arg = self._escape_shell_arg(path)
|
|
if body == "cat":
|
|
body_cmd = f"cat {arg} 2>/dev/null"
|
|
elif body == "sample":
|
|
body_cmd = f"head -c 4096 {arg} 2>/dev/null"
|
|
else:
|
|
body_cmd = ":"
|
|
return (
|
|
f"if [ -f {arg} ]; then "
|
|
f"head -c 3 {arg} 2>/dev/null | base64 2>/dev/null; echo {sentinel}; "
|
|
f"{body_cmd}; "
|
|
f"else echo {MISSING_SENTINEL}; fi")
|
|
|
|
def _probe_write_target(self, path: str, pre_content: Optional[str], want_pre: bool,
|
|
) -> tuple[bool, Optional[str], Optional[str]]:
|
|
"""``(has_bom, pre_content, original_line_ending)`` for ``path`` in ONE
|
|
round-trip (replaces ``cat`` when pre-content is wanted, a ``head -c 4096``
|
|
line-ending sample and a ``head -c 3`` BOM check). Semantics unchanged:
|
|
pre-content is read only when wanted and not supplied; the line ending comes
|
|
from pre-content when there is any, else from the sample; the BOM always comes
|
|
from disk. An unparseable reply falls back to the separate probes."""
|
|
if want_pre and pre_content is None:
|
|
body_mode: Optional[str] = "cat"
|
|
elif not pre_content:
|
|
body_mode = "sample"
|
|
else:
|
|
body_mode = None
|
|
|
|
sentinel = _new_sentinel(_WRITE_SENTINEL_PREFIX)
|
|
probe = self._exec(self._write_probe_cmd(path, sentinel, body_mode))
|
|
output = probe.stdout or ""
|
|
if sentinel not in output:
|
|
if _strip_terminal_fence_leaks(output).strip() == MISSING_SENTINEL:
|
|
ending = _detect_line_ending(pre_content) if pre_content else None
|
|
return False, pre_content, ending
|
|
logger.debug(
|
|
"write_file: pre-write probe reply for %s has no sentinel "
|
|
"(exit %s, %d chars); falling back to sequential probes",
|
|
path, probe.exit_code, len(output))
|
|
return self._probe_write_target_sequential(path, pre_content, want_pre)
|
|
|
|
segments = _split_segments(output, sentinel)
|
|
if probe.exit_code != 0 or len(segments) != 2:
|
|
logger.debug(
|
|
"write_file: pre-write probe for %s returned exit %s with %d "
|
|
"segments (want 2); falling back to sequential probes",
|
|
path, probe.exit_code, len(segments))
|
|
return self._probe_write_target_sequential(path, pre_content, want_pre)
|
|
head_seg, body = segments
|
|
|
|
head_bytes = self._decode_base64_sample(head_seg)
|
|
if head_bytes is None:
|
|
# No clean base64 on this shell; ask the way we used to.
|
|
logger.debug(
|
|
"write_file: no usable base64 head for %s; paying one extra "
|
|
"round-trip for the BOM probe", path)
|
|
has_bom = self._file_has_bom(path, pre_content)
|
|
else:
|
|
has_bom = head_bytes.startswith(_UTF8_BOM.encode("utf-8"))
|
|
|
|
if body_mode == "cat" and body:
|
|
pre_content = body
|
|
if pre_content:
|
|
ending = _detect_line_ending(pre_content)
|
|
elif body_mode == "sample" and body:
|
|
ending = _detect_line_ending(body)
|
|
else:
|
|
ending = None
|
|
return has_bom, pre_content, ending
|
|
|
|
def _probe_write_target_sequential(self, path: str, pre_content: Optional[str], want_pre: bool,
|
|
) -> tuple[bool, Optional[str], Optional[str]]:
|
|
"""Pre-compound form of ``_probe_write_target``: one exec per question. A
|
|
failed ``cat`` leaves pre_content None so the lint-delta and LSP consumers
|
|
degrade gracefully."""
|
|
if want_pre and pre_content is None:
|
|
read_result = self._cat(path)
|
|
if read_result.exit_code == 0 and read_result.stdout:
|
|
pre_content = read_result.stdout
|
|
if pre_content:
|
|
ending = _detect_line_ending(pre_content)
|
|
else:
|
|
head = self._head(path, 4096)
|
|
ending = _detect_line_ending(head.stdout) if head.exit_code == 0 and head.stdout else None
|
|
return self._file_has_bom(path, pre_content), pre_content, ending
|
|
|
|
def _verify_written_hash(self, path: str, content_bytes: bytes) -> tuple[Optional[bool], Optional[WriteResult]]:
|
|
"""Compare the on-disk sha256 to the intended bytes: ``(verified, error)``.
|
|
The explicit flag saves the model a confirming re-read; a mismatch is a hard
|
|
error. ``verified`` is None when the hash could not be taken."""
|
|
try:
|
|
hash_result = self._exec(f"sha256sum {self._escape_shell_arg(path)} 2>/dev/null")
|
|
if hash_result.exit_code == 0 and hash_result.stdout.strip():
|
|
disk_sha = hash_result.stdout.strip().split()[0]
|
|
if disk_sha != hashlib.sha256(content_bytes).hexdigest():
|
|
return False, WriteResult(error=(
|
|
f"Post-write verification failed for {path}: on-disk "
|
|
"content hash differs from the intended write. The "
|
|
"write did not persist correctly — re-read the file "
|
|
"and retry."))
|
|
return True, None
|
|
except Exception:
|
|
pass
|
|
return None, None
|
|
|
|
def write_file(self, path: str, content: str, pre_content: Optional[str] = None) -> WriteResult:
|
|
"""Write content atomically, creating parent directories as needed.
|
|
|
|
Order: deny list → lone-surrogate refusal → fail-closed syntax gate on the
|
|
CANDIDATE content (JSON/YAML/TOML) → one compound on-disk probe
|
|
(pre-content when wanted, CRLF, BOM; see ``_probe_write_target``) →
|
|
CRLF/BOM preservation → LSP baseline snapshot → atomic write (content rides
|
|
stdin: no ARG_MAX limit) → sha256 verification → lint delta → LSP
|
|
diagnostics when syntax is clean. ``pre_content``: pre-edit content the
|
|
caller already has (skips the read); BOM detection always probes disk.
|
|
"""
|
|
path = self._expand_path(path)
|
|
denied = get_write_denied_error(path)
|
|
if denied:
|
|
return WriteResult(error=denied)
|
|
refused = self._reject_unencodable(path, content)
|
|
if refused is not None:
|
|
return refused
|
|
ext = os.path.splitext(path)[1].lower()
|
|
refused = self._fail_closed_syntax_error(path, ext, content)
|
|
if refused is not None:
|
|
return refused
|
|
|
|
# Pre-content is read only for extensions in the UNION of in-process lint and
|
|
# LSP coverage (keeps the hot path fast for binaries).
|
|
want_pre = ext in LINTERS_INPROC or self._lsp_handles_extension(ext)
|
|
has_bom, pre_content, original_ending = self._probe_write_target(path, pre_content, want_pre)
|
|
# read_file strips the BOM and models send bare-LF text, so a round-trip would
|
|
# otherwise normalize CRLF files and drop the BOM (prepend only when absent).
|
|
if original_ending == "\r\n":
|
|
content = _normalize_line_endings(content, "\r\n")
|
|
if has_bom and not _has_bom(content):
|
|
content = _UTF8_BOM + content
|
|
# Best-effort snapshot so the LSP tier reports only this edit's diagnostics.
|
|
self._snapshot_lsp_baseline(path)
|
|
# ``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))
|
|
# 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:
|
|
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)
|
|
# 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
|
|
return WriteResult(
|
|
bytes_written=len(content_bytes), dirs_created=dirs_created, verified=content_verified,
|
|
_content_sha256=hashlib.sha256(content_bytes).hexdigest(),
|
|
lint=lint_result.to_dict() if lint_result else None, lsp_diagnostics=lsp_diagnostics)
|
|
|
|
# --- 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:
|
|
"""PatchResult for a failed fuzzy match. Already-applied detection first: the
|
|
most common production failure is a re-send of an edit that already landed,
|
|
and 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,
|
|
note=(
|
|
f"File already contains the target text — the edit "
|
|
f"appears to be already applied to {path}. No write "
|
|
"performed; do not re-send this patch."))
|
|
err_msg = error or f"Could not find match for old_string in {path}"
|
|
try:
|
|
err_msg += format_no_match_hint(err_msg, match_count, old_string, content)
|
|
except Exception:
|
|
pass
|
|
return PatchResult(error=err_msg)
|
|
|
|
def _verify_patch_persisted(self, path: str, new_content: str) -> Optional[PatchResult]:
|
|
"""Re-read ``path`` and confirm the intended bytes landed; error result or None.
|
|
Catches silent persistence failures (FS oddities, races, truncated pipe).
|
|
Line endings are normalized first (Windows text-mode ``open()`` writes LF as
|
|
CRLF) and the re-read's BOM stripped (``new_content`` is the BOM-less
|
|
string we matched against)."""
|
|
data, _failed = self._read_exact_bytes(path)
|
|
if data is None:
|
|
return PatchResult(error=f"Post-write verification failed: could not re-read {path}")
|
|
bomless, _ = _strip_bom(data.decode("utf-8", "surrogateescape"))
|
|
on_disk = bomless.replace("\r\n", "\n").replace("\r", "\n")
|
|
intended = new_content.replace("\r\n", "\n").replace("\r", "\n")
|
|
if on_disk != intended:
|
|
return PatchResult(error=(
|
|
f"Post-write verification failed for {path}: on-disk content "
|
|
f"differs from intended write "
|
|
f"(wrote {len(intended)} chars, read back "
|
|
f"{len(on_disk)} chars after normalizing line endings). "
|
|
"The patch did not persist. Re-read the file and try again."))
|
|
return None
|
|
|
|
def patch_replace(self, path: str, old_string: str, new_string: str,
|
|
replace_all: bool = False) -> PatchResult:
|
|
"""Replace text in a file using fuzzy matching (``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)
|
|
data, failed = self._read_exact_bytes(path)
|
|
if data is None:
|
|
return PatchResult(error=failed.cwd_error or f"Failed to read file: {path}")
|
|
# Every line the replacement does not touch is written back, so read the exact bytes;
|
|
# surrogateescape lets write_file restore any byte UTF-8 cannot decode (#79178).
|
|
# 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 = data.decode("utf-8", "surrogateescape")
|
|
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; 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)
|
|
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_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],
|
|
lint=lint_result.to_dict() if lint_result else None,
|
|
# 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:
|
|
"""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}")
|
|
return apply_v4a_operations(operations, self)
|
|
|
|
# --- SEARCH -------------------------------------------------------------
|
|
|
|
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,
|
|
order: str = "discovery") -> SearchResult:
|
|
"""Search for content (regex, ``target="content"``) or files (glob,
|
|
``target="files"``). ``output_mode``: "content", "files_only" or "count";
|
|
``context``: lines of context around matches; ``order``: file-search
|
|
ordering — fast "discovery" or exact "modified" time."""
|
|
offset, limit = normalize_search_pagination(offset, limit)
|
|
if target == "files" and order not in {"discovery", "modified"}:
|
|
return SearchResult(
|
|
error=(f"Invalid file search order {order!r}; expected "
|
|
"'discovery' or 'modified'."))
|
|
path = self._expand_path(path)
|
|
probe = self._path_exists_probe(path)
|
|
exists_probe = probe.stdout
|
|
if probe.cwd_error:
|
|
return SearchResult(error=probe.cwd_error)
|
|
if "exists" not in exists_probe and "not_found" not in exists_probe:
|
|
return SearchResult(error=(f"Terminal environment unavailable: could not stat {path} "
|
|
"(the sandbox may still be starting or was removed). Retry shortly."))
|
|
if "not_found" in exists_probe:
|
|
# 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, order)
|
|
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, order)
|
|
exclusions = self._macos_search_exclusions(path)
|
|
if exclusions and not result.error:
|
|
skipped = ", ".join(item.split("/")[-1] for item in exclusions)
|
|
result.warning = (
|
|
"Skipped macOS protected folders during broad search to avoid "
|
|
f"an unattended privacy prompt: {skipped}. Search a protected "
|
|
"folder directly when access is intentional.")
|
|
return result
|
|
|
|
|
|
# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
|
|
# Names external plugins imported from this module before the Sep 2026 decomposition.
|
|
# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
|
|
# The whole block is removed by reverting the commit that added it.
|
|
from typing import Any # noqa: F401,E402
|
|
from typing import ClassVar # noqa: F401,E402
|
|
from typing import List # noqa: F401,E402
|
|
from agent.file_safety import build_write_denied_paths # noqa: F401,E402
|
|
from agent.file_safety import build_write_denied_prefixes # noqa: F401,E402
|
|
from dataclasses import dataclass # noqa: F401,E402
|
|
from dataclasses import field # noqa: F401,E402
|
|
import posixpath # noqa: F401,E402
|
|
import threading # noqa: F401,E402
|
|
|
|
MAX_LINES = 2000
|
|
|
|
MAX_LINE_LENGTH = 2000
|
|
|
|
WRITE_DENIED_PATHS = build_write_denied_paths(_HOME)
|
|
|
|
WRITE_DENIED_PREFIXES = build_write_denied_prefixes(_HOME)
|
|
|
|
|
|
_PLUGIN_COMPAT_LAZY = {
|
|
'DEFAULT_READ_LIMIT': ('tools.file_operations_common', 'DEFAULT_READ_LIMIT'),
|
|
'DEFAULT_READ_OFFSET': ('tools.file_operations_common', 'DEFAULT_READ_OFFSET'),
|
|
'DEFAULT_SEARCH_LIMIT': ('tools.file_operations_common', 'DEFAULT_SEARCH_LIMIT'),
|
|
'DEFAULT_SEARCH_OFFSET': ('tools.file_operations_common', 'DEFAULT_SEARCH_OFFSET'),
|
|
'LINTERS': ('tools.file_operations_lint', 'LINTERS'),
|
|
'LintResult': ('tools.file_operations_common', 'LintResult'),
|
|
'MAX_FILE_SIZE': ('tools.transcription_common', 'MAX_FILE_SIZE'),
|
|
'SEARCH_PRUNE_DIR_NAMES': ('agent.search_policy', 'SEARCH_PRUNE_DIR_NAMES'),
|
|
'SearchMatch': ('tools.file_operations_common', 'SearchMatch'),
|
|
'build_write_denied_paths': ('agent.file_safety', 'build_write_denied_paths'),
|
|
'build_write_denied_prefixes': ('agent.file_safety', 'build_write_denied_prefixes'),
|
|
'tool_interrupt': ('tools', 'interrupt'),
|
|
}
|
|
|
|
|
|
def __getattr__(name): # PEP 562 — lazy so no import cycles
|
|
target = _PLUGIN_COMPAT_LAZY.get(name)
|
|
if target is None:
|
|
raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
|
|
import importlib
|
|
from hermes_cli.plugin_compat import warn_once
|
|
warn_once(__name__, name, *target)
|
|
return getattr(importlib.import_module(target[0]), target[1])
|
|
# ---- END PLUGIN-COMPAT ----
|