path: root/debug/risc-v-gsp-rm-vm/tests/test_mcp.py

"""Tests for the risc-v-gsp-rm-debugger MCP server plugin (roadmap R3).

The protocol and tool tests drive an in-memory Session over a stub RSP-shaped
client, so they need no QEMU and no simulator.  A standalone subprocess test
launches the native backend end-to-end and skips when that backend is absent.
"""
from __future__ import annotations

import io
import json
import os
import subprocess
import sys
import tempfile
import unittest

HERE = os.path.dirname(os.path.abspath(__file__))
TOOL_ROOT = os.path.abspath(os.path.join(HERE, ".."))
PLUGINS_DIR = os.path.join(TOOL_ROOT, "plugins")
MCP_SERVER = os.path.join(PLUGINS_DIR, "mcp_server.py")
DEMO_ELF = os.path.join(TOOL_ROOT, "firmware", "uproc_demo", "build", "uproc_demo.elf")
SERVICE_ELF = os.path.join(TOOL_ROOT, "sdk", "build", "injected_demo.elf")
for _path in (TOOL_ROOT, PLUGINS_DIR):
    if _path not in sys.path:
        sys.path.insert(0, _path)

import mcp_server as mcp  # noqa: E402
import greeneye as ge  # noqa: E402
from greeneye.inject import SERVICE_REGISTRY_ATTR, InjectedService  # noqa: E402
from greeneye.rsp import StopReply  # noqa: E402
from greeneye.session import Session  # noqa: E402
from greeneye.symbols import Symbol, SymbolDB  # noqa: E402

BASE = 0x80000000


class StubClient:
    """RSP-shaped in-memory client: enough for Session without an emulator."""

    def __init__(self, size: int = 0x20000):
        self.base = BASE
        self.mem = bytearray(size)
        self.regs = {"pc": BASE, "ra": 0, "sp": BASE + size, "a0": 0, "a1": 0}
        self.sw_bps: set[int] = set()
        self.hw_bps: set[int] = set()
        self.watches: dict[tuple[int, int, str], None] = {}
        self.closed = False
        self.features = "swbreak+;hwbreak+"

    def _span(self, addr: int, size: int) -> int:
        start = addr - self.base
        if start < 0 or start + size > len(self.mem):
            raise ValueError(f"out of range: {addr:#x}+{size}")
        return start

    def read_mem(self, addr: int, size: int, chunk: int = 0) -> bytes:
        start = self._span(addr, size)
        return bytes(self.mem[start : start + size])

    def write_mem(self, addr: int, data: bytes, chunk: int = 0) -> None:
        start = self._span(addr, len(data))
        self.mem[start : start + len(data)] = data

    def read_registers(self) -> dict[str, int]:
        return dict(self.regs)

    def read_reg(self, name: str) -> int:
        return int(self.regs.get(str(name), 0))

    def write_reg(self, name: str, value: int) -> None:
        self.regs[str(name)] = int(value)

    @property
    def pc(self) -> int:
        return self.regs["pc"]

    def set_sw_break(self, addr: int, kind: int = 4) -> None:
        self.sw_bps.add(addr)

    def clear_sw_break(self, addr: int, kind: int = 4) -> None:
        self.sw_bps.discard(addr)

    def set_hw_break(self, addr: int, kind: int = 4) -> None:
        self.hw_bps.add(addr)

    def clear_hw_break(self, addr: int, kind: int = 4) -> None:
        self.hw_bps.discard(addr)

    def set_watch(self, addr: int, size: int = 8, kind: str = "w") -> None:
        self.watches[(addr, size, kind)] = None

    def clear_watch(self, addr: int, size: int = 8, kind: str = "w") -> None:
        self.watches.pop((addr, size, kind), None)

    def continue_(self, timeout: float | None = None) -> StopReply:
        return StopReply(raw="T05", signal=5)

    def step(self, timeout: float | None = None) -> StopReply:
        return StopReply(raw="T05", signal=5)

    def interrupt(self, timeout: float | None = None) -> StopReply:
        return StopReply(raw="T02", signal=2)

    def close(self) -> None:
        self.closed = True


