Merge pull request #151 from inktomi/feat/auto-save-guard

feat: guard SWIG auto-save against external file changes

python/kicad_interface.py

@@ -7,11 +7,14 @@ and KiCAD's Python API (pcbnew). It receives commands via stdin as
 JSON and returns responses via stdout also as JSON.
 """
 
+import hashlib
 import json
 import logging
 import os
+import shutil
 import sys
 import traceback
+from datetime import datetime
 from pathlib import Path
 from typing import Any, Dict, List, Optional, Tuple
 
@@ -259,6 +262,12 @@ class KiCADInterface:
         """Initialize the interface and command handlers"""
         self.board = None
         self.project_filename = None
+        # On-disk signature (mtime_ns, sha256_hex) of self.board's file as of
+        # last load or successful auto-save.  Used by _auto_save_board() to
+        # detect external modifications and refuse to clobber them.
+        self._board_disk_signature: Optional[Tuple[int, str]] = None
+        # Number of timestamped backups to keep in .mcp-backups/ per board file.
+        self._auto_save_backup_keep = 20
         self.use_ipc = USE_IPC_BACKEND
         self.ipc_backend = ipc_backend
         self.ipc_board_api = None
@@ -539,11 +548,22 @@ class KiCADInterface:
                         # Get board from the project commands handler
                         self.board = self.project_commands.board
                         self._update_command_handlers()
+                        # Record the file's signature so subsequent auto-saves
+                        # can detect external modifications and refuse to
+                        # overwrite them.
+                        self._record_board_signature()
                     elif command in self._BOARD_MUTATING_COMMANDS:
                         # Auto-save after every board mutation via SWIG.
                         # Prevents data loss if Claude hits context limit before
-                        # an explicit save_project call.
-                        self._auto_save_board()
+                        # an explicit save_project call.  When auto-save refuses
+                        # because the on-disk file changed externally, surface
+                        # a warning to the caller so they don't believe their
+                        # mutation was persisted.
+                        save_status = self._auto_save_board()
+                        if isinstance(result, dict) and not save_status.get("saved"):
+                            if save_status.get("warning"):
+                                result.setdefault("warnings", []).append(save_status["warning"])
+                            result["autoSave"] = save_status
 
                 return result
             else:
@@ -587,19 +607,133 @@ class KiCADInterface:
         "connect_to_net",
     }
 
-    def _auto_save_board(self) -> None:
-        """Save board to disk after SWIG mutations.
-        Called automatically after every board-mutating SWIG command so that
-        data is not lost if Claude hits the context limit before save_project.
-        """
+    @staticmethod
+    def _disk_signature(path: str) -> Optional[Tuple[int, str]]:
+        """Return (mtime_ns, sha256_hex) for the file, or None if missing/unreadable."""
         try:
-            if self.board:
-                board_path = self.board.GetFileName()
-                if board_path:
-                    pcbnew.SaveBoard(board_path, self.board)
-                    logger.debug(f"Auto-saved board to: {board_path}")
+            st = os.stat(path)
+            h = hashlib.sha256()
+            with open(path, "rb") as f:
+                for chunk in iter(lambda: f.read(65536), b""):
+                    h.update(chunk)
+            return (st.st_mtime_ns, h.hexdigest())
+        except OSError:
+            return None
+
+    def _record_board_signature(self) -> None:
+        """Record the current on-disk signature of self.board's file.
+
+        Call this after a fresh load (open_project / create_project) or after
+        any save we perform ourselves, so that _auto_save_board() can detect
+        when an external actor has modified the file in between.
+        """
+        if not self.board:
+            self._board_disk_signature = None
+            return
+        try:
+            path = self.board.GetFileName()
+        except Exception:
+            path = None
+        self._board_disk_signature = self._disk_signature(path) if path else None
+
+    def _prune_auto_save_backups(self, backup_dir: str, base_name: str) -> None:
+        """Keep only the most recent `_auto_save_backup_keep` backups for `base_name`."""
+        try:
+            entries = [
+                os.path.join(backup_dir, f)
+                for f in os.listdir(backup_dir)
+                if f.startswith(base_name + ".")
+            ]
+            entries.sort(key=os.path.getmtime, reverse=True)
+            for old in entries[self._auto_save_backup_keep :]:
+                try:
+                    os.remove(old)
+                except OSError:
+                    pass
+        except OSError as e:
+            logger.debug(f"Backup pruning skipped: {e}")
+
+    def _auto_save_board(self) -> Dict[str, Any]:
+        """Save the in-memory board to disk after a SWIG-path mutation.
+
+        Behaviour:
+          * If the file's on-disk signature has diverged from the one we
+            recorded at load (or at our last successful save), refuse to
+            overwrite — an external actor (KiCad GUI, another process, git)
+            has touched the file and saving would clobber their changes.
+          * Otherwise, copy the existing file to ``<dir>/.mcp-backups/<name>.<ts>``
+            (rotating, keeps the most recent `_auto_save_backup_keep`),
+            then call pcbnew.SaveBoard().
+          * Update the recorded signature on success.
+
+        Returns a status dict that handle_command merges into the caller's
+        response so warnings about refused saves are visible:
+          {"saved": True,  "boardPath": ..., "backup": <path-or-None>}
+          {"saved": False, "skipped": <reason>}                      -- nothing to save
+          {"saved": False, "warning": ..., "diskChangedExternally": True, ...}
+          {"saved": False, "error": ...}                             -- pcbnew error
+        """
+        if not self.board:
+            return {"saved": False, "skipped": "no board loaded"}
+
+        try:
+            board_path = self.board.GetFileName()
+        except Exception as e:
+            return {"saved": False, "skipped": f"GetFileName failed: {e}"}
+
+        if not board_path:
+            return {"saved": False, "skipped": "no board path"}
+
+        expected = self._board_disk_signature
+        current = self._disk_signature(board_path)
+
+        # If we have a recorded signature and disk has diverged, refuse to save.
+        # (If expected is None we treat this as "first save" and proceed —
+        # otherwise users with pre-existing setups would never be able to save.)
+        if expected is not None and current is not None and expected != current:
+            warning = (
+                "Auto-save refused: the on-disk PCB file changed externally "
+                "since this MCP session loaded it. To avoid clobbering those "
+                "changes, the in-memory mutation has NOT been written to disk. "
+                "Reload via open_project to refresh, then re-apply the change."
+            )
+            logger.warning(f"{warning} ({board_path})")
+            logger.warning(f"  expected mtime_ns={expected[0]} sha256={expected[1][:12]}…")
+            logger.warning(f"  current  mtime_ns={current[0]} sha256={current[1][:12]}…")
+            return {
+                "saved": False,
+                "warning": warning,
+                "boardPath": board_path,
+                "diskChangedExternally": True,
+                "expectedMtimeNs": expected[0],
+                "currentMtimeNs": current[0],
+                "memChangesUnsaved": True,
+            }
+
+        # Make a rotating backup of the existing file (best-effort).
+        backup_path: Optional[str] = None
+        if current is not None:
+            try:
+                backup_dir = os.path.join(os.path.dirname(board_path) or ".", ".mcp-backups")
+                os.makedirs(backup_dir, exist_ok=True)
+                stamp = datetime.now().strftime("%Y%m%d-%H%M%S-%f")[:-3]
+                base = os.path.basename(board_path)
+                backup_path = os.path.join(backup_dir, f"{base}.{stamp}")
+                shutil.copy2(board_path, backup_path)
+                self._prune_auto_save_backups(backup_dir, base)
+            except OSError as e:
+                logger.warning(f"Auto-save backup failed (continuing): {e}")
+                backup_path = None
+
+        # Write the board.
+        try:
+            pcbnew.SaveBoard(board_path, self.board)
+            logger.debug(f"Auto-saved board to: {board_path}")
+            self._board_disk_signature = self._disk_signature(board_path)
+            return {"saved": True, "boardPath": board_path, "backup": backup_path}
         except Exception as e:
             logger.warning(f"Auto-save failed: {e}")
+            return {"saved": False, "error": str(e), "backup": backup_path}
 
     def _update_command_handlers(self) -> None:
         """Update board reference in all command handlers"""

tests/test_auto_save_guard.py

@@ -0,0 +1,226 @@
+"""
+Tests for the auto-save guard in kicad_interface._auto_save_board.
+
+The guard is meant to prevent the MCP server from silently overwriting a
+.kicad_pcb file that was modified externally between LoadBoard and
+SaveBoard (e.g. by KiCad GUI's own save, a git checkout, or another
+process). Behaviour exercised here:
+
+  - First-load semantics: with no recorded signature, auto-save proceeds.
+  - Detect external change: when the on-disk file has been altered since
+    the recorded signature, auto-save is refused and the in-memory
+    mutation is NOT written to disk.
+  - Backup creation: a successful save copies the prior file contents to
+    `.mcp-backups/<name>.<timestamp>` before overwriting.
+  - Backup pruning: only the most recent N backups are retained.
+  - Signature update: after a successful save, the recorded signature is
+    refreshed so subsequent saves are not falsely flagged.
+"""
+
+import os
+import sys
+import time
+from pathlib import Path
+from typing import Any
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+sys.path.insert(0, str(Path(__file__).parent.parent / "python"))
+
+
+def _make_iface() -> Any:
+    """Construct a KiCADInterface bypassing __init__ (avoids pcbnew / IPC)."""
+    with patch("kicad_interface.USE_IPC_BACKEND", False):
+        from kicad_interface import KiCADInterface
+
+        iface = KiCADInterface.__new__(KiCADInterface)
+    iface.board = None
+    iface._board_disk_signature = None
+    iface._auto_save_backup_keep = 5
+    return iface
+
+
+@pytest.fixture()
+def iface():
+    return _make_iface()
+
+
+@pytest.fixture()
+def board_file(tmp_path: Path) -> Path:
+    """A temp .kicad_pcb file with placeholder contents."""
+    f = tmp_path / "test.kicad_pcb"
+    f.write_text("(kicad_pcb (version 1) (generator test))\n")
+    return f
+
+
+def _fake_board(path: str) -> MagicMock:
+    """A MagicMock that quacks like a pcbnew BOARD for our helpers."""
+    b = MagicMock()
+    b.GetFileName.return_value = path
+    return b
+
+
+# ---------------------------------------------------------------------------
+# _disk_signature: read-only, no side effects
+# ---------------------------------------------------------------------------
+
+
+def test_disk_signature_returns_mtime_and_hash(iface, board_file):
+    sig = iface._disk_signature(str(board_file))
+    assert sig is not None
+    mtime_ns, sha = sig
+    assert isinstance(mtime_ns, int) and mtime_ns > 0
+    assert isinstance(sha, str) and len(sha) == 64  # sha256 hex
+
+
+def test_disk_signature_returns_none_for_missing_file(iface, tmp_path: Path):
+    assert iface._disk_signature(str(tmp_path / "does-not-exist.kicad_pcb")) is None
+
+
+def test_disk_signature_changes_when_file_changes(iface, board_file):
+    s1 = iface._disk_signature(str(board_file))
+    # ensure mtime tick (filesystems vary; nanoseconds usually suffice but
+    # add a small sleep for resolutions that don't)
+    time.sleep(0.01)
+    board_file.write_text(board_file.read_text() + "; modified\n")
+    s2 = iface._disk_signature(str(board_file))
+    assert s1 != s2
+    assert s1[1] != s2[1]  # hash differs
+
+
+# ---------------------------------------------------------------------------
+# _auto_save_board: skip cases (no board / no path)
+# ---------------------------------------------------------------------------
+
+
+def test_auto_save_skips_when_no_board(iface):
+    iface.board = None
+    result = iface._auto_save_board()
+    assert result == {"saved": False, "skipped": "no board loaded"}
+
+
+def test_auto_save_skips_when_no_path(iface):
+    iface.board = MagicMock()
+    iface.board.GetFileName.return_value = ""
+    result = iface._auto_save_board()
+    assert result["saved"] is False
+    assert "skipped" in result
+
+
+# ---------------------------------------------------------------------------
+# _auto_save_board: happy-path save with signature tracking + backup
+# ---------------------------------------------------------------------------
+
+
+def test_auto_save_with_matching_signature_proceeds(iface, board_file):
+    iface.board = _fake_board(str(board_file))
+    iface._record_board_signature()
+    pre_sig = iface._board_disk_signature
+    assert pre_sig is not None
+
+    save_calls = []
+
+    def fake_save(path, board):
+        save_calls.append((path, board))
+        # Simulate pcbnew rewriting the file
+        Path(path).write_text("(kicad_pcb (version 1) (generator test) ; saved)\n")
+
+    with patch("kicad_interface.pcbnew") as mock_pcb:
+        mock_pcb.SaveBoard.side_effect = fake_save
+        result = iface._auto_save_board()
+
+    assert result["saved"] is True
+    assert result["boardPath"] == str(board_file)
+    assert len(save_calls) == 1
+    # Signature should have been refreshed
+    assert iface._board_disk_signature is not None
+    assert iface._board_disk_signature != pre_sig
+
+
+def test_auto_save_creates_backup_before_writing(iface, board_file):
+    iface.board = _fake_board(str(board_file))
+    iface._record_board_signature()
+
+    original_contents = board_file.read_text()
+
+    def fake_save(path, board):
+        Path(path).write_text("(kicad_pcb ; overwritten)\n")
+
+    with patch("kicad_interface.pcbnew") as mock_pcb:
+        mock_pcb.SaveBoard.side_effect = fake_save
+        result = iface._auto_save_board()
+
+    assert result["saved"] is True
+    backup_dir = board_file.parent / ".mcp-backups"
+    assert backup_dir.is_dir()
+    backups = list(backup_dir.glob(f"{board_file.name}.*"))
+    assert len(backups) == 1
+    # Backup must contain the PRE-save contents (snapshot before overwrite)
+    assert backups[0].read_text() == original_contents
+    # Returned path matches the file we created
+    assert result["backup"] == str(backups[0])
+
+
+# ---------------------------------------------------------------------------
+# _auto_save_board: refuses when disk diverged from recorded signature
+# ---------------------------------------------------------------------------
+
+
+def test_auto_save_refuses_when_disk_changed_externally(iface, board_file):
+    iface.board = _fake_board(str(board_file))
+    iface._record_board_signature()
+
+    # Simulate an external actor (KiCad GUI, git, another process)
+    # writing the file after we loaded it.
+    time.sleep(0.01)
+    board_file.write_text("(kicad_pcb ; changed by someone else)\n")
+
+    with patch("kicad_interface.pcbnew") as mock_pcb:
+        result = iface._auto_save_board()
+        assert mock_pcb.SaveBoard.call_count == 0  # MUST NOT save
+
+    assert result["saved"] is False
+    assert result["diskChangedExternally"] is True
+    assert result["memChangesUnsaved"] is True
+    assert "warning" in result
+    # File on disk must still hold the external content, untouched
+    assert "changed by someone else" in board_file.read_text()
+
+
+def test_auto_save_first_save_with_no_recorded_signature_proceeds(iface, board_file):
+    """If we never loaded the file (e.g. first save_project after create),
+    treat it as a normal first save rather than refusing."""
+    iface.board = _fake_board(str(board_file))
+    iface._board_disk_signature = None  # explicit: nothing recorded yet
+
+    with patch("kicad_interface.pcbnew") as mock_pcb:
+        mock_pcb.SaveBoard.side_effect = lambda p, b: Path(p).write_text("first\n")
+        result = iface._auto_save_board()
+
+    assert result["saved"] is True
+    assert iface._board_disk_signature is not None  # now recorded
+
+
+# ---------------------------------------------------------------------------
+# Backup rotation: keep only N most-recent
+# ---------------------------------------------------------------------------
+
+
+def test_backup_pruning_keeps_only_n_most_recent(iface, board_file):
+    iface.board = _fake_board(str(board_file))
+    iface._auto_save_backup_keep = 3
+
+    def fake_save(path, board):
+        Path(path).write_text(f"(kicad_pcb ; save at {time.time_ns()})\n")
+
+    with patch("kicad_interface.pcbnew") as mock_pcb:
+        mock_pcb.SaveBoard.side_effect = fake_save
+        for _ in range(7):
+            iface._record_board_signature()
+            iface._auto_save_board()
+            time.sleep(0.005)  # ensure unique timestamps
+
+    backup_dir = board_file.parent / ".mcp-backups"
+    backups = sorted(backup_dir.glob(f"{board_file.name}.*"))
+    assert len(backups) == 3