fix(file-ops): both patch modes read their write-back source byte-exactly

Review follow-up (ehz0ah, teknium1). Keying the cleanup on the
__HERMES_FENCE_ marker still rewrote a real line that contains that
text, and nothing has emitted the wrapper since the spawn-per-call layer
(d684d7ee7e), so the cleanup can only ever eat the file's own bytes.

The same class also hit mode="replace", the default: patch_replace (and
V4A past the 1000-byte binary sample) read the file through the text
transport, which decodes with errors="replace", so any byte UTF-8 cannot
decode came back as U+FFFD and was persisted on lines the edit never
touched, while the diff and the post-write check (same lossy read) showed
nothing.

_read_exact_bytes reads natively on the local POSIX host (regular files
only) and over base64 elsewhere; read_file_raw, patch_replace and
_verify_patch_persisted decode it with surrogateescape, which write_file's
encode already inverts (#79178), so every untouched byte is written back
exactly and no readable file becomes a read error (V4A's Add/Move/Delete
existence checks are unchanged). A garbled transport reply refuses
instead of writing stray output into the file. The Python linter parses
bytes so a declared legacy coding still lints clean. Local edits spawn
two fewer shells (replace 5 -> 3, V4A 9 -> 7).

tests/tools/test_file_operations.py

@@ -1,5 +1,6 @@
 """Tests for tools/file_operations.py — deny list, result dataclasses, helpers."""
 
+import base64
 import os
 import pytest
 import subprocess
@@ -296,29 +297,6 @@ class TestShellFileOpsHelpers:
         assert "\x07" not in result.content
         assert "1|print('ok')" in result.content
 
-    def test_read_file_raw_strips_leaked_terminal_fence_markers(self, mock_env):
-        leaked = (
-            "__HERMES_FENCE_a9f7b3__\x07'\n"
-            "alpha\n"
-            "\x1b]0;cat '/tmp/test/a.txt'\x07__HERMES_FENCE_a9f7b3__\n"
-        )
-
-        def side_effect(command, **kwargs):
-            if command.startswith("if [ -f ") or command.startswith("wc -c"):
-                return {"output": "6\n", "returncode": 0}
-            if command.startswith("head -c"):
-                return {"output": "alpha\n", "returncode": 0}
-            if command.startswith("cat "):
-                return {"output": leaked, "returncode": 0}
-            return {"output": "", "returncode": 0}
-
-        mock_env.execute.side_effect = side_effect
-        ops = ShellFileOperations(mock_env)
-        result = ops.read_file_raw("/tmp/test/a.txt")
-
-        assert result.error is None
-        assert result.content == "alpha\n"
-
     def test_newline_terminated_content_has_no_phantom_line(self, file_ops):
         # A file ending in a newline (the normal, well-formed case) has its
         # last line terminated, NOT followed by an empty line. The gutter must
@@ -457,12 +435,11 @@ class TestPatchReplacePostWriteVerification:
         file_contents = {"/tmp/test/a.py": "hello world\n"}
 
         def side_effect(command, **kwargs):
-            # cat reads the file — both the initial read and the verify read
-            if command.startswith("cat "):
-                # Extract path from cat command (strip quotes)
+            # the byte-exact read (base64 over the transport) — both the initial read and the verify read
+            if command.startswith("base64 < "):
                 for path in file_contents:
                     if path in command:
-                        return {"output": file_contents[path], "returncode": 0}
+                        return {"output": base64.b64encode(file_contents[path].encode()).decode(), "returncode": 0}
                 return {"output": "", "returncode": 1}
             # mkdir for parent dir
             if command.startswith("mkdir "):
@@ -490,18 +467,18 @@ class TestPatchReplacePostWriteVerification:
 
     def test_patch_replace_fails_when_verify_read_errors(self, mock_env):
         """If the verify-read step itself fails (exit code != 0), return an error."""
-        call_count = {"cat": 0}
+        call_count = {"read": 0}
         state = {"content": "hello world\n"}
 
         def side_effect(command, stdin_data=None, **kwargs):
             if stdin_data is not None:  # write (atomic temp-file + mv script)
                 state["content"] = stdin_data
                 return {"output": "", "returncode": 0}
-            if command.startswith("cat "):  # read
-                call_count["cat"] += 1
+            if command.startswith("base64 < "):  # byte-exact read
+                call_count["read"] += 1
                 # First read (initial fetch) succeeds; second read (verify) fails
-                if call_count["cat"] == 1:
-                    return {"output": state["content"], "returncode": 0}
+                if call_count["read"] == 1:
+                    return {"output": base64.b64encode(state["content"].encode()).decode(), "returncode": 0}
                 return {"output": "", "returncode": 1}
             if command.startswith("mkdir "):
                 return {"output": "", "returncode": 0}

tests/tools/test_file_write_safety.py

@@ -372,11 +372,12 @@ class TestBomHandling:
 
 
     def test_v4a_update_keeps_terminal_escape_bytes_on_untouched_lines(self, ops, tmp_path: Path):
-        # read_file_raw feeds the V4A write-back; its leak cleanup must not eat the file's own
-        # OSC title escapes and BEL bytes on lines the patch never touched.
+        # read_file_raw feeds the V4A write-back: every byte on a line the patch never touched
+        # survives, including OSC escapes, BEL and literal fence-marker text.
         target = tmp_path / "prompt.sh"
         original = (b'set_title() { printf "\x1b]0;%s\x07" "$1"; }\n'
                     b'beep() { printf "\x07"; }\n'
+                    b'SENTINEL = "__HERMES_FENCE_a9f7b3__\x07"  # marker text is file content too\n'
                     b'VERSION=1\n')
         target.write_bytes(original)
         patch = (
@@ -391,6 +392,26 @@ class TestBomHandling:
         assert res.success, res.error
         assert target.read_bytes() == original.replace(b"VERSION=1", b"VERSION=2")
 
+    @pytest.mark.parametrize("mode", ["replace", "v4a"])
+    def test_edit_keeps_bytes_utf8_cannot_decode_on_untouched_lines(self, ops, tmp_path: Path, mode):
+        # Both edit paths write back every line they did not touch, so their source read must be
+        # byte-exact: the text transport decodes with errors="replace", which turned this legacy
+        # latin-1 byte into U+FFFD on disk. (Past the 1000-byte sample, where V4A reads it as text.)
+        target = tmp_path / "legacy.py"
+        original = (b"# -*- coding: latin-1 -*-\n" + b"# " + b"x" * 1100 + b"\n"
+                    b"name = 'caf\xe9'\n"
+                    b"x = 1\n")
+        target.write_bytes(original)
+        if mode == "replace":
+            res = ops.patch_replace(str(target), "x = 1", "x = 2")
+        else:
+            res = ops.patch_v4a(f"*** Begin Patch\n*** Update File: {target}\n@@\n-x = 1\n+x = 2\n*** End Patch")
+        assert res.success, res.error
+        assert target.read_bytes() == original.replace(b"x = 1", b"x = 2")
+        # The file declares its encoding, so it is valid Python and lints clean.
+        lints = res.lint.values() if mode == "v4a" else [res.lint]
+        assert [lint["status"] for lint in lints] == ["ok"], res.lint
+
 
 class TestProtectedInstructionFiles:
     """Writes to agent-instruction files ALWAYS require approval.

tools/file_operations.py

@@ -263,11 +263,37 @@ class ShellFileOperations(LintMixin, SearchMixin, FileOperations):
             return None
         return self._decode_base64_sample(result.stdout)
 
+    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:
+                if _stat.S_ISREG(os.stat(full).st_mode):
+                    with open(full, "rb") as fh:
+                        return fh.read(), None
+            except OSError:
+                pass  # missing/unreadable: the shell read below reports it the usual way
+        result = self._exec(f"base64 < {self._escape_shell_arg(path)}")
+        if result.exit_code != 0:
+            return None, result
+        data = self._decode_base64_sample(result.stdout)
+        if data is None:  # stray output in the payload: refuse rather than guess (never echo it back)
+            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 ``head -c N | base64`` sample. Whitespace-joins the whole text
-        first (``base64`` wraps at 76 columns), so callers hand over exactly one
-        segment; anything else fails validation → None (legacy text heuristic)."""
+        """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""
@@ -1080,13 +1106,15 @@ class ShellFileOperations(LintMixin, SearchMixin, FileOperations):
         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))
-        cat_result = self._exec(f"cat {self._escape_shell_arg(path)}")
-        if cat_result.exit_code != 0:
-            return ReadResult(error=f"Failed to read file: {cat_result.stdout}")
+        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.
-        # V4A writes this back, so only lines carrying a leaked fence are cleaned.
-        raw_content, _ = _strip_bom(_strip_terminal_fence_leaks(cat_result.stdout, fenced_lines_only=True))
+        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:
@@ -1401,10 +1429,10 @@ class ShellFileOperations(LintMixin, SearchMixin, FileOperations):
         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)."""
-        verify_result = self._cat(path)
-        if verify_result.exit_code != 0:
+        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(verify_result.stdout)
+        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:
@@ -1424,12 +1452,14 @@ class ShellFileOperations(LintMixin, SearchMixin, FileOperations):
         denied = get_write_denied_error(path)
         if denied:
             return PatchResult(error=denied)
-        read_result = self._cat(path)
-        if read_result.exit_code != 0:
-            return PatchResult(error=read_result.cwd_error or f"Failed to read file: {path}")
+        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 = read_result.stdout
+        raw_content = data.decode("utf-8", "surrogateescape")
         content, _ = _strip_bom(raw_content)
 
         from tools.fuzzy_match import fuzzy_find_and_replace

tools/file_operations_common.py

@@ -204,20 +204,13 @@ def count_conflict_blocks(formatted_content: str) -> int:
     return min(opens, len(_CONFLICT_CLOSE.findall(formatted_content))) if opens else 0
 
 
-def _strip_terminal_fence_leaks(text: str, *, fenced_lines_only: bool = False) -> str:
+def _strip_terminal_fence_leaks(text: str) -> str:
     """Strip leaked terminal fence wrappers (OSC sequences, fence markers) from
-    command output; drops lines that were nothing but wrapper.
-
-    ``fenced_lines_only`` is for file content that gets written back: a leak always carries the
-    fence marker, so a line without one is the file's own bytes (a prompt script's ``\x1b]0;``
-    title escape, a BEL) and is kept verbatim."""
+    command output; drops lines that were nothing but wrapper."""
     if not text:
         return text
     cleaned_lines: List[str] = []
     for line in text.splitlines(keepends=True):
-        if fenced_lines_only and "__HERMES_FENCE_" not in line:
-            cleaned_lines.append(line)
-            continue
         had_terminal_wrapper = "__HERMES_FENCE_" in line or "\x1b]" in line
         cleaned = _FENCE_MARKER_RE.sub("", _OSC_SEQUENCE_RE.sub("", line)).replace("\x07", "")
         if had_terminal_wrapper and cleaned.strip("'\r\n\t ") == "":

tools/file_operations_lint.py

@@ -96,9 +96,11 @@ def _lint_toml_inproc(content: str) -> tuple[bool, str]:
 
 
 def _lint_python_inproc(content: str) -> tuple[bool, str]:
-    """In-process Python syntax check via ast.parse (py_compile's scope, no subprocess)."""
+    """In-process Python syntax check via ast.parse (py_compile's scope, no subprocess). Parses the
+    bytes, as py_compile does, so a legacy ``coding:`` cookie is honoured for a file the edit paths
+    read with surrogateescape."""
     try:
-        ast.parse(content)
+        ast.parse(content.encode("utf-8", "surrogateescape"))
         return True, ""
     except SyntaxError as e:
         loc = f" (line {e.lineno}, column {e.offset})" if e.lineno else ""