diff options
Diffstat (limited to 'debug/risc-v-gsp-rm-vm/plugins')
| -rw-r--r-- | debug/risc-v-gsp-rm-vm/plugins/mcp_server.py | 434 |
1 files changed, 433 insertions, 1 deletions
diff --git a/debug/risc-v-gsp-rm-vm/plugins/mcp_server.py b/debug/risc-v-gsp-rm-vm/plugins/mcp_server.py index 0059da2..5acce65 100644 --- a/debug/risc-v-gsp-rm-vm/plugins/mcp_server.py +++ b/debug/risc-v-gsp-rm-vm/plugins/mcp_server.py @@ -29,6 +29,7 @@ import argparse import hashlib import json import os +import shutil import socket import subprocess import sys @@ -58,8 +59,12 @@ _FWIMAGE = os.path.join( ) MUTATION_SCRIPT = os.path.join(_TOOL_ROOT, "harness", "mutation_harness.py") MATRIX_SCRIPT = os.path.join(_TOOL_ROOT, "harness", "portability_matrix.py") +SERVICE_ELF = os.path.join(_TOOL_ROOT, "sdk", "build", "injected_demo.elf") LOOPBACK_HOSTS = ("127.0.0.1", "localhost", "::1") +#: Firmware digests are cached by (path, size, mtime) - the real image is 28 MiB. +_SHA_CACHE: dict[tuple, str | None] = {} + #: Per-response size caps so a single tool call can never flood a client. MAX_READ_BYTES = 1 << 20 MAX_HEXDUMP_BYTES = 1 << 20 @@ -138,6 +143,21 @@ def _tool(name: str, description: str, properties: dict | None = None, required: TOOLS: list[dict] = [ _tool("session_info", "Active session state: backend, PC, symbols, recording, breakpoints/watchpoints, enabled capabilities."), _tool( + "backends_list", + "List the execution backends this server can launch (QEMU and the native GSP simulator) and the active one.", + ), + _tool( + "session_launch", + "Launch a new session on a chosen backend (protocol-level backend selection) and serve it. Standalone mode only.", + { + "backend": {"type": "string", "enum": ["native", "qemu"], "default": "native", "description": "execution backend"}, + "kernel": _string("firmware ELF (default: bundled demo when built)"), + "record": {"type": ["boolean", "string"], "default": True, "description": "record to dumps/ (true) or an explicit directory"}, + "name": _string("session name", "mcp"), + "trace": _boolean("native only: enable the exec/access trace at launch", False), + }, + ), + _tool( "symbols_resolve", "Resolve a symbol name, 'symbol+0xNN' expression or hex address against the session symbol DB.", {"target": _address("symbol name, symbol+offset or address")}, @@ -222,6 +242,10 @@ TOOLS: list[dict] = [ ("service",), ), _tool( + "service_list", + "List services injected into this session (injection SDK registry) and the demo handler slot table.", + ), + _tool( "trace_start", "Start the native exec/access trace (native backend, recording session only).", { @@ -232,6 +256,24 @@ TOOLS: list[dict] = [ }, ), _tool("trace_stop", "Stop the native trace, flush it to blobs/trace.jsonl and return event counts."), + _tool( + "sim_cycles", + "Native simulator counters: instruction count, modeled cycle total, the cycle-model table and its fidelity label.", + ), + _tool( + "sim_device_state", + "Native simulator device state: unknown-MMIO policy/log, device-map size, and the sim/devices model when present.", + {"limit": _integer("most recent unmodelled MMIO entries to include", 16, 0, 1000)}, + ), + _tool( + "sim_profile", + "Native simulator boot/profile metadata: firmware, bias, stop info, trace config, model identity and fidelity.", + ), + _tool( + "sim_regions", + "Native simulator memory-region metadata: RAM/UART windows, loaded firmware segments, device-map entries and observed unmodelled MMIO pages.", + {"limit": _integer("entry cap for the device-map and observed-page lists", 16, 0, 1000)}, + ), _tool("analyze_current", "Render the symbol-annotated summary of the active recording.", {"max_rows": _integer("table row cap", 40, 1, 1000)}), _tool( "compare_sessions", @@ -493,9 +535,16 @@ class GreenEyeMCPServer: # -- tools: session ---------------------------------------------------- + def _active_backend(self) -> str: + if hasattr(self.session.client, "trace"): + return "native" + if self.session.target is not None: + return "qemu" + return "custom" + def tool_session_info(self, args: dict) -> dict: session = self.session - backend = "native" if hasattr(session.client, "trace") else ("qemu" if session.target is not None else "custom") + backend = self._active_backend() return { "name": session.name, "backend": backend, @@ -526,6 +575,76 @@ class GreenEyeMCPServer: sym = self.session.symbols.resolve(addr) return {"address": addr, "symbol": sym.name if sym else None, "annotation": self._annotate(addr)} + def tool_backends_list(self, args: dict) -> dict: + try: + import greeneye.sim_backend # noqa: F401 + + native_available, native_note = True, None + except Exception as exc: # pragma: no cover - only on a broken install + native_available, native_note = False, str(exc) + qemu_binary = shutil.which("qemu-system-riscv64") or os.environ.get("GREENEYE_QEMU") + return { + "active": self._active_backend(), + "launchable": self.owns_session, + "backends": [ + { + "name": "native", + "available": native_available, + "transport": "in-process deterministic RV64IMAC simulator (sim_backend.launch_native)", + "features": ["cycles", "trace", "device_state", "injection", "deterministic"], + "note": native_note, + }, + { + "name": "qemu", + "available": bool(qemu_binary), + "transport": "qemu-system-riscv64 gdbstub", + "binary": qemu_binary, + "features": ["trace_plugin", "injection"], + }, + ], + } + + def tool_session_launch(self, args: dict) -> dict: + if not self.owns_session: + raise _unavailable("session_launch is standalone-mode only; in plugin mode the REPL owns the session") + backend = str(args.get("backend", "native")) + if backend not in ("qemu", "native"): + raise ToolError("backend must be 'qemu' or 'native'", code="invalid_arguments") + kernel = args.get("kernel") + if kernel is not None: + kernel = str(kernel) + if not os.path.isfile(kernel): + raise ToolError(f"kernel not found: {kernel}", code="invalid_arguments") + elif os.path.isfile(_DEMO_ELF): + kernel = _DEMO_ELF + record = args.get("record", True) + if record is False: + record = None + trace = bool(args.get("trace", False)) + if trace and backend != "native": + raise ToolError("trace= is native-only (QEMU uses the TCG plugin)", code="invalid_arguments") + name = str(args.get("name", "mcp")) + try: + import greeneye as ge + + session = ge.launch( + kernel=kernel, + backend=backend, + record=record, + name=name, + trace=trace or None, + ) + except Exception as exc: + raise ToolError(f"launch failed: {exc}", code="launch_failed", backend=backend, kernel=kernel) + if self.session is not None and not self.session_closed: + try: + self.session.close() + except Exception as exc: # replacing a session must not fail the launch + self._log(f"closing previous session failed: {exc!r}") + self.session = session + self.session_closed = False + return {"launched": True, "backend": self._active_backend(), "session": self.tool_session_info({})} + # -- tools: instrumentation ------------------------------------------- def tool_bp_set(self, args: dict) -> dict: @@ -684,6 +803,39 @@ class GreenEyeMCPServer: result["firmware"] = invoke_via_firmware(self.session, service, cmd, a0, a1) return result + def tool_service_list(self, args: dict) -> dict: + try: + from greeneye.inject import CTRL_HANDLERS, INJECT_CMD_BASE, services + except Exception as exc: + raise _unavailable(f"injection support unavailable: {exc}") from exc + loaded = [] + for service in services(self.session): + view = getattr(service, "to_dict", None) + loaded.append(view() if callable(view) else {"name": getattr(service, "name", None)}) + slots: list[dict] = [] + ctrl = self.session.symbols.get("gsp_ctrl") + if ctrl is not None: + for index in range(8): + try: + value = self.session.memory.read_u64(ctrl.addr + CTRL_HANDLERS + index * 8) + except Exception: + slots = [] + break + slots.append( + { + "slot": index, + "cmd": INJECT_CMD_BASE + index, + "entry": value, + "annotation": self._annotate(value) if value else None, + } + ) + return { + "services": loaded, + "count": len(loaded), + "slots": slots, + "mailbox_cmd_base": INJECT_CMD_BASE, + } + # -- tools: native trace ---------------------------------------------- def tool_trace_start(self, args: dict) -> dict: @@ -715,6 +867,262 @@ class GreenEyeMCPServer: blob = tracer.flush(self.session.recorder) return {"tracing": False, "blob": blob, "counts": tracer.counts()} + # -- tools: native simulator ------------------------------------------ + + def _native_machine(self): + if not hasattr(self.session.client, "trace"): + raise _unavailable("this tool requires the native simulator backend (launch/serve with --backend native)") + machine = getattr(self.session.target, "machine", None) + if machine is None: + raise _unavailable("native backend exposes no Machine handle") + return machine + + def tool_sim_cycles(self, args: dict) -> dict: + machine = self._native_machine() + model = getattr(machine, "cycle_model", None) + cycles = int(getattr(machine, "cycles", 0)) + insns = int(getattr(machine, "insns", 0)) + describe = getattr(model, "describe", None) + model_info = describe() if callable(describe) else ({"name": model.name} if model is not None else None) + try: + from sim.cycles import CYCLE_FIDELITY + except Exception: # pragma: no cover - sim always ships cycles.py today + CYCLE_FIDELITY = "cycles are modeled defaults, not measured" + return { + "backend": "native", + "insns": insns, + "cycles": cycles, + "cycles_per_instruction": round(cycles / insns, 6) if insns else None, + "accounting_enabled": model is not None, + "cycle_model": model_info, + "fidelity": CYCLE_FIDELITY if model is not None else "cycle accounting disabled for this session", + } + + def _device_model_state(self, machine) -> dict: + """Defensive probe of an optional ``sim.devices`` device model. + + The S1 device model may land concurrently with this server; when its + module/entry point is absent the tool reports a structured + ``not_available`` sub-result instead of failing the import or call. + """ + import importlib + + try: + module = importlib.import_module("sim.devices") + except Exception as exc: + return { + "error": "not_available", + "message": f"sim.devices device model not present: {exc}", + } + for name in ("state_for", "describe", "snapshot", "state", "device_state"): + entry = getattr(module, name, None) + if not callable(entry): + continue + try: + return {"source": f"sim.devices.{name}", "state": entry(machine)} + except TypeError: + try: + return {"source": f"sim.devices.{name}", "state": entry()} + except Exception as exc: + return {"error": "probe_failed", "message": f"sim.devices.{name}() failed: {exc}"} + except Exception as exc: + return {"error": "probe_failed", "message": f"sim.devices.{name}(machine) failed: {exc}"} + devices = getattr(machine, "devices", None) + if devices is not None: + state = getattr(devices, "state", None) + if callable(state): + try: + return {"source": "machine.devices.state", "state": state()} + except Exception as exc: + return { + "error": "probe_failed", + "message": f"machine.devices.state() failed: {exc}", + } + return { + "error": "not_available", + "message": "machine.devices exposes no state() entry point", + } + return { + "error": "not_available", + "message": "sim.devices is present but exposes no state_for/describe/snapshot entry point", + } + + def tool_sim_device_state(self, args: dict) -> dict: + machine = self._native_machine() + limit = _bounded(_as_int(args.get("limit", 16), "limit"), 0, 1000, "limit") + target = self.session.target + log = getattr(target, "mmio_log", None) + if log is None: + log = getattr(machine, "mmio_log", []) or [] + device_map = getattr(machine, "device_map", None) + return { + "backend": "native", + "pc": self.session.pc, + "pc_symbol": self._annotate(self.session.pc), + "bias": getattr(machine, "bias", 0), + "unknown_mmio": getattr(machine, "unknown_mmio", None), + "mmio_unknown": int(getattr(machine, "mmio_unknown", 0)), + "mmio_log": list(log)[-limit:] if limit else [], + "device_map_entries": len(device_map) if device_map is not None else None, + "device_model": self._device_model_state(machine), + } + + def tool_sim_profile(self, args: dict) -> dict: + machine = self._native_machine() + meta = dict(getattr(self.session.recorder, "meta", {}) or {}) + model = getattr(machine, "cycle_model", None) + kernel = self._session_kernel() + device_map = getattr(machine, "device_map", None) + profile = { + "backend": "native", + "kernel": kernel, + "firmware_sha256": _file_sha256(kernel), + "bias": getattr(machine, "bias", 0), + "unknown_mmio": getattr(machine, "unknown_mmio", None), + "device_map_entries": len(device_map) if device_map is not None else None, + "insns": int(getattr(machine, "insns", 0)), + "cycles": int(getattr(machine, "cycles", 0)), + "cycle_model": model.name if model is not None else None, + "accounting_enabled": model is not None, + "mmio_unknown": int(getattr(machine, "mmio_unknown", 0)), + "recording": getattr(self.session.recorder, "dir", None), + "profile_kind": meta.get("kind"), + "trace": meta.get("trace"), + "session_meta": meta, + } + if meta.get("stop_reason") is not None: + profile["stop"] = { + "reason": meta.get("stop_reason"), + "pc": meta.get("stop_pc"), + "pc_fw": meta.get("stop_pc_fw"), + } + return profile + + def tool_sim_regions(self, args: dict) -> dict: + """Read-only region map of the native session: RAM, UART, loaded + firmware segments, the optional device map and observed unmodelled + MMIO pages. Metadata only - no execution, no timing claims.""" + machine = self._native_machine() + limit = _bounded(_as_int(args.get("limit", 16), "limit"), 0, 1000, "limit") + target = self.session.target + bias = int(getattr(machine, "bias", 0) or 0) + ram_base = getattr(machine, "ram_base", None) + if ram_base is None: + ram_base = getattr(target, "ram_base", None) + ram_size = getattr(machine, "ram_size", None) + if ram_size is None: + memory_mb = getattr(target, "memory_mb", None) + ram_size = int(memory_mb) * 1024 * 1024 if memory_mb else None + + regions: list[dict] = [] + if ram_base is not None and ram_size: + regions.append( + { + "name": "ram", + "kind": "ram", + "base": int(ram_base), + "size": int(ram_size), + "end": int(ram_base) + int(ram_size), + } + ) + try: + from sim.machine import UART_BASE, UART_SIZE + except Exception: # pragma: no cover - constants always ship with sim + UART_BASE, UART_SIZE = 0x10000000, 0x100 + regions.append( + { + "name": "uart", + "kind": "device", + "base": UART_BASE, + "size": UART_SIZE, + "end": UART_BASE + UART_SIZE, + "note": "NS16550A console", + } + ) + image = getattr(self.session, "image", None) + for index, segment in enumerate(getattr(image, "segments", []) or []): + if not getattr(segment, "is_load", False): + continue + vaddr = int(getattr(segment, "p_vaddr", 0)) + regions.append( + { + "name": f"image[{index}]", + "kind": "firmware", + "base": vaddr + bias, + "size": int(getattr(segment, "p_memsz", 0)), + "end": vaddr + bias + int(getattr(segment, "p_memsz", 0)), + "file_size": int(getattr(segment, "p_filesz", 0)), + "vaddr_fw": vaddr, + "flags": { + "read": True, + "write": bool(getattr(segment, "writable", False)), + "exec": bool(getattr(segment, "executable", False)), + }, + "source": getattr(image, "path", None), + } + ) + + device_map = getattr(machine, "device_map", None) + device_total = None + device_entries: list[dict] = [] + entries_fn = getattr(device_map, "entries", None) + if callable(entries_fn): + all_entries = list(entries_fn()) + device_total = len(all_entries) + for entry in all_entries[:limit] if limit else []: + device_entries.append( + { + "address": int(getattr(entry, "address", 0)), + "width": int(getattr(entry, "width", 0)), + "kind": getattr(entry, "kind", None), + "hits": int(getattr(entry, "hits", 0)), + "first_epc_fw": int(getattr(entry, "first_epc_fw", 0)), + "source": getattr(entry, "source", None), + } + ) + + pages: dict[int, dict] = {} + for record in getattr(machine, "mmio_log", []) or []: + addr = int(record.get("addr", 0)) + page = addr & ~0xFFF + info = pages.setdefault( + page, + {"page": page, "touches": 0, "reads": 0, "writes": 0, "addresses": set(), "pcs": set()}, + ) + info["touches"] += 1 + if record.get("rw") == "w": + info["writes"] += 1 + else: + info["reads"] += 1 + info["addresses"].add(addr) + if record.get("pc") is not None: + info["pcs"].add(int(record["pc"])) + observed = [ + { + "page": info["page"], + "touches": info["touches"], + "reads": info["reads"], + "writes": info["writes"], + "addresses": sorted(info["addresses"]), + "pcs": sorted(info["pcs"]), + } + for info in sorted(pages.values(), key=lambda item: (-item["touches"], item["page"]))[:limit] + if limit + ] + + return { + "backend": "native", + "pc": self.session.pc, + "pc_symbol": self._annotate(self.session.pc), + "bias": bias, + "unknown_mmio": getattr(machine, "unknown_mmio", None), + "regions": regions, + "device_map_entries": device_total, + "device_map": device_entries, + "observed_mmio_pages": observed, + "fidelity": "region metadata only; no timing or device-environment claims", + } + # -- tools: analysis --------------------------------------------------- def tool_analyze_current(self, args: dict) -> dict: @@ -1013,6 +1421,30 @@ def _version() -> str: return "unknown" +def _file_sha256(path: str | None) -> str | None: + """Digest of a firmware file, cached by (path, size, mtime).""" + if not path or not os.path.isfile(str(path)): + return None + path = str(path) + try: + stat = os.stat(path) + except OSError: + return None + key = (path, stat.st_size, stat.st_mtime_ns) + if key in _SHA_CACHE: + return _SHA_CACHE[key] + try: + digest = hashlib.sha256() + with open(path, "rb") as fh: + for chunk in iter(lambda: fh.read(1 << 20), b""): + digest.update(chunk) + value = digest.hexdigest() + except OSError: + value = None + _SHA_CACHE[key] = value + return value + + # --------------------------------------------------------------------------- # plugin entry point # --------------------------------------------------------------------------- |