fix(file-reads): read UTF-16 text files in read_file, summarize_file, and grep_code

REQUIREMENTS.md

@@ -2,7 +2,7 @@
 
 > **Note:** This document is automatically generated and verified against the live test suite by `scripts/generate_requirements.py` and `tests/backend/test_requirements_sync.py`.
 
-**Test Verification Baseline:** **1172 Automated Tests** (808 Pytest Backend + 314 Vitest Frontend + 50 Playwright E2E).
+**Test Verification Baseline:** **1176 Automated Tests** (812 Pytest Backend + 314 Vitest Frontend + 50 Playwright E2E).
 
 ---
 
@@ -1393,6 +1393,12 @@ over the PostgresConnectionAdapter using an initialized test engine._
 - `test_include_permalinks_returns_them_on_every_node`
 - `test_compact_answer_drops_the_four_node_permalinks`
 
+#### `tests/test_utf16_file_reads.py` (4 tests)
+- `test_text_samples_are_not_binary`
+- `test_real_binary_sample_is_binary`
+- `test_read_file_returns_the_text_of_a_utf16_file`
+- `test_read_file_still_refuses_real_binary_files`
+
 #### `tests/test_webhooks.py` (12 tests)
 - `test_github_webhook_no_secret`
 - `test_github_webhook_with_secret_valid_and_invalid`

app/services/adr.py

@@ -3,6 +3,7 @@
 import logging
 from typing import Dict, Any, Optional
 import frontmatter
+from app.services.text_decoding import read_source_text
 
 from app.services.database import upsert_adr
 
@@ -158,8 +159,7 @@ def sync_adr_file(filepath: str, repo: str, content: Optional[str] = None) -> Op
             norm_fp = os.path.normpath(os.path.abspath(filepath))
             if not os.path.exists(norm_fp):
                 return None
-            with open(norm_fp, "r", encoding="utf-8", errors="ignore") as f:
-                content = f.read()
+            content = read_source_text(norm_fp)
 
         parsed = parse_adr_markdown(content, filepath, repo)
         res = upsert_adr(

app/services/code_grep_scanner.py

@@ -5,6 +5,7 @@
 from typing import Any, Callable, Dict, List, Optional, Tuple
 
 from app.services.git_manager import format_git_permalink
+from app.services.text_decoding import read_source_text
 
 logger = logging.getLogger("contextcortex.file_reader")
 
@@ -117,8 +118,7 @@ def _matches_glob(path_str: str) -> bool:
         try:
             if not os.path.isfile(abs_p) or is_binary_fn(abs_p):
                 continue
-            with open(abs_p, "r", encoding="utf-8", errors="replace") as f:
-                file_text = f.read()
+            file_text = read_source_text(abs_p)
         except Exception:
             continue
 

app/services/file_reader.py

@@ -12,6 +12,7 @@
     validate_path_security,
 )
 from app.services.local_storage import get_default_storage_path
+from app.services.text_decoding import is_binary_file, read_source_text
 
 logger = logging.getLogger("contextcortex.file_reader")
 
@@ -55,10 +56,8 @@ def resolve_safe_path(self, path: str, repo: Optional[str] = None) -> Tuple[str,
         )
 
     def is_binary_file(self, abs_path: str) -> bool:
-        """Detects binary files by checking for null bytes in the initial sample."""
-        with open(abs_path, "rb") as f:
-            chunk = f.read(8192)
-            return b"\x00" in chunk
+        """Detects binary files: NUL bytes in the initial sample that are not UTF-16/UTF-32 text."""
+        return is_binary_file(abs_path)
 
     def _slice_lines(
         self,
@@ -229,8 +228,7 @@ def read_file(
             raise ValueError(f"Cannot read binary file: {path}")
 
         size_bytes = os.path.getsize(abs_path)
-        with open(abs_path, "r", encoding="utf-8", errors="replace") as f:
-            text = f.read()
+        text = read_source_text(abs_path)
 
         if not text:
             return {

app/services/local_storage.py

@@ -14,6 +14,7 @@
 from app.services.indexing.processor import MAX_FILE_SIZE_BYTES, process_file_content
 from app.services.indexing.state import trigger_list_changed_notification
 from app.services.pdf_extractor import MAX_PDF_SIZE_BYTES, extract_pdf_pages
+from app.services.text_decoding import read_source_text
 
 logger = logging.getLogger("contextcortex.storage")
 
@@ -153,8 +154,7 @@ def index_file(
             doc_type = "pdf"
             pdf_errors = pdf_res.errors
         else:
-            with open(abs_path, "r", encoding="utf-8", errors="replace") as f:
-                content = f.read()
+            content = read_source_text(abs_path)
             doc_type = "doc" if lower_abs.endswith(DOC_TEXT_EXTENSIONS) else "code"
 
         points, ast_symbols, summary_tuple, ast_rel, api_routes, api_calls = process_file_content(
@@ -395,8 +395,7 @@ def read_file_content(self, rel_path: str) -> Dict[str, Any]:
             return res
 
         try:
-            with open(target_path, "r", encoding="utf-8", errors="replace") as f:
-                text = f.read()
+            text = read_source_text(target_path)
         except Exception:
             text = ""
 

app/services/text_decoding.py

@@ -58,6 +58,22 @@ def _guess_utf16_without_bom(sample: bytes) -> Optional[str]:
     return None
 
 
+def is_binary_sample(sample: bytes) -> bool:
+    """True for binary data: the sample has NUL bytes and is not UTF-16/UTF-32 text (with or without a BOM)."""
+    if b"\x00" not in sample:
+        return False
+    bom_match = _detect_bom(sample)
+    if bom_match and bom_match[1] != "utf-8":
+        return False
+    return _guess_utf16_without_bom(sample) is None
+
+
+def is_binary_file(path: str, sample_size: int = 8192) -> bool:
+    """Reads the first bytes of a file and applies `is_binary_sample`."""
+    with open(path, "rb") as handle:
+        return is_binary_sample(handle.read(sample_size))
+
+
 def decode_source_bytes(raw: bytes) -> str:
     """Decodes file bytes as UTF-8, UTF-16, or UTF-32 (BOM or UTF-16 without BOM) and removes NUL characters."""
     bom_match = _detect_bom(raw)

app/services/topology/helpers.py

@@ -3,6 +3,7 @@
 import logging
 from typing import Optional, Dict, Any
 from app.services.git_manager import format_git_permalink
+from app.services.text_decoding import read_source_text
 
 logger = logging.getLogger("contextcortex.topology")
 
@@ -57,8 +58,7 @@ def _read_code_snippet(
         return None
 
     try:
-        with open(found_path, "r", encoding="utf-8", errors="replace") as f:
-            lines = f.readlines()
+        lines = read_source_text(found_path).splitlines(keepends=True)
 
         s_line = max(1, start_line or 1)
         e_line = min(len(lines), end_line or len(lines))

tests/test_utf16_file_reads.py

@@ -0,0 +1,64 @@
+"""Tests that read_file, grep_code, and the binary check treat UTF-16/UTF-32 text files as text, not binary."""
+
+import codecs
+
+import pytest
+
+from app.services.database.engine import get_db_engine, init_db
+from app.services.file_reader import FileReaderService
+from app.services.text_decoding import is_binary_sample
+
+PROC = "CREATE PROCEDURE dbo.GetLatestVersion\r\nAS\r\nSELECT TOP 1 Version FROM dbo.Releases\r\n"
+
+
+@pytest.mark.parametrize(
+    "sample",
+    [
+        codecs.BOM_UTF16_LE + PROC.encode("utf-16-le"),
+        PROC.encode("utf-16-le"),
+        PROC.encode("utf-16-be"),
+        codecs.BOM_UTF32_LE + PROC.encode("utf-32-le"),
+        PROC.encode("utf-8"),
+    ],
+    ids=["utf16le-bom", "utf16le-no-bom", "utf16be-no-bom", "utf32le-bom", "utf8"],
+)
+def test_text_samples_are_not_binary(sample):
+    assert is_binary_sample(sample) is False
+
+
+def test_real_binary_sample_is_binary():
+    png_like = b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR\x00\x00\x01\x00\x00\x00\x01\x00\x08\x06\x00\x00\x00\x5c\x72\xa8\x66" * 8
+    assert is_binary_sample(png_like) is True
+    assert is_binary_sample(codecs.BOM_UTF8 + b"text\x00\x01\x02\x03binary") is True
+
+
+@pytest.fixture
+def storage_with_utf16_proc(tmp_path, monkeypatch):
+    db_file = tmp_path / "utf16_reads.db"
+    db_url = f"sqlite:///{db_file}"
+    monkeypatch.setenv("DATABASE_URL", db_url)
+    monkeypatch.setattr("app.services.database.connection.CACHE_DB_PATH", str(db_file))
+    monkeypatch.setattr("app.services.database.CACHE_DB_PATH", str(db_file), raising=False)
+    init_db(engine=get_db_engine(db_url, reset=True))
+
+    storage_dir = tmp_path / "storage"
+    (storage_dir / "sql").mkdir(parents=True)
+    monkeypatch.setenv("LOCAL_STORAGE_PATH", str(storage_dir))
+    monkeypatch.setattr("app.services.local_storage.LOCAL_STORAGE_PATH", str(storage_dir), raising=False)
+    (storage_dir / "sql" / "GetLatestVersion.sql").write_bytes(codecs.BOM_UTF16_LE + PROC.encode("utf-16-le"))
+    return storage_dir
+
+
+def test_read_file_returns_the_text_of_a_utf16_file(storage_with_utf16_proc):
+    service = FileReaderService(storage_root=str(storage_with_utf16_proc))
+    result = service.read_file("sql/GetLatestVersion.sql")
+    assert result["content"].splitlines() == PROC.splitlines()
+    assert result["total_lines"] == 3
+    assert "\x00" not in result["content"]
+
+
+def test_read_file_still_refuses_real_binary_files(storage_with_utf16_proc):
+    (storage_with_utf16_proc / "logo.png").write_bytes(b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR\x00\x00\x01\x00" * 8)
+    service = FileReaderService(storage_root=str(storage_with_utf16_proc))
+    with pytest.raises(ValueError, match="binary"):
+        service.read_file("logo.png")