cowork-bot: atomic file writes for fix command (prevent config corruption on crash)

Add _atomic.py module with tempfile+fsync+os.replace pattern for safe writes. Replace all direct file writes in the fix command (JSON/YAML/TOML) with atomic helpers that serialize to buffer first, then write to temp file and atomically rename. Original config files are now preserved intact if the process crashes mid-write (disk full, SIGTERM, power loss). 7 new tests verify the atomic-write contract including failure-mode preservation.

Author: Coding-Dev-Tools

src/configdrift/_atomic.py

@@ -0,0 +1,72 @@
+"""Atomic file-write helpers.
+
+Write to a temporary file in the same directory, fsync, then os.replace()
+to atomically swap.  If the process crashes mid-write the original file
+is preserved intact.
+"""
+
+from __future__ import annotations
+
+import contextlib
+import os
+import tempfile
+from pathlib import Path
+
+
+def atomic_write_text(path: Path, text: str, encoding: str = "utf-8") -> None:
+    """Atomically write *text* to *path*.
+
+    Creates a temporary file beside *path*, writes + fsyncs, then
+    ``os.replace()`` for an atomic rename.
+    """
+    parent = path.parent
+    parent.mkdir(parents=True, exist_ok=True)
+    fd, tmp = tempfile.mkstemp(dir=parent, suffix=".tmp")
+    try:
+        with os.fdopen(fd, "w", encoding=encoding, newline="") as fh:
+            fh.write(text)
+            fh.flush()
+            os.fsync(fh.fileno())
+        os.replace(tmp, path)
+    except BaseException:
+        # Clean up temp file on any failure
+        with contextlib.suppress(OSError):
+            os.unlink(tmp)
+        raise
+
+
+def atomic_write_bytes(path: Path, data: bytes) -> None:
+    """Atomically write *data* to *path*."""
+    parent = path.parent
+    parent.mkdir(parents=True, exist_ok=True)
+    fd, tmp = tempfile.mkstemp(dir=parent, suffix=".tmp")
+    try:
+        with os.fdopen(fd, "wb") as fh:
+            fh.write(data)
+            fh.flush()
+            os.fsync(fh.fileno())
+        os.replace(tmp, path)
+    except BaseException:
+        with contextlib.suppress(OSError):
+            os.unlink(tmp)
+        raise
+
+
+def atomic_dump_yaml(path: Path, data: object, **dump_kwargs: object) -> None:
+    """Serialize *data* via ``yaml.dump`` into a temp file, then atomically rename."""
+    import io
+    import yaml
+
+    buf = io.StringIO()
+    yaml.dump(data, buf, **dump_kwargs)  # type: ignore[arg-type]
+    atomic_write_text(path, buf.getvalue())
+
+
+def atomic_dump_toml(path: Path, data: object) -> None:
+    """Serialize *data* via ``tomli_w.dump`` into a temp file, then atomically rename."""
+    import io
+    import tomli_w
+
+    buf = io.BytesIO()
+    tomli_w.dump(data, buf)  # type: ignore[arg-type]
+    atomic_write_bytes(path, buf.getvalue())

src/configdrift/cli.py

@@ -22,6 +22,7 @@ def require_license(product: str) -> None:  # type: ignore[misc]
 
 
 from configdrift import __version__
