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

1 files changed, 244 insertions, 0 deletions

diff --git a/debug/risc-v-gsp-rm-vm/tests/test_sim_backend.py b/debug/risc-v-gsp-rm-vm/tests/test_sim_backend.py
new file mode 100644
index 0000000..a0635ab
--- /dev/null
+++ b/debug/risc-v-gsp-rm-vm/tests/test_sim_backend.py
@@ -0,0 +1,244 @@
+"""Native (M1) backend tests: demo parity with QEMU and engine determinism.
+
+The native simulator is pure Python and always available; every test still
+skips cleanly when the demo ELF cannot be loaded.  The differential class
+additionally requires qemu-system-riscv64 and compares the scripted
+hook/watch/leave profile of both backends event for event.
+"""
+from __future__ import annotations
+
+import os
+import shutil
+import tempfile
+import unittest
+
+import greeneye as ge
+from greeneye.errors import RspTimeout
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+TOOL_ROOT = os.path.join(HERE, "..")
+DEMO_ELF = os.path.join(TOOL_ROOT, "firmware", "uproc_demo", "build", "uproc_demo.elf")
+RPC_HANDLERS = ["rpc_ping", "rpc_echo", "rpc_fill", "rpc_sum", "rpc_state"]
+
+
+def native_available() -> bool:
+    try:
+        import greeneye.sim_backend  # noqa: F401
+    except Exception:
+        return False
+    return True
+
+
+def qemu_available() -> bool:
+    return bool(os.environ.get("GREENEYE_QEMU") or shutil.which("qemu-system-riscv64"))
+
+
+def demo_available() -> bool:
+    return os.path.exists(DEMO_ELF) and native_available()
+
+
+def launch(backend: str, record=None, name: str = "sim"):
+    return ge.launch(kernel=DEMO_ELF, backend=backend, record=record, name=name)
+
+
+@unittest.skipUnless(native_available(), "native simulator unavailable")
+class NativeIsaTests(unittest.TestCase):
+    """Exercise RV64M/A/C paths the demo firmware does not reach."""
+
+    # Assembled with llvm-mc -triple=riscv64 -mattr=+m,+a,+c (encodings are
+    # data, so this runs without a toolchain):
+    #   li a0,6; li a1,7; mul a2,a0,a1; div a5,a1,a0; rem a6,a1,a0;
+    #   srai a4,a1,1; subw a3,a0,a1; lui s0,0x12345; addi sp,sp,-16;
+    #   sd a0,8(sp); ld s1,8(sp); lh s2,8(sp); lbu s3,8(sp);
+    #   amoadd.d s4,a0,(sp); lr.d s5,(sp); sc.d s6,a0,(sp);
+    #   li t0,1; c.addi t0,3; c.mv a2,t0; c.andi a2,15; ebreak
+    PROGRAM = bytes.fromhex(
+        "".join(
+            """
+            19 45 9d 45 33 06 b5 02 b3 c7 a5 02 33 e8 a5 02 13 d7 15 40
+            bb 06 b5 40 37 54 34 12 41 11 2a e4 a2 64 03 19 81 00 83 49 81 00
+            2f 3a a1 00 af 3a 01 10 2f 3b a1 18 85 42 8d 02 16 86 3d 8a 02 90
+            """.split()
+        )
+    )
+    BASE = 0x80000000
+    STACK = 0x80004000
+
+    def test_m_a_c_execution(self):
+        from sim.machine import Machine
+
+        machine = Machine()
+        machine.load_segments([(self.BASE, self.PROGRAM)], self.BASE)
+        machine.set_reg(2, self.STACK)  # sp
+        reason, pc, _, _ = machine.run(1000)
+        self.assertEqual(reason, 3)  # R_EBREAK: stops before executing it
+        self.assertEqual(pc, self.BASE + len(self.PROGRAM) - 2)
+        expected = {
+            5: 4,  # t0: c.addi
+            8: 0x12345000,  # s0: lui
+            9: 6,  # s1: c.sdsp/c.ldsp round-trip
+            10: 6,  # a0
+            11: 7,  # a1
+            12: 4,  # a2: mul then c.mv/c.andi
+            13: 0xFFFFFFFFFFFFFFFF,  # a3: subw sign extension
+            14: 3,  # a4: srai
+            15: 1,  # a5: div
+            16: 1,  # a6: rem
+            18: 6,  # s2: lh
+            19: 6,  # s3: lbu
+            20: 0,  # s4: amoadd old value
+            21: 6,  # s5: lr
+            22: 0,  # s6: sc success
+            2: self.STACK - 16,  # c.addi16sp
+        }
+        got = {index: machine.x[index] for index in expected}
+        self.assertEqual(got, expected)
+        self.assertEqual(machine.read_mem(self.STACK - 16, 8), (6).to_bytes(8, "little"))
+
+
+@unittest.skipUnless(demo_available(), "demo firmware or native simulator unavailable")
+class NativeSmokeTests(unittest.TestCase):
+    def test_start_sequence_sets_stack_and_enters_main(self):
+        session = launch("native", name="unittest-sim-start")
+        try:
+            session.step(4)  # auipc sp / addi sp / auipc ra / jalr main
+            stack_top = next(s.value for s in session.image.symbols if s.name == "_stack_top")
+            self.assertEqual(session.pc, session.symbols.lookup("main"))
+            self.assertEqual(session.read_regs()["sp"], stack_top)
+        finally:
+            session.close()
+
+    def test_boot_hook_watch_snapshot_analyze(self):
+        root = tempfile.mkdtemp(prefix="greeneye-sim-")
+        session = launch("native", record=root, name="unittest-sim-smoke")
+        hits = []
+        try:
+            ctrl = session.symbols.lookup("gsp_ctrl")
+            session.hook("rpc_sum", on_enter=lambda hit: hits.append(hit))
+            watch_hits = []
+            session.watch(ctrl + 16, size=8, kind="w", callback=lambda hit: watch_hits.append(hit))
+            session.run(duration=0.6)
+            self.assertTrue(session.target.wait_for_uart("GREEN-EYE UPROC DEMO", timeout=1.0))
+            self.assertGreaterEqual(len(hits), 1, "rpc_sum hook never fired")
+            self.assertGreaterEqual(len(watch_hits), 7, "rpc_count watchpoint never fired")
+            self.assertEqual(watch_hits[0].addr, ctrl + 16)
+            self.assertEqual(watch_hits[0].watch.kind, "w")
+            after = session.snapshot(ctrl, 0x188, "ctrl")
+            self.assertEqual(int.from_bytes(after[0:4], "little"), 0x47535031)
+            self.assertGreaterEqual(int.from_bytes(after[16:24], "little"), 7)
+        finally:
+            directory = getattr(session.recorder, "dir", None)
+            session.close()
+        from greeneye.analyze import render_summary
+
+        report = render_summary(directory)
+        self.assertIn("rpc_sum", report)
+        self.assertIn("watch", report)
+
+    def test_hook_args_and_leave_value(self):
+        session = launch("native", name="unittest-sim-hookargs")
+        events = []
+        try:
+            ctrl = session.symbols.lookup("gsp_ctrl")
+            session.hook(
+                "rpc_sum",
+                on_enter=lambda hit: events.append(("enter", hit.arg(0), hit.arg(1))),
+                on_leave=lambda hit: events.append(("leave", hit.retval)),
+            )
+            session.run(duration=0.2)
+            # Self-test: rpc_dispatch(4, ctl->buf, 16, 0) after fill(0x5a, 32).
+            self.assertIn(("enter", ctrl + 0x48, 16), events)
+            self.assertIn(("leave", 16 * 0x5A), events)
+        finally:
+            session.close()
+
+    def test_call_and_run_until(self):
+        session = launch("native", name="unittest-sim-call")
+        try:
+            before = session.pc
+            self.assertEqual(session.call("rpc_ping", (0x1111,)), 0x1111 ^ 0xBEEF)
+            self.assertEqual(session.pc, before)
+            hit = session.run_until("rpc_sum", timeout=2.0)
+            self.assertEqual(hit.kind, "breakpoint")
+            self.assertEqual(hit.addr, session.symbols.lookup("rpc_sum"))
+        finally:
+            session.close()
+
+    def test_run_until_timeout(self):
+        session = launch("native", name="unittest-sim-timeout")
+        try:
+            with self.assertRaises(RspTimeout):
+                session.run_until("rpc_fault", timeout=0.05)  # self-test never calls it
+        finally:
+            session.close()
+
+
+@unittest.skipUnless(demo_available(), "demo firmware or native simulator unavailable")
+class NativeDeterminismTests(unittest.TestCase):
+    def _scripted_profile(self):
+        """Drive fixed instruction budgets: no host clock in the result."""
+        session = ge.launch(kernel=DEMO_ELF, backend="native", name="unittest-sim-determinism")
+        events = []
+        stops = []
+        try:
+            session.hook(
+                "rpc_sum",
+                on_enter=lambda hit: events.append(("enter", hit.arg(0), hit.arg(1))),
+                on_leave=lambda hit: events.append(("leave", hit.retval)),
+            )
+            ctrl = session.symbols.lookup("gsp_ctrl")
+            session.watch(ctrl + 16, size=8, kind="w")
+            for _ in range(20):
+                try:
+                    stop = session.client.continue_(timeout=0.02)  # exactly 20k insns
+                except RspTimeout:
+                    stops.append(("budget", session.pc))
+                    continue
+                session._dispatch_stop(stop)
+                stops.append(("stop", session.pc))
+            ctrl_image = session.read_regs()
+            snapshot = session.client.read_mem(ctrl, 0x188)
+        finally:
+            session.close()
+        return events, stops, ctrl_image["pc"], snapshot
+
+    def test_two_runs_identical(self):
+        first = self._scripted_profile()
+        second = self._scripted_profile()
+        self.assertEqual(first, second)
+
+
+@unittest.skipUnless(demo_available() and qemu_available(), "native simulator and QEMU required")
+class DifferentialTests(unittest.TestCase):
+    """Run one scripted scenario on both backends and compare the profiles."""
+
+    def _profile(self, backend: str):
+        session = launch(backend, name=f"unittest-diff-{backend}")
+        events = []
+        watch_hits = []
+        try:
+            for name in RPC_HANDLERS:
+                session.hook(
+                    name,
+                    on_enter=lambda hit, n=name: events.append(("enter", n, hit.arg(0), hit.arg(1))),
+                    on_leave=lambda hit, n=name: events.append(("leave", n, hit.retval)),
+                )
+            ctrl = session.symbols.lookup("gsp_ctrl")
+            session.watch(ctrl + 16, size=8, kind="w", callback=lambda hit: watch_hits.append(hit.addr))
+            session.run(duration=0.3)
+            uart = session.target.read_uart()
+            bp_counts = [handle.hits for handle in session._bps.values()]
+        finally:
+            session.close()
+        return events, len(watch_hits), bp_counts, uart.splitlines()[0] if uart else ""
+
+    def test_hook_watch_profile_parity(self):
+        qemu = self._profile("qemu")
+        native = self._profile("native")
+        self.assertEqual(qemu, native)
+        self.assertEqual(native[1], 8)  # 7 self-test dispatches + ctrl init
+        self.assertEqual(native[2], [1, 1, 1, 1, 3])
+
+
+if __name__ == "__main__":
+    unittest.main()