fix(file-ops): byte-layer binary detection — stop flagging boundary-cut UTF-8 as binary (#80308 class)

tests/tools/test_file_operations.py

@@ -304,6 +304,9 @@ def side_effect(command, **kwargs):
             commands.append(command)
             if 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"):
@@ -318,7 +321,7 @@ def side_effect(command, **kwargs):
 
         assert result.error is None
         assert commands[0] == "wc -c < '/c/Users/alice/notes.txt' 2>/dev/null"
-        assert commands[1] == "head -c 1000 '/c/Users/alice/notes.txt' 2>/dev/null"
+        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'"
         assert commands[3] == "wc -l < '/c/Users/alice/notes.txt'"
 
@@ -673,3 +676,122 @@ def test_plain_utf8_text_not_flagged(self, tmp_path):
         ops = ShellFileOperations(make_real_subprocess_env(str(tmp_path)))
         # Proper UTF-8 (including non-ASCII) must still read as text.
         assert ops._is_likely_binary("notes.txt", "café résumé\nsecond\n") is False
+
+# =========================================================================
+# Byte-layer binary detection (#80308 class: CJK/multibyte text flagged
+# binary because the byte-boundary sample manufactured U+FFFD in transit)
+# =========================================================================
+
+class TestByteLayerBinaryDetection:
+    """Regression suite for the misclassification class behind #80308.
+
+    Fragment reports/fixes each caught one member: #80261, #80250, #80188,
+    #80349, #79834, #79534, #79408. The boundary contract: text = valid
+    UTF-8 allowing one incomplete multibyte sequence at the sample's end;
+    NUL or mid-stream invalid UTF-8 = read-only.
+    """
+
+    # --- unit: _is_likely_binary_bytes -----------------------------------
+
+    def test_cjk_text_cut_mid_character_is_text(self, file_ops):
+        # 999 ASCII bytes + a 3-byte CJK char cut after its first byte —
+        # exactly what `head -c 1000` does to a CJK file.
+        sample = (b"a" * 999 + "中".encode("utf-8"))[:1000]
+        assert sample[-1:] != b"a"  # the cut really is mid-character
+        assert file_ops._is_likely_binary_bytes(sample) is False
+
+    def test_pure_cjk_text_cut_mid_character_is_text(self, file_ops):
+        sample = ("汉字" * 400).encode("utf-8")[:1000]
+        assert file_ops._is_likely_binary_bytes(sample) is False
+
+    def test_emoji_cut_at_boundary_is_text(self, file_ops):
+        # 4-byte sequence cut after 2 bytes.
+        sample = (b"x" * 998 + "🎉".encode("utf-8"))[:1000]
+        assert file_ops._is_likely_binary_bytes(sample) is False
+
+    def test_utf8_bom_is_text(self, file_ops):
+        assert file_ops._is_likely_binary_bytes(b"\xef\xbb\xbfhello") is False
+
+    def test_file_containing_real_replacement_char_is_text(self, file_ops):
+        # A log file that legitimately stores U+FFFD is valid UTF-8. The old
+        # text-layer check could not tell it from transport damage.
+        assert file_ops._is_likely_binary_bytes("log: \ufffd bad byte\n".encode("utf-8")) is False
+
+    def test_nul_byte_is_binary(self, file_ops):
+        assert file_ops._is_likely_binary_bytes(b"MZ\x00\x01text") is True
+
+    def test_elf_header_is_binary(self, file_ops):
+        assert file_ops._is_likely_binary_bytes(b"\x7fELF\x02\x01\x01\x00" + b"\x00" * 8) is True
+
+    def test_latin1_text_stays_read_only(self, file_ops):
+        # Mid-stream invalid UTF-8 (0xE9 = latin-1 é). Reading it through the
+        # replace-decoding transport would mojibake a read→edit→write
+        # round-trip, so it must stay flagged (the old check's guarantee).
+        assert file_ops._is_likely_binary_bytes(b"caf\xe9 au lait, plus padding") is True
+
+    def test_empty_sample_is_text(self, file_ops):
+        assert file_ops._is_likely_binary_bytes(b"") is False
+
+    def test_short_ascii_is_text(self, file_ops):
+        assert file_ops._is_likely_binary_bytes(b"hello\n") is False
+
+    def test_truncated_garbage_tail_after_invalid_prefix_is_binary(self, file_ops):
+        # Error near the end but the prefix itself is not clean UTF-8.
+        assert file_ops._is_likely_binary_bytes(b"\xff\xfe" + b"a" * 10 + b"\xe4") is True
+
+    # --- transport: _sample_file_bytes ------------------------------------
+
+    def test_sample_decodes_base64_transport(self, mock_env):
+        import base64 as b64
+        payload = ("汉字" * 400).encode("utf-8")[:1000]
+        mock_env.execute.return_value = {
+            "output": b64.b64encode(payload).decode() + "\n",
+            "returncode": 0,
+        }
+        ops = ShellFileOperations(mock_env)
+        assert ops._sample_file_bytes("/tmp/x.txt") == payload
+
+    def test_sample_falls_back_on_non_base64_output(self, mock_env):
+        mock_env.execute.return_value = {"output": "not base64 at all!!", "returncode": 0}
+        ops = ShellFileOperations(mock_env)
+        assert ops._sample_file_bytes("/tmp/x.txt") is None
+
+    def test_sample_falls_back_on_nonzero_exit(self, mock_env):
+        mock_env.execute.return_value = {"output": "", "returncode": 127}
+        ops = ShellFileOperations(mock_env)
+        assert ops._sample_file_bytes("/tmp/x.txt") is None
+
+    # --- integration: read_file over the mocked terminal ------------------
+
+    def _dispatch(self, cjk_bytes):
+        import base64 as b64
+
+        def side_effect(command, **kwargs):
+            if 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}
+            return {"output": "", "returncode": 0}
+
+        return side_effect
+
+    def test_read_file_returns_cjk_content_instead_of_binary_error(self, mock_env):
+        content = ("汉字测试" * 300).encode("utf-8")  # > 1000 bytes, cut mid-char
+        mock_env.execute.side_effect = self._dispatch(content)
+        ops = ShellFileOperations(mock_env)
+        result = ops.read_file("/tmp/notes-中文.txt")
+        assert result.is_binary is False
+        assert result.error is None
+        assert "汉字测试" in (result.content or "")
+
+    def test_read_file_still_blocks_nul_binaries(self, mock_env):
+        content = b"\x7fELF\x00\x00binarybinary" + b"\x00" * 100
+        mock_env.execute.side_effect = self._dispatch(content)
+        ops = ShellFileOperations(mock_env)
+        result = ops.read_file("/tmp/a.out")
+        assert result.is_binary is True
+

