fix(compress): stop Windows data loss — UTF-8 + atomic writes (#652, #655, #686)

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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_016ySX6TBWZuvFze4ajf7Hpf

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 `<filename>.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 `<filename>.original.md`, but NOT beside the source file — it lives in an out-of-tree data dir (`$XDG_DATA_HOME/caveman-compress/backups/<parent-dir-name>/`, or `%LOCALAPPDATA%\caveman-compress\backups\<parent-dir-name>\` 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)

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)

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

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"
 

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)
 
 

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/<parent-dir-name>/` (macOS/Linux) or `%LOCALAPPDATA%\caveman-compress\backups\<parent-dir-name>\` (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
 

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/<parent-dir-name>/`, or `%LOCALAPPDATA%\caveman-compress\backups\<parent-dir-name>\` on Windows). Beyond the target file and that backup location, no files are read or written.
 
 ### What the skill does NOT do
 

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 `<filename>.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 `<filename>.original.md`, but NOT beside the source file — it lives in an out-of-tree data dir (`$XDG_DATA_HOME/caveman-compress/backups/<parent-dir-name>/`, or `%LOCALAPPDATA%\caveman-compress\backups\<parent-dir-name>\` 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)

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)

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

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"
 

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)
 
 

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()

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):