class McpTestCase(unittest.TestCase):
    def setUp(self):
        self.client = StubClient()
        symbols = SymbolDB()
        symbols.add(Symbol(addr=BASE + 0x100, name="rpc_ping", size=0x20, source="test"))
        symbols.add(Symbol(addr=BASE + 0x200, name="gsp_ctrl", size=0x40, source="test"))
        self.session = Session(client=self.client, symbols=symbols, record=False, name="mcp-test")
        self.addCleanup(self.session.close)

    def make_server(self, **kw):
        self.inp = io.StringIO()
        self.out = io.StringIO()
        self.log = io.StringIO()
        options = dict(input_stream=self.inp, output_stream=self.out, log_stream=self.log)
        options.update(kw)
        return mcp.GreenEyeMCPServer(self.session, **options)

    def request(self, server, method, params=None, request_id=1):
        message = {"jsonrpc": "2.0", "id": request_id, "method": method}
        if params is not None:
            message["params"] = params
        return server.handle_message(message)

    def call_tool(self, server, name, **arguments):
        response = self.request(server, "tools/call", {"name": name, "arguments": arguments}, request_id=9)
        self.assertIn("result", response, response)
        result = response["result"]
        payload = json.loads(result["content"][0]["text"])
        return payload, result.get("isError", False)


