debug: green-eye wave 3 - RM harness, local simulator, device map, MCP plugin, value-validation demos, CI pipeline, toolchain docs; add host RM tools rmtool/rmtop

1 files changed, 1155 insertions, 0 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
new file mode 100644
index 0000000..b5e8f90
--- /dev/null
+++ b/debug/risc-v-gsp-rm-vm/plugins/mcp_server.py
@@ -0,0 +1,1155 @@
+"""green-eye MCP server plugin (roadmap R3).
+
+A stdlib-only Model Context Protocol server (JSON-RPC 2.0) that exposes the
+*active* green-eye session as MCP tools.  Two ways to run it:
+
+* **plugin mode** - load it into a live session and serve that session::
+
+      ./greeneye.sh attach --plugin plugins/mcp_server.py
+      greeneye> mcp-serve [--allow-write] [--allow-inject]
+
+* **standalone mode** - launch its own session and serve it::
+
+      python3 plugins/mcp_server.py --stdio [--kernel <elf>]
+              [--backend qemu|native] [--record] [--plugins PATH]
+
+Transport is newline-delimited JSON-RPC over stdin/stdout by default; an
+optional TCP mode binds to loopback only (``--tcp 127.0.0.1:PORT``).  stdout
+carries protocol frames only - all logs go to stderr.
+
+Safety: read/observe tools are always available; memory writes and remote
+calls require ``--allow-write`` and service injection requires
+``--allow-inject``.  Capabilities owned by other components (the QEMU-only
+harnesses, the native trace) are probed defensively and return a structured
+``not_available`` tool error when absent.  See ``docs/20-MCP-SERVER.md``.
+"""
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import os
+import socket
+import subprocess
+import sys
+from typing import Iterable
+
+_TOOL_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+if _TOOL_ROOT not in sys.path:  # `python3 plugins/mcp_server.py` must import greeneye
+    sys.path.insert(0, _TOOL_ROOT)
+
+from greeneye.errors import GreenEyeError  # noqa: E402
+
+SERVER_NAME = "greeneye-mcp"
+JSONRPC_VERSION = "2.0"
+SUPPORTED_PROTOCOLS = ("2024-11-05", "2025-03-26", "2025-06-18")
+DEFAULT_PROTOCOL = "2024-11-05"
+
+PARSE_ERROR = -32700
+INVALID_REQUEST = -32600
+METHOD_NOT_FOUND = -32601
+INVALID_PARAMS = -32602
+
+_DUMPS = os.path.join(_TOOL_ROOT, "dumps")
+_PLUGINS_DIR = os.path.join(_TOOL_ROOT, "plugins")
+_DEMO_ELF = os.path.join(_TOOL_ROOT, "firmware", "uproc_demo", "build", "uproc_demo.elf")
+_FWIMAGE = os.path.join(
+    os.path.dirname(_TOOL_ROOT), "firmware", "57X-GSP", "570.144", "gsp-570.144.fwimage.bin"
+)
+MUTATION_SCRIPT = os.path.join(_TOOL_ROOT, "harness", "mutation_harness.py")
+MATRIX_SCRIPT = os.path.join(_TOOL_ROOT, "harness", "portability_matrix.py")
+LOOPBACK_HOSTS = ("127.0.0.1", "localhost", "::1")
+
+#: Per-response size caps so a single tool call can never flood a client.
+MAX_READ_BYTES = 1 << 20
+MAX_HEXDUMP_BYTES = 1 << 20
+MAX_SNAPSHOT_BYTES = 1 << 24
+
+
+class McpError(Exception):
+    """A JSON-RPC level error (unknown method/tool, bad params)."""
+
+    def __init__(self, code: int, message: str, data=None):
+        super().__init__(message)
+        self.code = code
+        self.message = message
+        self.data = data
+
+
+class ToolError(Exception):
+    """A tool-execution error returned inside a tools/call result (isError)."""
+
+    def __init__(self, message: str, *, code: str = "error", **details):
+        super().__init__(message)
+        self.payload = {"error": code, "message": message, **details}
+
+
+def _unavailable(message: str, **details) -> ToolError:
+    return ToolError(message, code="not_available", **details)
+
+
+# ---------------------------------------------------------------------------
+# tool catalogue
+# ---------------------------------------------------------------------------
+
+def _address(description: str) -> dict:
+    return {"type": ["string", "integer"], "description": description}
+
+
+def _integer(description: str, default=None, minimum=None, maximum=None) -> dict:
+    schema: dict = {"type": "integer", "description": description}
+    if default is not None:
+        schema["default"] = default
+    if minimum is not None:
+        schema["minimum"] = minimum
+    if maximum is not None:
+        schema["maximum"] = maximum
+    return schema
+
+
+def _number(description: str, default=None) -> dict:
+    schema: dict = {"type": "number", "description": description}
+    if default is not None:
+        schema["default"] = default
+    return schema
+
+
+def _string(description: str, default=None) -> dict:
+    schema: dict = {"type": "string", "description": description}
+    if default is not None:
+        schema["default"] = default
+    return schema
+
+
+def _boolean(description: str, default=None) -> dict:
+    schema: dict = {"type": "boolean", "description": description}
+    if default is not None:
+        schema["default"] = default
+    return schema
+
+
+def _tool(name: str, description: str, properties: dict | None = None, required: Iterable[str] = ()) -> dict:
+    schema: dict = {"type": "object", "properties": properties or {}}
+    if required:
+        schema["required"] = list(required)
+    return {"name": name, "description": description, "inputSchema": schema}
+
+
+TOOLS: list[dict] = [
+    _tool("session_info", "Active session state: backend, PC, symbols, recording, breakpoints/watchpoints, enabled capabilities."),
+    _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")},
+        ("target",),
+    ),
+    _tool(
+        "bp_set",
+        "Set a software or hardware breakpoint at a symbol/address; hits are recorded in the session.",
+        {"target": _address("symbol or address"), "kind": {"type": "string", "enum": ["sw", "hw"], "default": "sw"}},
+        ("target",),
+    ),
+    _tool("bp_clear", "Clear every breakpoint at a symbol/address.", {"target": _address("symbol or address")}, ("target",)),
+    _tool(
+        "watch_set",
+        "Set a read/write/access watchpoint; hits are recorded in the session.",
+        {
+            "target": _address("symbol or address"),
+            "size": _integer("watched region size in bytes", 8, 1, 4096),
+            "kind": {"type": "string", "enum": ["w", "r", "a"], "default": "w"},
+        },
+        ("target",),
+    ),
+    _tool(
+        "run",
+        "Resume and service hooks/watchpoints for a wall-clock budget (bounded; default 1.0s).",
+        {
+            "duration": _number("wall-clock budget in seconds", 1.0),
+            "max_stops": _integer("stop budget (optional)", None, 1),
+            "timeout": _number("alias for duration when duration is null"),
+        },
+    ),
+    _tool(
+        "run_until",
+        "Run until execution reaches a target address/symbol, servicing intervening stops.",
+        {"target": _address("symbol or address"), "timeout": _number("seconds to wait", 5.0)},
+        ("target",),
+    ),
+    _tool("step", "Single-step the target n times, dispatching hooks and watchpoints.", {"count": _integer("instructions", 1, 1, 10000)}),
+    _tool(
+        "read_mem",
+        "Read memory and return lowercase hex.",
+        {"addr": _address("address or symbol"), "size": _integer("bytes to read", 64, 1, MAX_READ_BYTES)},
+        ("addr",),
+    ),
+    _tool(
+        "write_mem",
+        "Write hex bytes to target memory. Requires --allow-write at server startup.",
+        {"addr": _address("address or symbol"), "hex": _string("hex bytes, e.g. '01 02 ff'")},
+        ("addr", "hex"),
+    ),
+    _tool(
+        "hexdump",
+        "Symbol-annotated hexdump of target memory.",
+        {"addr": _address("address or symbol"), "size": _integer("bytes", 64, 1, MAX_HEXDUMP_BYTES)},
+        ("addr",),
+    ),
+    _tool(
+        "snapshot",
+        "Capture a memory region into the recording (zlib + sha256) and return its digest.",
+        {"addr": _address("address or symbol"), "size": _integer("bytes", None, 1, MAX_SNAPSHOT_BYTES), "label": _string("snapshot label", "mcp")},
+        ("addr", "size"),
+    ),
+    _tool(
+        "call_function",
+        "Remote-call a target function (Frida-like). Requires --allow-write at server startup.",
+        {"target": _address("symbol or address"), "args": {"type": "array", "items": {"type": "integer"}, "description": "a0..a7"}, "timeout": _number("seconds", 5.0)},
+        ("target",),
+    ),
+    _tool(
+        "inject_service",
+        "Load and invoke an injected RV64 service blob. Requires --allow-inject at server startup.",
+        {
+            "service": _string("path to the service ELF"),
+            "slot": _integer("demo handler slot", 6, 0, 7),
+            "load_addr": _address("load address (optional)"),
+            "cmd": _integer("service command", 1, 0),
+            "arg0": _integer("argument 0", 0, 0),
+            "arg1": _integer("argument 1", 0, 0),
+            "path": {"type": "string", "enum": ["remote", "firmware", "both"], "default": "remote"},
+            "settle": _number("seconds to settle before the firmware mailbox call", 0.25),
+        },
+        ("service",),
+    ),
+    _tool(
+        "trace_start",
+        "Start the native exec/access trace (native backend, recording session only).",
+        {
+            "exec": _boolean("record exec events", True),
+            "start": _address("only PCs >= start"),
+            "end": _address("only PCs <= end"),
+            "max_events": _integer("ring budget (0 = unlimited)"),
+        },
+    ),
+    _tool("trace_stop", "Stop the native trace, flush it to blobs/trace.jsonl and return event counts."),
+    _tool("analyze_current", "Render the symbol-annotated summary of the active recording.", {"max_rows": _integer("table row cap", 40, 1, 1000)}),
+    _tool(
+        "compare_sessions",
+        "A/B two recorded session directories at the symbolic event level.",
+        {"a": _string("session A directory"), "b": _string("session B directory"), "symbols": _boolean("resolve with the active symbol DB", True), "max_rows": _integer("table row cap", 40, 1, 1000)},
+        ("a", "b"),
+    ),
+    _tool(
+        "export_hits",
+        "Export breakpoint/watchpoint hits as Ghidra's greeneye_hits.tsv.",
+        {"session": _string("session directory (default: active recording)"), "out": _string("output TSV path")},
+    ),
+    _tool(
+        "probe_firmware",
+        "Run the QEMU probe CLI on a firmware image and report the first divergence (QEMU required).",
+        {
+            "elf": _string("firmware ELF (default: the project fwimage)"),
+            "instructions": _integer("step budget (0 = wall-clock duration)", 2000, 0),
+            "duration": _number("wall-clock seconds when instructions=0", 1.0),
+            "no_relocate": _boolean("do not relocate to 0x80000000", False),
+            "symbols": _string("green-eye symbol DB JSON"),
+            "timeout": _number("subprocess timeout seconds", 300.0),
+        },
+    ),
+    _tool(
+        "mutation_run",
+        "Run harness/mutation_harness.py (QEMU-only) and return the outcome-class counts.",
+        {
+            "kernel": _string("demo/re-hosted ELF (default: active kernel or bundled demo)"),
+            "out": _string("output directory (default: dumps/mcp-mutation)"),
+            "iterations": _integer("mutation count", 10, 1, 10000),
+            "seed": _integer("deterministic seed", 1),
+            "timeout": _number("observation budget per mutation", 0.25),
+            "settle": _number("settle budget per attempt", 0.2),
+            "memory_region": _string("optional RAM region ADDR:SIZE"),
+            "allow_foreign_elf": _boolean("permit a non-demo ELF", False),
+        },
+    ),
+    _tool(
+        "portability_matrix_run",
+        "Run harness/portability_matrix.py (QEMU-only) over one or more firmware ELFs.",
+        {
+            "targets": {"type": "array", "items": {"type": "string"}, "description": "firmware ELF paths"},
+            "out": _string("output directory (default: dumps/mcp-matrix)"),
+            "probe_instructions": _integer("probe step budget", 2000, 0),
+            "scenario_duration": _number("per-phase wall budget", 0.4),
+            "symbols": _string("green-eye symbol DB JSON"),
+            "max_rows": _integer("report row cap", 40, 1, 1000),
+        },
+        ("targets",),
+    ),
+    _tool("plugins_list", "List bundled plugins, GREENEYE_PLUGINS and plugin-provided REPL commands."),
+    _tool("close", "Stop serving MCP; in standalone mode also close the launched session."),
+]
+
+_TOOL_NAMES = {tool["name"] for tool in TOOLS}
+
+
+# ---------------------------------------------------------------------------
+# server
+# ---------------------------------------------------------------------------
+
+
+class GreenEyeMCPServer:
+    """JSON-RPC 2.0 MCP server over newline-delimited streams."""
+
+    def __init__(
+        self,
+        session,
+        *,
+        allow_write: bool = False,
+        allow_inject: bool = False,
+        api=None,
+        owns_session: bool = False,
+        input_stream=None,
+        output_stream=None,
+        log_stream=None,
+    ):
+        self.session = session
+        self.allow_write = bool(allow_write)
+        self.allow_inject = bool(allow_inject)
+        self.api = api  # GreenEyeAPI when loaded as a plugin (plugins_list)
+        self.owns_session = bool(owns_session)
+        self._input = input_stream if input_stream is not None else sys.stdin
+        self._output = output_stream if output_stream is not None else sys.stdout
+        self._log_stream = log_stream if log_stream is not None else sys.stderr
+        self._stop = False
+        self.session_closed = False
+        self._initialized = False
+
+    # -- transport ---------------------------------------------------------
+
+    def serve_stdio(self) -> None:
+        self._log(f"serving {self.session.name!r} over stdio (write={self.allow_write}, inject={self.allow_inject})")
+        self.serve_stream(self._input, self._output)
+        self._log("stdio server stopped")
+
+    def serve_stream(self, stream_in, stream_out) -> None:
+        for raw in stream_in:
+            if self._stop:
+                break
+            if isinstance(raw, bytes):
+                raw = raw.decode("utf-8", "replace")
+            self.process_line(raw, stream_out)
+
+    def serve_tcp(self, host: str, port: int) -> None:
+        if host not in LOOPBACK_HOSTS:
+            raise ValueError(f"refusing non-loopback bind {host!r}; MCP is local-only")
+        server = socket.socket(socket.AF_INET if ":" not in host else socket.AF_INET6, socket.SOCK_STREAM)
+        server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+        server.bind((host, port))
+        server.listen(2)
+        bound = server.getsockname()
+        self._log(f"serving {self.session.name!r} on {bound[0]}:{bound[1]} (loopback only)")
+        try:
+            while not self._stop:
+                try:
+                    conn, _addr = server.accept()
+                except OSError:
+                    break
+                with conn:
+                    reader = conn.makefile("r", encoding="utf-8", newline="\n")
+                    writer = conn.makefile("w", encoding="utf-8", newline="\n")
+                    try:
+                        self.serve_stream(reader, writer)
+                    finally:
+                        reader.close()
+                        writer.close()
+        finally:
+            server.close()
+            self._log("tcp server stopped")
+
+    def process_line(self, line: str, stream_out=None) -> None:
+        stream = stream_out if stream_out is not None else self._output
+        line = line.strip()
+        if not line:
+            return
+        try:
+            message = json.loads(line)
+        except json.JSONDecodeError as exc:
+            self._write(stream, self._error(None, PARSE_ERROR, f"parse error: {exc}"))
+            return
+        try:
+            response = self.handle_message(message)
+        except McpError as exc:  # defensive: protocol bugs must not kill the loop
+            self._log(f"protocol error: {exc}")
+            response = self._error(message.get("id") if isinstance(message, dict) else None, exc.code, exc.message, exc.data)
+        if response is not None:
+            self._write(stream, response)
+
+    def _write(self, stream, payload: dict) -> None:
+        stream.write(json.dumps(payload, separators=(",", ":"), default=str) + "\n")
+        stream.flush()
+
+    # -- JSON-RPC ----------------------------------------------------------
+
+    @staticmethod
+    def _result(request_id, result: dict) -> dict:
+        return {"jsonrpc": JSONRPC_VERSION, "id": request_id, "result": result}
+
+    @staticmethod
+    def _error(request_id, code: int, message: str, data=None) -> dict:
+        payload = {"jsonrpc": JSONRPC_VERSION, "id": request_id, "error": {"code": code, "message": message}}
+        if data is not None:
+            payload["error"]["data"] = data
+        return payload
+
+    def handle_message(self, message) -> dict | None:
+        if not isinstance(message, dict):
+            return self._error(None, INVALID_REQUEST, "message must be a JSON object")
+        method = message.get("method")
+        request_id = message.get("id")
+        if not isinstance(method, str):
+            return self._error(request_id, INVALID_REQUEST, "missing method")
+        params = message.get("params")
+        if params is None:
+            params = {}
+        if not isinstance(params, dict):
+            return self._error(request_id, INVALID_PARAMS, "params must be an object")
+
+        if method.startswith("notifications/") or "id" not in message:
+            return None  # notifications never get a response
+
+        if method == "initialize":
+            return self._result(request_id, self._initialize(params))
+        if method == "ping":
+            return self._result(request_id, {})
+        if method == "tools/list":
+            return self._result(request_id, {"tools": TOOLS})
+        if method == "tools/call":
+            try:
+                result = self._dispatch_tool(params)
+            except McpError as exc:
+                return self._error(request_id, exc.code, exc.message, exc.data)
+            return self._result(request_id, result)
+        return self._error(request_id, METHOD_NOT_FOUND, f"unknown method {method!r}", {"method": method})
+
+    def _initialize(self, params: dict) -> dict:
+        requested = params.get("protocolVersion")
+        version = requested if requested in SUPPORTED_PROTOCOLS else DEFAULT_PROTOCOL
+        self._initialized = True
+        return {
+            "protocolVersion": version,
+            "capabilities": {"tools": {"listChanged": False}},
+            "serverInfo": {"name": SERVER_NAME, "version": _version()},
+            "instructions": (
+                "green-eye session server. Read/observe tools are always on; writes need "
+                "--allow-write and injection needs --allow-inject. Never claim timing, cycles "
+                "or hardware behaviour; sessions are the evidence."
+            ),
+        }
+
+    # -- tool dispatch -----------------------------------------------------
+
+    def _dispatch_tool(self, params: dict) -> dict:
+        name = params.get("name")
+        if not isinstance(name, str):
+            raise McpError(INVALID_PARAMS, "tools/call requires a tool name")
+        handler = getattr(self, f"tool_{name}", None)
+        if handler is None or name not in _TOOL_NAMES:
+            raise McpError(INVALID_PARAMS, f"unknown tool {name!r}", {"tool": name})
+        arguments = params.get("arguments")
+        if arguments is None:
+            arguments = {}
+        if not isinstance(arguments, dict):
+            raise McpError(INVALID_PARAMS, "tool arguments must be an object")
+        try:
+            result = handler(arguments)
+        except ToolError as exc:
+            return _tool_error(exc.payload)
+        except GreenEyeError as exc:
+            return _tool_error({"error": "greeneye", "message": str(exc)})
+        except Exception as exc:  # one bad call must never take the server down
+            self._log(f"tool {name} failed: {exc!r}")
+            return _tool_error({"error": "internal", "message": f"{type(exc).__name__}: {exc}"})
+        return {"content": [{"type": "text", "text": json.dumps(result, sort_keys=True, default=str)}]}
+
+    def _require_capability(self, name: str) -> None:
+        if name == "write" and not self.allow_write:
+            raise ToolError(
+                "target writes and remote calls are disabled; restart the server with --allow-write",
+                code="write_disabled",
+            )
+        if name == "inject" and not self.allow_inject:
+            raise ToolError(
+                "service injection is disabled; restart the server with --allow-inject",
+                code="inject_disabled",
+            )
+
+    def _annotate(self, addr: int) -> str:
+        return self.session.symbols.annotate(addr)
+
+    def _session_kernel(self) -> str | None:
+        target = self.session.target
+        kernel = getattr(target, "kernel", None)
+        if kernel:
+            return str(kernel)
+        return getattr(self.session.image, "path", None)
+
+    # -- tools: session ----------------------------------------------------
+
+    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")
+        return {
+            "name": session.name,
+            "backend": backend,
+            "pc": session.pc,
+            "pc_symbol": self._annotate(session.pc),
+            "symbols": len(session.symbols),
+            "recording": getattr(session.recorder, "dir", None),
+            "kernel": self._session_kernel(),
+            "closed": self.session_closed,
+            "capabilities": {
+                "write": self.allow_write,
+                "inject": self.allow_inject,
+                "trace": backend == "native",
+            },
+            "breakpoints": [
+                {"addr": bp.addr, "label": bp.label, "kind": bp.kind, "hits": bp.hits}
+                for bp in session._bps.values()  # noqa: SLF001 - the MCP server is a privileged client
+            ],
+            "watches": [
+                {"addr": wp.addr, "size": wp.size, "kind": wp.kind, "hits": wp.hits}
+                for wp in session._watches.values()  # noqa: SLF001
+            ],
+            "tools": len(TOOLS),
+        }
+
+    def tool_symbols_resolve(self, args: dict) -> dict:
+        addr = self.session.resolve(_require(args, "target"))
+        sym = self.session.symbols.resolve(addr)
+        return {"address": addr, "symbol": sym.name if sym else None, "annotation": self._annotate(addr)}
+
+    # -- tools: instrumentation -------------------------------------------
+
+    def tool_bp_set(self, args: dict) -> dict:
+        target = _require(args, "target")
+        kind = args.get("kind", "sw")
+        if kind not in ("sw", "hw"):
+            raise ToolError("kind must be 'sw' or 'hw'", code="invalid_arguments")
+        handle = self.session.hook(target, kind=kind, listener="mcp")
+        return {"addr": handle.addr, "label": handle.label, "kind": kind}
+
+    def tool_bp_clear(self, args: dict) -> dict:
+        addr = self.session.resolve(_require(args, "target"))
+        removed = 0
+        for bp in list(self.session._bps.values()):  # noqa: SLF001
+            if bp.addr == addr:
+                self.session.remove_breakpoint(bp)
+                removed += 1
+        return {"addr": addr, "cleared": removed}
+
+    def tool_watch_set(self, args: dict) -> dict:
+        kind = args.get("kind", "w")
+        if kind not in ("w", "r", "a"):
+            raise ToolError("kind must be 'w', 'r' or 'a'", code="invalid_arguments")
+        wp = self.session.watch(
+            _require(args, "target"),
+            size=_as_int(args.get("size", 8), "size"),
+            kind=kind,
+            listener="mcp",
+        )
+        return {"addr": wp.addr, "size": wp.size, "kind": wp.kind, "label": wp.label}
+
+    # -- tools: execution --------------------------------------------------
+
+    def tool_run(self, args: dict) -> dict:
+        duration = args.get("duration", 1.0)
+        timeout = args.get("timeout")
+        if duration is None:
+            duration = timeout if timeout is not None else 1.0
+        max_stops = args.get("max_stops")
+        stops = self.session.run(
+            duration=float(duration),
+            max_stops=_as_int(max_stops, "max_stops") if max_stops is not None else None,
+            timeout=float(timeout) if timeout is not None else None,
+        )
+        return {"stops": stops, "pc": self.session.pc, "pc_symbol": self._annotate(self.session.pc)}
+
+    def tool_run_until(self, args: dict) -> dict:
+        hit = self.session.run_until(_require(args, "target"), timeout=float(args.get("timeout", 5.0)))
+        return {
+            "addr": hit.addr,
+            "kind": hit.kind,
+            "symbol": hit.symbol.name if hit.symbol else None,
+            "pc": hit.pc,
+            "pc_symbol": self._annotate(hit.pc),
+        }
+
+    def tool_step(self, args: dict) -> dict:
+        count = _as_int(args.get("count", 1), "count")
+        if not 1 <= count <= 10000:
+            raise ToolError("count must be in 1..10000", code="invalid_arguments")
+        hits = self.session.step(count)
+        return {
+            "steps": len(hits),
+            "pc": self.session.pc,
+            "pc_symbol": self._annotate(self.session.pc),
+            "hits": [
+                {"kind": hit.kind, "addr": hit.addr, "symbol": hit.symbol.name if hit.symbol else None}
+                for hit in hits
+            ],
+        }
+
+    # -- tools: memory -----------------------------------------------------
+
+    def tool_read_mem(self, args: dict) -> dict:
+        addr = self.session.resolve(_require(args, "addr"))
+        size = _bounded(_as_int(args.get("size", 64), "size"), 1, MAX_READ_BYTES, "size")
+        data = self.session.memory.read(addr, size)
+        return {"addr": addr, "size": len(data), "hex": data.hex(), "annotation": self._annotate(addr)}
+
+    def tool_write_mem(self, args: dict) -> dict:
+        self._require_capability("write")
+        addr = self.session.resolve(_require(args, "addr"))
+        data = _hex_bytes(_require(args, "hex"))
+        self.session.memory.write(addr, data)
+        return {"addr": addr, "written": len(data), "hex": data.hex(), "annotation": self._annotate(addr)}
+
+    def tool_hexdump(self, args: dict) -> dict:
+        addr = self.session.resolve(_require(args, "addr"))
+        size = _bounded(_as_int(args.get("size", 64), "size"), 1, MAX_HEXDUMP_BYTES, "size")
+        return {"addr": addr, "size": size, "text": self.session.memory.hexdump(addr, size)}
+
+    def tool_snapshot(self, args: dict) -> dict:
+        addr = self.session.resolve(_require(args, "addr"))
+        size = _bounded(_as_int(_require(args, "size"), "size"), 1, MAX_SNAPSHOT_BYTES, "size")
+        label = str(args.get("label", "mcp"))
+        data = self.session.memory.snapshot(addr, size, name=label)
+        return {
+            "label": label,
+            "addr": addr,
+            "size": len(data),
+            "sha256": hashlib.sha256(data).hexdigest(),
+            "recording": getattr(self.session.recorder, "dir", None),
+        }
+
+    def tool_call_function(self, args: dict) -> dict:
+        self._require_capability("write")
+        target = _require(args, "target")
+        call_args = args.get("args") or []
+        if not isinstance(call_args, list):
+            raise ToolError("args must be an array of integers", code="invalid_arguments")
+        if len(call_args) > 8:
+            raise ToolError("at most 8 arguments (a0..a7)", code="invalid_arguments")
+        values = [_as_int(value, "args[]") for value in call_args]
+        retval = self.session.call(target, values, timeout=float(args.get("timeout", 5.0)))
+        return {"retval": retval, "retval_hex": hex(retval), "target": str(target)}
+
+    def tool_inject_service(self, args: dict) -> dict:
+        self._require_capability("inject")
+        service_path = str(_require(args, "service"))
+        if not os.path.isfile(service_path):
+            raise ToolError(
+                f"service not found: {service_path} (build it with `make sdk`)", code="invalid_arguments"
+            )
+        try:
+            from greeneye.inject import invoke_remote, invoke_via_firmware, load_service
+        except Exception as exc:
+            raise _unavailable(f"injection support unavailable: {exc}") from exc
+        load_addr = args.get("load_addr")
+        service = load_service(
+            self.session,
+            service_path,
+            load_addr=_as_int(load_addr, "load_addr") if load_addr is not None else None,
+            slot=_as_int(args.get("slot", 6), "slot"),
+        )
+        cmd = _as_int(args.get("cmd", 1), "cmd")
+        a0 = _as_int(args.get("arg0", 0), "arg0")
+        a1 = _as_int(args.get("arg1", 0), "arg1")
+        mode = args.get("path", "remote")
+        if mode not in ("remote", "firmware", "both"):
+            raise ToolError("path must be 'remote', 'firmware' or 'both'", code="invalid_arguments")
+        result = {
+            "service": {
+                "name": service.name,
+                "load_addr": service.load_addr,
+                "entry_addr": service.entry_addr,
+                "slot": service.slot,
+                "size": service.size,
+            },
+            "cmd": cmd,
+            "args": [a0, a1],
+        }
+        if mode in ("remote", "both"):
+            result["remote"] = invoke_remote(self.session, service, cmd, a0, a1)
+        if mode in ("firmware", "both"):
+            self.session.run(duration=float(args.get("settle", 0.25)))
+            result["firmware"] = invoke_via_firmware(self.session, service, cmd, a0, a1)
+        return result
+
+    # -- tools: native trace ----------------------------------------------
+
+    def tool_trace_start(self, args: dict) -> dict:
+        if not hasattr(self.session.client, "trace"):
+            raise _unavailable("trace start/stop is native-backend only (start the server with --backend native)")
+        if getattr(self.session.recorder, "dir", None) is None:
+            raise _unavailable("trace_start requires a recording session (start the server/session with --record)")
+        if getattr(self.session.client, "trace", None) is not None:
+            raise ToolError("a trace is already active; call trace_stop first", code="invalid_arguments")
+        try:
+            from sim.trace import make_trace
+        except Exception as exc:
+            raise _unavailable(f"native trace module unavailable: {exc}") from exc
+        spec: dict = {"exec": bool(args.get("exec", True))}
+        for key in ("start", "end", "max_events"):
+            if args.get(key) is not None:
+                spec[key] = _as_int(args[key], key)
+        tracer = make_trace(spec)
+        self.session.client.trace = tracer
+        self.session.client._recorder = self.session.recorder  # noqa: SLF001 - SimClient boundary
+        self.session.recorder.meta["trace"] = tracer.config()
+        return {"tracing": True, "config": tracer.config()}
+
+    def tool_trace_stop(self, args: dict) -> dict:
+        tracer = getattr(self.session.client, "trace", None)
+        if tracer is None:
+            raise ToolError("no active trace", code="invalid_arguments")
+        self.session.client.trace = None
+        blob = tracer.flush(self.session.recorder)
+        return {"tracing": False, "blob": blob, "counts": tracer.counts()}
+
+    # -- tools: analysis ---------------------------------------------------
+
+    def tool_analyze_current(self, args: dict) -> dict:
+        directory = getattr(self.session.recorder, "dir", None)
+        if not directory:
+            raise _unavailable("the session is not recording; start it with --record")
+        try:
+            from greeneye.analyze import load_events, load_meta, load_trace, render_summary_from
+        except Exception as exc:
+            raise _unavailable(f"analysis support unavailable: {exc}") from exc
+        # An active recording has no session.json until close(), so merge the
+        # live recorder metadata instead of requiring the closed-session files.
+        if os.path.isfile(os.path.join(directory, "session.json")):
+            meta = load_meta(directory)
+        else:
+            meta = dict(getattr(self.session.recorder, "meta", {}))
+            meta.setdefault("format", "greeneye-session/1")
+            meta.setdefault("name", self.session.name)
+            meta.setdefault("dir", directory)
+        events = load_events(directory)
+        trace = load_trace(directory, meta)
+        report = render_summary_from(
+            meta, events, symbols=self.session.symbols, max_rows=_as_int(args.get("max_rows", 40), "max_rows"), trace=trace
+        )
+        return {"session": directory, "report": report}
+
+    def tool_compare_sessions(self, args: dict) -> dict:
+        a = str(_require(args, "a"))
+        b = str(_require(args, "b"))
+        for path in (a, b):
+            if not os.path.isdir(path):
+                raise ToolError(f"session directory not found: {path}", code="invalid_arguments")
+        try:
+            from greeneye.analyze import compare_sessions
+        except Exception as exc:
+            raise _unavailable(f"analysis support unavailable: {exc}") from exc
+        symbols = self.session.symbols if args.get("symbols", True) else None
+        report = compare_sessions(a, b, symbols=symbols, max_rows=_as_int(args.get("max_rows", 40), "max_rows"))
+        return {"a": a, "b": b, "report": report}
+
+    def tool_export_hits(self, args: dict) -> dict:
+        directory = args.get("session") or getattr(self.session.recorder, "dir", None)
+        if not directory:
+            raise _unavailable("no session directory given and the active session is not recording")
+        if not os.path.isdir(str(directory)):
+            raise ToolError(f"session directory not found: {directory}", code="invalid_arguments")
+        out = args.get("out") or os.path.join(str(directory), "greeneye_hits.tsv")
+        try:
+            from greeneye.ghidra import export_hits
+        except Exception as exc:
+            raise _unavailable(f"Ghidra export unavailable: {exc}") from exc
+        path = export_hits(str(directory), str(out), symbols=self.session.symbols)
+        return {"session": str(directory), "path": path}
+
+    # -- tools: external harnesses (defensive) -----------------------------
+
+    def tool_probe_firmware(self, args: dict) -> dict:
+        elf = args.get("elf") or (_FWIMAGE if os.path.isfile(_FWIMAGE) else None)
+        if not elf:
+            raise _unavailable(f"default fwimage not found and no elf argument given ({_FWIMAGE})")
+        if not os.path.isfile(str(elf)):
+            raise ToolError(f"firmware not found: {elf}", code="invalid_arguments")
+        cmd = [sys.executable, "-m", "greeneye", "probe", str(elf), "--record"]
+        if args.get("instructions") is not None:
+            cmd += ["--instructions", str(_as_int(args["instructions"], "instructions"))]
+        elif args.get("duration") is not None:
+            cmd += ["--duration", str(float(args["duration"]))]
+        if args.get("no_relocate"):
+            cmd.append("--no-relocate")
+        if args.get("symbols"):
+            cmd += ["--symbols", str(args["symbols"])]
+        rc, stdout, stderr = self._run_cli(cmd, timeout=float(args.get("timeout", 300.0)))
+        if rc != 0:
+            raise ToolError(
+                f"probe failed (rc={rc})", code="probe_failed", elf=str(elf), stdout=stdout[-4000:], stderr=stderr[-4000:]
+            )
+        return {"elf": str(elf), "returncode": rc, "stdout": stdout[-20000:], "stderr": stderr[-4000:]}
+
+    def tool_mutation_run(self, args: dict) -> dict:
+        if not os.path.isfile(MUTATION_SCRIPT):
+            raise _unavailable(f"mutation harness not present: {MUTATION_SCRIPT}")
+        kernel = args.get("kernel") or self._session_kernel() or _DEMO_ELF
+        if not os.path.isfile(str(kernel)):
+            raise ToolError(f"kernel not found: {kernel}", code="invalid_arguments")
+        out = str(args.get("out") or os.path.join(_DUMPS, "mcp-mutation"))
+        cmd = [
+            sys.executable,
+            MUTATION_SCRIPT,
+            "--kernel",
+            str(kernel),
+            "--out",
+            out,
+            "--iterations",
+            str(_as_int(args.get("iterations", 10), "iterations")),
+            "--seed",
+            str(_as_int(args.get("seed", 1), "seed")),
+            "--timeout",
+            str(float(args.get("timeout", 0.25))),
+            "--settle",
+            str(float(args.get("settle", 0.2))),
+        ]
+        if args.get("memory_region"):
+            cmd += ["--memory-region", str(args["memory_region"])]
+        if args.get("allow_foreign_elf"):
+            cmd.append("--allow-foreign-elf")
+        rc, stdout, stderr = self._run_cli(cmd, timeout=float(args.get("subprocess_timeout", 1800.0)))
+        outcomes = _count_findings(os.path.join(out, "findings.jsonl"))
+        result = {
+            "kernel": str(kernel),
+            "out": out,
+            "returncode": rc,
+            "outcomes": outcomes,
+            "findings": sum(outcomes.values()),
+            "stdout": stdout[-4000:],
+            "stderr": stderr[-4000:],
+        }
+        if rc != 0:
+            raise ToolError("mutation harness failed", code="harness_failed", **result)
+        return result
+
+    def tool_portability_matrix_run(self, args: dict) -> dict:
+        if not os.path.isfile(MATRIX_SCRIPT):
+            raise _unavailable(f"portability matrix not present: {MATRIX_SCRIPT}")
+        targets = args.get("targets")
+        if not isinstance(targets, list) or not targets:
+            raise ToolError("targets must be a non-empty array of ELF paths", code="invalid_arguments")
+        for target in targets:
+            if not os.path.isfile(str(target)):
+                raise ToolError(f"target not found: {target}", code="invalid_arguments")
+        out = str(args.get("out") or os.path.join(_DUMPS, "mcp-matrix"))
+        cmd = [sys.executable, MATRIX_SCRIPT, "--out", out]
+        for target in targets:
+            cmd += ["--target", str(target)]
+        cmd += [
+            "--probe-instructions",
+            str(_as_int(args.get("probe_instructions", 2000), "probe_instructions")),
+            "--scenario-duration",
+            str(float(args.get("scenario_duration", 0.4))),
+            "--max-rows",
+            str(_as_int(args.get("max_rows", 40), "max_rows")),
+        ]
+        if args.get("symbols"):
+            cmd += ["--symbols", str(args["symbols"])]
+        rc, stdout, stderr = self._run_cli(cmd, timeout=float(args.get("subprocess_timeout", 3600.0)))
+        report_path = os.path.join(out, "portability-report.json")
+        summary = None
+        if os.path.isfile(report_path):
+            try:
+                with open(report_path, "r", encoding="utf-8") as fh:
+                    report = json.load(fh)
+                summary = {
+                    "targets": [
+                        {"name": t.get("name"), "status": t.get("status"), "sha256": (t.get("sha256") or "")[:16]}
+                        for t in report.get("targets", [])
+                    ],
+                    "comparisons": len(report.get("comparisons", [])),
+                }
+            except (OSError, ValueError) as exc:
+                summary = {"parse_error": str(exc)}
+        result = {
+            "targets": [str(target) for target in targets],
+            "out": out,
+            "returncode": rc,
+            "report": report_path if os.path.isfile(report_path) else None,
+            "summary": summary,
+            "stdout": stdout[-4000:],
+            "stderr": stderr[-4000:],
+        }
+        if rc != 0:
+            raise ToolError("portability matrix failed", code="harness_failed", **result)
+        return result
+
+    def tool_plugins_list(self, args: dict) -> dict:
+        bundled: list[str] = []
+        if os.path.isdir(_PLUGINS_DIR):
+            bundled = sorted(
+                name
+                for name in os.listdir(_PLUGINS_DIR)
+                if name.endswith(".py") and not name.startswith("_")
+            )
+        commands = sorted(self.api.commands) if self.api is not None else []
+        return {
+            "bundled": bundled,
+            "GREENEYE_PLUGINS": os.environ.get("GREENEYE_PLUGINS"),
+            "commands": commands,
+            "mcp_plugin": "mcp_server.py" in bundled,
+        }
+
+    def tool_close(self, args: dict) -> dict:
+        if self.owns_session and not self.session_closed:
+            self.session.close()
+            self.session_closed = True
+        self._stop = True
+        return {"closed": True, "session_closed": self.session_closed}
+
+    # -- helpers -----------------------------------------------------------
+
+    def _run_cli(self, cmd: list[str], timeout: float) -> tuple[int, str, str]:
+        env = dict(os.environ)
+        env.setdefault("GREENEYE_DUMPS", _DUMPS)
+        try:
+            completed = subprocess.run(
+                cmd,
+                cwd=_TOOL_ROOT,
+                env=env,
+                capture_output=True,
+                text=True,
+                timeout=timeout,
+            )
+        except FileNotFoundError as exc:
+            raise _unavailable(f"cannot run {cmd[0]}: {exc}") from exc
+        except subprocess.TimeoutExpired as exc:
+            raise ToolError(f"command timed out after {timeout}s", code="timeout", cmd=cmd) from exc
+        return completed.returncode, completed.stdout or "", completed.stderr or ""
+
+    def _log(self, message: str) -> None:
+        try:
+            print(f"[greeneye-mcp] {message}", file=self._log_stream, flush=True)
+        except Exception:  # a broken log stream must not break serving
+            pass
+
+
+# ---------------------------------------------------------------------------
+# small argument helpers
+# ---------------------------------------------------------------------------
+
+
+def _require(args: dict, key: str):
+    if key not in args or args[key] is None:
+        raise ToolError(f"missing required argument {key!r}", code="invalid_arguments")
+    return args[key]
+
+
+def _as_int(value, name: str) -> int:
+    if isinstance(value, bool):
+        raise ToolError(f"{name} must be an integer", code="invalid_arguments")
+    if isinstance(value, int):
+        return int(value)
+    if isinstance(value, str):
+        try:
+            return int(value, 0)
+        except ValueError as exc:
+            raise ToolError(f"{name}: cannot parse integer from {value!r}", code="invalid_arguments") from exc
+    raise ToolError(f"{name} must be an integer", code="invalid_arguments")
+
+
+def _bounded(value: int, minimum: int, maximum: int, name: str) -> int:
+    if not minimum <= value <= maximum:
+        raise ToolError(f"{name} must be in {minimum}..{maximum}", code="invalid_arguments")
+    return value
+
+
+def _hex_bytes(text: str) -> bytes:
+    cleaned = str(text).strip().replace(" ", "").replace(":", "").replace(",", "")
+    if cleaned.startswith("0x") or cleaned.startswith("0X"):
+        cleaned = cleaned[2:]
+    try:
+        return bytes.fromhex(cleaned)
+    except ValueError as exc:
+        raise ToolError(f"bad hex data: {exc}", code="invalid_arguments") from exc
+
+
+def _tool_error(payload: dict) -> dict:
+    return {
+        "content": [{"type": "text", "text": json.dumps(payload, sort_keys=True, default=str)}],
+        "isError": True,
+    }
+
+
+def _count_findings(path: str) -> dict:
+    counts: dict[str, int] = {}
+    if not os.path.isfile(path):
+        return counts
+    try:
+        with open(path, "r", encoding="utf-8") as fh:
+            for line in fh:
+                line = line.strip()
+                if not line:
+                    continue
+                try:
+                    outcome = json.loads(line).get("outcome", "?")
+                except ValueError:
+                    continue
+                counts[outcome] = counts.get(outcome, 0) + 1
+    except OSError:
+        return counts
+    return counts
+
+
+def _version() -> str:
+    try:
+        import greeneye
+
+        return greeneye.__version__
+    except Exception:  # pragma: no cover - only on a broken install
+        return "unknown"
+
+
+# ---------------------------------------------------------------------------
+# plugin entry point
+# ---------------------------------------------------------------------------
+
+
+def build_serve_parser(prog: str = "mcp-serve") -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(
+        prog=prog,
+        description="serve the active green-eye session over MCP (JSON-RPC over stdio or loopback TCP)",
+    )
+    parser.add_argument("--stdio", action="store_true", help="serve over stdin/stdout (default)")
+    parser.add_argument("--tcp", metavar="HOST:PORT", type=parse_tcp, help="serve over loopback TCP instead")
+    parser.add_argument("--allow-write", action="store_true", help="enable memory writes and remote calls")
+    parser.add_argument("--allow-inject", action="store_true", help="enable service injection")
+    return parser
+
+
+def register(ge) -> None:
+    """green-eye plugin hook: add the ``mcp-serve`` command."""
+
+    def mcp_serve(session, argv: list[str]) -> int:
+        parser = build_serve_parser()
+        try:
+            options = parser.parse_args(argv)
+        except SystemExit as exc:  # --help must not kill the REPL
+            return int(exc.code or 0)
+        server = GreenEyeMCPServer(
+            session,
+            allow_write=options.allow_write,
+            allow_inject=options.allow_inject,
+            api=ge,
+            owns_session=False,
+        )
+        ge.log(
+            "mcp server",
+            phase="start",
+            transport=f"tcp:{options.tcp[0]}:{options.tcp[1]}" if options.tcp else "stdio",
+            allow_write=options.allow_write,
+            allow_inject=options.allow_inject,
+        )
+        try:
+            if options.tcp:
+                server.serve_tcp(*options.tcp)
+            else:
+                server.serve_stdio()
+        finally:
+            if getattr(session.recorder, "dir", None):
+                ge.log("mcp server", phase="stop", requests="see events")
+        return 0
+
+    ge.add_command(
+        "mcp-serve",
+        mcp_serve,
+        help="serve this live session over MCP (stdlib JSON-RPC; --allow-write/--allow-inject to enable writes)",
+    )
+
+
+# ---------------------------------------------------------------------------
+# standalone entry point
+# ---------------------------------------------------------------------------
+
+
+def parse_tcp(text: str) -> tuple[str, int]:
+    host, _, port_text = text.rpartition(":")
+    host = host or "127.0.0.1"
+    if host not in LOOPBACK_HOSTS:
+        raise argparse.ArgumentTypeError(f"MCP is local-only; {host!r} is not loopback (use 127.0.0.1)")
+    try:
+        port = int(port_text)
+    except ValueError as exc:
+        raise argparse.ArgumentTypeError(f"expected HOST:PORT, got {text!r}") from exc
+    if not 0 <= port <= 65535:
+        raise argparse.ArgumentTypeError(f"port out of range: {port}")
+    return host, port
+
+
+def build_parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(prog="mcp_server", description="green-eye MCP server (roadmap R3)")
+    transport = parser.add_mutually_exclusive_group()
+    transport.add_argument("--stdio", action="store_true", help="serve over stdin/stdout (default)")
+    transport.add_argument("--tcp", metavar="HOST:PORT", type=parse_tcp, help="serve over loopback TCP instead")
+    parser.add_argument("--kernel", help="firmware ELF to launch (default: bundled demo when built)")
+    parser.add_argument("--backend", choices=("qemu", "native"), default="qemu", help="execution backend")
+    parser.add_argument("--record", nargs="?", const=True, default=None, help="record to a path (or dumps/)")
+    parser.add_argument("--name", default="mcp", help="session name")
+    parser.add_argument("--plugins", action="append", default=[], help="extra plugin path (repeatable)")
+    parser.add_argument("--allow-write", action="store_true", help="enable memory writes and remote calls")
+    parser.add_argument("--allow-inject", action="store_true", help="enable service injection")
+    return parser
+
+
+def _launch_standalone(args) -> tuple[object, object]:
+    import greeneye as ge
+    from greeneye.plugins import PluginManager
+
+    os.environ.setdefault("GREENEYE_DUMPS", _DUMPS)
+    kernel = args.kernel
+    if kernel and not os.path.isfile(kernel):
+        raise GreenEyeError(f"kernel not found: {kernel}")
+    if not kernel and os.path.isfile(_DEMO_ELF):
+        kernel = _DEMO_ELF
+    session = ge.launch(kernel=kernel, backend=args.backend, record=args.record, name=args.name)
+    manager = PluginManager(session)
+    manager.load_all(args.plugins)
+    return session, manager.api
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = build_parser()
+    args = parser.parse_args(argv)
+    try:
+        session, api = _launch_standalone(args)
+    except Exception as exc:
+        print(f"greeneye-mcp error: {exc}", file=sys.stderr)
+        return 2
+    server = GreenEyeMCPServer(
+        session,
+        allow_write=args.allow_write,
+        allow_inject=args.allow_inject,
+        api=api,
+        owns_session=True,
+    )
+    try:
+        if args.tcp:
+            server.serve_tcp(*args.tcp)
+        else:
+            server.serve_stdio()
+    except KeyboardInterrupt:
+        print("[greeneye-mcp] interrupted", file=sys.stderr)
+    finally:
+        if not server.session_closed:
+            try:
+                session.close()
+            except Exception:  # closing is best-effort on shutdown
+                pass
+    return 0
+
+
+if __name__ == "__main__":  # pragma: no cover - process entry point
+    raise SystemExit(main())