diff --git a/python/kicad_interface.py b/python/kicad_interface.py index c0ba8c6..55334ca 100644 --- a/python/kicad_interface.py +++ b/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 ``/.mcp-backups/.`` + (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": } + {"saved": False, "skipped": } -- 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""" diff --git a/tests/test_auto_save_guard.py b/tests/test_auto_save_guard.py new file mode 100644 index 0000000..fd2d3b5 --- /dev/null +++ b/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/.` 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