vgsp: host CUDA bridge + sema release recovery; greeneye simulator wave

- kernel: gsp_cuda_bridge (QMD launch capture, debugfs job/out/done, param+output resolution cached from inline writes), sema drain fixes (never hold/starve a channel), GR ring autopin + counter autoreg, per-page push reads, counter-derive gate; module builds clean
- tools/cuda-bridge: CUDA driver-API daemon executing mail/vadd kernels on the physical GPU, CUBB2/CBD2 v2 protocol, layout check + self-tests
- tools/guest-diag + tools/cuda-ab: scanner/pin fixes, victory push-only mode
- debug/risc-v-gsp-rm-vm: Phase S simulator wave (GSP device model, boot ladder, triage/re-demo harnesses, docs 30-41), GPGPU-sim scaffolding
- STATE-CURRENT/docs updates

7 files changed, 1869 insertions, 1 deletions

diff --git a/debug/risc-v-gsp-rm-vm/tests/test_gpgpu_sim_setup.py b/debug/risc-v-gsp-rm-vm/tests/test_gpgpu_sim_setup.py
new file mode 100644
index 0000000..c52f75f
--- /dev/null
+++ b/debug/risc-v-gsp-rm-vm/tests/test_gpgpu_sim_setup.py
@@ -0,0 +1,43 @@
+"""Regression guard for ``gpgpu-sim/setup.sh`` clone-source policy.
+
+The script must default to the upstream URL and use a local mirror only via
+the explicit ``GPGPUSIM_CLONE_SOURCE`` environment variable; no personal host
+path may be hardcoded.  These checks are static (plus ``bash -n``/``--help``)
+so they need neither a network nor a GPGPU-Sim checkout.
+"""
+import os
+import subprocess
+import sys
+import unittest
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+TOOL_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+SETUP = os.path.join(TOOL_ROOT, "gpgpu-sim", "setup.sh")
+PIN = "90ec3399763d7c8512cfe7dc193473086c38ca38"
+UPSTREAM_URL = "https://github.com/gpgpu-sim/gpgpu-sim_distribution.git"
+
+
+class SetupShTests(unittest.TestCase):
+    def test_clone_source_policy(self):
+        with open(SETUP, "r", encoding="utf-8") as fh:
+            text = fh.read()
+        self.assertNotIn("LOCAL_SRC=", text)
+        self.assertNotIn("/home/rmcore", text)
+        self.assertIn("GPGPUSIM_CLONE_SOURCE", text)
+        self.assertIn(UPSTREAM_URL, text)
+        self.assertIn(PIN, text)  # pin enforcement stays in the script
+
+    def test_script_syntax_and_help(self):
+        result = subprocess.run(["bash", "-n", SETUP],
+                                capture_output=True, text=True)
+        self.assertEqual(result.returncode, 0, result.stderr)
+        result = subprocess.run(["bash", SETUP, "--help"],
+                                capture_output=True, text=True)
+        self.assertEqual(result.returncode, 0, result.stderr)
+        self.assertIn("GPGPUSIM_CLONE_SOURCE", result.stdout)
+        self.assertIn("upstream URL", result.stdout)
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/debug/risc-v-gsp-rm-vm/tests/test_injection.py b/debug/risc-v-gsp-rm-vm/tests/test_injection.py
index 5dca9cb..33cc092 100644
--- a/debug/risc-v-gsp-rm-vm/tests/test_injection.py
+++ b/debug/risc-v-gsp-rm-vm/tests/test_injection.py
@@ -1,6 +1,9 @@
 """Integration tests for Frida-like remote calls, run-until and service injection.
 
 Skipped cleanly when QEMU, the demo firmware, clang/lld or the SDK are absent.
+The native simulator is a first-class injection target too: ``NativeInjectionTests``
+proves the demo service blob loads and that direct and firmware-mailbox
+invocations agree without QEMU.
 """
 from __future__ import annotations
 
@@ -24,12 +27,28 @@ def qemu_available() -> bool:
     return bool(os.environ.get("GREENEYE_QEMU") or shutil.which("qemu-system-riscv64"))
 
 
+def native_available() -> bool:
+    try:
+        import greeneye.sim_backend  # noqa: F401
+
+        return True
+    except Exception:
+        return False
+
+
 def toolchain_available() -> bool:
     return bool(shutil.which("clang")) and (
         shutil.which("ld.lld") or os.path.exists("/usr/lib/llvm-18/bin/ld.lld")
     )
 
 
+def build_sdk() -> bool:
+    if os.path.exists(SERVICE_ELF):
+        return True
+    result = subprocess.run(["make", "-C", os.path.join(TOOL_ROOT, "sdk")], capture_output=True, text=True)
+    return result.returncode == 0 and os.path.exists(SERVICE_ELF)
+
+
 @unittest.skipUnless(qemu_available(), "qemu-system-riscv64 not installed")
 @unittest.skipUnless(os.path.exists(DEMO_ELF), "demo firmware not built (make demo)")
 class CallTests(unittest.TestCase):
@@ -117,5 +136,47 @@ class InjectionTests(unittest.TestCase):
             session.close()
 
 
+@unittest.skipUnless(native_available(), "native backend unavailable")
+@unittest.skipUnless(os.path.exists(DEMO_ELF), "demo firmware not built (make demo)")
+@unittest.skipUnless(toolchain_available(), "clang/lld not available")
+class NativeInjectionTests(unittest.TestCase):
+    """Injection SDK against the native simulator (no QEMU required)."""
+
+    @classmethod
+    def setUpClass(cls):
+        if not build_sdk():
+            raise unittest.SkipTest("SDK build failed")
+
+    def test_native_direct_and_firmware_invocation(self):
+        from greeneye.inject import invoke_remote, invoke_via_firmware, load_service, services
+
+        session = ge.launch(kernel=DEMO_ELF, backend="native", record=True, name="unittest-native-inject")
+        try:
+            service = load_service(session, SERVICE_ELF, slot=6)
+            self.assertEqual(services(session), [service])
+            self.assertEqual(invoke_remote(session, service, 1, 0x41), 0x41 ^ 0xBEEF)
+            self.assertEqual(invoke_remote(session, service, 2, 0x10, 0x20), 0x30)
+            session.run(duration=0.3)
+            self.assertEqual(invoke_via_firmware(session, service, 1, 0x41), 0x41 ^ 0xBEEF)
+            self.assertEqual(invoke_via_firmware(session, service, 2, 0x10, 0x20), 0x30)
+        finally:
+            session.close()
+
+    def test_native_injection_replay_is_deterministic(self):
+        from greeneye.inject import invoke_remote, load_service
+
+        results = []
+        for _ in range(2):
+            session = ge.launch(kernel=DEMO_ELF, backend="native", record=False, name="unittest-native-inject-replay")
+            try:
+                service = load_service(session, SERVICE_ELF, slot=6)
+                results.append(invoke_remote(session, service, 2, 0x10, 0x20))
+                results.append(session.pc)
+            finally:
+                session.close()
+        self.assertEqual(results[0], 0x30)
+        self.assertEqual(results, results[:2] + results[:2])
+
+
 if __name__ == "__main__":
     unittest.main()
diff --git a/debug/risc-v-gsp-rm-vm/tests/test_mcp.py b/debug/risc-v-gsp-rm-vm/tests/test_mcp.py
index 2df8a4a..87a85a8 100644
--- a/debug/risc-v-gsp-rm-vm/tests/test_mcp.py
+++ b/debug/risc-v-gsp-rm-vm/tests/test_mcp.py
@@ -19,11 +19,14 @@ 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
@@ -167,8 +170,19 @@ class ProtocolTests(McpTestCase):
     def test_tools_list_catalogue(self):
         tools = self.request(self.make_server(), "tools/list")["result"]["tools"]
         names = [tool["name"] for tool in tools]
-        self.assertEqual(len(names), 24)
+        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"])
@@ -283,6 +297,328 @@ class ReadOnlyToolTests(McpTestCase):
                 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()
@@ -426,6 +762,90 @@ class StandaloneEntryTests(unittest.TestCase):
             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()
