From dcd51f16fe566fb7caef6cd801320d433a25a706 Mon Sep 17 00:00:00 2001 From: Julius Brussee <104168679+JuliusBrussee@users.noreply.github.com> Date: Tue, 21 Jul 2026 02:00:59 +0200 Subject: [PATCH] =?UTF-8?q?fix(compress):=20stop=20Windows=20data=20loss?= =?UTF-8?q?=20=E2=80=94=20UTF-8=20+=20atomic=20writes=20(#652,=20#655,=20#?= =?UTF-8?q?686)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Every read/write resolved to the locale codec (cp1252/cp949 on Windows): non-ASCII files were silently mojibake'd, and because Path.write_text truncates before encoding, a UnicodeEncodeError left the target at 0 bytes. The backup readback check couldn't catch it — it read back with the same wrong codec. - encoding=utf-8 pinned on every I/O call site (compress, validate, detect, benchmark); validate now decodes strict — it's the fidelity gate - write_text_atomic: encode first, temp file in same dir, fsync, preserve permissions, os.replace; temp unlinked on any failure - fix-retry pass gains the same empty-output guard as the first pass - fix-retry preamble leak (#588): output must start at the original's structural anchor (frontmatter/heading) or the attempt is rejected - primary-write failure now prints the backup path — users hitting the crash had no idea a backup existed - extract_inline_codes: strip fences via the CommonMark-aware extractor; the old column-0 regex leaked indented fences into inline-code pairing, causing false validation failures (extracted from PR #619's diagnosis) - SKILL.md/README/SECURITY corrected: backups live in the out-of-tree data dir (#420), not beside the source file Supersedes PRs #683 #678 #626 #534 and the fence fix from #619 with a local implementation. 58 python tests. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_016ySX6TBWZuvFze4ajf7Hpf --- .../caveman/skills/caveman-compress/SKILL.md | 4 +- .../caveman-compress/scripts/benchmark.py | 4 +- .../caveman-compress/scripts/compress.py | 84 ++++++++++++- .../skills/caveman-compress/scripts/detect.py | 2 +- .../caveman-compress/scripts/validate.py | 14 ++- skills/caveman-compress/README.md | 2 +- skills/caveman-compress/SECURITY.md | 2 +- skills/caveman-compress/SKILL.md | 4 +- skills/caveman-compress/scripts/benchmark.py | 4 +- skills/caveman-compress/scripts/compress.py | 84 ++++++++++++- skills/caveman-compress/scripts/detect.py | 2 +- skills/caveman-compress/scripts/validate.py | 14 ++- tests/test_compress_safety.py | 118 +++++++++++++++++- tests/test_validate_inline.py | 11 ++ 14 files changed, 318 insertions(+), 31 deletions(-) diff --git a/plugins/caveman/skills/caveman-compress/SKILL.md b/plugins/caveman/skills/caveman-compress/SKILL.md index 00ce454..0b95aab 100644 --- a/plugins/caveman/skills/caveman-compress/SKILL.md +++ b/plugins/caveman/skills/caveman-compress/SKILL.md @@ -11,7 +11,7 @@ description: > ## Purpose -Compress natural language files (CLAUDE.md, todos, preferences) into caveman-speak to reduce input tokens. Compressed version overwrites original. Human-readable backup saved as `.original.md`. +Compress natural language files (CLAUDE.md, todos, preferences) into caveman-speak to reduce input tokens. Compressed version overwrites original. Human-readable backup saved as `.original.md`, but NOT beside the source file — it lives in an out-of-tree data dir (`$XDG_DATA_HOME/caveman-compress/backups//`, or `%LOCALAPPDATA%\caveman-compress\backups\\` on Windows) so skill auto-loaders don't re-ingest it as a live file. ## Trigger @@ -107,5 +107,5 @@ Compressed: - NEVER modify: .py, .js, .ts, .json, .yaml, .yml, .toml, .env, .lock, .css, .html, .xml, .sql, .sh - If file has mixed content (prose + code), compress ONLY the prose sections - If unsure whether something is code or prose, leave it unchanged -- Original file is backed up as FILE.original.md before overwriting +- Original file is backed up as FILE.original.md before overwriting — in the out-of-tree backup data dir (see Purpose), not beside the source file - Never compress FILE.original.md (skip it) diff --git a/plugins/caveman/skills/caveman-compress/scripts/benchmark.py b/plugins/caveman/skills/caveman-compress/scripts/benchmark.py index f9e2ee0..97d081b 100644 --- a/plugins/caveman/skills/caveman-compress/scripts/benchmark.py +++ b/plugins/caveman/skills/caveman-compress/scripts/benchmark.py @@ -23,8 +23,8 @@ def count_tokens(text): def benchmark_pair(orig_path: Path, comp_path: Path): - orig_text = orig_path.read_text() - comp_text = comp_path.read_text() + orig_text = orig_path.read_text(encoding="utf-8", errors="ignore") + comp_text = comp_path.read_text(encoding="utf-8", errors="ignore") orig_tokens = count_tokens(orig_text) comp_tokens = count_tokens(comp_text) diff --git a/plugins/caveman/skills/caveman-compress/scripts/compress.py b/plugins/caveman/skills/caveman-compress/scripts/compress.py index e93d934..80da520 100644 --- a/plugins/caveman/skills/caveman-compress/scripts/compress.py +++ b/plugins/caveman/skills/caveman-compress/scripts/compress.py @@ -9,8 +9,10 @@ Usage: import os import re import shutil +import stat import subprocess import sys +import tempfile from pathlib import Path from typing import List @@ -110,6 +112,59 @@ def strip_llm_wrapper(text: str) -> str: return m.group(2) return text + +def write_text_atomic(path: Path, text: str) -> None: + """Write ``text`` to ``path`` atomically as UTF-8. + + Path.write_text() truncates the destination before encoding the string — + a UnicodeEncodeError (or any other failure) partway through leaves a + 0-byte file, destroying whatever was there before (issue #655). Encode + first, write the bytes to a sibling temp file, fsync, then os.replace() + so the destination only ever moves from one complete, valid file to + another. Preserves the original file's permission bits across the swap. + """ + data = text.encode("utf-8") + fd, tmp_name = tempfile.mkstemp( + dir=str(path.parent), prefix=path.name + ".", suffix=".tmp" + ) + tmp_path = Path(tmp_name) + try: + with os.fdopen(fd, "wb") as f: + f.write(data) + f.flush() + os.fsync(f.fileno()) + if path.exists(): + os.chmod(tmp_path, stat.S_IMODE(path.stat().st_mode)) + os.replace(tmp_path, path) + except Exception: + try: + tmp_path.unlink() + except OSError: + pass + raise + + +def first_nonblank_line(text: str) -> str: + """Return the first non-blank line, stripped — used to detect a prose + preamble smuggled in ahead of the real content (issue #588).""" + for line in text.splitlines(): + if line.strip(): + return line.strip() + return "" + + +def _write_target(filepath: Path, text: str, backup_path: Path) -> None: + """Write to the target file, surfacing the backup location if the write + itself fails. write_text_atomic already leaves the target untouched on + failure, but the caller still needs to know where the pre-compression + original lives instead of being left to guess (issue #652).""" + try: + write_text_atomic(filepath, text) + except Exception: + print(f"❌ Write to {filepath} failed. Original preserved at backup: {backup_path}") + raise + + from .detect import should_compress from .validate import validate @@ -246,7 +301,7 @@ def compress_file(filepath: Path) -> bool: print("Skipping (not natural language)") return False - original_text = filepath.read_text(errors="ignore") + original_text = filepath.read_text(encoding="utf-8", errors="ignore") # Store backup outside the source directory so skill auto-loaders don't # re-ingest the `.original.md` copy as a live file. Mirror the source's # parent-dir name + stem under a platform-aware base to reduce collisions. @@ -300,8 +355,8 @@ def compress_file(filepath: Path) -> bool: # touching the input file. If the filesystem dropped bytes (encoding, # antivirus, disk full), unlink the bad backup and abort instead of # leaving the user with a corrupt backup + compressed primary. - backup_path.write_text(original_text) - backup_readback = backup_path.read_text(errors="ignore") + write_text_atomic(backup_path, original_text) + backup_readback = backup_path.read_text(encoding="utf-8", errors="ignore") if backup_readback != original_text: print(f"❌ Backup write verification failed: {backup_path}") print(" In-memory original differs from on-disk backup. Aborting before touching the input file.") @@ -310,7 +365,7 @@ def compress_file(filepath: Path) -> bool: except OSError: pass return False - filepath.write_text(compressed) + _write_target(filepath, compressed, backup_path) # Step 2: Validate + Retry for attempt in range(MAX_RETRIES): @@ -328,7 +383,7 @@ def compress_file(filepath: Path) -> bool: if attempt == MAX_RETRIES - 1: # Restore original on failure - filepath.write_text(original_text) + _write_target(filepath, original_text, backup_path) backup_path.unlink(missing_ok=True) print("❌ Failed after retries — original restored") return False @@ -337,6 +392,23 @@ def compress_file(filepath: Path) -> bool: compressed = call_claude( build_fix_prompt(original_text, compressed, result.errors) ) - filepath.write_text(compressed) + + if compressed is None or not compressed.strip(): + print("❌ Fix attempt aborted: Claude returned an empty response.") + print(" Skipping this attempt.") + continue + + # Guard against a prose preamble smuggled in ahead of the real fixed + # content (issue #588). Only enforced when the original starts with a + # structural anchor (frontmatter `---` or a heading) — plain-prose + # first lines get legitimately rewritten by compression, and requiring + # them verbatim would reject every valid fix. + anchor = first_nonblank_line(original_text) + if anchor.startswith(("---", "#")) and first_nonblank_line(compressed) != anchor: + print("❌ Fix attempt aborted: output does not start with the original's first line.") + print(" Possible preamble leak. Skipping this attempt.") + continue + + _write_target(filepath, compressed, backup_path) return True diff --git a/plugins/caveman/skills/caveman-compress/scripts/detect.py b/plugins/caveman/skills/caveman-compress/scripts/detect.py index 5114d04..6a468d5 100644 --- a/plugins/caveman/skills/caveman-compress/scripts/detect.py +++ b/plugins/caveman/skills/caveman-compress/scripts/detect.py @@ -90,7 +90,7 @@ def detect_file_type(filepath: Path) -> str: # Extensionless files (like CLAUDE.md, TODO) — check content if not ext: try: - text = filepath.read_text(errors="ignore") + text = filepath.read_text(encoding="utf-8", errors="ignore") except (OSError, PermissionError): return "unknown" diff --git a/plugins/caveman/skills/caveman-compress/scripts/validate.py b/plugins/caveman/skills/caveman-compress/scripts/validate.py index dc07307..dcd2a5d 100644 --- a/plugins/caveman/skills/caveman-compress/scripts/validate.py +++ b/plugins/caveman/skills/caveman-compress/scripts/validate.py @@ -28,7 +28,7 @@ class ValidationResult: def read_file(path: Path) -> str: - return path.read_text(errors="ignore") + return path.read_text(encoding="utf-8") # ---------- Extractors ---------- @@ -95,8 +95,16 @@ def count_bullets(text): def extract_inline_codes(text): - text_without_fences = re.sub(r"^```[\s\S]*?^```", "", text, flags=re.MULTILINE) - text_without_fences = re.sub(r"^~~~[\s\S]*?^~~~", "", text_without_fences, flags=re.MULTILINE) + """Backtick-delimited inline spans, with fenced code blocks stripped first. + + Previously used a column-0-anchored regex to strip fences, which misses + fences indented 1-3 spaces (valid CommonMark). Reuse extract_code_blocks + (FENCE_OPEN_REGEX-based, indentation-aware) instead so an indented fence's + body backticks don't leak into inline-code pairing. + """ + text_without_fences = text + for block in extract_code_blocks(text): + text_without_fences = text_without_fences.replace(block, "", 1) return re.findall(r"`([^`]+)`", text_without_fences) diff --git a/skills/caveman-compress/README.md b/skills/caveman-compress/README.md index 3ef6922..4aa149a 100644 --- a/skills/caveman-compress/README.md +++ b/skills/caveman-compress/README.md @@ -25,7 +25,7 @@ CLAUDE.md ← compressed (Claude reads this — fewer tokens every sess CLAUDE.original.md ← human-readable backup (you edit this) ``` -Original never lost. You can read and edit `.original.md`. Run skill again to re-compress after edits. +Original never lost. Backup lives in a data dir, not next to your file — `$XDG_DATA_HOME/caveman-compress/backups//` (macOS/Linux) or `%LOCALAPPDATA%\caveman-compress\backups\\` (Windows) — so skill auto-loaders don't re-read it as a live file. You can read and edit `.original.md` there. Run skill again to re-compress after edits. ## Benchmarks diff --git a/skills/caveman-compress/SECURITY.md b/skills/caveman-compress/SECURITY.md index 693108c..0efa9fe 100644 --- a/skills/caveman-compress/SECURITY.md +++ b/skills/caveman-compress/SECURITY.md @@ -8,7 +8,7 @@ 1. **subprocess usage**: The skill calls the `claude` CLI via `subprocess.run()` as a fallback when `ANTHROPIC_API_KEY` is not set. The subprocess call uses a fixed argument list — no shell interpolation occurs. User file content is passed via stdin, not as a shell argument. -2. **File read/write**: The skill reads the file the user explicitly points it at, compresses it, and writes the result back to the same path. A `.original.md` backup is saved alongside it. No files outside the user-specified path are read or written. +2. **File read/write**: The skill reads the file the user explicitly points it at, compresses it, and writes the result back to the same path. A `.original.md` backup is saved to an out-of-tree data dir (`$XDG_DATA_HOME/caveman-compress/backups//`, or `%LOCALAPPDATA%\caveman-compress\backups\\` on Windows). Beyond the target file and that backup location, no files are read or written. ### What the skill does NOT do diff --git a/skills/caveman-compress/SKILL.md b/skills/caveman-compress/SKILL.md index 00ce454..0b95aab 100644 --- a/skills/caveman-compress/SKILL.md +++ b/skills/caveman-compress/SKILL.md @@ -11,7 +11,7 @@ description: > ## Purpose -Compress natural language files (CLAUDE.md, todos, preferences) into caveman-speak to reduce input tokens. Compressed version overwrites original. Human-readable backup saved as `.original.md`. +Compress natural language files (CLAUDE.md, todos, preferences) into caveman-speak to reduce input tokens. Compressed version overwrites original. Human-readable backup saved as `.original.md`, but NOT beside the source file — it lives in an out-of-tree data dir (`$XDG_DATA_HOME/caveman-compress/backups//`, or `%LOCALAPPDATA%\caveman-compress\backups\\` on Windows) so skill auto-loaders don't re-ingest it as a live file. ## Trigger @@ -107,5 +107,5 @@ Compressed: - NEVER modify: .py, .js, .ts, .json, .yaml, .yml, .toml, .env, .lock, .css, .html, .xml, .sql, .sh - If file has mixed content (prose + code), compress ONLY the prose sections - If unsure whether something is code or prose, leave it unchanged -- Original file is backed up as FILE.original.md before overwriting +- Original file is backed up as FILE.original.md before overwriting — in the out-of-tree backup data dir (see Purpose), not beside the source file - Never compress FILE.original.md (skip it) diff --git a/skills/caveman-compress/scripts/benchmark.py b/skills/caveman-compress/scripts/benchmark.py index f9e2ee0..97d081b 100644 --- a/skills/caveman-compress/scripts/benchmark.py +++ b/skills/caveman-compress/scripts/benchmark.py @@ -23,8 +23,8 @@ def count_tokens(text): def benchmark_pair(orig_path: Path, comp_path: Path): - orig_text = orig_path.read_text() - comp_text = comp_path.read_text() + orig_text = orig_path.read_text(encoding="utf-8", errors="ignore") + comp_text = comp_path.read_text(encoding="utf-8", errors="ignore") orig_tokens = count_tokens(orig_text) comp_tokens = count_tokens(comp_text) diff --git a/skills/caveman-compress/scripts/compress.py b/skills/caveman-compress/scripts/compress.py index e93d934..80da520 100644 --- a/skills/caveman-compress/scripts/compress.py +++ b/skills/caveman-compress/scripts/compress.py @@ -9,8 +9,10 @@ Usage: import os import re import shutil +import stat import subprocess import sys +import tempfile from pathlib import Path from typing import List @@ -110,6 +112,59 @@ def strip_llm_wrapper(text: str) -> str: return m.group(2) return text + +def write_text_atomic(path: Path, text: str) -> None: + """Write ``text`` to ``path`` atomically as UTF-8. + + Path.write_text() truncates the destination before encoding the string — + a UnicodeEncodeError (or any other failure) partway through leaves a + 0-byte file, destroying whatever was there before (issue #655). Encode + first, write the bytes to a sibling temp file, fsync, then os.replace() + so the destination only ever moves from one complete, valid file to + another. Preserves the original file's permission bits across the swap. + """ + data = text.encode("utf-8") + fd, tmp_name = tempfile.mkstemp( + dir=str(path.parent), prefix=path.name + ".", suffix=".tmp" + ) + tmp_path = Path(tmp_name) + try: + with os.fdopen(fd, "wb") as f: + f.write(data) + f.flush() + os.fsync(f.fileno()) + if path.exists(): + os.chmod(tmp_path, stat.S_IMODE(path.stat().st_mode)) + os.replace(tmp_path, path) + except Exception: + try: + tmp_path.unlink() + except OSError: + pass + raise + + +def first_nonblank_line(text: str) -> str: + """Return the first non-blank line, stripped — used to detect a prose + preamble smuggled in ahead of the real content (issue #588).""" + for line in text.splitlines(): + if line.strip(): + return line.strip() + return "" + + +def _write_target(filepath: Path, text: str, backup_path: Path) -> None: + """Write to the target file, surfacing the backup location if the write + itself fails. write_text_atomic already leaves the target untouched on + failure, but the caller still needs to know where the pre-compression + original lives instead of being left to guess (issue #652).""" + try: + write_text_atomic(filepath, text) + except Exception: + print(f"❌ Write to {filepath} failed. Original preserved at backup: {backup_path}") + raise + + from .detect import should_compress from .validate import validate @@ -246,7 +301,7 @@ def compress_file(filepath: Path) -> bool: print("Skipping (not natural language)") return False - original_text = filepath.read_text(errors="ignore") + original_text = filepath.read_text(encoding="utf-8", errors="ignore") # Store backup outside the source directory so skill auto-loaders don't # re-ingest the `.original.md` copy as a live file. Mirror the source's # parent-dir name + stem under a platform-aware base to reduce collisions. @@ -300,8 +355,8 @@ def compress_file(filepath: Path) -> bool: # touching the input file. If the filesystem dropped bytes (encoding, # antivirus, disk full), unlink the bad backup and abort instead of # leaving the user with a corrupt backup + compressed primary. - backup_path.write_text(original_text) - backup_readback = backup_path.read_text(errors="ignore") + write_text_atomic(backup_path, original_text) + backup_readback = backup_path.read_text(encoding="utf-8", errors="ignore") if backup_readback != original_text: print(f"❌ Backup write verification failed: {backup_path}") print(" In-memory original differs from on-disk backup. Aborting before touching the input file.") @@ -310,7 +365,7 @@ def compress_file(filepath: Path) -> bool: except OSError: pass return False - filepath.write_text(compressed) + _write_target(filepath, compressed, backup_path) # Step 2: Validate + Retry for attempt in range(MAX_RETRIES): @@ -328,7 +383,7 @@ def compress_file(filepath: Path) -> bool: if attempt == MAX_RETRIES - 1: # Restore original on failure - filepath.write_text(original_text) + _write_target(filepath, original_text, backup_path) backup_path.unlink(missing_ok=True) print("❌ Failed after retries — original restored") return False @@ -337,6 +392,23 @@ def compress_file(filepath: Path) -> bool: compressed = call_claude( build_fix_prompt(original_text, compressed, result.errors) ) - filepath.write_text(compressed) + + if compressed is None or not compressed.strip(): + print("❌ Fix attempt aborted: Claude returned an empty response.") + print(" Skipping this attempt.") + continue + + # Guard against a prose preamble smuggled in ahead of the real fixed + # content (issue #588). Only enforced when the original starts with a + # structural anchor (frontmatter `---` or a heading) — plain-prose + # first lines get legitimately rewritten by compression, and requiring + # them verbatim would reject every valid fix. + anchor = first_nonblank_line(original_text) + if anchor.startswith(("---", "#")) and first_nonblank_line(compressed) != anchor: + print("❌ Fix attempt aborted: output does not start with the original's first line.") + print(" Possible preamble leak. Skipping this attempt.") + continue + + _write_target(filepath, compressed, backup_path) return True diff --git a/skills/caveman-compress/scripts/detect.py b/skills/caveman-compress/scripts/detect.py index 5114d04..6a468d5 100644 --- a/skills/caveman-compress/scripts/detect.py +++ b/skills/caveman-compress/scripts/detect.py @@ -90,7 +90,7 @@ def detect_file_type(filepath: Path) -> str: # Extensionless files (like CLAUDE.md, TODO) — check content if not ext: try: - text = filepath.read_text(errors="ignore") + text = filepath.read_text(encoding="utf-8", errors="ignore") except (OSError, PermissionError): return "unknown" diff --git a/skills/caveman-compress/scripts/validate.py b/skills/caveman-compress/scripts/validate.py index dc07307..dcd2a5d 100644 --- a/skills/caveman-compress/scripts/validate.py +++ b/skills/caveman-compress/scripts/validate.py @@ -28,7 +28,7 @@ class ValidationResult: def read_file(path: Path) -> str: - return path.read_text(errors="ignore") + return path.read_text(encoding="utf-8") # ---------- Extractors ---------- @@ -95,8 +95,16 @@ def count_bullets(text): def extract_inline_codes(text): - text_without_fences = re.sub(r"^```[\s\S]*?^```", "", text, flags=re.MULTILINE) - text_without_fences = re.sub(r"^~~~[\s\S]*?^~~~", "", text_without_fences, flags=re.MULTILINE) + """Backtick-delimited inline spans, with fenced code blocks stripped first. + + Previously used a column-0-anchored regex to strip fences, which misses + fences indented 1-3 spaces (valid CommonMark). Reuse extract_code_blocks + (FENCE_OPEN_REGEX-based, indentation-aware) instead so an indented fence's + body backticks don't leak into inline-code pairing. + """ + text_without_fences = text + for block in extract_code_blocks(text): + text_without_fences = text_without_fences.replace(block, "", 1) return re.findall(r"`([^`]+)`", text_without_fences) diff --git a/tests/test_compress_safety.py b/tests/test_compress_safety.py index b11cf5c..08f7627 100644 --- a/tests/test_compress_safety.py +++ b/tests/test_compress_safety.py @@ -9,6 +9,7 @@ bytes is detected before the input is overwritten. """ import os +import stat import sys import tempfile import unittest @@ -24,7 +25,7 @@ from scripts import compress as compress_mod # noqa: E402 class CompressSafetyTests(unittest.TestCase): def _file_with(self, dirpath: Path, text: str) -> Path: path = dirpath / "task.md" - path.write_text(text) + path.write_text(text, encoding="utf-8") return path def test_empty_input_refused(self): @@ -88,6 +89,121 @@ class CompressSafetyTests(unittest.TestCase): self.assertEqual(backup.read_text(), original) self.assertFalse((Path(tmp) / "task.original.md").exists()) + def test_utf8_roundtrip_survives_compression(self): + # Path.read_text() without encoding= would decode with the system + # locale codec (cp1252/cp949 on Windows) and could silently mangle + # non-ASCII bytes. Read raw bytes and decode strictly as UTF-8 so the + # assertion is locale-independent (issue #686). + with tempfile.TemporaryDirectory() as tmp, \ + tempfile.TemporaryDirectory() as data_home, \ + mock.patch.dict(os.environ, {"XDG_DATA_HOME": data_home, "LOCALAPPDATA": data_home}): + original = "# Heading\n\nCafé, 中文, and an arrow → here.\n" + compressed = "# Heading\n\nCafé 中文 arrow → here.\n" + path = self._file_with(Path(tmp), original) + with mock.patch.object(compress_mod, "call_claude", return_value=compressed), \ + mock.patch.object(compress_mod, "validate") as v: + v.return_value = mock.Mock(is_valid=True, errors=[], warnings=[]) + ok = compress_mod.compress_file(path) + self.assertTrue(ok) + self.assertEqual(path.read_bytes().decode("utf-8"), compressed) + backup = compress_mod.backup_dir_for(path.resolve()) / "task.original.md" + self.assertEqual(backup.read_bytes().decode("utf-8"), original) + + def test_write_text_atomic_leaves_destination_untouched_on_encode_failure(self): + # Direct unit test of the atomic-write primitive: an encode failure + # partway through must not truncate the destination or leave a *.tmp + # file behind (issue #655). + class ExplodingStr(str): + def encode(self, *args, **kwargs): + raise UnicodeEncodeError("utf-8", self, 0, 1, "forced failure") + + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "task.md" + path.write_text("original content", encoding="utf-8") + + with self.assertRaises(UnicodeEncodeError): + compress_mod.write_text_atomic(path, ExplodingStr("new content")) + + self.assertEqual(path.read_text(encoding="utf-8"), "original content") + self.assertEqual(list(Path(tmp).glob("*.tmp")), []) + + def test_forced_primary_write_failure_leaves_original_and_backup_intact(self): + # Same failure, exercised through the full compress_file pipeline: + # the backup must already exist and be intact, the target must be + # untouched, and no *.tmp litter must remain in either directory. + with tempfile.TemporaryDirectory() as tmp, \ + tempfile.TemporaryDirectory() as data_home, \ + mock.patch.dict(os.environ, {"XDG_DATA_HOME": data_home, "LOCALAPPDATA": data_home}): + original = "# Heading\n\nProse to compress.\n" + compressed = "# Heading\n\nProse.\n" + path = self._file_with(Path(tmp), original) + target = path.resolve() + real_write_text_atomic = compress_mod.write_text_atomic + + def flaky_write(write_path, text): + if write_path == target: + raise UnicodeEncodeError("utf-8", text, 0, 1, "forced failure") + return real_write_text_atomic(write_path, text) + + with mock.patch.object(compress_mod, "call_claude", return_value=compressed), \ + mock.patch.object(compress_mod, "validate") as v, \ + mock.patch.object(compress_mod, "write_text_atomic", side_effect=flaky_write): + v.return_value = mock.Mock(is_valid=True, errors=[], warnings=[]) + with self.assertRaises(UnicodeEncodeError): + compress_mod.compress_file(path) + + self.assertEqual(path.read_text(encoding="utf-8"), original) + backup_dir = compress_mod.backup_dir_for(target) + backup = backup_dir / "task.original.md" + self.assertEqual(backup.read_text(encoding="utf-8"), original) + self.assertEqual(list(Path(tmp).glob("*.tmp")), []) + self.assertEqual(list(backup_dir.glob("*.tmp")), []) + + def test_permission_preserved_across_compression(self): + with tempfile.TemporaryDirectory() as tmp, \ + tempfile.TemporaryDirectory() as data_home, \ + mock.patch.dict(os.environ, {"XDG_DATA_HOME": data_home, "LOCALAPPDATA": data_home}): + original = "# Heading\n\nProse to compress.\n" + compressed = "# Heading\n\nProse.\n" + path = self._file_with(Path(tmp), original) + path.chmod(0o644) + with mock.patch.object(compress_mod, "call_claude", return_value=compressed), \ + mock.patch.object(compress_mod, "validate") as v: + v.return_value = mock.Mock(is_valid=True, errors=[], warnings=[]) + ok = compress_mod.compress_file(path) + self.assertTrue(ok) + self.assertEqual(stat.S_IMODE(path.stat().st_mode), 0o644) + + def test_retry_preamble_output_rejected_and_not_written(self): + # A fix-retry response with a prose preamble ahead of the real content + # must never reach disk — only the restore-on-failure write should + # land, and it must restore the original (issue #588). + with tempfile.TemporaryDirectory() as tmp, \ + tempfile.TemporaryDirectory() as data_home, \ + mock.patch.dict(os.environ, {"XDG_DATA_HOME": data_home, "LOCALAPPDATA": data_home}): + original = "# Heading\n\nProse that fails validation.\n" + first_pass = "# Heading\n\nCompressed prose.\n" + preamble_fix = "Here is the fixed file:\n\n# Heading\n\nCompressed prose, fixed.\n" + path = self._file_with(Path(tmp), original) + + invalid = mock.Mock(is_valid=False, errors=["some validation error"], warnings=[]) + written_texts = [] + real_write_target = compress_mod._write_target + + def spy_write_target(target_path, text, backup_path): + written_texts.append(text) + return real_write_target(target_path, text, backup_path) + + with mock.patch.object( + compress_mod, "call_claude", side_effect=[first_pass, preamble_fix] + ), mock.patch.object(compress_mod, "validate", return_value=invalid), \ + mock.patch.object(compress_mod, "_write_target", side_effect=spy_write_target): + ok = compress_mod.compress_file(path) + + self.assertFalse(ok) + self.assertNotIn(preamble_fix, written_texts) + self.assertEqual(path.read_text(encoding="utf-8"), original) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_validate_inline.py b/tests/test_validate_inline.py index 4fec1ad..d6ffeb0 100644 --- a/tests/test_validate_inline.py +++ b/tests/test_validate_inline.py @@ -41,6 +41,17 @@ More text with `inline3`. def test_empty(self): self.assertEqual(extract_inline_codes("no backticks here"), []) + def test_indented_fence_backtick_not_leaked_as_inline(self): + # A fence indented 1-3 spaces is valid CommonMark and already handled + # by extract_code_blocks/FENCE_OPEN_REGEX. The old column-0-anchored + # strip regex missed it, so a backtick inside the indented fence body + # leaked out and got paired with the next real inline span (issue + # from PR #619 review). Only the real trailing inline span should + # come back. + text = " ```\n `weird`\n ```\nReal `inline` span here." + result = extract_inline_codes(text) + self.assertEqual(result, ["inline"]) + class TestValidateInlineCodes(unittest.TestCase): def test_match(self):