diff --git a/tests/tools/test_file_operations.py b/tests/tools/test_file_operations.py index 5269370196..8302ee059c 100644 --- a/tests/tools/test_file_operations.py +++ b/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} diff --git a/tests/tools/test_file_write_safety.py b/tests/tools/test_file_write_safety.py index 58441f8188..c1cfdab091 100644 --- a/tests/tools/test_file_write_safety.py +++ b/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. diff --git a/tools/file_operations.py b/tools/file_operations.py index ffd578fdfe..240cb4f7ab 100644 --- a/tools/file_operations.py +++ b/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 diff --git a/tools/file_operations_common.py b/tools/file_operations_common.py index 324cdef2fa..fa616482a0 100644 --- a/tools/file_operations_common.py +++ b/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 ") == "": diff --git a/tools/file_operations_lint.py b/tools/file_operations_lint.py index d875f98a84..816254c7e1 100644 --- a/tools/file_operations_lint.py +++ b/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 ""