class ProtocolTests(McpTestCase):
    def test_initialize(self):
        response = self.request(
            self.make_server(),
            "initialize",
            {"protocolVersion": "2024-11-05", "capabilities": {}, "clientInfo": {"name": "t", "version": "0"}},
        )
        result = response["result"]
        self.assertEqual(response["jsonrpc"], "2.0")
        self.assertEqual(result["protocolVersion"], "2024-11-05")
        self.assertEqual(result["serverInfo"]["name"], "greeneye-mcp")
        self.assertIn("tools", result["capabilities"])

    def test_initialize_negotiation_and_fallback(self):
        server = self.make_server()
        self.assertEqual(
            self.request(server, "initialize", {"protocolVersion": "2025-03-26"})["result"]["protocolVersion"],
            "2025-03-26",
        )
        self.assertEqual(
            self.request(server, "initialize", {"protocolVersion": "1999-01-01"})["result"]["protocolVersion"],
            mcp.DEFAULT_PROTOCOL,
        )

    def test_initialized_notification_has_no_response(self):
        server = self.make_server()
        self.assertIsNone(server.handle_message({"jsonrpc": "2.0", "method": "notifications/initialized"}))

    def test_ping(self):
        self.assertEqual(self.request(self.make_server(), "ping")["result"], {})

    def test_tools_list_catalogue(self):
        tools = self.request(self.make_server(), "tools/list")["result"]["tools"]
        names = [tool["name"] for tool in tools]
        self.assertGreaterEqual(len(names), 24)
        self.assertEqual(len(names), len(mcp.TOOLS))
        self.assertEqual(set(names), mcp._TOOL_NAMES)
        for expected in (
            "backends_list",
            "session_launch",
            "service_list",
            "sim_cycles",
            "sim_device_state",
            "sim_profile",
            "sim_regions",
        ):
            self.assertIn(expected, names)
        for tool in tools:
            self.assertEqual(tool["inputSchema"]["type"], "object")
            self.assertTrue(tool["description"])

    def test_unknown_method(self):
        response = self.request(self.make_server(), "no/such")
        self.assertEqual(response["error"]["code"], mcp.METHOD_NOT_FOUND)
        self.assertEqual(response["id"], 1)

    def test_unknown_tool(self):
        response = self.request(self.make_server(), "tools/call", {"name": "nope", "arguments": {}})
        self.assertEqual(response["error"]["code"], mcp.INVALID_PARAMS)

    def test_tools_call_before_initialize_is_allowed(self):
        payload, is_error = self.call_tool(self.make_server(), "session_info")
        self.assertFalse(is_error)
        self.assertEqual(payload["name"], "mcp-test")

    def test_parse_error_is_reported(self):
        server = self.make_server()
        server.process_line("{not json")
        line = self.out.getvalue().strip()
        self.assertEqual(json.loads(line)["error"]["code"], mcp.PARSE_ERROR)

    def test_serve_stream_end_to_end_and_no_stdout_pollution(self):
        server = self.make_server()
        lines = [
            json.dumps({"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {}}),
            json.dumps({"jsonrpc": "2.0", "method": "notifications/initialized"}),
            json.dumps({"jsonrpc": "2.0", "id": 2, "method": "tools/list"}),
            json.dumps(
                {"jsonrpc": "2.0", "id": 3, "method": "tools/call", "params": {"name": "session_info", "arguments": {}}}
            ),
        ]
        server.serve_stream(io.StringIO("\n".join(lines) + "\n"), self.out)
        raw = self.out.getvalue()
        responses = [json.loads(line) for line in raw.splitlines()]
        self.assertEqual([response["id"] for response in responses], [1, 2, 3])
        self.assertTrue(all(response["jsonrpc"] == "2.0" for response in responses))
        self.assertNotIn("[greeneye-mcp]", raw)
        self.assertNotIn("[greeneye-mcp]", self.inp.getvalue())


class ReadOnlyToolTests(McpTestCase):
    def test_symbols_resolve(self):
        payload, is_error = self.call_tool(self.make_server(), "symbols_resolve", target="rpc_ping")
        self.assertFalse(is_error)
        self.assertEqual(payload["address"], BASE + 0x100)
        self.assertEqual(payload["symbol"], "rpc_ping")

    def test_read_mem_and_hexdump(self):
        self.client.mem[0x10:0x14] = b"\xde\xad\xbe\xef"
        server = self.make_server()
        payload, is_error = self.call_tool(server, "read_mem", addr=BASE + 0x10, size=4)
        self.assertFalse(is_error)
        self.assertEqual(payload["hex"], "deadbeef")
        payload, is_error = self.call_tool(server, "hexdump", addr=BASE + 0x10, size=4)
        self.assertFalse(is_error)
        self.assertIn("de ad be ef", payload["text"])

    def test_bp_set_and_clear(self):
        server = self.make_server()
        payload, is_error = self.call_tool(server, "bp_set", target="rpc_ping")
        self.assertFalse(is_error)
        self.assertIn(BASE + 0x100, self.client.sw_bps)
        payload, is_error = self.call_tool(server, "bp_clear", target="rpc_ping")
        self.assertFalse(is_error)
        self.assertEqual(payload["cleared"], 1)
        self.assertNotIn(BASE + 0x100, self.client.sw_bps)

    def test_watch_set(self):
        payload, is_error = self.call_tool(self.make_server(), "watch_set", target="gsp_ctrl", size=8, kind="w")
        self.assertFalse(is_error)
        self.assertEqual(payload["addr"], BASE + 0x200)
        self.assertIn((BASE + 0x200, 8, "w"), self.client.watches)

    def test_snapshot_digest(self):
        payload, is_error = self.call_tool(self.make_server(), "snapshot", addr=BASE, size=16, label="unit")
        self.assertFalse(is_error)
        self.assertEqual(payload["size"], 16)
        self.assertEqual(len(payload["sha256"]), 64)

    def test_session_info_capabilities(self):
        payload, is_error = self.call_tool(
            self.make_server(allow_write=True, allow_inject=True), "session_info"
        )
        self.assertFalse(is_error)
        self.assertEqual(payload["name"], "mcp-test")
        self.assertTrue(payload["capabilities"]["write"])
        self.assertTrue(payload["capabilities"]["inject"])
        self.assertFalse(payload["capabilities"]["trace"])

    def test_plugins_list(self):
        payload, is_error = self.call_tool(self.make_server(), "plugins_list")
        self.assertFalse(is_error)
        self.assertIn("mcp_server.py", payload["bundled"])

    def test_analyze_current_reads_active_recording(self):
        with tempfile.TemporaryDirectory(prefix="mcp-rec-") as directory:
            session = Session(client=StubClient(), symbols=SymbolDB(), record=directory, name="active")
            try:
                server = mcp.GreenEyeMCPServer(
                    session,
                    input_stream=io.StringIO(),
                    output_stream=io.StringIO(),
                    log_stream=io.StringIO(),
                )
                payload, is_error = self.call_tool(server, "analyze_current")
                self.assertFalse(is_error)
                self.assertIn("# risc-v-gsp-rm-debugger session summary", payload["report"])
            finally:
                session.close()


class StubCycleModel:
    """Minimal CycleModel shape for MCP sim-tool tests (no sim/ import needed)."""

    name = "rv64imac-modeled/1"

    def describe(self) -> dict:
        return {"name": self.name, "alu": 1, "memory_latency": 3, "fidelity": "modeled"}


class StubMachine:
    """Machine-shaped object exposing the public surface the sim tools read."""

    def __init__(self):
        self.cycles = 120
        self.insns = 100
        self.cycle_model = StubCycleModel()
        self.unknown_mmio = "lenient"
        self.bias = 0x80000000
        self.ram_base = BASE
        self.ram_size = 0x20000
        self.mmio_log = [{"pc": BASE + 4, "addr": 0x2000000000000A00, "size": 8, "rw": "r"}]
        self.mmio_unknown = 1
        self.device_map = None


class StubTarget:
    """SimTarget-shaped handle so the server classifies the session as native."""

    keep_alive = True

    def __init__(self, machine):
        self.machine = machine
        self.mmio_log = machine.mmio_log
        self.mmio_unknown = machine.mmio_unknown
        self.ram_base = BASE
        self.memory_mb = 256

    def kill(self):
        pass


class NativeSimToolTests(McpTestCase):
    def setUp(self):
        super().setUp()
        self.client.trace = None  # backend detection: native
        self.machine = StubMachine()
        self.session.target = StubTarget(self.machine)

    def test_backends_list(self):
        payload, is_error = self.call_tool(self.make_server(owns_session=True), "backends_list")
        self.assertFalse(is_error)
        self.assertEqual(payload["active"], "native")
        self.assertTrue(payload["launchable"])
        names = {entry["name"] for entry in payload["backends"]}
        self.assertEqual(names, {"native", "qemu"})

    def test_sim_cycles(self):
        payload, is_error = self.call_tool(self.make_server(), "sim_cycles")
        self.assertFalse(is_error)
        self.assertEqual(payload["backend"], "native")
        self.assertEqual(payload["insns"], 100)
        self.assertEqual(payload["cycles"], 120)
        self.assertEqual(payload["cycles_per_instruction"], 1.2)
        self.assertEqual(payload["cycle_model"]["name"], "rv64imac-modeled/1")
        self.assertIn("modeled", payload["fidelity"])

    def test_sim_device_state_reports_absent_device_model(self):
        payload, is_error = self.call_tool(self.make_server(), "sim_device_state", limit=4)
        self.assertFalse(is_error)
        self.assertEqual(payload["mmio_unknown"], 1)
        self.assertEqual(len(payload["mmio_log"]), 1)
        self.assertEqual(payload["unknown_mmio"], "lenient")
        self.assertIn("device_model", payload)
        if "error" in payload["device_model"]:
            self.assertEqual(payload["device_model"]["error"], "not_available")

    def test_sim_device_state_serialises_attached_registry(self):
        class Registry:
            def state(self):
                return {"schema": "greeneye-sim-devices/1", "devices": []}

        self.machine.devices = Registry()
        payload, is_error = self.call_tool(self.make_server(), "sim_device_state")
        self.assertFalse(is_error)
        self.assertEqual(payload["device_model"]["source"], "machine.devices.state")
        self.assertEqual(payload["device_model"]["state"]["schema"], "greeneye-sim-devices/1")
        json.dumps(payload)  # the tool result must stay JSON-serialisable

    def test_sim_profile(self):
        payload, is_error = self.call_tool(self.make_server(), "sim_profile")
        self.assertFalse(is_error)
        self.assertEqual(payload["backend"], "native")
        self.assertEqual(payload["bias"], 0x80000000)
        self.assertEqual(payload["cycles"], 120)
        self.assertIn("session_meta", payload)

    def test_sim_regions(self):
        payload, is_error = self.call_tool(self.make_server(), "sim_regions", limit=4)
        self.assertFalse(is_error)
        self.assertEqual(payload["backend"], "native")
        self.assertEqual(payload["bias"], 0x80000000)
        self.assertEqual(payload["unknown_mmio"], "lenient")
        regions = {region["name"]: region for region in payload["regions"]}
        self.assertIn("ram", regions)
        self.assertIn("uart", regions)
        self.assertEqual(regions["ram"]["base"], BASE)
        self.assertEqual(regions["ram"]["size"], 0x20000)
        self.assertEqual(regions["ram"]["end"], BASE + 0x20000)
        self.assertEqual(regions["uart"]["base"], 0x10000000)
        self.assertEqual(regions["uart"]["size"], 0x100)
        self.assertIsNone(payload["device_map_entries"])
        self.assertEqual(payload["device_map"], [])
        self.assertEqual(len(payload["observed_mmio_pages"]), 1)
        page = payload["observed_mmio_pages"][0]
        self.assertEqual(page["page"], 0x2000000000000A00 & ~0xFFF)
        self.assertEqual(page["reads"], 1)
        self.assertEqual(page["writes"], 0)
        self.assertEqual(page["addresses"], [0x2000000000000A00])
        self.assertEqual(page["pcs"], [BASE + 4])

    def test_sim_regions_device_map_entries(self):
        class Entry:
            address = 0x2000000000000A00
            width = 8
            kind = "r"
            hits = 3
            first_epc_fw = 0x4009014
            source = "probe"

        class Map:
            def entries(self):
                return [Entry()]

        self.machine.device_map = Map()
        payload, is_error = self.call_tool(self.make_server(), "sim_regions", limit=8)
        self.assertFalse(is_error)
        self.assertEqual(payload["device_map_entries"], 1)
        self.assertEqual(payload["device_map"][0]["address"], 0x2000000000000A00)
        self.assertEqual(payload["device_map"][0]["first_epc_fw"], 0x4009014)
        self.assertEqual(payload["device_map"][0]["source"], "probe")

    def test_sim_tools_refuse_non_native(self):
        # a fresh stub client without .trace is not a native session
        client = StubClient()
        session = Session(client=client, symbols=self.session.symbols, record=False, name="stub-qemu")
        server = mcp.GreenEyeMCPServer(
            session, input_stream=io.StringIO(), output_stream=io.StringIO(), log_stream=io.StringIO()
        )
        try:
            for name in ("sim_cycles", "sim_device_state", "sim_profile", "sim_regions"):
                payload, is_error = self.call_tool(server, name)
                self.assertTrue(is_error, name)
                self.assertEqual(payload["error"], "not_available", name)
        finally:
            session.close()


class ServiceListTests(McpTestCase):
    def test_service_list_empty_with_slots(self):
        self.session.memory.write_u64(BASE + 0x200 + 0x150 + 6 * 8, 0x81000000)
        payload, is_error = self.call_tool(self.make_server(), "service_list")
        self.assertFalse(is_error)
        self.assertEqual(payload["count"], 0)
        self.assertEqual(payload["mailbox_cmd_base"], 16)
        self.assertEqual(payload["slots"][6]["entry"], 0x81000000)
        self.assertEqual(payload["slots"][6]["cmd"], 22)

    def test_service_list_reports_registry(self):
        service = InjectedService(
            path="/tmp/svc.elf",
            name="svc.elf",
            load_addr=0x81000000,
            entry_addr=0x81000100,
            dispatch_offset=0x100,
            slot=6,
            size=64,
        )
        setattr(self.session, SERVICE_REGISTRY_ATTR, [service])
        payload, is_error = self.call_tool(self.make_server(), "service_list")
        self.assertFalse(is_error)
        self.assertEqual(payload["count"], 1)
        self.assertEqual(payload["services"][0]["entry_addr"], 0x81000100)
        self.assertEqual(payload["services"][0]["slot"], 6)


class SessionLaunchTests(McpTestCase):
    def test_refused_in_plugin_mode(self):
        payload, is_error = self.call_tool(self.make_server(owns_session=False), "session_launch", backend="native")
        self.assertTrue(is_error)
        self.assertEqual(payload["error"], "not_available")

    def test_invalid_backend(self):
        payload, is_error = self.call_tool(self.make_server(owns_session=True), "session_launch", backend="mips")
        self.assertTrue(is_error)
        self.assertEqual(payload["error"], "invalid_arguments")

    def test_missing_kernel(self):
        payload, is_error = self.call_tool(
            self.make_server(owns_session=True), "session_launch", backend="native", kernel="/nonexistent.elf"
        )
        self.assertTrue(is_error)
        self.assertEqual(payload["error"], "invalid_arguments")

    def test_qemu_trace_argument_refused(self):
        payload, is_error = self.call_tool(
            self.make_server(owns_session=True), "session_launch", backend="qemu", trace=True
        )
        self.assertTrue(is_error)
        self.assertEqual(payload["error"], "invalid_arguments")

    @unittest.skipUnless(os.path.isfile(DEMO_ELF), "demo firmware not built (make demo)")
    def test_launch_native_replaces_owned_session(self):
        try:
            import greeneye.sim_backend  # noqa: F401
        except Exception as exc:
            self.skipTest(f"native backend unavailable: {exc}")
        server = self.make_server(owns_session=True)
        # Contract (docs/20 §5/§7): session_launch is session-class and allowed
        # without --allow-write; it only replaces the local served session.
        self.assertFalse(server.allow_write)
        with tempfile.TemporaryDirectory(prefix="mcp-launch-") as directory:
            payload, is_error = self.call_tool(
                server, "session_launch", backend="native", kernel=DEMO_ELF, record=directory
            )
            self.assertFalse(is_error, payload)
            self.assertEqual(payload["backend"], "native")
            self.assertEqual(payload["session"]["kernel"], DEMO_ELF)
            self.assertTrue(self.client.closed)
            cycles, is_error = self.call_tool(server, "sim_cycles")
            self.assertFalse(is_error, cycles)
            self.assertEqual(cycles["backend"], "native")
            self.assertIsNotNone(cycles["cycle_model"])
            regions, is_error = self.call_tool(server, "sim_regions")
            self.assertFalse(is_error, regions)
            self.assertIn("ram", {region["name"] for region in regions["regions"]})
            server.session.close()


@unittest.skipUnless(os.path.isfile(DEMO_ELF), "demo firmware not built (make demo)")
class NativeSessionToolTests(unittest.TestCase):
    """End-to-end tool checks against a real native session (demo firmware)."""

    def setUp(self):
        try:
            import greeneye.sim_backend  # noqa: F401
        except Exception as exc:
            self.skipTest(f"native backend unavailable: {exc}")
        self.session = ge.launch(kernel=DEMO_ELF, backend="native", record=None, name="native-mcp")
        self.machine = self.session.target.machine
        self.entry = self.machine.pc
        self.addCleanup(self.session.close)
        self.inp = io.StringIO()
        self.out = io.StringIO()
        self.log = io.StringIO()

    def make_server(self, **kw):
        return mcp.GreenEyeMCPServer(
            self.session, input_stream=self.inp, output_stream=self.out, log_stream=self.log, **kw
        )

    def call_tool(self, server, name, **arguments):
        response = server.handle_message(
            {"jsonrpc": "2.0", "id": 9, "method": "tools/call", "params": {"name": name, "arguments": arguments}}
        )
        self.assertIn("result", response, response)
        result = response["result"]
        return json.loads(result["content"][0]["text"]), result.get("isError", False)

    def test_native_step_budget(self):
        server = self.make_server()
        self.assertEqual(self.machine.insns, 0)
        payload, is_error = self.call_tool(server, "step", count=3)
        self.assertFalse(is_error, payload)
        self.assertEqual(payload["steps"], 3)
        self.assertEqual(self.machine.insns, 3)
        self.assertNotEqual(self.machine.pc, self.entry)
        self.assertEqual(payload["pc"], self.machine.pc)

    def test_native_step_budget_cap(self):
        payload, is_error = self.call_tool(self.make_server(), "step", count=10001)
        self.assertTrue(is_error)
        self.assertEqual(payload["error"], "invalid_arguments")

    def test_native_run_budget_and_cycles(self):
        server = self.make_server()
        budget = 0.02
        before = self.machine.insns
        payload, is_error = self.call_tool(server, "run", duration=budget)
        self.assertFalse(is_error, payload)
        self.assertGreaterEqual(payload["stops"], 1)
        insns = self.machine.insns
        self.assertGreater(insns, before)
        self.assertLessEqual(insns, int(budget * 1_000_000) + 2000)
        cycles, is_error = self.call_tool(server, "sim_cycles")
        self.assertFalse(is_error, cycles)
        self.assertEqual(cycles["backend"], "native")
        self.assertEqual(cycles["insns"], insns)
        self.assertGreater(cycles["cycles"], 0)
        self.assertEqual(cycles["accounting_enabled"], True)

    def test_native_sim_regions(self):
        payload, is_error = self.call_tool(self.make_server(), "sim_regions", limit=8)
        self.assertFalse(is_error, payload)
        self.assertEqual(payload["backend"], "native")
        regions = {region["name"]: region for region in payload["regions"]}
        self.assertEqual(regions["ram"]["base"], 0x80000000)
        self.assertEqual(regions["ram"]["size"], 256 * 1024 * 1024)
        self.assertEqual(regions["uart"]["base"], 0x10000000)
        image_regions = [region for region in payload["regions"] if region["kind"] == "firmware"]
        self.assertTrue(image_regions)
        self.assertGreaterEqual(image_regions[0]["base"], 0x80000000)
        self.assertTrue(image_regions[0]["flags"]["exec"])
        self.assertIsNone(payload["device_map_entries"])
        self.assertEqual(payload["observed_mmio_pages"], [])

    def test_native_session_info_reports_native(self):
        payload, is_error = self.call_tool(self.make_server(), "session_info")
        self.assertFalse(is_error)
        self.assertEqual(payload["backend"], "native")
        self.assertTrue(payload["capabilities"]["trace"])


class GatingTests(McpTestCase):
    def test_write_mem_refused_without_allow_write(self):
        server = self.make_server()
        payload, is_error = self.call_tool(server, "write_mem", addr=BASE + 0x10, hex="0102")
        self.assertTrue(is_error)
        self.assertEqual(payload["error"], "write_disabled")
        self.assertEqual(bytes(self.client.mem[0x10:0x12]), b"\x00\x00")

    def test_write_mem_enabled(self):
        server = self.make_server(allow_write=True)
        payload, is_error = self.call_tool(server, "write_mem", addr=BASE + 0x10, hex="0x01 02")
        self.assertFalse(is_error)
        self.assertEqual(payload["written"], 2)
        self.assertEqual(bytes(self.client.mem[0x10:0x12]), b"\x01\x02")

    def test_call_function_refused_without_allow_write(self):
        payload, is_error = self.call_tool(self.make_server(), "call_function", target="rpc_ping", args=[1])
        self.assertTrue(is_error)
        self.assertEqual(payload["error"], "write_disabled")

    def test_inject_refused_without_allow_inject(self):
        payload, is_error = self.call_tool(self.make_server(), "inject_service", service="/nonexistent.elf")
        self.assertTrue(is_error)
        self.assertEqual(payload["error"], "inject_disabled")

    def test_inject_missing_file_after_gate(self):
        payload, is_error = self.call_tool(
            self.make_server(allow_inject=True), "inject_service", service="/nonexistent.elf"
        )
        self.assertTrue(is_error)
        self.assertEqual(payload["error"], "invalid_arguments")


class AvailabilityTests(McpTestCase):
    def test_analyze_current_without_recording(self):
        payload, is_error = self.call_tool(self.make_server(), "analyze_current")
        self.assertTrue(is_error)
        self.assertEqual(payload["error"], "not_available")

    def test_trace_start_on_non_native_backend(self):
        payload, is_error = self.call_tool(self.make_server(), "trace_start")
        self.assertTrue(is_error)
        self.assertEqual(payload["error"], "not_available")

    def test_trace_stop_without_active_trace(self):
        payload, is_error = self.call_tool(self.make_server(), "trace_stop")
        self.assertTrue(is_error)
        self.assertEqual(payload["error"], "invalid_arguments")

    def test_mutation_missing_script_is_structured(self):
        server = self.make_server()
        original = mcp.MUTATION_SCRIPT
        mcp.MUTATION_SCRIPT = "/nonexistent/mutation_harness.py"
        try:
            payload, is_error = self.call_tool(server, "mutation_run")
        finally:
            mcp.MUTATION_SCRIPT = original
        self.assertTrue(is_error)
        self.assertEqual(payload["error"], "not_available")

    def test_portability_missing_script_is_structured(self):
        server = self.make_server()
        original = mcp.MATRIX_SCRIPT
        mcp.MATRIX_SCRIPT = "/nonexistent/portability_matrix.py"
        try:
            payload, is_error = self.call_tool(server, "portability_matrix_run", targets=["/x.elf"])
        finally:
            mcp.MATRIX_SCRIPT = original
        self.assertTrue(is_error)
        self.assertEqual(payload["error"], "not_available")

    def test_probe_missing_elf(self):
        payload, is_error = self.call_tool(self.make_server(), "probe_firmware", elf="/nonexistent/fw.elf")
        self.assertTrue(is_error)
        self.assertEqual(payload["error"], "invalid_arguments")


class CloseAndTransportTests(McpTestCase):
    def test_close_stops_serving_but_leaves_plugin_session(self):
        server = self.make_server(owns_session=False)
        payload, is_error = self.call_tool(server, "close")
        self.assertFalse(is_error)
        self.assertFalse(payload["session_closed"])
        self.assertTrue(server._stop)
        self.assertFalse(self.client.closed)

    def test_close_closes_owned_session(self):
        server = self.make_server(owns_session=True)
        payload, is_error = self.call_tool(server, "close")
        self.assertFalse(is_error)
        self.assertTrue(payload["session_closed"])
        self.assertTrue(self.client.closed)

    def test_parse_tcp_is_loopback_only(self):
        self.assertEqual(mcp.parse_tcp("127.0.0.1:8765"), ("127.0.0.1", 8765))
        self.assertEqual(mcp.parse_tcp("localhost:0")[0], "localhost")
        with self.assertRaises(Exception):
            mcp.parse_tcp("0.0.0.0:8765")


class StandaloneEntryTests(unittest.TestCase):
    @unittest.skipUnless(os.path.isfile(MCP_SERVER), "mcp_server.py missing")
    def test_help(self):
        result = subprocess.run(
            [sys.executable, MCP_SERVER, "--help"], capture_output=True, text=True, cwd=TOOL_ROOT
        )
        self.assertEqual(result.returncode, 0, result.stderr)
        self.assertIn("--stdio", result.stdout)
        self.assertIn("--allow-write", result.stdout)

    @unittest.skipUnless(os.path.isfile(MCP_SERVER), "mcp_server.py missing")
    def test_stdio_native_session(self):
        try:
            import greeneye.sim_backend  # noqa: F401
        except Exception as exc:
            self.skipTest(f"native backend unavailable: {exc}")
        with tempfile.TemporaryDirectory(prefix="mcp-native-") as dumps:
            env = dict(os.environ, GREENEYE_DUMPS=dumps)
            requests = [
                {"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {"protocolVersion": "2024-11-05"}},
                {"jsonrpc": "2.0", "method": "notifications/initialized"},
                {"jsonrpc": "2.0", "id": 2, "method": "tools/call", "params": {"name": "session_info", "arguments": {}}},
                {"jsonrpc": "2.0", "id": 3, "method": "tools/call", "params": {"name": "close", "arguments": {}}},
            ]
            payload = "".join(json.dumps(request) + "\n" for request in requests)
            result = subprocess.run(
                [sys.executable, MCP_SERVER, "--stdio", "--backend", "native", "--record"],
                input=payload,
                capture_output=True,
                text=True,
                cwd=TOOL_ROOT,
                env=env,
                timeout=180,
            )
            self.assertEqual(result.returncode, 0, result.stderr[-2000:])
            responses = [json.loads(line) for line in result.stdout.splitlines() if line.strip()]
            self.assertEqual(len(responses), 3)
            by_id = {response.get("id"): response for response in responses}
            info = json.loads(by_id[2]["result"]["content"][0]["text"])
            self.assertEqual(info["backend"], "native")
            self.assertFalse(by_id[3]["result"].get("isError", False))
            self.assertTrue(any(name.endswith("-mcp") for name in os.listdir(dumps)))

    @unittest.skipUnless(os.path.isfile(MCP_SERVER), "mcp_server.py missing")
    @unittest.skipUnless(os.path.isfile(DEMO_ELF), "demo firmware not built (make demo)")
    def test_stdio_native_sim_tools_injection_and_trace(self):
        try:
            import greeneye.sim_backend  # noqa: F401
        except Exception as exc:
            self.skipTest(f"native backend unavailable: {exc}")
        if not os.path.isfile(SERVICE_ELF):
            build = subprocess.run(
                ["make", "-C", os.path.join(TOOL_ROOT, "sdk")], capture_output=True, text=True
            )
            if build.returncode != 0 or not os.path.isfile(SERVICE_ELF):
                self.skipTest(f"SDK build failed: {build.stderr[-200:]}")
        with tempfile.TemporaryDirectory(prefix="mcp-native-sim-") as dumps:
            env = dict(os.environ, GREENEYE_DUMPS=dumps)
            requests = [
                {"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {"protocolVersion": "2024-11-05"}},
                {"jsonrpc": "2.0", "method": "notifications/initialized"},
                {"jsonrpc": "2.0", "id": 2, "method": "tools/call", "params": {"name": "backends_list", "arguments": {}}},
                {"jsonrpc": "2.0", "id": 3, "method": "tools/call", "params": {"name": "sim_cycles", "arguments": {}}},
                {"jsonrpc": "2.0", "id": 4, "method": "tools/call", "params": {"name": "sim_profile", "arguments": {}}},
                {"jsonrpc": "2.0", "id": 5, "method": "tools/call", "params": {"name": "sim_device_state", "arguments": {"limit": 2}}},
                {"jsonrpc": "2.0", "id": 6, "method": "tools/call", "params": {"name": "trace_start", "arguments": {"exec": False, "max_events": 10000}}},
                {"jsonrpc": "2.0", "id": 7, "method": "tools/call", "params": {"name": "trace_stop", "arguments": {}}},
                {
                    "jsonrpc": "2.0",
                    "id": 8,
                    "method": "tools/call",
                    "params": {
                        "name": "inject_service",
                        "arguments": {"service": SERVICE_ELF, "slot": 6, "cmd": 1, "arg0": 65, "path": "both"},
                    },
                },
                {"jsonrpc": "2.0", "id": 9, "method": "tools/call", "params": {"name": "service_list", "arguments": {}}},
                {"jsonrpc": "2.0", "id": 10, "method": "tools/call", "params": {"name": "close", "arguments": {}}},
            ]
            payload = "".join(json.dumps(request) + "\n" for request in requests)
            result = subprocess.run(
                [sys.executable, MCP_SERVER, "--stdio", "--backend", "native", "--record", "--allow-inject"],
                input=payload,
                capture_output=True,
                text=True,
                cwd=TOOL_ROOT,
                env=env,
                timeout=300,
            )
            self.assertEqual(result.returncode, 0, result.stderr[-2000:])
            responses = [json.loads(line) for line in result.stdout.splitlines() if line.strip()]
            by_id = {response.get("id"): response for response in responses}
            self.assertEqual(sorted(by_id), [1, 2, 3, 4, 5, 6, 7, 8, 9, 10])

            def call_result(request_id):
                response = by_id[request_id]
                self.assertIn("result", response, response)
                body = response["result"]
                self.assertFalse(body.get("isError", False), body)
                return json.loads(body["content"][0]["text"])

            backends = call_result(2)
            self.assertEqual(backends["active"], "native")
            cycles = call_result(3)
            self.assertEqual(cycles["backend"], "native")
            self.assertIsNotNone(cycles["cycle_model"])
            profile = call_result(4)
            self.assertEqual(profile["backend"], "native")
            self.assertEqual(profile["kernel"], DEMO_ELF)
            device_state = call_result(5)
            self.assertIn("device_model", device_state)
            trace_start = call_result(6)
            self.assertTrue(trace_start["tracing"])
            trace_stop = call_result(7)
            self.assertFalse(trace_stop["tracing"])
            self.assertTrue(str(trace_stop["blob"]).endswith("trace.jsonl"))
            injected = call_result(8)
            self.assertEqual(injected["remote"], 0x41 ^ 0xBEEF)
            self.assertEqual(injected["firmware"], 0x41 ^ 0xBEEF)
            services = call_result(9)
            self.assertEqual(services["count"], 1)
            self.assertEqual(services["services"][0]["entry_addr"], injected["service"]["entry_addr"])
            slots = services["slots"]
            if slots:
                self.assertEqual(slots[6]["entry"], injected["service"]["entry_addr"])
            self.assertFalse(by_id[10]["result"].get("isError", False))


if __name__ == "__main__":
    unittest.main()