diff --git a/debug/risc-v-gsp-rm-vm/tests/test_sim_cycles.py b/debug/risc-v-gsp-rm-vm/tests/test_sim_cycles.py
new file mode 100644
index 0000000..9bb69ad
--- /dev/null
+++ b/debug/risc-v-gsp-rm-vm/tests/test_sim_cycles.py
@@ -0,0 +1,299 @@
+"""Cycle model (sim/cycles.py) and GSP cycle profile (sim/gsp_profile.py) tests.
+
+All assertions here are about a *modeled* cycle count and its determinism, not
+about silicon timings.  Everything is stdlib-only and offline: no hardware,
+no GPU, no QEMU, no network.  The real-image tests are bounded (200k
+instructions), read-only, and skip cleanly when the 570.144 fwimage is absent.
+"""
+from __future__ import annotations
+
+import json
+import os
+import tempfile
+import unittest
+
+from greeneye.device_map import QEMU_REFERENCE
+from sim.cycles import (
+    CLASS_ALU,
+    CLASS_DIV,
+    CLASS_LOAD,
+    CLASS_MUL,
+    CYCLE_FIDELITY,
+    DEFAULT_MODEL,
+    NOTE,
+    CycleModel,
+)
+from sim.decode import (
+    A_ADD,
+    A_DIV,
+    A_MUL,
+    A_SUB,
+    K_AMO,
+    K_BRANCH,
+    K_CSR,
+    K_EBREAK,
+    K_ECALL,
+    K_ILLEGAL,
+    K_JAL,
+    K_JALR,
+    K_LOAD,
+    K_OP,
+    K_OP32,
+    decode32,
+)
+from sim.gsp_profile import DEFAULT_IMAGE, run_profile
+from sim.machine import R_EBREAK, Machine
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+TOOL_ROOT = os.path.normpath(os.path.join(HERE, ".."))
+DEMO_ELF = os.path.join(TOOL_ROOT, "firmware", "uproc_demo", "build", "uproc_demo.elf")
+REAL_IMAGE = os.path.normpath(
+    os.path.join(
+        HERE, "..", "..", "..", "firmware", "57X-GSP", "570.144", "gsp-570.144.fwimage.bin"
+    )
+)
+
+BASE = 0x80000000
+TOUCH_A = QEMU_REFERENCE["access_addr"]  # 0x2000000000000A00 (570.144 first touch)
+
+# addi a0,a0,1 x3; ebreak
+PROGRAM_ALU = bytes.fromhex("13051500" * 3 + "73001000")
+# lui t0,0x80000; lw a0,0(t0); ebreak  (one completed RAM load)
+PROGRAM_RAM_LOAD = bytes.fromhex("b702008003a5020073001000")
+# lui t0,0x20000; slli t0,t0,32; addi t0,t0,0x600; addi t0,t0,0x400;
+# lw a0,0(t0); sw a0,0(t0); ebreak  (one unknown-MMIO read + one write)
+PROGRAM_MMIO = bytes.fromhex("b702002093920202938202609382024003a5020023a0a20073001000")
+
+
+def decoded_cost(program: bytes, model: CycleModel, pc0: int = BASE) -> int:
+    """Instruction cost of a straight-line 4-byte-encoded program (no memory)."""
+    total = 0
+    pc = pc0
+    while pc < pc0 + len(program):
+        word = int.from_bytes(program[pc - pc0 : pc - pc0 + 4], "little")
+        kind, sub, _rd, _r1, _r2, _imm, _nxt = decode32(pc, word)
+        total += model.instruction_cost(kind, sub)
+        pc += 4
+    return total
+
+
+class CycleTableTests(unittest.TestCase):
+    def test_every_decoded_kind_has_a_positive_cost(self):
+        subs = {K_OP: range(18), K_OP32: range(18)}
+        for kind in range(18):
+            for sub in subs.get(kind, (0,)):
+                cost = DEFAULT_MODEL.instruction_cost(kind, sub)
+                self.assertIsInstance(cost, int)
+                self.assertGreaterEqual(cost, 1, f"kind={kind} sub={sub}")
+
+    def test_alu_branch_and_jump_are_unit_cost(self):
+        model = DEFAULT_MODEL
+        self.assertEqual(model.instruction_cost(K_OP, A_ADD), model.alu)
+        self.assertEqual(model.instruction_cost(K_OP, A_SUB), model.alu)
+        self.assertEqual(model.instruction_cost(K_BRANCH), model.branch)
+        self.assertEqual(model.instruction_cost(K_JAL), model.jump)
+        self.assertEqual(model.instruction_cost(K_JALR), model.jump)
+
+    def test_m_extension_is_conservative(self):
+        model = DEFAULT_MODEL
+        mul = model.instruction_cost(K_OP, A_MUL)
+        div = model.instruction_cost(K_OP, A_DIV)
+        self.assertGreater(mul, model.alu)
+        self.assertGreater(div, mul)
+        # Same table is used for the 32-bit M forms.
+        self.assertEqual(model.instruction_cost(K_OP32, A_MUL), mul)
+        self.assertEqual(model.instruction_cost(K_OP32, A_DIV), div)
+
+    def test_memory_latency_is_a_parameter(self):
+        model = DEFAULT_MODEL
+        self.assertGreaterEqual(model.memory_cost(False), 1)
+        self.assertGreaterEqual(model.memory_cost(True), model.memory_cost(False))
+        custom = CycleModel(memory_latency=9, mmio_latency=40)
+        self.assertEqual(custom.memory_cost(False), 9)
+        self.assertEqual(custom.memory_cost(True), 40)
+        self.assertEqual(custom.instruction_cost(K_LOAD), model.instruction_cost(K_LOAD))
+
+    def test_csr_fence_amo_and_trap_are_costs(self):
+        model = DEFAULT_MODEL
+        for kind in (K_CSR, K_AMO, K_EBREAK, K_ECALL, K_ILLEGAL):
+            self.assertGreaterEqual(model.instruction_cost(kind), 1)
+
+    def test_classify_reports_classes(self):
+        self.assertEqual(CycleModel.classify(K_OP, A_ADD), CLASS_ALU)
+        self.assertEqual(CycleModel.classify(K_OP, A_MUL), CLASS_MUL)
+        self.assertEqual(CycleModel.classify(K_OP, A_DIV), CLASS_DIV)
+        self.assertEqual(CycleModel.classify(K_LOAD), CLASS_LOAD)
+
+    def test_model_is_labelled_modeled_not_measured(self):
+        self.assertIn("modeled", NOTE.lower())
+        self.assertIn("not measured", NOTE.lower())
+        self.assertIn("L2", CYCLE_FIDELITY)
+        self.assertIn("modeled", DEFAULT_MODEL.name)
+        described = DEFAULT_MODEL.describe()
+        self.assertEqual(described["name"], DEFAULT_MODEL.name)
+        self.assertEqual(described["note"], NOTE)
+        self.assertIn("memory_latency", described)
+
+    def test_invalid_parameters_rejected(self):
+        with self.assertRaises(ValueError):
+            CycleModel(alu=0)
+        with self.assertRaises(ValueError):
+            CycleModel(memory_latency=-1)
+        with self.assertRaises(ValueError):
+            CycleModel(div=True)
+
+
+class MachineCycleTests(unittest.TestCase):
+    def _machine(self, program: bytes, model=None, **kwargs) -> Machine:
+        machine = Machine(cycle_model=model, **kwargs)
+        machine.load_segments([(BASE, program)], BASE)
+        return machine
+
+    def test_counting_enabled_by_default(self):
+        machine = self._machine(PROGRAM_ALU)
+        machine.run(3)  # three addi, budget stop
+        self.assertEqual(machine.insns, 3)
+        self.assertEqual(machine.cycles, 3)
+
+    def test_ram_load_adds_the_modelled_memory_latency(self):
+        machine = self._machine(PROGRAM_RAM_LOAD)
+        reason, _pc, _a, _b = machine.run(100)
+        self.assertEqual(reason, R_EBREAK)
+        expected = decoded_cost(PROGRAM_RAM_LOAD, DEFAULT_MODEL) + DEFAULT_MODEL.memory_latency
+        self.assertEqual(machine.cycles, expected)
+
+    def test_mmio_accesses_add_the_modelled_mmio_latency(self):
+        machine = self._machine(PROGRAM_MMIO, unknown_mmio="lenient")
+        reason, _pc, _a, _b = machine.run(100)
+        self.assertEqual(reason, R_EBREAK)
+        self.assertEqual(machine.mmio_unknown, 2)
+        self.assertEqual(machine.mmio_log[0]["addr"], TOUCH_A)
+        expected = decoded_cost(PROGRAM_MMIO, DEFAULT_MODEL) + 2 * DEFAULT_MODEL.mmio_latency
+        self.assertEqual(machine.cycles, expected)
+
+    def test_variable_memory_latency_changes_only_the_counter(self):
+        slow = CycleModel(memory_latency=9)
+        quick = self._machine(PROGRAM_RAM_LOAD, model=DEFAULT_MODEL)
+        slower = self._machine(PROGRAM_RAM_LOAD, model=slow)
+        quick.run(100)
+        slower.run(100)
+        self.assertEqual(slower.insns, quick.insns)
+        self.assertEqual(slower.x, quick.x)
+        self.assertEqual(slower.cycles - quick.cycles, 9 - DEFAULT_MODEL.memory_latency)
+
+    def test_disabled_model_leaves_the_counter_at_zero(self):
+        machine = self._machine(PROGRAM_ALU, model=False)
+        machine.run(3)
+        self.assertIsNone(machine.cycle_model)
+        self.assertEqual(machine.insns, 3)
+        self.assertEqual(machine.cycles, 0)
+
+    def test_model_does_not_change_execution(self):
+        without = self._machine(PROGRAM_MMIO, model=False, unknown_mmio="lenient")
+        with_model = self._machine(PROGRAM_MMIO, unknown_mmio="lenient")
+        result_without = without.run(100)
+        result_with_model = with_model.run(100)
+        self.assertEqual(result_without, result_with_model)
+        self.assertEqual(without.insns, with_model.insns)
+        self.assertEqual(without.x, with_model.x)
+        self.assertEqual(without.mmio_log, with_model.mmio_log)
+        self.assertGreater(with_model.cycles, 0)
+
+    def test_double_run_identical(self):
+        def profile():
+            machine = self._machine(PROGRAM_MMIO, unknown_mmio="lenient")
+            result = machine.run(100)
+            return result, machine.insns, machine.cycles, machine.pc, tuple(machine.x)
+
+        self.assertEqual(profile(), profile())
+
+    @unittest.skipUnless(os.path.isfile(DEMO_ELF), "demo firmware absent")
+    def test_demo_execution_identical_with_and_without_model(self):
+        from greeneye.elf import load_elf
+
+        image = load_elf(DEMO_ELF)
+        plan = image.flat_load_plan(bias=0)
+        entry = image.relocated_entry(bias=0)
+
+        def run(model):
+            machine = Machine(cycle_model=model)
+            machine.load_segments(plan, entry)
+            result = machine.run(20_000)
+            return result, machine.insns, tuple(machine.x), machine.uart.text()
+
+        self.assertEqual(run(False), run(None))
+
+
+class GspProfileTests(unittest.TestCase):
+    def test_default_image_path_matches_workspace_firmware(self):
+        self.assertEqual(
+            os.path.normpath(DEFAULT_IMAGE),
+            os.path.normpath(REAL_IMAGE),
+        )
+
+    def test_absent_image_skips_cleanly(self):
+        with tempfile.TemporaryDirectory() as root:
+            missing = os.path.join(root, "no-such-image.bin")
+            result = run_profile(image=missing, instructions=1000)
+        self.assertFalse(result.ran)
+        self.assertIn("absent", result.reason)
+        self.assertIsNone(result.session_dir)
+
+    def test_bad_budget_rejected(self):
+        from greeneye.errors import GreenEyeError
+
+        with self.assertRaises(GreenEyeError):
+            run_profile(image=REAL_IMAGE, instructions=0)
+
+
+@unittest.skipUnless(os.path.isfile(REAL_IMAGE), "TU104 570.144 fwimage absent")
+class RealImageCycleProfileTests(unittest.TestCase):
+    def test_bounded_profile_first_touch_and_metadata(self):
+        with tempfile.TemporaryDirectory() as root:
+            result = run_profile(
+                image=REAL_IMAGE, instructions=200_000, record=root, name="unittest-cycle0"
+            )
+            self.assertTrue(result.ran, result.reason)
+            self.assertEqual(result.instructions, 200_000)
+            self.assertEqual(result.stop_reason, "budget")
+            self.assertGreater(result.cycles, result.instructions)
+            self.assertEqual(result.cycle_model, DEFAULT_MODEL.name)
+            self.assertEqual(result.first_unknown_mmio["addr"], TOUCH_A)
+            self.assertEqual(result.first_unknown_mmio["rw"], "r")
+            self.assertEqual(result.first_unknown_mmio["pc"], 0x84009014)
+            self.assertEqual(result.stop_pc, 0x84009070)
+            self.assertEqual(result.stop_pc_fw, 0x4009070)
+
+            with open(os.path.join(result.session_dir, "session.json"), "r", encoding="utf-8") as fh:
+                meta = json.load(fh)
+            self.assertEqual(meta["cycles"], result.cycles)
+            self.assertEqual(meta["instructions_executed"], 200_000)
+            self.assertEqual(meta["cycle_model"], DEFAULT_MODEL.name)
+            self.assertEqual(meta["firmware_sha256"], result.image_sha256)
+            self.assertIn("modeled", meta["cycle_fidelity"])
+            self.assertEqual(meta["first_unknown_mmio"]["addr"], TOUCH_A)
+            self.assertEqual(meta["cycle_model_table"]["name"], DEFAULT_MODEL.name)
+
+    def test_two_runs_identical_cycles(self):
+        with tempfile.TemporaryDirectory() as root:
+            first = run_profile(
+                image=REAL_IMAGE,
+                instructions=200_000,
+                record=os.path.join(root, "a"),
+                name="unittest-cycle-a",
+            )
+            second = run_profile(
+                image=REAL_IMAGE,
+                instructions=200_000,
+                record=os.path.join(root, "b"),
+                name="unittest-cycle-b",
+            )
+        self.assertTrue(first.ran and second.ran)
+        self.assertEqual(first.instructions, second.instructions)
+        self.assertEqual(first.cycles, second.cycles)
+        self.assertEqual(first.stop_pc, second.stop_pc)
+        self.assertEqual(first.image_sha256, second.image_sha256)
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/debug/risc-v-gsp-rm-vm/tests/test_sim_devices.py b/debug/risc-v-gsp-rm-vm/tests/test_sim_devices.py
new file mode 100644
index 0000000..8063ee9
--- /dev/null
+++ b/debug/risc-v-gsp-rm-vm/tests/test_sim_devices.py
@@ -0,0 +1,458 @@
+"""Device-model tests: registry, per-block semantics, profile and boot ladder.
+
+Every modelled register assertion here is tied to the evidence documented in
+``sim/devices/*.py`` and ``docs/36-GSP-DEVICE-MODEL.md``.  The real-firmware
+tests are bounded (<= 20k instructions) and assert *observed* rungs only.
+"""
+import os
+import sys
+import unittest
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+from sim.devices import (  # noqa: E402
+    CsbWindow,
+    DmaTranslator,
+    FalconBlock,
+    IrqRouter,
+    PmcBlock,
+    PmuStub,
+    PriscvBlock,
+    PtimerBlock,
+    Wpr2Block,
+    load_profile,
+    registry_from_profile,
+)
+from sim.devices.base import DeviceRegistry  # noqa: E402
+from sim.devices import falcon as falcon_mod  # noqa: E402
+from sim.machine import Machine  # noqa: E402
+
+TOOL_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+DEFAULT_IMAGE = os.path.normpath(
+    os.path.join(TOOL_ROOT, "..", "..", "firmware", "57X-GSP", "570.144",
+                 "gsp-570.144.fwimage.bin"))
+DEFAULT_PROFILE = os.path.join(TOOL_ROOT, "sim", "profiles", "tu104-570.144.json")
+
+
+class RegistryTests(unittest.TestCase):
+    def test_containment_and_unhandled_log(self):
+        reg = DeviceRegistry()
+        dev = PmcBlock(base=0x1000, size=0x1000)
+        reg.add(dev)
+        ok, value = reg.read(0x1A00, 4)
+        self.assertTrue(ok)
+        self.assertEqual(value, dev.boot_42)
+        # unmodelled offset inside the window: unhandled but window-matched
+        ok, _ = reg.read(0x1A04, 4)
+        self.assertFalse(ok)
+        self.assertEqual(reg.unhandled, 1)
+        # outside every window
+        ok, _ = reg.read(0x2000, 4)
+        self.assertFalse(ok)
+        self.assertEqual(reg.unhandled, 2)
+        # Only window-matched accesses are recorded; outside-window accesses
+        # are counted but left to the machine's unknown log.
+        self.assertEqual(len(reg.accesses), 2)
+        self.assertFalse(reg.accesses[1]["handled"])
+
+    def test_overlapping_windows_fall_back_on_refusal(self):
+        # A later window shadows an earlier one, but an offset the overlay does
+        # not model (here the PMU stub refuses everything) falls back to the
+        # wider device instead of being logged as unknown.
+        reg = DeviceRegistry()
+        fal = FalconBlock(base=0x1000, size=0x1000)
+        reg.add(fal)
+        reg.add(PmuStub(base=0x1000, size=0x1000))
+        self.assertTrue(reg.write(0x1040, 4, 0xCA))  # MAILBOX0 via fallback
+        self.assertEqual(fal.regs[falcon_mod.MAILBOX0], 0xCA)
+        ok, value = reg.read(0x1040, 4)
+        self.assertTrue(ok)
+        self.assertEqual(value, 0xCA)
+        self.assertEqual(reg.unhandled, 0)
+        self.assertTrue(all(a["handled"] for a in reg.accesses))
+
+    def test_overlapping_windows_all_refuse_records_once(self):
+        reg = DeviceRegistry()
+        reg.add(PmuStub(base=0x1000, size=0x1000))
+        reg.add(PmuStub(base=0x1000, size=0x1000))
+        ok, _ = reg.read(0x1000, 4)
+        self.assertFalse(ok)
+        self.assertEqual(reg.unhandled, 1)
+        self.assertEqual(len(reg.accesses), 1)
+        self.assertFalse(reg.accesses[0]["handled"])
+        self.assertFalse(reg.write(0x1000, 4, 1))
+        self.assertEqual(reg.unhandled, 2)
+        self.assertEqual(len(reg.accesses), 2)
+
+    def test_straddling_access_is_not_served(self):
+        reg = DeviceRegistry()
+        reg.add(PmcBlock(base=0x1000, size=0x10))
+        ok, _ = reg.read(0x1008, 8)  # ends at 0x1010, outside the window
+        self.assertFalse(ok)
+
+    def test_state_round_trip(self):
+        reg = DeviceRegistry()
+        fal = FalconBlock(base=0x1000, size=0x1000)
+        reg.add(fal)
+        reg.write(0x1040, 4, 0xCA)
+        snap = reg.state()
+        reg2 = DeviceRegistry()
+        fal2 = FalconBlock(base=0x1000, size=0x1000)
+        reg2.add(fal2)
+        reg2.load_state(snap)
+        self.assertEqual(fal2.regs[falcon_mod.MAILBOX0], 0xCA)
+        self.assertEqual(fal2.state(), fal.state())
+
+
+class PmcTests(unittest.TestCase):
+    def test_tu104_value_fields(self):
+        dev = PmcBlock(chip_id=0x164)
+        value = dev.read(0xA00, 4)
+        self.assertEqual((value >> 20) & 0x1FF, 0x164)
+        self.assertEqual((value >> 24) & 0x1F, 0x16)  # ARCHITECTURE TU100
+        self.assertEqual((value >> 20) & 0xF, 0x4)    # IMPLEMENTATION
+        self.assertEqual(dev.read(0xA04, 4), None)
+        self.assertFalse(dev.write(0xA00, 4, 0))
+
+    def test_profile_profile_value_override(self):
+        dev = PmcBlock(boot_42=0x00123456)
+        self.assertEqual(dev.read(0xA00, 4), 0x00123456)
+
+    def test_chip_id_constants_are_distinct(self):
+        from sim.devices import pmc as pmc_mod
+        ids = [pmc_mod.TU102, pmc_mod.TU104, pmc_mod.TU105, pmc_mod.TU106,
+               pmc_mod.TU116, pmc_mod.TU117]
+        self.assertEqual(len(ids), len(set(ids)))
+        self.assertEqual(pmc_mod.TU117, 0x167)
+        self.assertFalse(hasattr(pmc_mod, "TU107"))
+
+
+class FalconTests(unittest.TestCase):
+    def test_irq_set_clear_status(self):
+        dev = FalconBlock()
+        dev.write(falcon_mod.IRQSSET, 4, falcon_mod.IRQ_HALT)
+        self.assertEqual(dev.read(falcon_mod.IRQSTAT, 4), falcon_mod.IRQ_HALT)
+        dev.write(falcon_mod.IRQSCLR, 4, falcon_mod.IRQ_HALT)
+        self.assertEqual(dev.read(falcon_mod.IRQSTAT, 4), 0)
+        # write-only set/clear registers have no read-back
+        self.assertIsNone(dev.read(falcon_mod.IRQSSET, 4))
+
+    def test_mailbox_rw(self):
+        dev = FalconBlock()
+        dev.write(falcon_mod.MAILBOX0, 4, 0xCA)
+        dev.write(falcon_mod.MAILBOX1, 4, 0x11DEAD)
+        self.assertEqual(dev.read(falcon_mod.MAILBOX0, 4), 0xCA)
+        self.assertEqual(dev.read(falcon_mod.MAILBOX1, 4), 0x11DEAD)
+
+    def test_debuginfo_storage_without_shim(self):
+        dev = FalconBlock(debuginfo_host_consume=False)
+        dev.write(falcon_mod.DEBUGINFO, 4, 0x11DEAD)
+        self.assertEqual(dev.read(falcon_mod.DEBUGINFO, 4), 0x11DEAD)
+
+    def test_debuginfo_host_consume_shim(self):
+        dev = FalconBlock(debuginfo_host_consume=True)
+        dev.write(falcon_mod.DEBUGINFO, 4, 0x11DEAD)
+        value = dev.read(falcon_mod.DEBUGINFO, 4)
+        self.assertEqual(value, 0x11DEAD)          # value preserved for the reader
+        self.assertEqual(dev.read(falcon_mod.DEBUGINFO, 4), 0x1DEAD)  # stage cleared
+        self.assertEqual(dev.debuginfo_consumed, 1)
+        # debugger peek never consumes
+        dev.write(falcon_mod.DEBUGINFO, 4, 0x21DEAD)
+        self.assertEqual(dev.peek(falcon_mod.DEBUGINFO, 4), 0x21DEAD)
+        self.assertEqual(dev.debuginfo_consumed, 1)
+
+    def test_dma_transfer_through_translator(self):
+        ram = bytearray(0x10000)
+        ram[0x100:0x108] = b"DEADBEEF"
+        dma = DmaTranslator(ram=ram, ram_base=0, ram_size=len(ram),
+                            windows=[{"pa": 0x0, "sim": 0, "size": 0x10000}])
+        dev = FalconBlock(dma=dma)
+        # Program: destination = 0x200, source = IMEM + 0x100.  The register
+        # encodes ((base1 << 32) | base) << 8; the profile's DMA windows decide
+        # which aperture that address resolves to (aperture selection itself is
+        # not proven yet and is documented as an unknown).
+        dev.write(falcon_mod.DMATRFBASE1, 4, 0)
+        dev.write(falcon_mod.DMATRFBASE, 4, 0)
+        dev.write(falcon_mod.DMATRFFBOFFS, 4, 0x200)
+        dev.write(falcon_mod.DMATRFMOFFS, 4, 0x100)
+        cmd = (falcon_mod.CMD_IMEM | falcon_mod.CMD_WRITE
+               | (2 << falcon_mod.CMD_SIZE_SHIFT))  # IMEM -> FB, 16 bytes
+        dev.write(falcon_mod.DMATRFCMD, 4, cmd)
+        self.assertEqual(ram[0x200:0x210], b"DEADBEEF" + bytes(8))
+        self.assertTrue(dev.regs[falcon_mod.DMATRFCMD] & falcon_mod.CMD_IDLE)
+        self.assertEqual(dev.dma_transfers, 1)
+
+    def test_dma_refused_when_unmapped(self):
+        ram = bytearray(0x1000)
+        dma = DmaTranslator(ram=ram, ram_base=0, ram_size=len(ram), windows=[])
+        dev = FalconBlock(dma=dma)
+        dev.write(falcon_mod.DMATRFBASE, 4, 0)
+        dev.write(falcon_mod.DMATRFCMD, 4, falcon_mod.CMD_IMEM)
+        self.assertFalse(dev.regs[falcon_mod.DMATRFCMD] & falcon_mod.CMD_IDLE)
+        self.assertTrue(dev.regs[falcon_mod.DMATRFCMD] & falcon_mod.CMD_ERROR)
+        self.assertEqual(dev.dma_refused, 1)
+
+
+class PriscvTests(unittest.TestCase):
+    def test_irq_mask_semantics(self):
+        dev = PriscvBlock()
+        dev.write(0x2AC, 4, 0x5)
+        dev.write(0x2B0, 4, 0x1)
+        self.assertEqual(dev.read(0x2B4, 4), 0x4)
+        self.assertFalse(dev.write(0x2B4, 4, 0xFFFF))  # read-only
+
+    def test_trace_and_error_registers(self):
+        dev = PriscvBlock()
+        dev.write(0x344, 4, 0x1234)
+        self.assertEqual(dev.read(0x344, 4), 0x1234)
+        # observed boot panic reads (R-I4R): read-back is zero, writes refused
+        self.assertEqual(dev.read(0x364, 4), 0)
+        self.assertEqual(dev.read(0x368, 4), 0)
+        self.assertFalse(dev.write(0x368, 4, 0x1))
+
+
+class TimerTests(unittest.TestCase):
+    def test_modeled_time_reads(self):
+        clock = {"insns": 0}
+        dev = PtimerBlock(ns_per_insn=2.0, clock=lambda: clock["insns"])
+        self.assertEqual(dev.read(0x400, 4), 0)
+        self.assertEqual(dev.read(0x410, 4), 0)
+        clock["insns"] = 16
+        # 32 ns -> TIME_0 NSEC field (bits 31:5) = 1
+        self.assertEqual(dev.read(0x400, 4), 1)
+        self.assertEqual(dev.read(0x410, 4), 32)
+
+    def test_profile_window_reaches_time_registers(self):
+        # Regression: the profile window is PRIV+0x9000, so TIME_0/TIME_1 must
+        # be block-relative (0x400/0x410); aperture-absolute 0x9400/0x9410 fall
+        # outside the declared window and serve nothing.
+        profile = {
+            "schema": "greeneye-sim-profile/1",
+            "devices": {
+                "timer": {
+                    "ns_per_insn": 1.0,
+                    "windows": [{"base": "0x9000", "size": "0x1000"}],
+                }
+            },
+        }
+        machine = Machine(memory_mb=1, ram_base=0)
+        reg = registry_from_profile(profile, machine=machine)
+        machine.insns = 32
+        ok, value = reg.read(0x9000 + 0x400, 4, pc=0)
+        self.assertTrue(ok)
+        self.assertEqual(value, 1)  # (32 ns >> 5)
+        ok, value = reg.read(0x9000 + 0x410, 4, pc=0)
+        self.assertTrue(ok)
+        self.assertEqual(value, 32)
+
+    def test_timer_registration_is_opt_in(self):
+        machine = Machine(memory_mb=1, ram_base=0)
+        no_section = {
+            "schema": "greeneye-sim-profile/1",
+            "devices": {"pmc": {"windows": [{"base": "0x1000", "size": "0x1000"}]}},
+        }
+        reg = registry_from_profile(no_section, machine=machine)
+        self.assertIsNone(reg.device_by_name("timer"))
+        disabled = {
+            "schema": "greeneye-sim-profile/1",
+            "devices": {"timer": {"enabled": False,
+                                  "windows": [{"base": "0x9000", "size": "0x1000"}]}},
+        }
+        reg = registry_from_profile(disabled, machine=machine)
+        self.assertIsNone(reg.device_by_name("timer"))
+        enabled = {
+            "schema": "greeneye-sim-profile/1",
+            "devices": {"timer": {"ns_per_insn": 1.0,
+                                  "windows": [{"base": "0x9000", "size": "0x1000"}]}},
+        }
+        reg = registry_from_profile(enabled, machine=machine)
+        self.assertIsNotNone(reg.device_by_name("timer"))
+
+
+class CsbTests(unittest.TestCase):
+    def test_undeclared_window_serves_nothing(self):
+        dev = CsbWindow(base=0x1000, size=0x1000)
+        self.assertIsNone(dev.read(0x0, 4))
+        self.assertFalse(dev.write(0x0, 4, 1))
+
+    def test_declared_semantics(self):
+        dev = CsbWindow(base=0x1000, size=0x1000, registers={
+            0x00: {"width": 4, "value": 0xF, "semantics": "w1c"},
+            0x04: {"width": 4, "value": 0, "semantics": "rw"},
+            0x08: {"width": 4, "value": 0, "semantics": "ro"},
+        })
+        dev.write(0x00, 4, 0x3)
+        self.assertEqual(dev.read(0x00, 4), 0xC)
+        dev.write(0x04, 4, 0x77)
+        self.assertEqual(dev.read(0x04, 4), 0x77)
+        self.assertFalse(dev.write(0x08, 4, 1))
+
+
+class Wpr2Tests(unittest.TestCase):
+    def test_range_encoding(self):
+        dev = Wpr2Block()
+        dev.set_range(0x2000000, 0x6000000)
+        lo = dev.read(0x1FA824, 4)
+        hi = dev.read(0x1FA828, 4)
+        self.assertEqual((lo << 12), 0x2000000)
+        self.assertEqual((hi << 12) + 0x20000, 0x6000000)
+        self.assertIsNone(dev.read(0x1FA800, 4))
+
+    def test_state_derives_range_after_writes(self):
+        # Regression: writes update regs only; state() must derive start/end
+        # from the registers rather than reporting a stale set_range range.
+        dev = Wpr2Block()
+        dev.write(0x1FA824, 4, 0x2000000 >> 12)
+        dev.write(0x1FA828, 4, (0x6000000 - 0x20000) >> 12)
+        state = dev.state()
+        self.assertEqual(state["start"], 0x2000000)
+        self.assertEqual(state["end"], 0x6000000)
+        restored = Wpr2Block()
+        restored.load_state(state)
+        self.assertEqual(restored.start, 0x2000000)
+        self.assertEqual(restored.end, 0x6000000)
+        # A stale serialised start/end never wins over the registers.
+        stale = Wpr2Block()
+        stale.load_state({"regs": {str(0x1FA824): 0x2000},
+                          "start": 0xDEAD, "end": 0xBEEF})
+        self.assertEqual(stale.start, 0x2000000)
+        self.assertIsNone(stale.end)
+
+
+class PmuTests(unittest.TestCase):
+    def test_stub_never_fakes(self):
+        dev = PmuStub(base=0x1000, size=0x1000)
+        self.assertIsNone(dev.read(0x0, 4))
+        self.assertFalse(dev.write(0x0, 4, 1))
+        self.assertEqual(dev.seen["r:0x0:4"], 1)
+
+
+class IrqRouterTests(unittest.TestCase):
+    def test_aggregation(self):
+        fal = FalconBlock()
+        pri = PriscvBlock()
+        router = IrqRouter(falcon=fal, priscv=pri)
+        fal.write(falcon_mod.IRQSSET, 4, falcon_mod.IRQ_HALT)
+        pri.write(0x2AC, 4, 0x2)
+        state = router.pending()
+        self.assertTrue(state["falcon_halt"])
+        self.assertEqual(state["riscv_irq_mask"], 0x2)
+
+
+class DmaTranslatorTests(unittest.TestCase):
+    def test_translate_and_refuse(self):
+        ram = bytearray(0x100)
+        dma = DmaTranslator(ram=ram, ram_base=0, ram_size=len(ram),
+                            windows=[{"pa": 0x1000, "sim": 0, "size": 0x100}])
+        self.assertEqual(dma.translate(0x1000), 0)
+        self.assertEqual(dma.translate(0x10FF), 0xFF)
+        self.assertIsNone(dma.translate(0x2000))
+        self.assertTrue(dma.write(0x1000, b"AB"))
+        self.assertEqual(dma.read(0x1000, 2), b"AB")
+        self.assertIsNone(dma.read(0x2000, 2))
+        self.assertEqual(dma.refused, 1)
+
+    def test_copy_counts_only_completed_transfers_and_exports_events(self):
+        ram = bytearray(0x100)
+        ram[0:4] = b"ABCD"
+        dma = DmaTranslator(ram=ram, ram_base=0, ram_size=len(ram),
+                            windows=[{"pa": 0x1000, "sim": 0, "size": 0x100}])
+        # Destination outside every window: refused, not a transfer.
+        self.assertFalse(dma.copy(0x1000, 0x2000, 4))
+        self.assertEqual(dma.transfers, 0)
+        self.assertEqual(dma.refused, 1)
+        # Both mapped: one completed transfer.
+        self.assertTrue(dma.copy(0x1000, 0x1004, 4))
+        self.assertEqual(dma.transfers, 1)
+        self.assertEqual(ram[0x4:0x8], b"ABCD")
+        state = dma.state()
+        self.assertIn("events", state)
+        self.assertEqual(state["events"][0]["kind"], "w")
+        self.assertEqual(len(state["events"]), 1)
+
+
+class MachineDeviceIntegrationTests(unittest.TestCase):
+    def _machine(self):
+        reg = DeviceRegistry()
+        reg.add(PmcBlock(base=0x2000000000000000, size=0x1000))
+        m = Machine(memory_mb=1, ram_base=0, unknown_mmio="lenient", devices=reg)
+        return m, reg
+
+    def test_modeled_device_takes_precedence_over_lenient(self):
+        m, reg = self._machine()
+        self.assertEqual(m._mmio_load(0x2000000000000A00, 4), PmcBlock().boot_42)
+        self.assertEqual(m.mmio_unknown, 0)
+        self.assertEqual(len(reg.accesses), 1)
+
+    def test_unknown_inside_window_is_logged(self):
+        m, reg = self._machine()
+        self.assertEqual(m._mmio_load(0x2000000000000A04, 4), 0)
+        self.assertEqual(m.mmio_unknown, 1)
+        self.assertFalse(reg.accesses[0]["handled"])
+
+    def test_debugger_peek_and_device_state(self):
+        m, reg = self._machine()
+        self.assertEqual(m.read_mem(0x2000000000000A00, 4),
+                         PmcBlock().boot_42.to_bytes(4, "little"))
+        self.assertIsInstance(m.device_state(), dict)
+        self.assertEqual(m.device_state()["schema"], "greeneye-sim-devices/1")
+
+    def test_reset_regs_names(self):
+        m = Machine(memory_mb=1, ram_base=0, reset_regs={"a0": 0x1234, "sp": 0x8000})
+        self.assertEqual(m.x[10], 0x1234)
+        self.assertEqual(m.x[2], 0x8000)
+
+
+@unittest.skipUnless(os.path.isfile(DEFAULT_IMAGE), "real GSP firmware absent")
+class BootLadderTests(unittest.TestCase):
+    """Bounded real-firmware runs: observed rungs only."""
+
+    def test_profile_loads_and_builds_windows(self):
+        profile = load_profile(DEFAULT_PROFILE)
+        self.assertEqual(profile["schema"], "greeneye-sim-profile/1")
+        m = Machine(memory_mb=16, ram_base=0, unknown_mmio="lenient")
+        reg = registry_from_profile(profile, machine=m)
+        names = {d.name for d in reg.devices}
+        self.assertIn("pmc", names)
+        self.assertIn("falcon", names)
+        self.assertIn("priscv", names)
+        self.assertIsNotNone(reg.dma)
+
+    def test_no_shim_b1_device_model_serves_boot(self):
+        from sim.gsp_boot import run_boot
+        r = run_boot(image=DEFAULT_IMAGE, profile_path=DEFAULT_PROFILE,
+                     instructions=20_000, record=False, use_shim=False)
+        self.assertTrue(r.ran)
+        self.assertEqual(r.rung, "B1")
+        self.assertIn("bootargs_consumer", r.milestones)
+        self.assertEqual(r.mmio_unknown, 0)
+        self.assertEqual(r.device_unhandled, 0)
+
+    def test_shim_reaches_relocation_b2(self):
+        from sim.gsp_boot import run_boot
+        r = run_boot(image=DEFAULT_IMAGE, profile_path=DEFAULT_PROFILE,
+                     instructions=20_000, record=False, use_shim=True)
+        self.assertTrue(r.ran)
+        self.assertEqual(r.rung, "B2")
+        self.assertIn("relocation_copy", r.milestones)
+        self.assertEqual(r.mmio_unknown, 0)
+        # Known unknown: the DMEM runspace content the app relocates from is
+        # not reconstructed; the run stops there rather than faking it.
+        self.assertEqual(r.stop_reason, "fault")
+        self.assertEqual(r.stop_pc, 0x400EA0C)
+
+    def test_two_runs_identical(self):
+        from sim.gsp_boot import run_boot
+        kwargs = dict(image=DEFAULT_IMAGE, profile_path=DEFAULT_PROFILE,
+                      instructions=20_000, record=False, use_shim=True)
+        a = run_boot(**kwargs)
+        b = run_boot(**kwargs)
+        self.assertEqual(a.instructions, b.instructions)
+        self.assertEqual(a.stop_pc, b.stop_pc)
+        self.assertEqual(a.stop_reason, b.stop_reason)
+        self.assertEqual(a.device_state_sha256, b.device_state_sha256)
+        self.assertEqual(a.registers, b.registers)
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/debug/risc-v-gsp-rm-vm/tests/test_sim_qemu_line.py b/debug/risc-v-gsp-rm-vm/tests/test_sim_qemu_line.py
new file mode 100644
index 0000000..5a798f2
--- /dev/null
+++ b/debug/risc-v-gsp-rm-vm/tests/test_sim_qemu_line.py
@@ -0,0 +1,323 @@
+"""Tests for the S4 QEMU-line evaluation harness (harness/sim_qemu_line.py).
+
+The pure-function tests need no firmware.  The integration tests use the real
+TU104 570.144 fwimage read-only and **skip cleanly** when it (or
+qemu-system-riscv64, or the bundled demo firmware) is absent.  All runs are
+bounded (<= 20k native instructions in tests, <= 1 s QEMU) and touch no
+hardware, GPU, module, guest or network.
+
+Run::
+
+    python3 -m unittest tests.test_sim_qemu_line -v
+"""
+from __future__ import annotations
+
+import json
+import os
+import shutil
+import sys
+import tempfile
+import unittest
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+TOOL_ROOT = os.path.abspath(os.path.join(HERE, ".."))
+REAL_IMAGE = os.path.normpath(
+    os.path.join(TOOL_ROOT, "..", "..", "firmware", "57X-GSP", "570.144", "gsp-570.144.fwimage.bin")
+)
+DEMO_ELF = os.path.join(TOOL_ROOT, "firmware", "uproc_demo", "build", "uproc_demo.elf")
+if TOOL_ROOT not in sys.path:
+    sys.path.insert(0, TOOL_ROOT)
+
+from greeneye.device_map import QEMU_REFERENCE, firmware_sha256  # noqa: E402
+from harness.common import qemu_available  # noqa: E402
+from harness.sim_qemu_line import FROZEN_THRESHOLDS  # noqa: E402
+
+
+class PureUnitTests(unittest.TestCase):
+    """No firmware required: rung maths, thresholds and report rendering."""
+
+    def test_rung_ordinal_and_above(self):
+        from harness.sim_qemu_line import above_qemu_line, rung_ordinal
+
+        self.assertEqual(rung_ordinal("pre-B0"), 0)
+        self.assertEqual(rung_ordinal("B0"), 1)
+        self.assertIsNone(rung_ordinal("B9"))
+        self.assertTrue(above_qemu_line("pre-B0", "B0"))
+        self.assertFalse(above_qemu_line("pre-B0", "pre-B0"))
+        self.assertFalse(above_qemu_line("B1", "B0"))
+        self.assertFalse(above_qemu_line(None, "B0"))
+
+    def test_threshold_checks_require_device_model(self):
+        from harness.sim_qemu_line import threshold_checks
+
+        qemu = {"ran": True, "first_touch_match": True, "rung": "pre-B0"}
+        native = {
+            "ran": True,
+            "rung": "B0",
+            "repeat": 2,
+            "identical": True,
+            "pair_wall_s": 1.0,
+            "device_model_used": False,
+        }
+        checks = threshold_checks({"qemu": qemu, "native": native})
+        self.assertTrue(checks["qemu_first_fault_reproduced"])
+        self.assertTrue(checks["native_above_qemu"])
+        self.assertTrue(checks["identical_runs"])
+        self.assertFalse(checks["device_model_documented"])
+        self.assertFalse(checks["all"])
+        native["device_model_used"] = True
+        self.assertTrue(threshold_checks({"qemu": qemu, "native": native})["all"])
+
+    def test_build_report_verdicts(self):
+        from harness.sim_qemu_line import build_report
+
+        qemu = {"ran": True, "first_touch_match": True, "rung": "pre-B0"}
+        native_ok = {
+            "ran": True,
+            "rung": "B1",
+            "repeat": 2,
+            "identical": True,
+            "pair_wall_s": 1.0,
+            "device_model_used": True,
+            "runs": [],
+        }
+        report = build_report(
+            image=os.path.join(TOOL_ROOT, "does-not-exist.bin"),
+            qemu=qemu,
+            native=native_ok,
+            features=None,
+            s1={"source": None, "declared_rung": None, "notes": []},
+            thresholds=dict(FROZEN_THRESHOLDS),
+        )
+        self.assertEqual(report["verdict"], "above")
+        self.assertTrue(report["above_qemu_line"])
+        self.assertTrue(report["threshold_checks"]["all"])
+
+        lenient = dict(native_ok, device_model_used=False)
+        report = build_report(
+            image="missing.bin",
+            qemu=qemu,
+            native=lenient,
+            features=None,
+            s1={"source": None, "declared_rung": None, "notes": []},
+            thresholds=dict(FROZEN_THRESHOLDS),
+        )
+        self.assertEqual(report["verdict"], "above_lenient_only")
+        self.assertFalse(report["above_qemu_line"])
+
+        report = build_report(
+            image="missing.bin",
+            qemu={"ran": False, "rung": None},
+            native=native_ok,
+            features=None,
+            s1={"source": None, "declared_rung": None, "notes": []},
+            thresholds=dict(FROZEN_THRESHOLDS),
+        )
+        self.assertEqual(report["verdict"], "unverified_qemu_absent")
+
+        report = build_report(
+            image="missing.bin",
+            qemu={"ran": True, "first_touch_match": False, "rung": "pre-B0"},
+            native=native_ok,
+            features=None,
+            s1={"source": None, "declared_rung": None, "notes": []},
+            thresholds=dict(FROZEN_THRESHOLDS),
+        )
+        self.assertEqual(report["verdict"], "unverified_qemu_mismatch")
+
+    def test_render_markdown_shape(self):
+        from harness.sim_qemu_line import build_report, render_markdown
+
+        report = build_report(
+            image="missing.bin",
+            qemu={
+                "ran": True,
+                "first_touch_match": True,
+                "rung": "pre-B0",
+                "wall_s": 1.0,
+                "first_fault": {"desc": "fault_load", "epc_fw": 0x4009014,
+                                "access_addr": QEMU_REFERENCE["access_addr"],
+                                "size": 4, "rw": "read"},
+                "session_hashes": {"session.json": "ab" * 32},
+                "session": "harness/qemu_line/sessions/qemu-probe",
+            },
+            native={
+                "ran": True,
+                "rung": "B0",
+                "rung_basis": "test",
+                "repeat": 2,
+                "identical": True,
+                "pair_wall_s": 1.0,
+                "device_model_used": False,
+                "runs": [
+                    {
+                        "rung": "B0",
+                        "instructions": 10,
+                        "cycles": 12,
+                        "wall_s": 0.1,
+                        "stop": {"reason": "budget", "pc_fw": 0x4009070},
+                        "session": "harness/qemu_line/sessions/native-1",
+                        "session_hashes": {"session.json": "cd" * 32},
+                    }
+                ],
+                "compare": {"ran": True, "headers": []},
+            },
+            features={"features": {"mcp": {"status": "not_run", "reason": "S3"}}},
+            s1={"source": None, "declared_rung": None, "notes": []},
+            thresholds=dict(FROZEN_THRESHOLDS),
+        )
+        text = render_markdown(report)
+        self.assertIn("Rung table", text)
+        self.assertIn("Session hashes", text)
+        self.assertIn("Frozen thresholds", text)
+        self.assertIn("above_lenient_only", text)
+        self.assertIn("native run 1", text)
+
+
+@unittest.skipUnless(os.path.isfile(REAL_IMAGE), "TU104 570.144 fwimage absent")
+class NativeIntegrationTests(unittest.TestCase):
+    """Native-only run on the real image: shape + two-run determinism."""
+
+    def setUp(self):
+        self.tmp = tempfile.mkdtemp(prefix="sim-qemu-line-test-")
+
+    def tearDown(self):
+        shutil.rmtree(self.tmp, ignore_errors=True)
+
+    def test_profile_discovery_when_s1_present(self):
+        from harness.sim_qemu_line import discover_s1_device_model
+
+        profile = os.path.join(TOOL_ROOT, "sim", "profiles", "tu104-570.144.json")
+        if not os.path.isfile(profile):
+            self.skipTest("S1 profile absent")
+        s1 = discover_s1_device_model(REAL_IMAGE, firmware_sha256(REAL_IMAGE))
+        self.assertEqual(os.path.abspath(s1.get("profile") or ""), profile)
+        self.assertTrue(any("greeneye-sim-profile" in note for note in s1["notes"]))
+
+    def test_native_report_shape_and_determinism(self):
+        from harness.sim_qemu_line import run_harness
+
+        report = run_harness(
+            image=REAL_IMAGE,
+            out_dir=self.tmp,
+            instructions=20_000,
+            repeat=2,
+            features=False,
+            run_qemu=False,
+        )
+        self.assertEqual(report["schema"], "greeneye-sim-qemu-line/1")
+        self.assertTrue(report["firmware"]["sha256"])
+        native = report["native"]
+        self.assertTrue(native["ran"])
+        self.assertEqual(native["repeat"], 2)
+        self.assertTrue(native["identical"])
+        runs = native["runs"]
+        self.assertEqual(len(runs), 2)
+        keys = (
+            "instructions", "cycles", "stop", "first_unknown_mmio", "mmio_unknown",
+            "rung", "device_state_sha256", "milestones",
+        )
+        self.assertEqual(
+            [{k: r.get(k) for k in keys} for r in runs],
+            [{k: runs[0].get(k) for k in keys}] * 2,
+        )
+        for run in runs:
+            self.assertGreater(run["instructions"], 0)
+            self.assertLessEqual(run["instructions"], 20_000)
+            self.assertGreater(run["cycles"], 0)
+            self.assertTrue(run["session_hashes"]["session.json"])
+        if native.get("device_model_used"):
+            # S1 semantic path: device model + recorded shim, B2 observed.
+            self.assertEqual(native["rung"], "B2")
+            self.assertIsNone(runs[0]["first_unknown_mmio"])
+            self.assertEqual(runs[0]["mmio_unknown"], 0)
+            self.assertEqual(runs[0]["milestones"], ["bootargs_consumer", "relocation_copy"])
+            self.assertTrue(runs[0]["device_state_sha256"])
+        else:
+            # Lenient path (no S1 in the tree): B0 observation baseline.
+            self.assertEqual(native["rung"], "B0")
+            self.assertEqual(runs[0]["first_unknown_mmio"]["addr"], QEMU_REFERENCE["access_addr"])
+        self.assertEqual(report["verdict"], "unverified_qemu_absent")
+        for name in ("sim-qemu-line.json", "sim-qemu-line.md"):
+            path = os.path.join(self.tmp, name)
+            self.assertTrue(os.path.isfile(path), path)
+        with open(os.path.join(self.tmp, "sim-qemu-line.json"), encoding="utf-8") as fh:
+            payload = json.load(fh)
+        self.assertEqual(payload["schema"], report["schema"])
+        with open(os.path.join(self.tmp, "sim-qemu-line.md"), encoding="utf-8") as fh:
+            md = fh.read()
+        self.assertIn("| native run 1 |", md)
+        self.assertIn("Session hashes", md)
+
+    def test_native_lenient_path_still_available(self):
+        from harness.sim_qemu_line import run_native_baseline
+
+        native = run_native_baseline(
+            image=REAL_IMAGE,
+            out_dir=self.tmp,
+            instructions=20_000,
+            repeat=1,
+            profile=None,
+            device_map=None,
+        )
+        self.assertTrue(native["ran"])
+        self.assertFalse(native["device_model_used"])
+        self.assertEqual(native["rung"], "B0")
+        self.assertEqual(
+            native["runs"][0]["first_unknown_mmio"]["addr"], QEMU_REFERENCE["access_addr"]
+        )
+
+
+@unittest.skipUnless(
+    os.path.isfile(REAL_IMAGE) and qemu_available(),
+    "TU104 570.144 fwimage or qemu-system-riscv64 absent",
+)
+class QemuIntegrationTests(unittest.TestCase):
+    """QEMU probe on the real image must reproduce its documented first fault."""
+
+    def setUp(self):
+        self.tmp = tempfile.mkdtemp(prefix="sim-qemu-line-test-")
+
+    def tearDown(self):
+        shutil.rmtree(self.tmp, ignore_errors=True)
+
+    def test_qemu_first_fault_reproduced(self):
+        from harness.sim_qemu_line import run_qemu_baseline
+
+        result = run_qemu_baseline(image=REAL_IMAGE, out_dir=self.tmp, duration=0.5)
+        self.assertTrue(result["ran"], result.get("reason"))
+        self.assertEqual(result["rung"], "pre-B0")
+        self.assertTrue(result["first_touch_match"], "documented first fault not reproduced")
+        first = result["first_fault"]
+        self.assertEqual(first["epc_fw"], QEMU_REFERENCE["firmware_vaddr"])
+        self.assertEqual(first["access_addr"], QEMU_REFERENCE["access_addr"])
+        self.assertTrue(result["session_hashes"]["session.json"])
+        self.assertTrue(os.path.isfile(os.path.join(self.tmp, "sessions", "qemu-probe", "session.json")))
+
+
+@unittest.skipUnless(os.path.isfile(DEMO_ELF), "demo firmware absent (run make demo)")
+class FeatureMatrixTests(unittest.TestCase):
+    """The native feature matrix runs on the bundled demo firmware."""
+
+    def setUp(self):
+        self.tmp = tempfile.mkdtemp(prefix="sim-qemu-line-test-")
+
+    def tearDown(self):
+        shutil.rmtree(self.tmp, ignore_errors=True)
+
+    def test_feature_matrix_shape(self):
+        from harness.sim_qemu_line import run_feature_matrix
+
+        result = run_feature_matrix(self.tmp)
+        self.assertTrue(result["ran"], result.get("reason"))
+        features = result["features"]
+        for name in ("hooks", "breakpoints", "watch", "memory", "struct", "sessions", "trace"):
+            self.assertIn(name, features)
+            self.assertEqual(features[name]["status"], "pass", features[name])
+        self.assertEqual(features["mcp"]["status"], "not_run")
+        self.assertIn(features["injection"]["status"], ("pass", "not_run"))
+        self.assertIn(features["analyze_compare"]["status"], ("pass", "not_run"))
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/debug/risc-v-gsp-rm-vm/tests/test_triage.py b/debug/risc-v-gsp-rm-vm/tests/test_triage.py
new file mode 100644
index 0000000..35ac128
--- /dev/null
+++ b/debug/risc-v-gsp-rm-vm/tests/test_triage.py
@@ -0,0 +1,264 @@
+"""Tests for the new-version triage workflow (harness/triage.py, docs/40).
+
+The orchestration tests use injected fake step runners: no QEMU, no native
+boot, no firmware.  The integration test uses the real TU104 570.144 fwimage
+read-only with a tiny budget and skips cleanly when it (or
+qemu-system-riscv64) is absent.  Everything is bounded and touches no
+hardware, GPU, module, guest or network.
+
+Run::
+
+    python3 -m unittest tests.test_triage -v
+"""
+from __future__ import annotations
+
+import contextlib
+import io
+import json
+import os
+import sys
+import tempfile
+import unittest
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+TOOL_ROOT = os.path.abspath(os.path.join(HERE, ".."))
+REAL_IMAGE = os.path.normpath(
+    os.path.join(TOOL_ROOT, "..", "..", "firmware", "57X-GSP", "570.144", "gsp-570.144.fwimage.bin")
+)
+if TOOL_ROOT not in sys.path:
+    sys.path.insert(0, TOOL_ROOT)
+
+from harness.common import qemu_available  # noqa: E402
+from harness.triage import (  # noqa: E402
+    DEFAULT_RUNNERS,
+    FORMAT,
+    PLAN_FORMAT,
+    TriageOptions,
+    build_plan,
+    extract_version,
+    main,
+    run_triage,
+)
+
+
+def _run_main(argv: list[str]) -> tuple[int, str, str]:
+    out, err = io.StringIO(), io.StringIO()
+    with contextlib.redirect_stdout(out), contextlib.redirect_stderr(err):
+        rc = main(argv)
+    return rc, out.getvalue(), err.getvalue()
+
+
+def _fake_runners(fail: str | None = None, session_dir: str | None = None):
+    """Deterministic step runners; optionally one raises."""
+    calls: list[str] = []
+
+    def make(name: str, **detail):
+        def fn(options, out):
+            calls.append(name)
+            payload = {"status": "ok"}
+            payload.update(detail)
+            if session_dir and name in ("probe", "native_boot"):
+                os.makedirs(session_dir, exist_ok=True)
+                with open(os.path.join(session_dir, "session.json"), "w", encoding="utf-8") as fh:
+                    fh.write("{}\n")
+                payload["sessions"] = [{"name": name, "dir": session_dir}]
+                payload["session_hashes"] = {"session.json": "deadbeef"}
+            return payload
+
+        return fn
+
+    def boom(options, out):
+        calls.append(fail)
+        raise RuntimeError("injected failure")
+
+    runners = [
+        ("identity", boom if fail == "identity" else make("identity")),
+        ("probe", boom if fail == "probe" else make("probe", rung="pre-B0")),
+        ("native_boot", boom if fail == "native_boot" else make("native_boot", rung="B2")),
+        ("ab_eval", boom if fail == "ab_eval" else make("ab_eval", verdict="above_lenient_only", above_qemu_line=False)),
+        ("version_diff", boom if fail == "version_diff" else make("version_diff", sections={"image": "ok"})),
+        ("features", boom if fail == "features" else make("features", unavailable=[])),
+    ]
+    return runners, calls
+
+
+class VersionExtractionTests(unittest.TestCase):
+    def test_explicit_label_wins(self):
+        version, source = extract_version("/does/not/matter.bin", "999.1")
+        self.assertEqual((version, source), ("999.1", "explicit --version"))
+
+    def test_sidecar_convention(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            fw = os.path.join(tmp, "gsp-x.fwimage.bin")
+            open(fw, "wb").close()
+            with open(os.path.join(tmp, "gsp_version.txt"), "w", encoding="utf-8") as fh:
+                fh.write("570.211W\n")
+            version, source = extract_version(fw)
+            self.assertEqual(version, "570.211W")
+            self.assertEqual(source, "gsp_version.txt sidecar")
+
+    def test_parent_directory_fallback(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            parent = os.path.join(tmp, "580.10")
+            os.makedirs(parent)
+            fw = os.path.join(parent, "gsp-580.10.fwimage.bin")
+            open(fw, "wb").close()
+            self.assertEqual(extract_version(fw)[0], "580.10")
+
+    def test_stem_fallback(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            fw = os.path.join(tmp, "mystery-image.bin")
+            open(fw, "wb").close()
+            version, source = extract_version(fw)
+            self.assertEqual(version, "mystery-image")
+            self.assertEqual(source, "file stem")
+
+
+class OrchestrationTests(unittest.TestCase):
+    def _options(self, tmp: str, **overrides) -> TriageOptions:
+        fw = os.path.join(tmp, "new.fwimage.bin")
+        if not os.path.exists(fw):
+            with open(fw, "wb") as fh:
+                fh.write(b"\x00" * 16)
+        values = dict(firmware=fw, out=os.path.join(tmp, "triage-out"), budget=1000)
+        values.update(overrides)
+        return TriageOptions(**values)
+
+    def test_report_shape_and_artifacts(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            session_dir = os.path.join(tmp, "session")
+            runners, calls = _fake_runners(session_dir=session_dir)
+            report = run_triage(self._options(tmp), runners=runners)
+            self.assertEqual(report["format"], FORMAT)
+            self.assertEqual([s["name"] for s in report["steps"]], [n for n, _ in DEFAULT_RUNNERS])
+            self.assertTrue(all(s["status"] == "ok" for s in report["steps"]))
+            self.assertEqual(report["errors"], [])
+            self.assertTrue(report["sessions"])
+            self.assertEqual(report["sessions"][0]["hashes"]["session.json"], "deadbeef")
+            self.assertEqual(report["support_readiness"]["verdict"], "sim-observable")
+            self.assertFalse(report["support_readiness"]["baseline_compared"])
+            self.assertEqual(report["threshold_checks"]["required_steps_missing"], [])
+            self.assertTrue(report["threshold_checks"]["beats_manual_baseline"])
+            for key in ("report_json", "report_md"):
+                self.assertTrue(os.path.isfile(report["artifacts"][key]), key)
+            with open(report["artifacts"]["report_json"], "r", encoding="utf-8") as fh:
+                loaded = json.load(fh)
+            self.assertEqual(loaded["format"], FORMAT)
+            self.assertEqual(len(loaded["steps"]), len(DEFAULT_RUNNERS))
+            with open(report["artifacts"]["report_md"], "r", encoding="utf-8") as fh:
+                md = fh.read()
+            self.assertIn("New-version triage", md)
+            self.assertIn("Support readiness", md)
+
+    def test_partial_failure_does_not_crash(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            runners, _calls = _fake_runners(fail="probe")
+            report = run_triage(self._options(tmp), runners=runners)
+            self.assertEqual(report["errors"], ["probe"])
+            step = next(s for s in report["steps"] if s["name"] == "probe")
+            self.assertEqual(step["status"], "error")
+            self.assertIn("injected failure", step["reason"])
+            self.assertEqual(
+                next(s for s in report["steps"] if s["name"] == "native_boot")["status"], "ok"
+            )
+            self.assertTrue(any("QEMU probe failed" in b for b in report["support_readiness"]["blockers"]))
+            self.assertTrue(os.path.isfile(report["artifacts"]["report_json"]))
+
+    def test_missing_firmware_is_reported_not_raised(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            out = os.path.join(tmp, "out")
+            report = run_triage(TriageOptions(firmware=os.path.join(tmp, "nope.bin"), out=out))
+            step = report["steps"][0]
+            self.assertEqual(step["name"], "identity")
+            self.assertEqual(step["status"], "error")
+            self.assertEqual(report["support_readiness"]["verdict"], "blocked")
+            self.assertTrue(os.path.isfile(report["artifacts"]["report_json"]))
+
+    def test_budget_bounds(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            options = self._options(tmp, budget=0)
+            with self.assertRaises(ValueError):
+                run_triage(options)
+            with self.assertRaises(ValueError):
+                run_triage(self._options(tmp, budget=3_000_000))
+
+    def test_plan_shape(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            plan = build_plan(self._options(tmp))
+            self.assertEqual(plan["format"], PLAN_FORMAT)
+            self.assertEqual([s["name"] for s in plan["steps"]], [n for n, _ in DEFAULT_RUNNERS])
+            self.assertEqual(plan["side_effects"]["hardware"], "none")
+
+
+class CliTests(unittest.TestCase):
+    def test_rm610_refusal(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            fw = os.path.join(tmp, "gsp-rm610.fwimage.bin")
+            open(fw, "wb").close()
+            out = os.path.join(tmp, "out")
+            rc, _stdout, stderr = _run_main(
+                ["--firmware", fw, "--version", "RM-610", "--out", out]
+            )
+            self.assertEqual(rc, 3)
+            self.assertIn("RM610", stderr)
+            self.assertFalse(os.path.exists(out))
+
+    def test_dry_run_has_no_side_effects(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            fw = os.path.join(tmp, "new.fwimage.bin")
+            open(fw, "wb").close()
+            out = os.path.join(tmp, "out")
+            rc, stdout, _stderr = _run_main(["--firmware", fw, "--out", out, "--dry-run"])
+            self.assertEqual(rc, 0)
+            plan = json.loads(stdout[stdout.index("{"):stdout.rindex("}") + 1])
+            self.assertEqual(plan["format"], PLAN_FORMAT)
+            self.assertFalse(os.path.exists(out))
+
+    def test_missing_firmware_exits_2(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            out = os.path.join(tmp, "out")
+            rc, _stdout, stderr = _run_main(["--firmware", os.path.join(tmp, "nope.bin"), "--out", out])
+            self.assertEqual(rc, 2)
+            self.assertIn("firmware absent", stderr)
+            self.assertFalse(os.path.exists(out))
+
+    def test_bad_budget_is_a_parser_error(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            fw = os.path.join(tmp, "new.fwimage.bin")
+            open(fw, "wb").close()
+            with self.assertRaises(SystemExit):
+                _run_main(["--firmware", fw, "--out", os.path.join(tmp, "out"), "--budget", "9999999"])
+
+
+@unittest.skipUnless(os.path.isfile(REAL_IMAGE), "real 570.144 fwimage absent")
+class RealImageIntegrationTests(unittest.TestCase):
+    def test_triage_real_image_tiny_budget(self):
+        if not qemu_available():
+            self.skipTest("qemu-system-riscv64 not installed")
+        with tempfile.TemporaryDirectory() as tmp:
+            report = run_triage(
+                TriageOptions(
+                    firmware=REAL_IMAGE,
+                    out=os.path.join(tmp, "out"),
+                    budget=3000,
+                    qemu_duration=0.2,
+                    step_timeout=120.0,
+                )
+            )
+            identity = next(s for s in report["steps"] if s["name"] == "identity")
+            self.assertEqual(identity["status"], "ok")
+            self.assertIn("570.144", identity["firmware"]["version"])
+            probe = next(s for s in report["steps"] if s["name"] == "probe")
+            self.assertEqual(probe["status"], "ok")
+            native = next(s for s in report["steps"] if s["name"] == "native_boot")
+            self.assertIn(native["status"], ("ok", "skipped"))
+            self.assertIn(
+                report["support_readiness"]["verdict"], ("sim-observable", "first-touch-only", "blocked")
+            )
+            version_diff = next(s for s in report["steps"] if s["name"] == "version_diff")
+            self.assertEqual(version_diff["status"], "skipped")
+            self.assertTrue(os.path.isfile(report["artifacts"]["report_json"]))
+
+
+if __name__ == "__main__":
+    unittest.main()