tools/file_operations.py

@@ -25,6 +25,8 @@
     result = file_ops.search("TODO", path=".", file_glob="*.py")
 """
 
+import base64
+import binascii
 import os
 import re
 import difflib
@@ -882,6 +884,72 @@ def _has_command(self, cmd: str) -> bool:
             self._command_cache[cmd] = result.stdout.strip() == 'yes'
         return self._command_cache[cmd]
 
+    def _sample_file_bytes(self, path: str, length: int = 1000):
+        """Fetch the first ``length`` raw bytes of a file through the terminal.
+
+        File operations run through a terminal backend (possibly remote), so
+        raw bytes cannot cross the transport directly — the terminal decodes
+        stdout with ``errors="replace"`` and manufactures U+FFFD at every
+        byte it cannot decode, including a multibyte character cut in half by
+        ``head -c``. Wrapping the sample in base64 lets the original bytes
+        survive the transport, so binary detection can happen at the byte
+        layer where it is well-defined (#80308 and friends).
+
+        Returns the sample bytes, or ``None`` when the transport could not
+        produce clean base64 (exotic shells without ``base64``); callers fall
+        back to the legacy text-sample heuristic in that case.
+        """
+        result = self._exec(
+            f"head -c {length} {self._escape_shell_arg(path)} 2>/dev/null | base64"
+        )
+        if result.exit_code != 0:
+            return None
+        encoded = _strip_terminal_fence_leaks(result.stdout)
+        encoded = "".join(encoded.split())
+        if not encoded:
+            return b""
+        if not re.fullmatch(r"[A-Za-z0-9+/]+={0,2}", encoded):
+            return None
+        try:
+            return base64.b64decode(encoded, validate=True)
+        except (binascii.Error, ValueError):
+            return None
+
+    @staticmethod
+    def _is_likely_binary_bytes(sample: bytes) -> bool:
+        """Byte-layer binary detection (the boundary for the #80308 class).
+
+        Contract: a file is text when its sample is valid UTF-8, allowing one
+        incomplete multibyte sequence at the very end (an artifact of cutting
+        the sample at a byte boundary, not a property of the file). Anything
+        else — NUL bytes, mid-stream invalid UTF-8 such as latin-1 or true
+        binaries — stays read-only, preserving the anti-mojibake guarantee
+        the old U+FFFD check existed for: a read→edit→write round-trip must
+        never rewrite undecodable bytes with replacement characters.
+
+        A file that legitimately *contains* U+FFFD (EF BF BD — e.g. logs of
+        lossy output) is valid UTF-8 and reads as text; the old text-layer
+        check misclassified it because it could not tell a stored replacement
+        character from a transport-manufactured one.
+        """
+        if not sample:
+            return False
+        if b"\x00" in sample:
+            return True
+        try:
+            sample.decode("utf-8")
+            return False
+        except UnicodeDecodeError as exc:
+            # UTF-8 sequences are at most 4 bytes: an error starting in the
+            # last 3 bytes with a clean prefix is a boundary cut, not binary.
+            if exc.start >= len(sample) - 3:
+                try:
+                    sample[: exc.start].decode("utf-8")
+                    return False
+                except UnicodeDecodeError:
+                    pass
+            return True
+
     def _is_likely_binary(self, path: str, content_sample: str = None) -> bool:
         """
         Check if a file is likely binary.
@@ -1188,12 +1256,19 @@ def read_file(self, path: str, offset: int = 1, limit: int = 2000) -> ReadResult
                 ),
             )
 
