Port from MoonshotAI/kimi-code#2647: read UTF-16 text files by transcoding to UTF-8
UTF-16 text files (Windows Notepad .txt, PowerShell > redirects) were refused as binary: the terminal env decodes stdout as UTF-8 with errors=replace, so their content arrived mangled with U+FFFD and tripped the binary guard. ShellFileOperations.read_file now probes the raw bytes via the backend's Python when the binary guard fires: a BOM or the zero-byte parity heuristic (derived from VS Code's encoding sniffer, tolerant of mixed Latin/CJK content) identifies UTF-16 LE/BE, and the file is transcoded to UTF-8 with CRLF normalized and the BOM stripped. Real binaries (zeros at both parities), binary extensions, files over 10 MiB, and legacy 8-bit encodings (GBK, Big5) still refuse — a wrong silent guess is worse than a clear refusal. Works on every shell backend (local/docker/ssh) since the probe runs via python3 -c. Tests run against a real LocalEnvironment (E2E, no mocks); sabotage run confirmed 6/9 fail without the fix.
This commit is contained in:
Teknium
parent
226b095a59
commit
6c564a81db
|
|
@ -0,0 +1,106 @@
|
|||
"""Tests for UTF-16 text file reading (transcode to UTF-8).
|
||||
|
||||
Ported from MoonshotAI/kimi-code#2647: UTF-16 text files (Windows Notepad
|
||||
.txt, PowerShell `>` redirects) previously tripped the binary-file guard
|
||||
because the terminal env decodes stdout as UTF-8 with errors="replace",
|
||||
mangling the content with U+FFFD. ShellFileOperations now probes raw bytes
|
||||
via the backend's Python and transcodes UTF-16 (BOM or zero-byte parity
|
||||
heuristic) to UTF-8.
|
||||
|
||||
These run against a real LocalEnvironment so the actual shell + subprocess
|
||||
path executes (E2E, no mocks).
|
||||
"""
|
||||
|
||||
import pytest
|
||||
|
||||
from tools.environments.local import LocalEnvironment
|
||||
from tools.file_operations import ShellFileOperations
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def fops(tmp_path):
|
||||
env = LocalEnvironment(cwd=str(tmp_path))
|
||||
return ShellFileOperations(env, cwd=str(tmp_path))
|
||||
|
||||
|
||||
def _write(tmp_path, name: str, text: str, encoding: str) -> str:
|
||||
p = tmp_path / name
|
||||
p.write_bytes(text.encode(encoding))
|
||||
return str(p)
|
||||
|
||||
|
||||
class TestUtf16Read:
|
||||
def test_utf16le_bom(self, fops, tmp_path):
|
||||
path = _write(tmp_path, "notepad.txt", "hello\nworld\n", "utf-16-le")
|
||||
# prepend BOM manually via utf-16 (writes native-endian BOM); use explicit
|
||||
raw = "\ufefflíne one\nlíne two\n".encode("utf-16-le")
|
||||
(tmp_path / "bom-le.txt").write_bytes(raw)
|
||||
result = fops.read_file(str(tmp_path / "bom-le.txt"))
|
||||
assert result.error is None
|
||||
assert "líne one" in result.content
|
||||
assert "\ufeff" not in result.content
|
||||
assert "Transcoded from UTF-16-LE" in (result.hint or "")
|
||||
|
||||
def test_utf16be_bom(self, fops, tmp_path):
|
||||
raw = "\ufeffbig endian text\nsecond line\n".encode("utf-16-be")
|
||||
(tmp_path / "bom-be.txt").write_bytes(raw)
|
||||
result = fops.read_file(str(tmp_path / "bom-be.txt"))
|
||||
assert result.error is None
|
||||
assert "big endian text" in result.content
|
||||
assert "UTF-16-BE" in (result.hint or "")
|
||||
|
||||
def test_utf16le_bomless(self, fops, tmp_path):
|
||||
path = _write(tmp_path, "bomless.txt", "plain ascii saved as utf16\n", "utf-16-le")
|
||||
result = fops.read_file(path)
|
||||
assert result.error is None
|
||||
assert "plain ascii saved as utf16" in result.content
|
||||
|
||||
def test_utf16le_mixed_cjk(self, fops, tmp_path):
|
||||
# Mixed Latin/CJK: CJK UTF-16 units carry no zero byte — the parity
|
||||
# heuristic must still detect from the Latin characters present.
|
||||
path = _write(tmp_path, "mixed.txt", "log: 你好世界 done\n", "utf-16-le")
|
||||
result = fops.read_file(path)
|
||||
assert result.error is None
|
||||
assert "你好世界" in result.content
|
||||
|
||||
def test_crlf_normalized(self, fops, tmp_path):
|
||||
path = _write(tmp_path, "crlf.txt", "\ufeffa\r\nb\r\nc", "utf-16-le")
|
||||
result = fops.read_file(path)
|
||||
assert result.error is None
|
||||
assert result.total_lines == 3
|
||||
assert "1|a" in result.content and "2|b" in result.content
|
||||
|
||||
def test_pagination(self, fops, tmp_path):
|
||||
text = "\ufeff" + "\n".join(f"line{i}" for i in range(1, 21)) + "\n"
|
||||
path = str(tmp_path / "paged.txt")
|
||||
(tmp_path / "paged.txt").write_bytes(text.encode("utf-16-le"))
|
||||
result = fops.read_file(path, offset=5, limit=3)
|
||||
assert result.error is None
|
||||
assert "5|line5" in result.content
|
||||
assert "7|line7" in result.content
|
||||
assert "line8" not in result.content
|
||||
assert result.truncated is True
|
||||
assert "offset=8" in (result.hint or "")
|
||||
|
||||
def test_real_binary_still_refused(self, fops, tmp_path):
|
||||
# Zero bytes at BOTH parities → not UTF-16 → stays binary.
|
||||
p = tmp_path / "blob.dat"
|
||||
p.write_bytes(bytes([0x00, 0x01, 0x02, 0x00, 0xFF, 0x00, 0x00, 0xFE]) * 40)
|
||||
result = fops.read_file(str(p))
|
||||
assert result.is_binary is True
|
||||
assert result.error is not None
|
||||
|
||||
def test_binary_extension_not_rescued(self, fops, tmp_path):
|
||||
# A .png is never probed for UTF-16 even if its bytes look like it.
|
||||
p = tmp_path / "img.png"
|
||||
p.write_bytes("fake image".encode("utf-16-le"))
|
||||
result = fops.read_file(str(p))
|
||||
assert result.is_binary is True or result.error is not None
|
||||
|
||||
def test_utf8_file_unaffected(self, fops, tmp_path):
|
||||
p = tmp_path / "normal.txt"
|
||||
p.write_text("just utf-8\n", encoding="utf-8")
|
||||
result = fops.read_file(str(p))
|
||||
assert result.error is None
|
||||
assert "just utf-8" in result.content
|
||||
assert "Transcoded" not in (result.hint or "")
|
||||
|
|
@ -29,6 +29,7 @@ import os
|
|||
import re
|
||||
import difflib
|
||||
import hashlib
|
||||
import json
|
||||
from abc import ABC, abstractmethod
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Optional, List, Dict, Any, ClassVar
|
||||
|
|
@ -1140,6 +1141,112 @@ class ShellFileOperations(FileOperations):
|
|||
# READ Implementation
|
||||
# =========================================================================
|
||||
|
||||
# UTF-16 rescue constants (ported from MoonshotAI/kimi-code#2647,
|
||||
# detection derived from VS Code's encoding sniffer): sample the leading
|
||||
# bytes; trust a BOM first, then a zero-byte parity heuristic — zeros
|
||||
# clustering at odd indices mean UTF-16 LE (`0xAA 0x00`), at even indices
|
||||
# UTF-16 BE (`0x00 0xAA`). Only the *placement* of zeros is checked, not
|
||||
# density, so mixed Latin/CJK content (whose CJK units carry no zero
|
||||
# byte) still detects. Zeros at both parities, or a single isolated
|
||||
# zero, mean real binary. Legacy 8-bit encodings (GBK, Big5, ...) are
|
||||
# never guessed — a wrong silent guess is worse than a clear refusal.
|
||||
_UTF16_MAX_BYTES = 10 * 1024 * 1024
|
||||
_UTF16_SAMPLE_BYTES = 512
|
||||
|
||||
def _try_read_utf16(self, path: str, offset: int, limit: int,
|
||||
file_size: int) -> "Optional[ReadResult]":
|
||||
"""Attempt to read ``path`` as UTF-16 text, transcoded to UTF-8.
|
||||
|
||||
Returns a populated ``ReadResult`` when the file is UTF-16 (BOM or
|
||||
zero-byte parity heuristic), else ``None`` so the caller falls back
|
||||
to the binary-file error. Files over 10 MiB are not rescued.
|
||||
``path`` must already be expanded (caller ran ``_expand_path``).
|
||||
"""
|
||||
# Extensions that are definitively binary (images, archives, ...)
|
||||
# never contain UTF-16 text worth rescuing — skip the subprocess.
|
||||
ext = os.path.splitext(path)[1].lower()
|
||||
if ext in BINARY_EXTENSIONS:
|
||||
return None
|
||||
if file_size > self._UTF16_MAX_BYTES:
|
||||
return None
|
||||
|
||||
snippet = (
|
||||
"import sys, json, os\n"
|
||||
f"p = {path!r}\n"
|
||||
f"offset = {int(offset)}\n"
|
||||
f"limit = {int(limit)}\n"
|
||||
f"MAX = {self._UTF16_MAX_BYTES}\n"
|
||||
f"SAMPLE = {self._UTF16_SAMPLE_BYTES}\n"
|
||||
"try:\n"
|
||||
" size = os.path.getsize(p)\n"
|
||||
" if size > MAX:\n"
|
||||
" print('HERMES_UTF16:NO'); sys.exit(0)\n"
|
||||
" with open(p, 'rb') as f:\n"
|
||||
" data = f.read()\n"
|
||||
" sample = data[:SAMPLE]\n"
|
||||
" enc = None\n"
|
||||
" if sample[:2] == b'\\xfe\\xff':\n"
|
||||
" enc = 'utf-16-be'\n"
|
||||
" elif sample[:2] == b'\\xff\\xfe':\n"
|
||||
" enc = 'utf-16-le'\n"
|
||||
" else:\n"
|
||||
" odd = sum(1 for i in range(1, len(sample), 2) if sample[i] == 0)\n"
|
||||
" even = sum(1 for i in range(0, len(sample), 2) if sample[i] == 0)\n"
|
||||
" if even == 0 and odd >= 2:\n"
|
||||
" enc = 'utf-16-le'\n"
|
||||
" elif odd == 0 and even >= 2:\n"
|
||||
" enc = 'utf-16-be'\n"
|
||||
" if enc is None:\n"
|
||||
" print('HERMES_UTF16:NO'); sys.exit(0)\n"
|
||||
" text = data.decode(enc, 'replace')\n"
|
||||
" if text[:1] == '\\ufeff':\n"
|
||||
" text = text[1:]\n"
|
||||
" text = text.replace('\\r\\n', '\\n')\n"
|
||||
" lines = text.split('\\n')\n"
|
||||
" total = len(lines)\n"
|
||||
" sel = lines[offset - 1: offset - 1 + limit]\n"
|
||||
" out = {'total_lines': total, 'encoding': enc,\n"
|
||||
" 'content': '\\n'.join(sel)}\n"
|
||||
" print('HERMES_UTF16:OK')\n"
|
||||
" print(json.dumps(out, ensure_ascii=True))\n"
|
||||
"except Exception:\n"
|
||||
" print('HERMES_UTF16:NO'); sys.exit(0)\n"
|
||||
)
|
||||
|
||||
result = self._exec(f"python3 -c {self._escape_shell_arg(snippet)}")
|
||||
if result.exit_code != 0 and "python3" in (result.stdout or ""):
|
||||
result = self._exec(f"python -c {self._escape_shell_arg(snippet)}")
|
||||
|
||||
stdout = _strip_terminal_fence_leaks(result.stdout or "")
|
||||
marker = stdout.find("HERMES_UTF16:OK")
|
||||
if result.exit_code != 0 or marker < 0:
|
||||
return None
|
||||
payload = stdout[marker + len("HERMES_UTF16:OK"):].strip()
|
||||
try:
|
||||
data = json.loads(payload.split("\n", 1)[0] if "\n" in payload else payload)
|
||||
content = data["content"]
|
||||
total_lines = int(data["total_lines"])
|
||||
encoding = str(data.get("encoding", "utf-16"))
|
||||
except (ValueError, KeyError, TypeError):
|
||||
return None
|
||||
|
||||
end_line = offset + limit - 1
|
||||
truncated = total_lines > end_line
|
||||
hint_parts = [f"Transcoded from {encoding.upper()} to UTF-8 for display. "
|
||||
"Text edits via patch/write_file would re-encode as UTF-8."]
|
||||
if truncated:
|
||||
hint_parts.append(
|
||||
f"Use offset={end_line + 1} to continue reading "
|
||||
f"(showing {offset}-{end_line} of {total_lines} lines)"
|
||||
)
|
||||
return ReadResult(
|
||||
content=self._add_line_numbers(content, offset),
|
||||
total_lines=total_lines,
|
||||
file_size=file_size,
|
||||
truncated=truncated,
|
||||
hint=" ".join(hint_parts),
|
||||
)
|
||||
|
||||
def read_file(self, path: str, offset: int = 1, limit: int = 2000) -> ReadResult:
|
||||
"""
|
||||
Read a file with pagination, binary detection, and line numbers.
|
||||
|
|
@ -1194,6 +1301,16 @@ class ShellFileOperations(FileOperations):
|
|||
sample_output = _strip_terminal_fence_leaks(sample_result.stdout)
|
||||
|
||||
if self._is_likely_binary(path, sample_output):
|
||||
# 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,
|
||||
|
|
|
|||
Loading…
Reference in New Issue