from __future__ import annotations from typing import Any from pathlib import Path from core.tools.base import BaseTool, ToolContext from core.tools.registry import registry from core.safety import safety from core.events import bus from core.subprocess import run_command class BochsTool(BaseTool): name = "bochs" description = "Bochs emulator control (run, validate, debug, config execution)" # ========================================================= # EXECUTE ROUTER # ========================================================= def execute(self, payload: dict[str, Any], ctx: ToolContext): action = str(payload.get("action", "")).strip() bus.log( "BOCHS", "bochs_execute", "INFO", {"action": action} ) match action: case "run": return self.run_vm(payload) case "validate": return self.validate_config(payload) case "debug": return self.debug_vm(payload) case _: raise ValueError(f"Unknown bochs action: {action}") # ========================================================= # HELPERS # ========================================================= def _path(self, value: str) -> Path: return safety.validate_path(value) # ========================================================= # RUN VM # ========================================================= def run_vm(self, payload: dict[str, Any]): config = payload.get("config") if not isinstance(config, str): raise ValueError("config must be string") config_path = self._path(config) if not config_path.exists(): raise ValueError(f"Bochs config not found: {config}") result = run_command( cmd=["bochs", "-f", str(config_path), "-q"], ) return { "action": "run", "config": str(config_path), "status": "success" if result.get("return_code") == 0 else "error", "stdout": result.get("stdout", ""), "stderr": result.get("stderr", "") } # ========================================================= # VALIDATE CONFIG # ========================================================= def validate_config(self, payload: dict[str, Any]): config = payload.get("config") if not isinstance(config, str): raise ValueError("config must be string") config_path = self._path(config) if not config_path.exists(): return { "status": "error", "error": "config file not found" } # Bochs has no strict "validate" mode, so we simulate dry run parse result = run_command( cmd=["bochs", "-f", str(config_path), "-n"], ) return { "action": "validate", "config": str(config_path), "status": "ok" if result.get("return_code") == 0 else "warning", "stdout": result.get("stdout", ""), "stderr": result.get("stderr", "") } # ========================================================= # DEBUG MODE # ========================================================= def debug_vm(self, payload: dict[str, Any]): config = payload.get("config") if not isinstance(config, str): raise ValueError("config must be string") config_path = self._path(config) if not config_path.exists(): raise ValueError("config not found") result = run_command( cmd=["bochs", "-f", str(config_path), "-q", "-debug"], ) return { "action": "debug", "config": str(config_path), "status": "success" if result.get("return_code") == 0 else "error", "stdout": result.get("stdout", ""), "stderr": result.get("stderr", "") } # ========================================================= # REGISTER TOOL # ========================================================= registry.register(BochsTool())