+from configdrift._atomic import atomic_dump_toml, atomic_dump_yaml, atomic_write_text
 from configdrift.diff import (
     Severity,
     diff_environments,
@@ -340,10 +341,8 @@ def fix(
         if ext == ".json":
             import json as _json
 
-            target_path.write_text(_json.dumps(target_data, indent=2) + "\n")
+            atomic_write_text(target_path, _json.dumps(target_data, indent=2) + "\n")
         elif ext in (".yaml", ".yml"):
-            import yaml as _yaml
-
             # Reconstruct nested structure from flat keys for YAML output
             nested: dict[str, Any] = {}
             for k, v in target_data.items():
@@ -352,11 +351,10 @@ def fix(
                 for part in parts[:-1]:
                     d = d.setdefault(part, {})
                 d[parts[-1]] = v
-            with open(target_path, "w", encoding="utf-8") as f:
-                _yaml.dump(nested, f, default_flow_style=False, sort_keys=False)
+            atomic_dump_yaml(target_path, nested, default_flow_style=False, sort_keys=False)
         elif ext == ".toml":
             try:
-                import tomli_w
+                import tomli_w  # noqa: F401
 
                 nested_toml: dict[str, Any] = {}
                 for k, v in target_data.items():
@@ -365,8 +363,7 @@ def fix(
                     for part in parts[:-1]:
                         d = d.setdefault(part, {})
                     d[parts[-1]] = v
-                with open(target_path, "wb") as f:
-                    tomli_w.dump(nested_toml, f)
+                atomic_dump_toml(target_path, nested_toml)
             except ImportError:
                 console.print("[yellow]Warning: tomli-w not installed; writing raw TOML not supported.[/yellow]")
                 raise typer.Exit(code=1) from None

tests/test_atomic_write.py

@@ -0,0 +1,86 @@
+"""Tests for atomic write helpers — verify original file survives write failure."""
+
+import os
+import pytest
+from configdrift._atomic import atomic_write_bytes, atomic_write_text
+from pathlib import Path
+from unittest.mock import patch
+
+
+class TestAtomicWriteText:
+    """Verify atomic_write_text preserves original on failure."""
+
+    def test_successful_write(self, tmp_path: Path):
+        target = tmp_path / "config.json"
+        target.write_text("original")
+        atomic_write_text(target, '{"key": "value"}\n')
+        assert target.read_text() == '{"key": "value"}\n'
+
+    def test_creates_file_if_missing(self, tmp_path: Path):
+        target = tmp_path / "new.json"
+        atomic_write_text(target, "hello")
+        assert target.read_text() == "hello"
+
+    def test_original_preserved_on_oserror(self, tmp_path: Path):
+        """If os.replace fails, the original file must remain intact."""
+        target = tmp_path / "config.json"
+        target.write_text("original-content")
+
+        with (
+            patch("os.replace", side_effect=OSError("Simulated disk full")),
+            pytest.raises(OSError, match="Simulated disk full"),
+        ):
+            atomic_write_text(target, "new-content")
+
+        # Original must be untouched
+        assert target.read_text() == "original-content"
+
+    def test_no_temp_files_left_on_failure(self, tmp_path: Path):
+        """Temp files must be cleaned up after a failed write."""
+        target = tmp_path / "config.json"
+        target.write_text("original")
+
+        with patch("os.replace", side_effect=OSError("fail")), pytest.raises(OSError):
+            atomic_write_text(target, "new")
+
+        temps = list(tmp_path.glob("*.tmp"))
+        assert temps == [], f"Leftover temp files: {temps}"
+
+    def test_truncation_does_not_corrupt_original(self, tmp_path: Path):
+        """Even if the temp-file write itself fails mid-stream, original is safe."""
+        target = tmp_path / "config.json"
+        original = "important-data-that-must-survive"
+        target.write_text(original)
+
+        real_fdopen = os.fdopen
+
+        def failing_fdopen(fd, *args, **kwargs):
+            fh = real_fdopen(fd, *args, **kwargs)  # noqa: F841
+            # Simulate failure after opening but before writing
+            raise OSError("Disk full during write")
+
+        with (
+            patch("os.fdopen", side_effect=failing_fdopen),
+            pytest.raises(OSError, match="Disk full during write"),
+        ):
+            atomic_write_text(target, "replacement")
+
+        assert target.read_text() == original
+
+
+class TestAtomicWriteBytes:
+    """Verify atomic_write_bytes preserves original on failure."""
+
+    def test_successful_binary_write(self, tmp_path: Path):
+        target = tmp_path / "data.bin"
+        atomic_write_bytes(target, b"\x00\x01\x02")
+        assert target.read_bytes() == b"\x00\x01\x02"
+
+    def test_original_preserved_on_oserror(self, tmp_path: Path):
+        target = tmp_path / "data.bin"
+        target.write_bytes(b"original-bytes")
+
+        with patch("os.replace", side_effect=OSError("fail")), pytest.raises(OSError):
+            atomic_write_bytes(target, b"new-bytes")
+
+        assert target.read_bytes() == b"original-bytes"