path: root/debug/risc-v-gsp-rm-vm/plugins/mcp_server.py

"""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())