From 21fc0702ac35325b545696906294e9a02c0f1a82 Mon Sep 17 00:00:00 2001 From: jbjardine <167578676+jbjardine@users.noreply.github.com> Date: Sat, 23 May 2026 23:08:23 +0200 Subject: [PATCH] Add backend state MCP tool --- python/kicad_interface.py | 133 ++++++++++++++++++++++++++++++- python/schemas/tool_schemas.py | 6 ++ src/tools/registry.ts | 1 + src/tools/ui.ts | 19 +++++ tests/test_backend_metadata.py | 139 +++++++++++++++++++++++++++++++++ tests/test_ts_tool_registry.py | 7 ++ 6 files changed, 304 insertions(+), 1 deletion(-) diff --git a/python/kicad_interface.py b/python/kicad_interface.py index b48bec6..ed570ec 100644 --- a/python/kicad_interface.py +++ b/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: diff --git a/python/schemas/tool_schemas.py b/python/schemas/tool_schemas.py index f841bec..1da88bf 100644 --- a/python/schemas/tool_schemas.py +++ b/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", diff --git a/src/tools/registry.ts b/src/tools/registry.ts index f5bbce6..445f395 100644 --- a/src/tools/registry.ts +++ b/src/tools/registry.ts @@ -165,6 +165,7 @@ export const directToolNames = [ "sync_schematic_to_board", // UI management + "get_backend_state", "check_kicad_ui", ]; diff --git a/src/tools/ui.ts b/src/tools/ui.ts index 561110b..6b2e830 100644 --- a/src/tools/ui.ts +++ b/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"); diff --git a/tests/test_backend_metadata.py b/tests/test_backend_metadata.py index fc4bf33..0906431 100644 --- a/tests/test_backend_metadata.py +++ b/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 diff --git a/tests/test_ts_tool_registry.py b/tests/test_ts_tool_registry.py index bcf5b1f..eb57375 100644 --- a/tests/test_ts_tool_registry.py +++ b/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