Merge pull request #202 from jbjardine/cojbdev/kicad-mcp-backend-state

Add backend state MCP tool

python/kicad_interface.py

@@ -282,6 +282,7 @@ class KiCADInterface:
         # 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
+        self._last_auto_save_status: Optional[Dict[str, Any]] = 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
@@ -459,6 +460,7 @@ class KiCADInterface:
             "add_sheet_pin": self._handle_add_sheet_pin,
             "import_svg_logo": self._handle_import_svg_logo,
             # UI/Process management commands
+            "get_backend_state": self._handle_get_backend_state,
             "check_kicad_ui": self._handle_check_kicad_ui,
             "launch_kicad_ui": self._handle_launch_kicad_ui,
             # Internal warm-up (pays wxApp init cost during startup)
@@ -599,7 +601,12 @@ class KiCADInterface:
 
     def _result_backend_for_command(self, command: str, result: Dict[str, Any]) -> str:
         """Return the backend label for a command result."""
-        if command in {"get_backend_info", "check_kicad_ui", "launch_kicad_ui"}:
+        if command in {
+            "get_backend_info",
+            "get_backend_state",
+            "check_kicad_ui",
+            "launch_kicad_ui",
+        }:
             return result.get("backend", "ipc" if self.use_ipc else "swig")
 
         if command in self.IPC_DIRECT_COMMANDS:
@@ -707,6 +714,10 @@ class KiCADInterface:
                         # can detect external modifications and refuse to
                         # overwrite them.
                         self._record_board_signature()
+                        self._last_auto_save_status = None
+                    elif command == "save_project":
+                        self._record_board_signature()
+                        self._last_auto_save_status = None
                     elif command in self._BOARD_MUTATING_COMMANDS:
                         # Auto-save after every board mutation via SWIG.
                         # Prevents data loss if Claude hits context limit before
@@ -715,6 +726,7 @@ class KiCADInterface:
                         # a warning to the caller so they don't believe their
                         # mutation was persisted.
                         save_status = self._auto_save_board()
+                        self._last_auto_save_status = save_status
                         if isinstance(result, dict) and not save_status.get("saved"):
                             if save_status.get("warning"):
                                 result.setdefault("warnings", []).append(save_status["warning"])
@@ -800,6 +812,91 @@ class KiCADInterface:
             path = None
         self._board_disk_signature = self._disk_signature(path) if path else None
 
+    def _current_board_path(self) -> Optional[str]:
+        """Return the current board file path, if a healthy board is loaded."""
+        board = getattr(self, "board", None)
+        if not board or not self._is_board_healthy(board):
+            return None
+        try:
+            path = board.GetFileName()
+        except Exception:
+            return None
+        return os.path.abspath(path) if path else None
+
+    def _current_project_file_path(self, board_path: Optional[str]) -> Optional[str]:
+        """Best-effort project file path for the currently loaded board."""
+        candidates = []
+        project_path = getattr(self, "_current_project_path", None)
+
+        if project_path:
+            project_path = Path(project_path)
+            if project_path.suffix == ".kicad_pro":
+                candidates.append(project_path)
+            elif board_path:
+                candidates.append(project_path / (Path(board_path).stem + ".kicad_pro"))
+            elif project_path.is_dir():
+                candidates.extend(project_path.glob("*.kicad_pro"))
+
+        if board_path and board_path.endswith(".kicad_pcb"):
+            candidates.append(Path(board_path).with_suffix(".kicad_pro"))
+
+        for candidate in candidates:
+            if candidate.exists():
+                return str(candidate.resolve())
+
+        return str(Path(candidates[0]).resolve()) if candidates else None
+
+    def _dirty_state(self, board_path: Optional[str]) -> Dict[str, Any]:
+        """Return the best-known dirty state for the loaded board.
+
+        dirty is intentionally tri-state: True/False when the MCP has evidence,
+        None when no reliable disk signature exists.
+        """
+        if not board_path:
+            return {
+                "dirty": False,
+                "dirtyReason": "No board is loaded",
+                "diskChangedExternally": False,
+            }
+
+        last_auto_save = getattr(self, "_last_auto_save_status", None) or {}
+        if last_auto_save.get("memChangesUnsaved"):
+            return {
+                "dirty": True,
+                "dirtyReason": "Auto-save refused after a board mutation; memory changes are not saved",
+                "diskChangedExternally": bool(last_auto_save.get("diskChangedExternally")),
+            }
+
+        expected = getattr(self, "_board_disk_signature", None)
+        current = self._disk_signature(board_path)
+
+        if expected is None:
+            return {
+                "dirty": None,
+                "dirtyReason": "No recorded disk signature for the loaded board",
+                "diskChangedExternally": False,
+            }
+
+        if current is None:
+            return {
+                "dirty": None,
+                "dirtyReason": "Board file is missing or unreadable on disk",
+                "diskChangedExternally": False,
+            }
+
+        if expected[1] != current[1]:
+            return {
+                "dirty": True,
+                "dirtyReason": "Board file contents changed on disk since this MCP session loaded it",
+                "diskChangedExternally": True,
+            }
+
+        return {
+            "dirty": False,
+            "dirtyReason": "Board file matches the MCP recorded disk signature",
+            "diskChangedExternally": False,
+        }
+
     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:
@@ -5969,6 +6066,7 @@ print("ok")
         tools are registered with the MCP client.
         """
         import time
+
         start = time.monotonic()
         try:
             # pcbnew.BOARD() triggers wxApp creation on macOS.
@@ -5984,6 +6082,7 @@ print("ok")
             elapsed = time.monotonic() - start
             logger.error(f"Warm-up failed after {elapsed:.1f}s: {exc}")
             return {"success": False, "message": str(exc), "elapsed_s": round(elapsed, 1)}
+
     # =========================================================================
 
     def _handle_get_backend_info(self, params: Dict[str, Any]) -> Dict[str, Any]:
@@ -6003,6 +6102,38 @@ print("ok")
             ),
         }
 
+    def _handle_get_backend_state(self, params: Dict[str, Any]) -> Dict[str, Any]:
+        """Return the MCP/KiCad backend state and currently loaded file state."""
+        if KiCADProcessManager.is_running():
+            self._try_enable_ipc_backend()
+
+        status = self._backend_status()
+        board_path = self._current_board_path()
+        project_path = self._current_project_file_path(board_path)
+        dirty_state = self._dirty_state(board_path)
+        loaded_board = board_path is not None
+        loaded_project = project_path is not None
+
+        return {
+            "success": True,
+            "backend": status["backend"],
+            "realtime": status["realtime_sync"],
+            "realtime_sync": status["realtime_sync"],
+            "ipcConnected": status["ipc_connected"],
+            "ipc_connected": status["ipc_connected"],
+            "loadedProject": loaded_project,
+            "loadedBoard": loaded_board,
+            "projectPath": project_path,
+            "boardPath": board_path,
+            "dirty": dirty_state["dirty"],
+            "dirtyReason": dirty_state["dirtyReason"],
+            "diskChangedExternally": dirty_state["diskChangedExternally"],
+            "message": (
+                f"{status['backend']} backend; "
+                f"{'board loaded' if loaded_board else 'no board loaded'}"
+            ),
+        }
+
     def _handle_ipc_add_track(self, params: Dict[str, Any]) -> Dict[str, Any]:
         """Add a track using IPC backend (real-time)"""
         if not self.use_ipc or not self.ipc_board_api:

python/schemas/tool_schemas.py

@@ -2196,6 +2196,12 @@ SCHEMATIC_TOOLS = [
 # =============================================================================
 
 UI_TOOLS = [
+    {
+        "name": "get_backend_state",
+        "title": "Get Backend State",
+        "description": ("Returns backend, realtime, loaded project/board paths, and dirty state."),
+        "inputSchema": {"type": "object", "properties": {}},
+    },
     {
         "name": "check_kicad_ui",
         "title": "Check KiCAD UI Status",

src/tools/registry.ts

@@ -165,6 +165,7 @@ export const directToolNames = [
   "sync_schematic_to_board",
 
   // UI management
+  "get_backend_state",
   "check_kicad_ui",
 ];
 

src/tools/ui.ts

@@ -7,6 +7,25 @@ import { z } from "zod";
 import { logger } from "../logger.js";
 
 export function registerUITools(server: McpServer, callKicadScript: Function) {
+  // Get MCP/KiCAD backend and loaded file state
+  server.tool(
+    "get_backend_state",
+    "Return the active backend, realtime status, loaded project/board paths, and dirty state.",
+    {},
+    async () => {
+      logger.info("Getting KiCAD backend state");
+      const result = await callKicadScript("get_backend_state", {});
+      return {
+        content: [
+          {
+            type: "text",
+            text: JSON.stringify(result, null, 2),
+          },
+        ],
+      };
+    },
+  );
+
   // Check if KiCAD UI is running
   server.tool("check_kicad_ui", "Check if KiCAD UI is currently running", {}, async () => {
     logger.info("Checking KiCAD UI status");

tests/test_backend_metadata.py

@@ -14,7 +14,11 @@ def _make_iface(command_routes, use_ipc=False):
     iface.use_ipc = use_ipc
     iface.ipc_backend = None
     iface.ipc_board_api = None
+    iface.board = None
     iface.command_routes = command_routes
+    iface._board_disk_signature = None
+    iface._current_project_path = None
+    iface._last_auto_save_status = None
     return iface
 
 
@@ -47,6 +51,20 @@ class _FakeIPCBoardAPI:
         return ["F.Cu", "B.Cu"]
 
 
+class _FakeBoard:
+    def __init__(self, filename):
+        self._filename = str(filename)
+
+    def GetFileName(self):
+        return self._filename
+
+    def GetDesignSettings(self):
+        return object()
+
+    def GetBoardEdgesBoundingBox(self):
+        return object()
+
+
 class _FakeIPCBackend:
     def __init__(self):
         self.connected = False
@@ -214,6 +232,127 @@ def test_backend_info_uses_reported_backend_for_metadata():
     assert result["_realtime"] is True
 
 
+def test_backend_state_without_board_reports_loaded_flags(monkeypatch):
+    """Backend state must explicitly say when no board/project is loaded."""
+    import kicad_interface
+
+    monkeypatch.setattr(kicad_interface.KiCADProcessManager, "is_running", lambda: False)
+
+    iface = _make_iface({}, use_ipc=False)
+    iface.command_routes["get_backend_state"] = iface._handle_get_backend_state
+
+    result = iface.handle_command("get_backend_state", {})
+
+    assert result["success"] is True
+    assert result["backend"] == "swig"
+    assert result["realtime"] is False
+    assert result["loadedProject"] is False
+    assert result["loadedBoard"] is False
+    assert result["projectPath"] is None
+    assert result["boardPath"] is None
+    assert result["dirty"] is False
+    assert result["_backend"] == "swig"
+    assert result["_realtime"] is False
+
+
+def test_backend_state_reports_loaded_project_board_and_clean_signature(tmp_path, monkeypatch):
+    """A loaded board with a matching disk signature should be visible and clean."""
+    import kicad_interface
+
+    monkeypatch.setattr(kicad_interface.KiCADProcessManager, "is_running", lambda: False)
+
+    board_path = tmp_path / "demo.kicad_pcb"
+    project_path = tmp_path / "demo.kicad_pro"
+    board_path.write_text("(kicad_pcb demo)\n", encoding="utf-8")
+    project_path.write_text("{}\n", encoding="utf-8")
+
+    iface = _make_iface({}, use_ipc=False)
+    iface.board = _FakeBoard(board_path)
+    iface._current_project_path = tmp_path
+    iface._record_board_signature()
+    iface.command_routes["get_backend_state"] = iface._handle_get_backend_state
+
+    result = iface.handle_command("get_backend_state", {})
+
+    assert result["loadedProject"] is True
+    assert result["loadedBoard"] is True
+    assert result["projectPath"] == str(project_path.resolve())
+    assert result["boardPath"] == str(board_path.resolve())
+    assert result["dirty"] is False
+    assert result["diskChangedExternally"] is False
+
+
+def test_backend_state_reports_disk_divergence_as_dirty(tmp_path, monkeypatch):
+    """If the board file changed after load, callers need a loud dirty signal."""
+    import kicad_interface
+
+    monkeypatch.setattr(kicad_interface.KiCADProcessManager, "is_running", lambda: False)
+
+    board_path = tmp_path / "demo.kicad_pcb"
+    board_path.write_text("(kicad_pcb original)\n", encoding="utf-8")
+
+    iface = _make_iface({}, use_ipc=False)
+    iface.board = _FakeBoard(board_path)
+    iface._record_board_signature()
+    board_path.write_text("(kicad_pcb changed)\n", encoding="utf-8")
+    iface.command_routes["get_backend_state"] = iface._handle_get_backend_state
+
+    result = iface.handle_command("get_backend_state", {})
+
+    assert result["loadedBoard"] is True
+    assert result["dirty"] is True
+    assert result["diskChangedExternally"] is True
+    assert "changed on disk" in result["dirtyReason"]
+
+
+def test_backend_state_reports_unsaved_memory_after_refused_autosave(tmp_path, monkeypatch):
+    """Auto-save refusal means the MCP has memory changes that are not persisted."""
+    import kicad_interface
+
+    monkeypatch.setattr(kicad_interface.KiCADProcessManager, "is_running", lambda: False)
+
+    board_path = tmp_path / "demo.kicad_pcb"
+    board_path.write_text("(kicad_pcb demo)\n", encoding="utf-8")
+
+    iface = _make_iface({}, use_ipc=False)
+    iface.board = _FakeBoard(board_path)
+    iface._record_board_signature()
+    iface._last_auto_save_status = {
+        "saved": False,
+        "memChangesUnsaved": True,
+        "diskChangedExternally": True,
+    }
+    iface.command_routes["get_backend_state"] = iface._handle_get_backend_state
+
+    result = iface.handle_command("get_backend_state", {})
+
+    assert result["dirty"] is True
+    assert result["diskChangedExternally"] is True
+    assert "memory changes" in result["dirtyReason"]
+
+
+def test_backend_state_reports_ipc_connection_without_loaded_board(monkeypatch):
+    """Backend state separates IPC connectivity from whether a board is loaded."""
+    import kicad_interface
+
+    monkeypatch.setattr(kicad_interface.KiCADProcessManager, "is_running", lambda: False)
+
+    backend = _FakeIPCBackend()
+    backend.connected = True
+    iface = _make_iface({}, use_ipc=True)
+    iface.ipc_backend = backend
+    iface.command_routes["get_backend_state"] = iface._handle_get_backend_state
+
+    result = iface.handle_command("get_backend_state", {})
+
+    assert result["backend"] == "ipc"
+    assert result["realtime"] is True
+    assert result["ipcConnected"] is True
+    assert result["loadedBoard"] is False
+    assert result["_backend"] == "ipc"
+    assert result["_realtime"] is True
+
+
 def test_ipc_capable_command_reconnects_when_kicad_is_running(monkeypatch):
     import kicad_interface
 

tests/test_ts_tool_registry.py

@@ -59,3 +59,10 @@ class TestTsToolRegistry:
         assert SRC_TOOLS_DIR.is_dir(), f"src/tools/ not found at {SRC_TOOLS_DIR}"
         ts_files = list(SRC_TOOLS_DIR.glob("**/*.ts"))
         assert ts_files, "No .ts files found in src/tools/"
+
+    def test_backend_state_tool_is_registered(self):
+        """Backend observability must be exposed as a first-class MCP tool."""
+        registrations = self._collect_registrations()
+        tool_names = {name for name, _, _ in registrations}
+
+        assert "get_backend_state" in tool_names