-        # Read a sample to check for binary content
-        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)
-
-        if self._is_likely_binary(path, sample_output):
+        # 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)
+        if sample_bytes is not None:
+            ext_binary = os.path.splitext(path)[1].lower() in BINARY_EXTENSIONS
+            is_binary = ext_binary or 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 ReadResult(
                 is_binary=True,
                 file_size=file_size,
@@ -1307,9 +1382,15 @@ def read_file_raw(self, path: str) -> ReadResult:
             file_size = 0
         if self._is_image(path):
             return ReadResult(is_image=True, is_binary=True, file_size=file_size)
-        sample_result = self._exec(f"head -c 1000 {self._escape_shell_arg(path)} 2>/dev/null")
-        sample_output = _strip_terminal_fence_leaks(sample_result.stdout)
-        if self._is_likely_binary(path, sample_output):
+        sample_bytes = self._sample_file_bytes(path)
+        if sample_bytes is not None:
+            ext_binary = os.path.splitext(path)[1].lower() in BINARY_EXTENSIONS
+            is_binary = ext_binary or self._is_likely_binary_bytes(sample_bytes)
+        else:
+            sample_result = self._exec(f"head -c 1000 {self._escape_shell_arg(path)} 2>/dev/null")
+            sample_output = _strip_terminal_fence_leaks(sample_result.stdout)
+            is_binary = self._is_likely_binary(path, sample_output)
+        if is_binary:
             return ReadResult(
                 is_binary=True, file_size=file_size,
                 error="Binary file — cannot display as text."