perf(file-ops): collapse read_file's shell probes into one compound command

(cherry picked from commit 4e6c0beff77e24c46641d09ce77258c819ea9098)

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}"

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

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()

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

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 —