diff --git a/python/commands/design_rules.py b/python/commands/design_rules.py index 2c679b6..2c106b1 100644 --- a/python/commands/design_rules.py +++ b/python/commands/design_rules.py @@ -187,6 +187,13 @@ class DesignRuleCommands: } report_path = params.get("reportPath") + # Caller-overridable timeout (seconds). Defaults to 600s for big boards + # but smaller MCP transport budgets (e.g. 120s) can lower it explicitly. + try: + timeout_sec = int(params.get("timeoutSec", 600)) + except (TypeError, ValueError): + timeout_sec = 600 + timeout_sec = max(10, min(timeout_sec, 1800)) # clamp to [10, 1800] # Get the board file path board_file = self.board.GetFileName() @@ -225,14 +232,14 @@ class DesignRuleCommands: board_file, ] - logger.info(f"Running DRC command: {' '.join(cmd)}") + logger.info(f"Running DRC command (timeout={timeout_sec}s): {' '.join(cmd)}") - # Run DRC + # Run DRC. subprocess.run kills the child on TimeoutExpired. result = subprocess.run( cmd, capture_output=True, text=True, - timeout=600, # 10 minute timeout for large boards (21MB PCB needs time) + timeout=timeout_sec, ) if result.returncode != 0: @@ -318,7 +325,7 @@ class DesignRuleCommands: "mm", board_file, ] - subprocess.run(cmd_report, capture_output=True, timeout=600) + subprocess.run(cmd_report, capture_output=True, timeout=timeout_sec) # Return summary only (not full violations list) return { @@ -339,11 +346,14 @@ class DesignRuleCommands: os.unlink(json_output) except subprocess.TimeoutExpired: - logger.error("DRC command timed out") + logger.error(f"DRC command timed out after {timeout_sec}s") return { "success": False, "message": "DRC command timed out", - "errorDetails": "Command took longer than 600 seconds (10 minutes)", + "errorDetails": ( + f"Command took longer than {timeout_sec} seconds; " + "raise timeoutSec param for very large boards" + ), } except Exception as e: logger.error(f"Error running DRC: {str(e)}") diff --git a/python/kicad_interface.py b/python/kicad_interface.py index 75d3ea4..b798672 100644 --- a/python/kicad_interface.py +++ b/python/kicad_interface.py @@ -567,6 +567,47 @@ class KiCADInterface: logger.info("Updating board reference...") # Get board from the project commands handler self.board = self.project_commands.board + + # Detect SWIG dehydration before claiming success. + # Without this, every later board op sees a raw + # SwigPyObject and raises AttributeError, while the + # MCP keeps reporting "Opened project" — the exact + # symptom users hit on KiCAD nightlies. + if not self._is_board_healthy(): + board_path = (result.get("project") or {}).get("boardPath") + recovered = None + if board_path: + logger.warning( + "Board after %s is SWIG-dehydrated; attempting recovery", + command, + ) + recovered = self._safe_load_board(board_path) + if recovered is not None: + self.board = recovered + self.project_commands.board = recovered + result.setdefault("warnings", []).append( + "SWIG board proxy was dehydrated on load; " + "recovered via pcbnew module reload" + ) + else: + # Surface the truth — never claim success when + # the board is unusable. + return { + "success": False, + "message": ( + f"{command} loaded the board but the SWIG " + "proxy is dehydrated and recovery failed" + ), + "errorDetails": ( + "pcbnew.LoadBoard returned a BOARD whose " + "method dispatch is missing (raw SwigPyObject). " + "This indicates SWIG state corruption in the " + "current Python process — restart the MCP " + "server to recover." + ), + "_backend": "swig", + "_realtime": False, + } self._update_command_handlers() # Record the file's signature so subsequent auto-saves # can detect external modifications and refuse to @@ -694,6 +735,9 @@ class KiCADInterface: (rotating, keeps the most recent `_auto_save_backup_keep`), then call pcbnew.SaveBoard(). * Update the recorded signature on success. + * If SaveBoard leaves the in-memory BOARD dehydrated (observed on + KiCAD nightlies after delete_trace + auto-save), reload from disk + so the next command sees a usable proxy instead of a SwigPyObject. Returns a status dict that handle_command merges into the caller's response so warnings about refused saves are visible: @@ -772,11 +816,30 @@ class KiCADInterface: 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} + # Post-save dehydration check. If the BOARD lost its bindings during + # save, reload from disk while we still know the path. board_path is + # guaranteed non-empty here (we returned early above otherwise). + if not self._is_board_healthy(): + logger.warning( + "Board became dehydrated during auto-save; reloading from %s", + board_path, + ) + recovered = self._safe_load_board(board_path) + if recovered is not None: + self.board = recovered + self._update_command_handlers() + else: + logger.error( + "Board dehydration after auto-save is unrecoverable — " + "subsequent commands will fail until MCP restart" + ) + + return {"saved": True, "boardPath": board_path, "backup": backup_path} + def _update_command_handlers(self) -> None: """Update board reference in all command handlers""" logger.debug("Updating board reference in command handlers") @@ -788,6 +851,70 @@ class KiCADInterface: self.export_commands.board = self.board self.freerouting_commands.board = self.board + # Stable BOARD methods used to detect SWIG dehydration. Newer KiCAD nightly + # builds occasionally return a raw SwigPyObject from pcbnew.LoadBoard after + # certain mutating sequences (delete_trace, refill_zones, …) — the proxy + # type-checks but every method access raises AttributeError. Probing for + # these methods catches that state without segfaulting. + _BOARD_HEALTH_METHODS = ( + "GetDesignSettings", + "GetBoardEdgesBoundingBox", + "GetFileName", + ) + + def _is_board_healthy(self, board: Optional[Any] = None) -> bool: + """Return True if the board (default self.board) has live SWIG dispatch.""" + target = board if board is not None else self.board + if target is None: + return False + return all(hasattr(target, m) for m in self._BOARD_HEALTH_METHODS) + + def _safe_load_board(self, path: str) -> Optional[Any]: + """Load a board from disk, recovering from SWIG dehydration if pcbnew is broken. + + If pcbnew.LoadBoard returns a dehydrated proxy, reload the pcbnew + module once and retry. Returns the new board, or None if recovery + is impossible (caller must surface a real failure rather than fake + success). + """ + global pcbnew + try: + board = pcbnew.LoadBoard(path) + except Exception as e: + logger.error(f"LoadBoard({path!r}) raised: {e}") + return None + + if self._is_board_healthy(board): + return board + + logger.warning( + f"LoadBoard({path!r}) returned a dehydrated SWIG proxy; " + "reloading pcbnew module and retrying" + ) + try: + import importlib + + pcbnew = importlib.reload(pcbnew) + except Exception as e: + logger.error(f"pcbnew module reload failed: {e}") + return None + + try: + board = pcbnew.LoadBoard(path) + except Exception as e: + logger.error(f"LoadBoard retry after pcbnew reload failed: {e}") + return None + + if not self._is_board_healthy(board): + logger.error( + "Board still dehydrated after pcbnew reload; SWIG state is " + "unrecoverable in this process — restart the MCP server" + ) + return None + + logger.info("Recovered from SWIG dehydration via pcbnew reload") + return board + # Schematic command handlers def _handle_create_schematic(self, params: Dict[str, Any]) -> Dict[str, Any]: """Create a new schematic""" @@ -921,17 +1048,19 @@ class KiCADInterface: current_board_file = str(Path(self.board.GetFileName()).resolve()) if self.board else "" if board_path_norm != current_board_file: logger.info(f"boardPath differs from current board — reloading: {board_path}") - try: - self.board = pcbnew.LoadBoard(board_path) - self._update_command_handlers() - logger.info("Board reloaded from boardPath") - except Exception as e: - logger.error(f"Failed to reload board from boardPath: {e}") + reloaded = self._safe_load_board(board_path) + if reloaded is None: return { "success": False, "message": f"Could not load board from boardPath: {board_path}", - "errorDetails": str(e), + "errorDetails": ( + "pcbnew.LoadBoard failed or returned a dehydrated " + "SWIG proxy that could not be recovered" + ), } + self.board = reloaded + self._update_command_handlers() + logger.info("Board reloaded from boardPath") project_path = Path(board_path).parent if project_path != getattr(self, "_current_project_path", None): @@ -4197,7 +4326,16 @@ class KiCADInterface: # Determine board to work with board = None if board_path: - board = pcbnew.LoadBoard(board_path) + board = self._safe_load_board(board_path) + if board is None: + return { + "success": False, + "message": f"Could not load board from {board_path}", + "errorDetails": ( + "pcbnew.LoadBoard failed or returned a dehydrated " + "SWIG proxy that could not be recovered" + ), + } elif self.board: board = self.board board_path = board.GetFileName() if not board_path else board_path @@ -4764,14 +4902,15 @@ class KiCADInterface: # call would overwrite the file with the stale in-memory state, erasing the # logo. Reload the board from disk so pcbnew's memory matches the file. if result.get("success") and self.board: - try: - self.board = pcbnew.LoadBoard(pcb_path) - # Propagate updated board reference to all command handlers + reloaded = self._safe_load_board(pcb_path) + if reloaded is not None: + self.board = reloaded self._update_command_handlers() logger.info("Reloaded board into pcbnew after SVG logo import") - except Exception as reload_err: + else: logger.warning( - f"Board reload after SVG import failed (non-fatal): {reload_err}" + "Board reload after SVG import failed (non-fatal); " + "next mutation may operate on stale in-memory state" ) return result @@ -4868,12 +5007,19 @@ class KiCADInterface: return {"success": False, "message": str(e)} def _handle_check_kicad_ui(self, params: Dict[str, Any]) -> Dict[str, Any]: - """Check if KiCAD UI is running""" + """Check if KiCAD UI is running. + + `processes` is the single source of truth — `running` is derived from + its length so the two fields cannot disagree. Previously they came + from separate detection methods (pgrep regex vs. ps-aux substring) and + could race or use different filters, producing the confusing + `running=True, processes=[]` state users hit after quitting KiCAD. + """ logger.info("Checking if KiCAD UI is running") try: manager = KiCADProcessManager() - is_running = manager.is_running() - processes = manager.get_process_info() if is_running else [] + processes = manager.get_process_info() + is_running = len(processes) > 0 return { "success": True, @@ -4956,7 +5102,18 @@ print("ok") ) if result.returncode == 0 and "ok" in result.stdout: # Reload board after subprocess modified it - self.board = pcbnew.LoadBoard(board_path) + reloaded = self._safe_load_board(board_path) + if reloaded is None: + return { + "success": False, + "message": ( + "Zone fill subprocess succeeded but the board " + "could not be reloaded into pcbnew (SWIG state " + "is corrupt — restart the MCP server)" + ), + "zoneCount": zone_count, + } + self.board = reloaded self._update_command_handlers() logger.info("Zone fill subprocess succeeded") return { diff --git a/tests/test_swig_dehydration.py b/tests/test_swig_dehydration.py new file mode 100644 index 0000000..6c65ded --- /dev/null +++ b/tests/test_swig_dehydration.py @@ -0,0 +1,376 @@ +""" +Tests for SWIG board-dehydration detection and recovery in KiCADInterface. + +Reproduces the failure mode users hit on KiCAD nightly builds: after sequences +like delete_trace + add_via, pcbnew.LoadBoard returns a "dehydrated" BOARD — +a SWIG proxy whose method dispatch table is missing, so every method access +raises AttributeError. Without recovery, every subsequent call fails and +open_project keeps reporting fake success because LoadBoard didn't raise. + +Also covers the check_kicad_ui consistency fix where running and processes +could disagree. +""" + +import sys +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")) + + +# --------------------------------------------------------------------------- +# Fixtures: stand-ins for SWIG BOARD proxies +# --------------------------------------------------------------------------- + + +def _make_dehydrated_board() -> Any: + """A bare object simulating the post-dehydration SwigPyObject — no methods.""" + + class _Dehydrated: + # Intentionally empty: hasattr(...) for the health-check methods + # returns False, matching the symptom users hit. + pass + + return _Dehydrated() + + +def _make_healthy_board() -> Any: + """A MagicMock that satisfies _is_board_healthy.""" + board = MagicMock(name="HealthyBoard") + board.GetDesignSettings = MagicMock(return_value=MagicMock()) + board.GetBoardEdgesBoundingBox = MagicMock(return_value=MagicMock()) + board.GetFileName = MagicMock(return_value="/tmp/test.kicad_pcb") + return board + + +def _make_iface() -> Any: + """Build a KiCADInterface skipping __init__ — same pattern as test_erc_handler. + + handle_command reads `use_ipc` and `ipc_board_api` early; set them so a + test exercising the routing layer doesn't crash on the IPC short-circuit. + """ + with patch("kicad_interface.USE_IPC_BACKEND", False): + from kicad_interface import KiCADInterface + + iface = KiCADInterface.__new__(KiCADInterface) + iface.use_ipc = False + iface.ipc_board_api = None + return iface + + +# --------------------------------------------------------------------------- +# _is_board_healthy +# --------------------------------------------------------------------------- + + +def test_is_board_healthy_returns_false_for_none(): + iface = _make_iface() + iface.board = None + assert iface._is_board_healthy() is False + + +def test_is_board_healthy_returns_false_for_dehydrated_proxy(): + iface = _make_iface() + iface.board = _make_dehydrated_board() + assert iface._is_board_healthy() is False + + +def test_is_board_healthy_returns_true_for_real_board(): + iface = _make_iface() + iface.board = _make_healthy_board() + assert iface._is_board_healthy() is True + + +def test_is_board_healthy_accepts_explicit_target(): + iface = _make_iface() + iface.board = None # self.board is broken + healthy = _make_healthy_board() + # Explicit argument bypasses self.board + assert iface._is_board_healthy(healthy) is True + assert iface._is_board_healthy(_make_dehydrated_board()) is False + + +# --------------------------------------------------------------------------- +# _safe_load_board +# --------------------------------------------------------------------------- + + +def test_safe_load_board_returns_loaded_board_when_healthy(): + iface = _make_iface() + healthy = _make_healthy_board() + with patch("kicad_interface.pcbnew") as mock_pcbnew: + mock_pcbnew.LoadBoard = MagicMock(return_value=healthy) + result = iface._safe_load_board("/tmp/x.kicad_pcb") + assert result is healthy + + +def test_safe_load_board_retries_after_pcbnew_reload_when_dehydrated(): + iface = _make_iface() + dehydrated = _make_dehydrated_board() + healthy = _make_healthy_board() + + # First LoadBoard returns dehydrated → reload pcbnew → second LoadBoard returns healthy + with patch("kicad_interface.pcbnew") as mock_pcbnew, patch("importlib.reload") as mock_reload: + mock_pcbnew.LoadBoard = MagicMock(side_effect=[dehydrated, healthy]) + mock_reload.return_value = mock_pcbnew # reload returns the (mock) module + result = iface._safe_load_board("/tmp/x.kicad_pcb") + + assert result is healthy + assert mock_pcbnew.LoadBoard.call_count == 2 + mock_reload.assert_called_once() + + +def test_safe_load_board_returns_none_when_recovery_fails(): + iface = _make_iface() + dehydrated = _make_dehydrated_board() + + with patch("kicad_interface.pcbnew") as mock_pcbnew, patch("importlib.reload") as mock_reload: + mock_pcbnew.LoadBoard = MagicMock(return_value=dehydrated) # always dehydrated + mock_reload.return_value = mock_pcbnew + result = iface._safe_load_board("/tmp/x.kicad_pcb") + + assert result is None + + +def test_safe_load_board_returns_none_when_loadboard_raises(): + iface = _make_iface() + with patch("kicad_interface.pcbnew") as mock_pcbnew: + mock_pcbnew.LoadBoard = MagicMock(side_effect=RuntimeError("io error")) + result = iface._safe_load_board("/tmp/missing.kicad_pcb") + assert result is None + + +# --------------------------------------------------------------------------- +# handle_command("open_project") — surfacing dehydration vs. recovery +# --------------------------------------------------------------------------- + + +def _wire_open_project(iface: Any, board_to_assign: Any, board_path: str) -> None: + """Make iface.project_commands.open_project return success and assign a board.""" + iface.project_commands = MagicMock() + + def _fake(params): + iface.project_commands.board = board_to_assign + return { + "success": True, + "message": f"Opened project: {Path(board_path).name}", + "project": { + "name": Path(board_path).stem, + "path": board_path, + "boardPath": board_path, + }, + } + + iface.project_commands.open_project = _fake + # Stub the rest of _update_command_handlers' targets + for attr in ( + "board_commands", + "component_commands", + "routing_commands", + "design_rule_commands", + "export_commands", + "freerouting_commands", + ): + setattr(iface, attr, MagicMock()) + # Provide minimal command_routes so handle_command can route + iface.command_routes = {"open_project": iface.project_commands.open_project} + + +def test_handle_open_project_surfaces_dehydration_when_recovery_fails(): + """If LoadBoard returns a dehydrated proxy and recovery fails, the MCP must + return success=False — never claim "Opened project" while the board is + unusable. This is the fix for the silent-success bug users hit.""" + iface = _make_iface() + dehydrated = _make_dehydrated_board() + _wire_open_project(iface, dehydrated, "/tmp/test.kicad_pcb") + iface._safe_load_board = MagicMock(return_value=None) # recovery impossible + + result = iface.handle_command("open_project", {"filename": "/tmp/test.kicad_pro"}) + + assert result["success"] is False, f"Expected failure, got: {result}" + combined = (result.get("errorDetails", "") + result.get("message", "")).lower() + assert "dehydrated" in combined or "swigpyobject" in combined + + +def test_handle_open_project_recovers_via_safe_load_board(): + """When _safe_load_board succeeds, open_project should report success + with a warning and the recovered board should be installed.""" + iface = _make_iface() + dehydrated = _make_dehydrated_board() + healthy = _make_healthy_board() + _wire_open_project(iface, dehydrated, "/tmp/test.kicad_pcb") + iface._safe_load_board = MagicMock(return_value=healthy) + + result = iface.handle_command("open_project", {"filename": "/tmp/test.kicad_pro"}) + + assert result["success"] is True, f"Expected recovery success, got: {result}" + assert iface.board is healthy + iface._safe_load_board.assert_called_once_with("/tmp/test.kicad_pcb") + warnings = result.get("warnings", []) + assert any("dehydrated" in w.lower() for w in warnings) + + +def test_handle_open_project_passes_through_when_already_healthy(): + """The fast path: LoadBoard returned a healthy board, no recovery needed.""" + iface = _make_iface() + healthy = _make_healthy_board() + _wire_open_project(iface, healthy, "/tmp/test.kicad_pcb") + iface._safe_load_board = MagicMock( + side_effect=AssertionError("should not be called when board is healthy") + ) + + result = iface.handle_command("open_project", {"filename": "/tmp/test.kicad_pro"}) + + assert result["success"] is True + assert iface.board is healthy + assert "warnings" not in result # no recovery happened + + +# --------------------------------------------------------------------------- +# _auto_save_board — recover if save dehydrates the board in-memory +# --------------------------------------------------------------------------- + + +def test_auto_save_recovers_when_save_leaves_board_dehydrated(): + """If pcbnew.SaveBoard somehow corrupts the in-memory BOARD (observed on + nightlies after delete_trace + auto-save), _auto_save_board reloads from + disk so the next command sees a usable proxy.""" + iface = _make_iface() + + # Start with a healthy board that becomes dehydrated after SaveBoard runs + healthy = _make_healthy_board() + healthy.GetFileName = MagicMock(return_value="/tmp/test.kicad_pcb") + iface.board = healthy + + recovered = _make_healthy_board() + iface._safe_load_board = MagicMock(return_value=recovered) + # Stub _update_command_handlers' downstream targets + for attr in ( + "project_commands", + "board_commands", + "component_commands", + "routing_commands", + "design_rule_commands", + "export_commands", + "freerouting_commands", + ): + setattr(iface, attr, MagicMock()) + + def _save_then_dehydrate(path, board): + # SaveBoard succeeds but leaves iface.board unusable + iface.board = _make_dehydrated_board() + iface.board.GetFileName = MagicMock(return_value="/tmp/test.kicad_pcb") + + with patch("kicad_interface.pcbnew") as mock_pcbnew: + mock_pcbnew.SaveBoard = MagicMock(side_effect=_save_then_dehydrate) + iface._auto_save_board() + + assert iface.board is recovered + iface._safe_load_board.assert_called_once_with("/tmp/test.kicad_pcb") + + +# --------------------------------------------------------------------------- +# check_kicad_ui consistency fix +# --------------------------------------------------------------------------- + + +def test_check_kicad_ui_running_false_when_no_processes(): + iface = _make_iface() + with patch("kicad_interface.KiCADProcessManager") as MockMgr: + MockMgr.return_value.get_process_info.return_value = [] + result = iface._handle_check_kicad_ui({}) + assert result["success"] is True + assert result["running"] is False + assert result["processes"] == [] + + +def test_check_kicad_ui_running_true_when_processes_present(): + iface = _make_iface() + with patch("kicad_interface.KiCADProcessManager") as MockMgr: + MockMgr.return_value.get_process_info.return_value = [ + {"pid": "1234", "name": "kicad", "command": "/Applications/KiCad/.../kicad"} + ] + result = iface._handle_check_kicad_ui({}) + assert result["success"] is True + assert result["running"] is True + assert len(result["processes"]) == 1 + + +def test_check_kicad_ui_running_and_processes_never_disagree(): + """Regression test for the old race where running=True coexisted with + processes=[] because is_running() and get_process_info() used different + detection methods.""" + iface = _make_iface() + with patch("kicad_interface.KiCADProcessManager") as MockMgr: + # Even if some hypothetical separate is_running() returned True, + # only get_process_info matters now. + MockMgr.return_value.is_running.return_value = True + MockMgr.return_value.get_process_info.return_value = [] + result = iface._handle_check_kicad_ui({}) + assert result["running"] is False + assert result["processes"] == [] + + +# --------------------------------------------------------------------------- +# run_drc — caller-overridable timeout +# --------------------------------------------------------------------------- + + +def test_run_drc_honors_timeout_sec_param(): + """timeoutSec param must be passed to the subprocess.run timeout argument.""" + import subprocess as _sp + + from commands.design_rules import DesignRuleCommands + + cmds = DesignRuleCommands(board=_make_healthy_board()) + cmds.board.GetFileName = MagicMock(return_value="/tmp/exists.kicad_pcb") + + captured: dict = {} + + def _fake_run(cmd, **kwargs): + captured["timeout"] = kwargs.get("timeout") + # Raise TimeoutExpired so we don't need to fake JSON output + raise _sp.TimeoutExpired(cmd, kwargs.get("timeout", 0)) + + with ( + patch("os.path.exists", return_value=True), + patch("subprocess.run", side_effect=_fake_run), + patch.object(cmds, "_find_kicad_cli", return_value="/usr/bin/kicad-cli"), + ): + result = cmds.run_drc({"timeoutSec": 45}) + + assert captured["timeout"] == 45 + assert result["success"] is False + assert "45" in result["errorDetails"] + + +def test_run_drc_clamps_extreme_timeout_values(): + """Bad timeoutSec values must be clamped, not crash.""" + import subprocess as _sp + + from commands.design_rules import DesignRuleCommands + + cmds = DesignRuleCommands(board=_make_healthy_board()) + cmds.board.GetFileName = MagicMock(return_value="/tmp/exists.kicad_pcb") + + captured: dict = {} + + def _fake_run(cmd, **kwargs): + captured.setdefault("timeouts", []).append(kwargs.get("timeout")) + raise _sp.TimeoutExpired(cmd, kwargs.get("timeout", 0)) + + with ( + patch("os.path.exists", return_value=True), + patch("subprocess.run", side_effect=_fake_run), + patch.object(cmds, "_find_kicad_cli", return_value="/usr/bin/kicad-cli"), + ): + cmds.run_drc({"timeoutSec": -100}) # below clamp + cmds.run_drc({"timeoutSec": 99999}) # above clamp + cmds.run_drc({"timeoutSec": "garbage"}) # unparseable → default 600 + + assert captured["timeouts"][0] == 10 # clamped to min + assert captured["timeouts"][1] == 1800 # clamped to max + assert captured["timeouts"][2] == 600 # fallback default