diff --git a/tests/tools/file_ops_fakes.py b/tests/tools/file_ops_fakes.py new file mode 100644 index 0000000000..4cbb2d992b --- /dev/null +++ b/tests/tools/file_ops_fakes.py @@ -0,0 +1,61 @@ +"""Fakes for ``ShellFileOperations``' compound shell probes. + +``read_file`` and ``write_file`` ask the shell everything in ONE command +whose stdout is split on a per-call random sentinel line. Test doubles that +script ``env.execute`` / ``_exec`` need to answer that command with exactly +the stream the shell would produce; these helpers build it. Match the +sentinel out of the command first (it is random), then compose: + + m = READ_SENTINEL_RE.search(command) + if m: + return {"output": compound_read_output(m.group(0), size=5, sample=b"hello", + content="hello\\n", total_lines=1), + "returncode": 0} +""" + +import base64 +import re +from typing import Optional + +READ_SENTINEL_RE = re.compile(r"__HERMES_RF_[0-9a-f]{32}__") +WRITE_SENTINEL_RE = re.compile(r"__HERMES_WF_[0-9a-f]{32}__") + + +def compound_read_output( + sentinel: str, + *, + size: int, + sample: Optional[bytes], + content: str, + total_lines: int, + trailing_newline: bool = True, + sample_rc: int = 0, + read_rc: int = 0, +) -> str: + """Stdout of ``_read_probe_cmd`` for a regular file. + + ``content`` is the ``sed | cut`` page exactly as the shell prints it: + every line newline-terminated (``cut`` always adds one), or ``""`` for a + page past EOF. ``sample`` is the raw first-1000-bytes slice (``None`` + emits an empty base64 segment, e.g. a shell without ``base64``). + """ + sample_seg = base64.b64encode(sample).decode() + "\n" if sample else "" + return ( + f"{size}\n{sentinel}\n" + f"{sample_seg}{sentinel}\n" + f"{content}{sentinel}\n" + f"{total_lines}\n{sentinel}\n" + f"{1 if trailing_newline else 0}\n{sentinel}\n" + f"{sample_rc} {read_rc}\n" + ) + + +def compound_write_probe_output(sentinel: str, *, head3: bytes, body: str) -> str: + """Stdout of ``_write_probe_cmd`` for an existing file. + + ``head3`` is the first three bytes on disk (BOM detection); ``body`` is + the second segment: the whole file when pre-content was wanted, else + the 4 KB line-ending sample. + """ + head_seg = base64.b64encode(head3).decode() + "\n" if head3 else "" + return f"{head_seg}{sentinel}\n{body}" diff --git a/tests/tools/test_file_operations.py b/tests/tools/test_file_operations.py index 9af6dcb73e..48d2cad681 100644 --- a/tests/tools/test_file_operations.py +++ b/tests/tools/test_file_operations.py @@ -7,6 +7,7 @@ import subprocess from pathlib import Path from unittest.mock import MagicMock +from tests.tools.file_ops_fakes import READ_SENTINEL_RE, compound_read_output from tools.file_operations import ( _is_write_denied, ReadResult, @@ -303,19 +304,14 @@ class TestShellFileOpsHelpers: def side_effect(command, **kwargs): commands.append(command) - # The size probe gates `wc -c` behind `[ -f ]` so a FIFO or device - # cannot block the read; it still reports a plain byte count. - if command.startswith("if [ -f ") or command.startswith("wc -c"): - return {"output": "5\n", "returncode": 0} - if command.startswith("head -c") and "| base64" in command: - import base64 as b64 - return {"output": b64.b64encode(b"hello").decode(), "returncode": 0} - if command.startswith("head -c"): - return {"output": "hello", "returncode": 0} - if command.startswith("sed -n"): - return {"output": "hello\n", "returncode": 0} - if command.startswith("wc -l"): - return {"output": "1\n", "returncode": 0} + m = READ_SENTINEL_RE.search(command) + if m: + return { + "output": compound_read_output( + m.group(0), size=5, sample=b"hello", content="hello\n", total_lines=1 + ), + "returncode": 0, + } return {"output": "", "returncode": 0} mock_env.execute.side_effect = side_effect @@ -323,16 +319,22 @@ class TestShellFileOpsHelpers: result = ops.read_file(r"C:\Users\alice\notes.txt") assert result.error is None - assert commands[0] == ( + # One compound probe carries every stage; each embeds the MSYS path. + # The size probe gates `wc -c` behind `[ -f ]` so a FIFO or device + # cannot block the read; it still reports a plain byte count. + assert len(commands) == 1 + probe = commands[0] + assert probe.startswith( "if [ -f '/c/Users/alice/notes.txt' ]; " "then wc -c < '/c/Users/alice/notes.txt' 2>/dev/null; " + ) + assert "head -c 1000 '/c/Users/alice/notes.txt' 2>/dev/null | base64" in probe + assert "sed -n '1,2000p' '/c/Users/alice/notes.txt' 2>/dev/null | cut -b1-8001" in probe + assert "wc -l < '/c/Users/alice/notes.txt'" in probe + assert ( "elif [ -e '/c/Users/alice/notes.txt' ]; " "then echo __hermes_not_regular__; " - "else exit 1; fi" - ) - assert commands[1] == "head -c 1000 '/c/Users/alice/notes.txt' 2>/dev/null | base64" - assert commands[2] == "sed -n '1,2000p' '/c/Users/alice/notes.txt' | cut -b1-8001" - assert commands[3] == "wc -l < '/c/Users/alice/notes.txt'" + ) in probe def test_is_likely_binary_by_extension(self, file_ops): assert file_ops._is_likely_binary("photo.png") is True @@ -355,14 +357,15 @@ class TestShellFileOpsHelpers: ) def side_effect(command, **kwargs): - if command.startswith("if [ -f ") or command.startswith("wc -c"): - return {"output": "12\n", "returncode": 0} - if command.startswith("head -c"): - return {"output": "print('ok')\n", "returncode": 0} - if command.startswith("sed -n"): - return {"output": leaked, "returncode": 0} - if command.startswith("wc -l"): - return {"output": "1\n", "returncode": 0} + m = READ_SENTINEL_RE.search(command) + if m: + return { + "output": compound_read_output( + m.group(0), size=12, sample=b"print('ok')\n", + content=leaked, total_lines=1, + ), + "returncode": 0, + } return {"output": "", "returncode": 0} mock_env.execute.side_effect = side_effect @@ -773,17 +776,19 @@ class TestByteLayerBinaryDetection: # --- integration: read_file over the mocked terminal ------------------ def _dispatch(self, cjk_bytes): - import base64 as b64 - def side_effect(command, **kwargs): - if command.startswith("if [ -f ") or command.startswith("wc -c"): - return {"output": f"{len(cjk_bytes)}\n", "returncode": 0} - if command.startswith("head -c") and "| base64" in command: - return {"output": b64.b64encode(cjk_bytes[:1000]).decode(), "returncode": 0} - if command.startswith("sed -n"): - return {"output": cjk_bytes.decode("utf-8", errors="replace"), "returncode": 0} - if command.startswith("wc -l"): - return {"output": "1\n", "returncode": 0} + m = READ_SENTINEL_RE.search(command) + if m: + return { + "output": compound_read_output( + m.group(0), + size=len(cjk_bytes), + sample=cjk_bytes[:1000], + content=cjk_bytes.decode("utf-8", errors="replace"), + total_lines=1, + ), + "returncode": 0, + } return {"output": "", "returncode": 0} return side_effect diff --git a/tests/tools/test_file_operations_edge_cases.py b/tests/tools/test_file_operations_edge_cases.py index 0865801911..e9875f236e 100644 --- a/tests/tools/test_file_operations_edge_cases.py +++ b/tests/tools/test_file_operations_edge_cases.py @@ -8,6 +8,7 @@ Covers: import pytest from unittest.mock import MagicMock, patch +from tests.tools.file_ops_fakes import READ_SENTINEL_RE, compound_read_output from tools.file_operations import ShellFileOperations, _parse_search_context_line @@ -205,14 +206,15 @@ class TestPaginationBounds: def fake_exec(command, *args, **kwargs): commands.append(command) - if command.startswith("if [ -f ") or command.startswith("wc -c"): - return MagicMock(exit_code=0, stdout="12") - if command.startswith("head -c"): - return MagicMock(exit_code=0, stdout="line1\nline2\n") - if command.startswith("sed -n"): - return MagicMock(exit_code=0, stdout="line1\n") - if command.startswith("wc -l"): - return MagicMock(exit_code=0, stdout="2") + m = READ_SENTINEL_RE.search(command) + if m: + return MagicMock( + exit_code=0, + stdout=compound_read_output( + m.group(0), size=12, sample=b"line1\nline2\n", + content="line1\n", total_lines=2, + ), + ) return MagicMock(exit_code=0, stdout="") with patch.object(ops, "_exec", side_effect=fake_exec): @@ -220,8 +222,9 @@ class TestPaginationBounds: assert result.error is None assert "1|line1" in result.content - sed_commands = [cmd for cmd in commands if cmd.startswith("sed -n")] - assert sed_commands == ["sed -n '1,1p' 'notes.txt' | cut -b1-8001"] + # The clamped range rides the single compound probe. + assert len(commands) == 1 + assert "sed -n '1,1p' 'notes.txt' 2>/dev/null | cut -b1-8001" in commands[0] def test_search_clamps_offset_and_limit_before_building_head_pipeline(self): env = MagicMock() diff --git a/tests/tools/test_file_ops_single_roundtrip.py b/tests/tools/test_file_ops_single_roundtrip.py new file mode 100644 index 0000000000..8c0f6dab7c --- /dev/null +++ b/tests/tools/test_file_ops_single_roundtrip.py @@ -0,0 +1,203 @@ +"""``read_file`` / ``write_file`` cost one shell round-trip, not four. + +Real ``LocalEnvironment`` against ``tmp_path`` (no mocks), with a spy on +``env.execute`` counting round-trips. The cases below are exactly the ones +that used to need their own probe (existence, size, binary sample, page, +line count, trailing newline), so each proves the compound reply carries +that answer. +""" + +import os +import sys +import threading +import unicodedata +from unittest.mock import patch + +import pytest + +from tools.environments.local import LocalEnvironment +from tools.file_operations import ExecuteResult, ShellFileOperations + +pytestmark = pytest.mark.skipif(sys.platform == "win32", reason="POSIX shell probes") + +READ_PROBE_MARK = "__HERMES_RF_" + + +@pytest.fixture +def shell(tmp_path, monkeypatch): + """(ops, calls): file ops over a real local shell, every execute recorded.""" + # Pin the shell path even where a native fast path exists. + monkeypatch.setenv("HERMES_NATIVE_FILE_READ", "0") + env = LocalEnvironment(cwd=str(tmp_path)) + calls = [] + real_execute = env.execute + + def spy(command, *args, **kwargs): + calls.append(command) + return real_execute(command, *args, **kwargs) + + env.execute = spy + return ShellFileOperations(env, cwd=str(tmp_path)), calls + + +def _write(tmp_path, name, data: bytes): + p = tmp_path / name + p.write_bytes(data) + return str(p) + + +class TestReadFileOneRoundTrip: + def test_text_read_is_one_round_trip(self, shell, tmp_path): + ops, calls = shell + p = _write(tmp_path, "a.txt", b"one\ntwo\nthree\n") + r = ops.read_file(p) + assert len(calls) == 1 and READ_PROBE_MARK in calls[0] + assert r.error is None + # ``_add_line_numbers`` numbers the empty tail after the final + # newline: long-standing behaviour, preserved byte for byte. + assert r.content == "1|one\n2|two\n3|three\n4|" + assert (r.total_lines, r.file_size, r.truncated) == (3, 14, False) + + def test_no_trailing_newline_needs_no_extra_probe(self, shell, tmp_path): + ops, calls = shell + p = _write(tmp_path, "b.txt", b"a\nb") + r = ops.read_file(p) + assert len(calls) == 1 + # ``cut`` newline-terminates the last line; the artifact is stripped + # from the same reply that used to need a fifth ``tail -c 1`` call. + assert r.content == "1|a\n2|b" + assert r.total_lines == 1 # wc -l semantics, unchanged + + def test_pagination_window_and_hint(self, shell, tmp_path): + ops, calls = shell + p = _write(tmp_path, "c.txt", b"".join(b"l%d\n" % i for i in range(1, 11))) + r = ops.read_file(p, offset=3, limit=2) + assert len(calls) == 1 + assert r.content == "3|l3\n4|l4\n5|" + assert r.truncated is True and r.total_lines == 10 + assert "offset=5" in r.hint + + def test_offset_past_eof_note(self, shell, tmp_path): + ops, calls = shell + p = _write(tmp_path, "c.txt", b"".join(b"l%d\n" % i for i in range(1, 6))) + r = ops.read_file(p, offset=50) + assert len(calls) == 1 + assert r.content == "" and r.error is None + assert "beyond the end" in r.hint and "5" in r.hint + + def test_empty_file(self, shell, tmp_path): + ops, calls = shell + r = ops.read_file(_write(tmp_path, "e.txt", b"")) + assert len(calls) == 1 + assert r.error is None and r.content == "" and r.total_lines == 0 + assert "empty" in r.hint + + def test_bom_stripped_on_first_page(self, shell, tmp_path): + ops, calls = shell + r = ops.read_file(_write(tmp_path, "f.txt", "hello\n".encode("utf-8"))) + assert len(calls) == 1 + assert r.content == "1|hello\n2|" + + def test_crlf_bytes_survive(self, shell, tmp_path): + ops, calls = shell + r = ops.read_file(_write(tmp_path, "g.txt", b"x\r\ny\r\n")) + assert r.content == "1|x\r\n2|y\r\n3|" + + def test_long_line_clamped_and_marked(self, shell, tmp_path): + ops, calls = shell + r = ops.read_file(_write(tmp_path, "L.txt", b"a" * 9000 + b"\nshort\n")) + assert len(calls) == 1 + first, second, tail = r.content.split("\n") + assert first.endswith("... [truncated]") and len(first) < 9000 + assert second == "2|short" and tail == "3|" + + def test_relative_path_resolves_against_env_cwd(self, shell, tmp_path): + ops, calls = shell + _write(tmp_path, "rel.txt", b"here\n") + r = ops.read_file("rel.txt") + assert r.error is None and r.content == "1|here\n2|" + + def test_sentinel_lookalike_in_content_reads_intact(self, shell, tmp_path): + ops, calls = shell + lookalike = "__HERMES_RF_" + "ab" * 16 + "__" + p = _write(tmp_path, "s.txt", f"x\n{lookalike}\ny\n".encode("utf-8")) + r = ops.read_file(p) + assert r.error is None and r.total_lines == 3 + assert r.content == f"1|x\n2|{lookalike}\n3|y\n4|" + + +class TestReadFileNonTextPaths: + def test_missing_file_probes_once_then_suggests(self, shell, tmp_path): + ops, calls = shell + _write(tmp_path, "notes.txt", b"x\n") + r = ops.read_file(str(tmp_path / "note.txt")) + assert READ_PROBE_MARK in calls[0] + assert r.error and "File not found" in r.error + assert any(s.endswith("notes.txt") for s in r.similar_files) + + def test_unicode_variant_retry_still_works(self, shell, tmp_path): + ops, calls = shell + nfc = unicodedata.normalize("NFC", "café.txt") + nfd = unicodedata.normalize("NFD", "café.txt") + assert nfc != nfd + _write(tmp_path, nfc, b"accent\n") + r = ops.read_file(str(tmp_path / nfd)) + assert r.error is None and r.content == "1|accent\n2|" + assert "unicode-equivalent" in r.hint + + def test_directory_is_not_regular(self, shell, tmp_path): + ops, calls = shell + r = ops.read_file(str(tmp_path)) + assert len(calls) == 1 + assert r.error and "not a regular file" in r.error + + def test_binary_sample_detected_in_same_reply(self, shell, tmp_path): + ops, calls = shell + p = _write(tmp_path, "blob", b"\x00\x01\x02" + b"\x00" * 50) + r = ops.read_file(p) + assert READ_PROBE_MARK in calls[0] + assert r.is_binary is True and r.error + # Only the UTF-16 rescue may add round-trips, never a second sample. + assert not any("head -c 1000" in c for c in calls[1:]) + + def test_image_extension_stops_at_size_probe(self, shell, tmp_path): + ops, calls = shell + r = ops.read_file(_write(tmp_path, "p.png", b"\x89PNG\r\n")) + assert len(calls) == 1 and READ_PROBE_MARK not in calls[0] + assert r.is_image is True and r.file_size == 6 + + @pytest.mark.linux_only + def test_fifo_returns_not_regular_without_blocking(self, shell, tmp_path): + if not hasattr(os, "mkfifo"): + pytest.skip("no mkfifo") + ops, calls = shell + fifo = tmp_path / "pipe" + os.mkfifo(fifo) + box = {} + + def run(): + box["r"] = ops.read_file(str(fifo)) + + t = threading.Thread(target=run, daemon=True) + t.start() + t.join(20) + assert not t.is_alive(), "read_file blocked on a writer-less FIFO" + assert "not a regular file" in box["r"].error + assert len(calls) == 1 + + +class TestCompoundFallback: + def test_unparseable_reply_falls_back_to_sequential_probes(self, shell, tmp_path): + ops, calls = shell + p = _write(tmp_path, "a.txt", b"one\ntwo\n") + real_exec = ops._exec + + def garbled(command, *args, **kwargs): + if READ_PROBE_MARK in command: + return ExecuteResult(stdout="[Command timed out after 1s]\n", exit_code=124) + return real_exec(command, *args, **kwargs) + + with patch.object(ops, "_exec", side_effect=garbled): + r = ops.read_file(p) + assert r.error is None and r.content == "1|one\n2|two\n3|" + assert r.total_lines == 2 diff --git a/tools/file_operations.py b/tools/file_operations.py index fbfd06cab8..ae1a2ce5bb 100644 --- a/tools/file_operations.py +++ b/tools/file_operations.py @@ -29,6 +29,7 @@ import base64 import binascii import os import re +import secrets import sys import difflib import hashlib @@ -848,6 +849,36 @@ DEFAULT_SEARCH_LIMIT = 50 # `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 ``_size_probe_cmd`` carries in ``exit 1`` has to travel in-band. +MISSING_SENTINEL = "__hermes_missing__" + +_READ_SENTINEL_PREFIX = "__HERMES_RF_" +_WRITE_SENTINEL_PREFIX = "__HERMES_WF_" + + +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, and the + underscores keep the token outside the base64 alphabet, so a sentinel + that ever leaked 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 its + output or prints nothing, so the separator is always ``sentinel + "\\n"`` + on a line of its own. The text after the final sentinel is the status + segment. + """ + return output.split(sentinel + "\n") + def _coerce_int(value: Any, default: int) -> int: """Best-effort integer coercion for tool pagination inputs.""" @@ -1032,7 +1063,18 @@ class ShellFileOperations(FileOperations): ) if result.exit_code != 0: return None - encoded = _strip_terminal_fence_leaks(result.stdout) + return self._decode_base64_sample(result.stdout) + + @staticmethod + def _decode_base64_sample(text: str) -> Optional[bytes]: + """Decode one base64 sample as emitted by ``head -c N | base64``. + + Whitespace-joins the whole text first (``base64`` wraps at 76 + columns), so callers must hand over exactly one segment; anything + else in the text fails validation and yields ``None``, which sends + the caller to the legacy text-sample heuristic. + """ + encoded = _strip_terminal_fence_leaks(text) encoded = "".join(encoded.split()) if not encoded: return b"" @@ -1507,42 +1549,190 @@ class ShellFileOperations(FileOperations): def read_file(self, path: str, offset: int = 1, limit: int = 2000) -> ReadResult: """ Read a file with pagination, binary detection, and line numbers. - + Args: path: File path (absolute or relative to cwd) offset: Line number to start from (1-indexed, default 1) limit: Maximum lines to return (default 500, max 2000) - + Returns: ReadResult with content, metadata, or error info + + One shell round-trip answers every question the read needs: + existence, size, binary sample, the page, line count, trailing + newline (see ``_read_probe_cmd``). A reply that cannot be parsed + falls back to ``_read_file_sequential``, the one-probe-per-call + form, so an exotic shell can never do worse than before. """ # Expand ~ and other shell paths path = self._expand_path(path) - + offset, limit = normalize_read_pagination(offset, limit) - + + # Images and known-binary extensions never inline content; the + # sequential path stops at the probes for them, so nothing is gained + # by streaming their bytes through the page pipeline. + if self._is_image(path) or os.path.splitext(path)[1].lower() in BINARY_EXTENSIONS: + 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) + return self._read_file_sequential(path, offset, limit) + + segments = _split_segments(output, sentinel) + if probe.exit_code != 0 or len(segments) != 6: + 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): + 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, + # paid only on 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: + sample_cmd = f"head -c 1000 {self._escape_shell_arg(path)} 2>/dev/null" + sample_result = self._exec(sample_cmd) + sample_output = _strip_terminal_fence_leaks(sample_result.stdout) + is_binary = self._is_likely_binary(path, sample_output) + + 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 _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. The probes run only inside ``[ -f ]``, the same + stat-not-open guard as ``_size_probe_cmd``, so a FIFO or device + never reaches ``head``/``sed``. A missing path echoes + ``MISSING_SENTINEL`` instead of exiting non-zero, because a compound + command only reports its last status. Every stage silences stderr: + the local backend merges stderr into stdout and a stray diagnostic + would otherwise land inside a segment. + + The page clamp is byte-based on purpose; see ``_read_file_sequential`` + for why it is ``4 * max_line_length + 1``. + """ + 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} ]; 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. + + Before failing, try unicode-equivalent spellings: NFC/NFD, narrow + no-break space, curly quotes render identically in a terminal, so + the model retyping a visually-correct path can never discover the + byte mismatch on its own (retrying is the tool's job, not the + model's). 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 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. + + ``read_file`` lands here for image / known-binary extensions (only + the probes matter) and whenever the compound reply cannot be parsed. + ``path`` is already expanded and ``offset``/``limit`` normalized. + """ # Check if file exists and get size (POSIX, works on Linux + macOS) stat_result = self._exec(self._size_probe_cmd(path)) if stat_result.exit_code != 0: - # File not found. Before failing, try unicode-equivalent - # spellings — NFC/NFD, narrow no-break space, curly quotes - # render identically in a terminal, so the model retyping a - # visually-correct path can never discover the byte mismatch - # on its own (retrying is the tool's job, not the model's). - 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 - # No equivalent spelling — suggest similar files - return self._suggest_similar_files(path) + return self._read_file_missing(path, offset, limit) stat_output = _strip_terminal_fence_leaks(stat_result.stdout) if stat_output.strip() == NOT_REGULAR_SENTINEL: @@ -1551,12 +1741,12 @@ class ShellFileOperations(FileOperations): file_size = int(stat_output.strip()) except ValueError: file_size = 0 - + # Check if file is too large if file_size > MAX_FILE_SIZE: # Still try to read, but warn pass - + # Images are never inlined — redirect to the vision tool if self._is_image(path): return ReadResult( @@ -1568,7 +1758,7 @@ class ShellFileOperations(FileOperations): "Use vision_analyze with this file path to inspect the image contents." ), ) - + # Read a sample to check for binary content — at the byte layer when # the transport allows, falling back to the legacy text heuristic. sample_bytes = self._sample_file_bytes(path) @@ -1582,22 +1772,8 @@ class ShellFileOperations(FileOperations): is_binary = self._is_likely_binary(path, sample_output) if is_binary: - # UTF-16 rescue (ported from MoonshotAI/kimi-code#2647): the - # terminal env decodes stdout as UTF-8 with errors="replace", so - # a UTF-16 text file (Windows Notepad .txt, PowerShell `>` - # redirects) arrives mangled with U+FFFD and trips the binary - # guard. Probe the raw bytes via the backend's Python and - # transcode to UTF-8 when a BOM or the zero-byte parity - # heuristic identifies UTF-16. - 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), - ) - + return self._read_binary_file(path, offset, limit, file_size, sample_bytes) + # Read with pagination using sed, clamping each line to a byte # budget IN THE SHELL so a pathological single-line file (e.g. one # 400MB minified line) never crosses the exec transport. The Python @@ -1630,16 +1806,11 @@ class ShellFileOperations(FileOperations): f" | cut -b1-{line_clamp_bytes}" ) read_result = self._exec(read_cmd) - + if read_result.exit_code != 0: return ReadResult(error=f"Failed to read file: {read_result.stdout}") read_output = _strip_terminal_fence_leaks(read_result.stdout) - # Strip a leading UTF-8 BOM so the model never sees a phantom U+FEFF - # before the first real character. Only meaningful on the first - # chunk (the marker lives at byte 0); later pages can't carry it. - if offset == 1: - read_output, _ = _strip_bom(read_output) - + # Get total line count wc_cmd = f"wc -l < {self._escape_shell_arg(path)}" wc_result = self._exec(wc_cmd) @@ -1648,7 +1819,50 @@ class ShellFileOperations(FileOperations): total_lines = int(wc_output.strip()) except ValueError: total_lines = 0 - + + # Only the page that reaches the file's final line can carry the + # ``cut`` newline artifact (see _assemble_read_result); probe the + # last byte just for that case, exactly as before. + file_ends_with_newline: Optional[bool] = None + if not total_lines > end_line and read_output.endswith('\n'): + tail_cmd = f"tail -c 1 {self._escape_shell_arg(path)} | wc -l" + tail_result = self._exec(tail_cmd) + tail_output = _strip_terminal_fence_leaks(tail_result.stdout) + if tail_result.exit_code == 0: + file_ends_with_newline = tail_output.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, the + ``cut`` newline artifact fix and the ambiguous-silence guards can + never drift apart. ``file_ends_with_newline`` is ``None`` when the + caller could not tell (the artifact is then left alone, as before). + """ + # Strip a leading UTF-8 BOM so the model never sees a phantom U+FEFF + # before the first real character. Only meaningful on the first + # chunk (the marker lives at byte 0); later pages can't carry it. + if offset == 1: + read_output, _ = _strip_bom(read_output) + # Check if truncated truncated = total_lines > end_line hint = None @@ -1658,13 +1872,13 @@ class ShellFileOperations(FileOperations): # ``cut`` (unlike sed -n p) always newline-terminates its output, # so a file whose final line has no trailing newline would grow a # phantom empty last line. Only possible when this page reaches the - # file's final line; probe the last byte and strip the artifact. - if not truncated and read_output.endswith('\n'): - tail_cmd = f"tail -c 1 {self._escape_shell_arg(path)} | wc -l" - tail_result = self._exec(tail_cmd) - tail_output = _strip_terminal_fence_leaks(tail_result.stdout) - if tail_result.exit_code == 0 and tail_output.strip() == "0": - read_output = read_output[:-1] + # file's final line; strip the artifact 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] # Ambiguous-silence guards: an empty content string is # indistinguishable, from inside the model, from a broken tool —