risc-v-gsp-rm-debugger: land docs 45-53 + harness/sim tool set with tests

rm_capture (chunked decode), launch_audit, state_seed/state_decode,
valprobe, rehost, contracts+run_contracts, host_traffic, gmmu_walk,
msgq_attach, libos_port, log_decode, server_snapshot; doc index updates
in the tool overview, operations pages and the repo README.

13 files changed, 4435 insertions, 0 deletions

diff --git a/debug/risc-v-gsp-rm-vm/tests/test_contracts.py b/debug/risc-v-gsp-rm-vm/tests/test_contracts.py
new file mode 100644
index 0000000..3199b09
--- /dev/null
+++ b/debug/risc-v-gsp-rm-vm/tests/test_contracts.py
@@ -0,0 +1,171 @@
+#!/usr/bin/env python3
+"""Tests for the wall-contract registry and runner (``harness/contracts.py``).
+
+The fast contracts execute the proven offline entry points (valprobe matrix,
+GMMU golden vector, launch audit, rehost demo, RM-capture demo) and are
+asserted to pass; prerequisites that are genuinely absent are reported as
+skips, never as passes.  No hardware, no module, no guest, no network.
+
+Run::
+
+    python3 -m unittest tests.test_contracts -v
+"""
+from __future__ import annotations
+
+import contextlib
+import importlib
+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, ".."))
+if TOOL_ROOT not in sys.path:
+    sys.path.insert(0, TOOL_ROOT)
+
+from harness import run_contracts as runner  # noqa: E402
+from harness.contracts import (  # noqa: E402
+    FAST,
+    SLOW,
+    WALL_CLASSES,
+    Contract,
+    ContractSkip,
+    all_contracts,
+    fast_contracts,
+    run_contracts,
+    summarize,
+)
+
+MISSING_PATH = os.path.join(TOOL_ROOT, "contract-test-missing-prerequisite")
+
+
+def _entry_target(entry: str) -> tuple[str, str]:
+    module, _, rest = entry.partition(":")
+    return module, rest.split("(", 1)[0].strip()
+
+
+def _raise(exc: Exception):
+    def _run():
+        raise exc
+
+    return _run
+
+
+def _fake(name: str, **overrides) -> Contract:
+    fields = {
+        "name": name,
+        "wall_class": "a",
+        "doc": "docs/50",
+        "entry": "tests:fake",
+        "speed": FAST,
+        "expect": "test expectation",
+        "run": lambda: {"ok": True},
+        "check": lambda report: (report.get("ok") is True, "ok"),
+        "requires": (),
+    }
+    fields.update(overrides)
+    return Contract(**fields)
+
+
+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):
+        try:
+            rc = runner.main(argv)
+        except SystemExit as exc:
+            rc = int(exc.code or 0)
+    return rc, out.getvalue(), err.getvalue()
+
+
+class RegistryShapeTests(unittest.TestCase):
+    def test_registry_shape(self):
+        contracts = all_contracts()
+        names = [contract.name for contract in contracts]
+        self.assertEqual(len(names), len(set(names)), "contract names must be unique")
+        self.assertGreaterEqual(len(contracts), 3, "keep at least the proven contracts")
+        for contract in contracts:
+            with self.subTest(contract=contract.name):
+                self.assertIn(contract.wall_class, WALL_CLASSES)
+                self.assertIn(contract.speed, (FAST, SLOW))
+                self.assertTrue(contract.doc.startswith("docs/"))
+                self.assertTrue(contract.entry)
+                self.assertTrue(contract.expect)
+        self.assertTrue(fast_contracts(), "registry must have fast contracts")
+
+    def test_entries_resolve_to_real_callables(self):
+        for contract in all_contracts():
+            with self.subTest(contract=contract.name):
+                module_name, func_name = _entry_target(contract.entry)
+                module = importlib.import_module(module_name)
+                self.assertTrue(callable(getattr(module, func_name, None)), contract.entry)
+
+
+class RunnerSemanticsTests(unittest.TestCase):
+    def test_classifies_pass_fail_skip(self):
+        results = run_contracts(
+            [
+                _fake("good"),
+                _fake("raising", run=_raise(RuntimeError("boom"))),
+                _fake("missing-prereq", requires=(MISSING_PATH,)),
+                _fake("declared-skip", run=_raise(ContractSkip("not applicable here"))),
+                _fake("failed-check", check=lambda report: (False, "checker rejected")),
+            ]
+        )
+        by_name = {result.name: result for result in results}
+        self.assertEqual(by_name["good"].status, "pass")
+        self.assertEqual(by_name["missing-prereq"].status, "skip")
+        self.assertEqual(by_name["declared-skip"].status, "skip")
+        self.assertEqual(by_name["raising"].status, "fail")
+        self.assertIn("boom", by_name["raising"].detail)
+        self.assertEqual(by_name["failed-check"].status, "fail")
+
+        summary = summarize(results)
+        self.assertFalse(summary["ok"])
+        self.assertEqual(summary["counts"], {"pass": 1, "fail": 2, "skip": 2})
+        domains = {domain["wall_class"]: domain for domain in summary["domains"]}
+        self.assertEqual(set(domains), set(WALL_CLASSES))
+        self.assertEqual(domains["a"]["total"], 5)
+        self.assertEqual(domains["d"]["total"], 0)
+
+    def test_fast_contracts_pass(self):
+        contracts = fast_contracts()
+        self.assertGreaterEqual(len(contracts), 3)
+        results = run_contracts(contracts)
+        for result in results:
+            with self.subTest(contract=result.name):
+                self.assertNotEqual(result.status, "fail", result.detail)
+                if result.status == "skip":
+                    self.skipTest(result.detail)
+        self.assertTrue(any(result.status == "pass" for result in results))
+
+
+class RunnerCliTests(unittest.TestCase):
+    def test_single_contract_json_summary(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            path = os.path.join(tmp, "contracts.json")
+            rc, out, err = _run_main(
+                ["--fast", "--name", "gmmu-golden-vector", "--json", path]
+            )
+            self.assertEqual(rc, 0, err)
+            self.assertIn("PASS", out)
+            with open(path, "r", encoding="utf-8") as fh:
+                summary = json.load(fh)
+            self.assertTrue(summary["ok"])
+            self.assertEqual(summary["counts"]["pass"], 1)
+            self.assertEqual(
+                [contract["name"] for contract in summary["contracts"]],
+                ["gmmu-golden-vector"],
+            )
+            self.assertEqual(len(summary["domains"]), len(WALL_CLASSES))
+
+    def test_unknown_contract_name_is_usage_error(self):
+        rc, _out, err = _run_main(["--fast", "--name", "no-such-contract"])
+        self.assertEqual(rc, 2)
+        self.assertIn("unknown contract", err)
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/debug/risc-v-gsp-rm-vm/tests/test_gmmu_walk.py b/debug/risc-v-gsp-rm-vm/tests/test_gmmu_walk.py
new file mode 100644
index 0000000..f438068
--- /dev/null
+++ b/debug/risc-v-gsp-rm-vm/tests/test_gmmu_walk.py
@@ -0,0 +1,243 @@
+"""Tests for the functional GMMU walker (sim/gmmu_walk.py, docs/48).
+
+The golden vector is a captured channel PDB (numeric entries); the synthetic
+trees are built with the module's own encoders and must round-trip.  No
+hardware, no modules, no guest, no network.
+
+Run::
+
+    python3 -m unittest tests.test_gmmu_walk -v
+"""
+from __future__ import annotations
+
+import contextlib
+import io
+import json
+import os
+import struct
+import sys
+import unittest
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+TOOL_ROOT = os.path.abspath(os.path.join(HERE, ".."))
+if TOOL_ROOT not in sys.path:
+    sys.path.insert(0, TOOL_ROOT)
+
+from sim.gmmu_walk import (  # noqa: E402
+    FAULT_INVALID_APERTURE,
+    FAULT_PTE_INVALID,
+    FAULT_TABLE_MISSING,
+    FAULT_UNSUPPORTED,
+    FAULT_OUT_OF_RANGE,
+    WALK_FORMAT,
+    MemoryView,
+    WalkError,
+    channel_pdb_fixture,
+    encode_dual_slot,
+    encode_pde_pointer,
+    encode_pte,
+    main,
+    walk,
+)
+
+
+def _tree(root_pa: int, va: int, target_pa: int | None, *, base: int = 0x50000000,
+          read_only: bool = False) -> MemoryView:
+    i0 = (va >> 47) & 0x3
+    i1 = (va >> 38) & 0x1FF
+    i2 = (va >> 29) & 0x1FF
+    i3 = (va >> 21) & 0xFF
+    i4 = (va >> 12) & 0x1FF
+    pd2, pd1, pd0, pt = base + 0x1000, base + 0x2000, base + 0x3000, base + 0x4000
+    mem = MemoryView()
+    page = bytearray(0x1000)
+    struct.pack_into("<Q", page, i0 * 8, encode_pde_pointer(pd2))
+    mem.add(root_pa, bytes(page))
+    page = bytearray(0x1000)
+    struct.pack_into("<Q", page, i1 * 8, encode_pde_pointer(pd1))
+    mem.add(pd2, bytes(page))
+    page = bytearray(0x1000)
+    struct.pack_into("<Q", page, i2 * 8, encode_pde_pointer(pd0))
+    mem.add(pd1, bytes(page))
+    page = bytearray(0x1000)
+    struct.pack_into("<16s", page, i3 * 16, encode_dual_slot(None, pt))
+    mem.add(pd0, bytes(page))
+    page = bytearray(0x1000)
+    if target_pa is not None:
+        struct.pack_into("<Q", page, i4 * 8, encode_pte(target_pa, read_only=read_only))
+    mem.add(pt, bytes(page))
+    return mem
+
+
+def _run_main(argv):
+    out, err = io.StringIO(), io.StringIO()
+    with contextlib.redirect_stdout(out), contextlib.redirect_stderr(err):
+        rc = main(argv)
+    return rc, out.getvalue(), err.getvalue()
+
+
+class GoldenVectorTests(unittest.TestCase):
+    def test_captured_channel_pdb_walks_to_logged_dmas(self):
+        root_pa, mem = channel_pdb_fixture()
+        expected = {
+            0x100000: 0x2045D9000,
+            0x101000: 0x121BD9000,
+            0x102000: 0x152115000,
+            0x103000: 0x15A5F5000,
+            0x104000: 0x10CF27000,
+        }
+        for va, pa in expected.items():
+            report = walk(va, root_pa=root_pa, mem=mem)
+            self.assertEqual(report["status"], "ok", report)
+            self.assertEqual(report["format"], WALK_FORMAT)
+            self.assertEqual(int(report["pa"], 16), pa)
+            self.assertEqual(report["page_size"], "4k")
+            self.assertTrue(report["writable"])
+            self.assertEqual(report["evidence"], "validated")
+            self.assertEqual(len(report["steps"]), 5)
+
+    def test_walk_is_deterministic(self):
+        root_pa, mem = channel_pdb_fixture()
+        first = json.dumps(walk(0x100000, root_pa=root_pa, mem=mem), sort_keys=True)
+        second = json.dumps(walk(0x100000, root_pa=root_pa, mem=mem), sort_keys=True)
+        self.assertEqual(first, second)
+
+
+class SyntheticRoundTripTests(unittest.TestCase):
+    def test_encoder_round_trip_4k(self):
+        va, pa = 0x204608000, 0x15E8FA000
+        mem = _tree(0x60000000, va, pa)
+        report = walk(va, root_pa=0x60000000, mem=mem)
+        self.assertEqual(report["status"], "ok")
+        self.assertEqual(int(report["pa"], 16), pa)
+        self.assertEqual(report["page_size"], "4k")
+        self.assertTrue(report["writable"])
+
+    def test_read_only_leaf_is_reported(self):
+        va, pa = 0x204608000, 0x15E8FA000
+        mem = _tree(0x60000000, va, pa, read_only=True)
+        report = walk(va, root_pa=0x60000000, mem=mem)
+        self.assertEqual(report["status"], "ok")
+        self.assertTrue(report["read_only"])
+        self.assertFalse(report["writable"])
+
+    def test_2m_leaf(self):
+        va, base_pa = 0x40000000, 0x20000000
+        mem = MemoryView()
+        i0 = (va >> 47) & 0x3
+        i1 = (va >> 38) & 0x1FF
+        i2 = (va >> 29) & 0x1FF
+        i3 = (va >> 21) & 0xFF
+        pd2, pd1, pd0 = 0x51001000, 0x51002000, 0x51003000
+        page = bytearray(0x1000)
+        struct.pack_into("<Q", page, i0 * 8, encode_pde_pointer(pd2))
+        mem.add(0x51000000, bytes(page))
+        page = bytearray(0x1000)
+        struct.pack_into("<Q", page, i1 * 8, encode_pde_pointer(pd1))
+        mem.add(pd2, bytes(page))
+        page = bytearray(0x1000)
+        struct.pack_into("<Q", page, i2 * 8, encode_pde_pointer(pd0))
+        mem.add(pd1, bytes(page))
+        page = bytearray(0x1000)
+        struct.pack_into("<Q", page, i3 * 16, encode_pte(base_pa, page_size="2m"))
+        mem.add(pd0, bytes(page))
+        report = walk(va + 0x12345, root_pa=0x51000000, mem=mem)
+        self.assertEqual(report["status"], "ok")
+        self.assertEqual(report["page_size"], "2m")
+        self.assertEqual(int(report["pa"], 16), base_pa + 0x12345)
+
+    def test_auto_page_mode_prefers_prefers_small_when_big_invalid(self):
+        va = 0x100000  # 64K-aligned; the dual BIG half is invalid (0)
+        root_pa, mem = channel_pdb_fixture()
+        report = walk(va, root_pa=root_pa, mem=mem, page_mode="auto")
+        self.assertEqual(report["status"], "ok")
+        self.assertEqual(report["page_mode_used"], "4k")
+
+    def test_auto_page_mode_uses_big_when_valid(self):
+        va, pa = 0x100000, 0x30000000
+        # build a tree whose dual BIG half points at a 64K-granularity table
+        i0 = (va >> 47) & 0x3
+        i1 = (va >> 38) & 0x1FF
+        i2 = (va >> 29) & 0x1FF
+        i3 = (va >> 21) & 0xFF
+        pd2, pd1, pd0, pt = 0x52001000, 0x52002000, 0x52003000, 0x52004000
+        mem = MemoryView()
+        page = bytearray(0x1000)
+        struct.pack_into("<Q", page, i0 * 8, encode_pde_pointer(pd2))
+        mem.add(0x52000000, bytes(page))
+        page = bytearray(0x1000)
+        struct.pack_into("<Q", page, i1 * 8, encode_pde_pointer(pd1))
+        mem.add(pd2, bytes(page))
+        page = bytearray(0x1000)
+        struct.pack_into("<Q", page, i2 * 8, encode_pde_pointer(pd0))
+        mem.add(pd1, bytes(page))
+        page = bytearray(0x1000)
+        struct.pack_into("<16s", page, i3 * 16, encode_dual_slot(pt, pt))
+        mem.add(pd0, bytes(page))
+        page = bytearray(0x1000)
+        # BIG path indexes the 64K table by VA bits 20:16
+        struct.pack_into("<Q", page, ((va >> 16) & 0x1FF) * 8,
+                         encode_pte(pa, page_size="64k"))
+        mem.add(pt, bytes(page))
+        report = walk(va, root_pa=0x52000000, mem=mem, page_mode="auto")
+        self.assertEqual(report["status"], "ok")
+        self.assertEqual(report["page_mode_used"], "64k")
+        self.assertEqual(report["evidence"], "modeled")
+
+
+class FaultTests(unittest.TestCase):
+    def test_missing_table_page(self):
+        root_pa, mem = channel_pdb_fixture()
+        report = walk(0x100000, root_pa=root_pa, mem=MemoryView())
+        self.assertEqual(report["status"], "fault")
+        self.assertEqual(report["fault"], FAULT_TABLE_MISSING)
+
+    def test_invalid_leaf(self):
+        mem = _tree(0x60000000, 0x204608000, None)  # leaf entry stays zero
+        report = walk(0x204608000, root_pa=0x60000000, mem=mem)
+        self.assertEqual(report["status"], "fault")
+        self.assertEqual(report["fault"], FAULT_PTE_INVALID)
+
+    def test_invalid_pde_aperture(self):
+        root_pa = 0x60000000
+        mem = MemoryView()
+        page = bytearray(0x1000)
+        struct.pack_into("<Q", page, 0, 0)  # aperture 0 = invalid
+        mem.add(root_pa, bytes(page))
+        report = walk(0x100000, root_pa=root_pa, mem=mem)
+        self.assertEqual(report["status"], "fault")
+        self.assertEqual(report["fault"], FAULT_INVALID_APERTURE)
+
+    def test_unsupported_kind_fails_closed(self):
+        root_pa, mem = channel_pdb_fixture()
+        report = walk(0x100000, root_pa=root_pa, mem=mem, kind="ver9")
+        self.assertEqual(report["status"], "fault")
+        self.assertEqual(report["fault"], FAULT_UNSUPPORTED)
+
+    def test_va_out_of_range(self):
+        root_pa, mem = channel_pdb_fixture()
+        report = walk(1 << 60, root_pa=root_pa, mem=mem)
+        self.assertEqual(report["status"], "fault")
+        self.assertEqual(report["fault"], FAULT_OUT_OF_RANGE)
+
+    def test_bad_page_mode_is_clean_error(self):
+        root_pa, mem = channel_pdb_fixture()
+        with self.assertRaises(WalkError):
+            walk(0x100000, root_pa=root_pa, mem=mem, page_mode="8k")
+
+
+class CliTests(unittest.TestCase):
+    def test_demo_returns_zero(self):
+        rc, out, _ = _run_main(["demo"])
+        self.assertEqual(rc, 0)
+        self.assertIn('"ok": true', out)
+        self.assertIn("PASS", out)
+
+    def test_run_returns_three_on_fault(self):
+        rc, out, _ = _run_main(["run", "--pdb-pa", "0x60000000", "--va", "0x1000"])
+        self.assertEqual(rc, 3)
+        self.assertIn("FAULT", out)
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/debug/risc-v-gsp-rm-vm/tests/test_host_traffic.py b/debug/risc-v-gsp-rm-vm/tests/test_host_traffic.py
new file mode 100644
index 0000000..93024d2
--- /dev/null
+++ b/debug/risc-v-gsp-rm-vm/tests/test_host_traffic.py
@@ -0,0 +1,639 @@
+"""Host-traffic message-loop transport tests (sim/host_traffic.py, docs/51).
+
+All fixtures are synthetic and fully offline: no hardware, no kernel module,
+no QEMU, no guest, no network.  The tests drive the modelled doorbell, the
+msgq rings, the queue-element framing and the stub dispatcher against a small
+in-process Machine (or a bare region) and assert framing, ordering and
+determinism only.  The dispatcher handlers are stubs: nothing here claims
+firmware behaviour.
+
+Run::
+
+    python3 -m unittest tests.test_host_traffic -v
+"""
+from __future__ import annotations
+
+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, ".."))
+if TOOL_ROOT not in sys.path:
+    sys.path.insert(0, TOOL_ROOT)
+
+from greeneye.record import Recorder, load_events  # noqa: E402
+from sim.devices import DeviceRegistry, FalconBlock  # noqa: E402
+from sim.host_traffic import (  # noqa: E402
+    DEMO_FORMAT,
+    DOORBELL_BASE,
+    DOORBELL_SIZE,
+    GSP_MSG_QUEUE_ELEMENT_HDR_SIZE,
+    GSP_MSG_QUEUE_ELEMENT_SIZE_MIN,
+    GSP_RM_ALLOC,
+    GSP_RM_CONTROL,
+    MSGQ_FLAGS_SWAP_RX,
+    MSGQ_ENTRY_OFF,
+    MSGQ_RX_HDR_OFF,
+    NV_ERR_NOT_SUPPORTED,
+    NV_OK,
+    REPLAY_FORMAT,
+    RPC_FN_ECHO,
+    RPC_FN_STATUS_QUERY,
+    RPC_HEADER_BYTES,
+    RPC_HEADER_VERSION,
+    RPC_SIGNATURE,
+    SHARED_MEM_BASE,
+    STATUS_MAGIC,
+    HostTraffic,
+    HostTrafficError,
+    MsgqQueue,
+    RegionMemory,
+    SharedQueues,
+    build_queue_element,
+    build_rpc_message,
+    compute_message_checksum,
+    load_stimulus,
+    main,
+    parse_queue_element,
+    parse_rpc_message,
+    replay_stimulus,
+    run_demo,
+)
+from sim.machine import R_EBREAK, Machine  # noqa: E402
+
+FALCON_BASE = 0x2000000000110000
+FALCON_DEBUGINFO = 0x94
+
+# lui t0,0x20000; slli t0,t0,32; lui t1,0x111; addi t1,t1,-0x400;
+# add t0,t0,t1; lw a0,0(t0); ebreak  -- read doorbell queue 0.
+PROGRAM_READ_DOORBELL = bytes.fromhex(
+    "b70200209392020237131100130303c0b382620003a5020073001000"
+)
+# lui t0,0x20000; slli t0,t0,32; lui t1,0x111; addi t1,t1,-0x3f0;
+# add t0,t0,t1; sw a1,0(t0); ebreak  -- write doorbell queue 2.
+PROGRAM_WRITE_DOORBELL = bytes.fromhex(
+    "b70200209392020237131100130303c1b382620023a0b20073001000"
+)
+
+
+def _sample_capture_log() -> str:
+    payload = (
+        (0xC2000006).to_bytes(4, "little")
+        + (0xABCD0080).to_bytes(4, "little")
+        + (0x2080012B).to_bytes(4, "little")
+        + bytes(4)
+    )
+    return "\n".join(
+        [
+            "[ 1.0] NVRM: GE-GOLDEN-BEGIN",
+            "[ 1.1] NVRM: GE-CTRL cmd=0x2080012b hClient=0xc2000006 "
+            "hObject=0xabcd0080 size=0x8 paramp=0x0 ret=0x0",
+            "[ 1.1] NVRM: GE-CTRL 000:" + "".join(f" {b:02x}" for b in bytes(8)),
+            "[ 1.2] NVRM: GE-SEND fn=0x0000004c len=0x00000028 ret=0x0",
+            "[ 1.2] NVRM: GE-SEND 000:" + "".join(f" {b:02x}" for b in payload),
+            "[ 1.3] NVRM: GE-GOLDEN-END status=0x0",
+        ]
+    ) + "\n"
+
+
+class FramingTests(unittest.TestCase):
+    def test_rpc_constants_match_capture_tool(self):
+        from harness.rm_capture import (
+            GSP_RM_ALLOC as capture_alloc,
+            GSP_RM_CONTROL as capture_control,
+            RPC_HEADER_BYTES as capture_header_bytes,
+        )
+
+        self.assertEqual(RPC_HEADER_BYTES, capture_header_bytes)
+        self.assertEqual(GSP_RM_CONTROL, capture_control)
+        self.assertEqual(GSP_RM_ALLOC, capture_alloc)
+        self.assertEqual(RPC_HEADER_BYTES, 32)
+
+    def test_rpc_message_roundtrip(self):
+        params = bytes(range(64))
+        raw = build_rpc_message(RPC_FN_ECHO, params, sequence=7)
+        self.assertEqual(len(raw), RPC_HEADER_BYTES + len(params))
+        message = parse_rpc_message(raw)
+        self.assertEqual(message.header_version, RPC_HEADER_VERSION)
+        self.assertEqual(message.signature, RPC_SIGNATURE)
+        self.assertEqual(message.function, RPC_FN_ECHO)
+        self.assertEqual(message.sequence, 7)
+        self.assertEqual(message.rpc_result, NV_OK)
+        self.assertEqual(message.length, len(raw))
+        self.assertEqual(message.params, params)
+        self.assertEqual(message.to_bytes(), raw)
+
+    def test_rpc_bad_signature_is_refused(self):
+        raw = bytearray(build_rpc_message(RPC_FN_ECHO, b"\x01", sequence=1))
+        raw[4:8] = (0xDEADBEEF).to_bytes(4, "little")
+        with self.assertRaises(HostTrafficError):
+            parse_rpc_message(bytes(raw))
+
+    def test_rpc_short_and_overlong_length_is_refused(self):
+        raw = build_rpc_message(RPC_FN_ECHO, b"\x01\x02", sequence=1)
+        with self.assertRaises(HostTrafficError):
+            parse_rpc_message(raw[: RPC_HEADER_BYTES - 1])
+        overlong = bytearray(raw)
+        overlong[8:12] = (len(raw) + 16).to_bytes(4, "little")
+        with self.assertRaises(HostTrafficError):
+            parse_rpc_message(bytes(overlong))
+
+    def test_queue_element_layout(self):
+        params = bytes(range(16))
+        rpc = build_rpc_message(RPC_FN_ECHO, params, sequence=3)
+        element = build_queue_element(rpc, sequence=9)
+        self.assertEqual(GSP_MSG_QUEUE_ELEMENT_HDR_SIZE, 0x30)
+        self.assertEqual(element[0x30:0x34], rpc[:4])
+        self.assertEqual(element[0x38:0x3C], len(rpc).to_bytes(4, "little"))
+        self.assertEqual(element[0x34:0x38], RPC_SIGNATURE.to_bytes(4, "little"))
+        self.assertEqual(element[0x24:0x28], (9).to_bytes(4, "little"))
+        self.assertEqual(element[0x28:0x2C], (1).to_bytes(4, "little"))
+        self.assertEqual(element[0x2C:0x30], bytes(4))
+        self.assertEqual(element[0x50:], params)
+        prefix, parsed = parse_queue_element(element)
+        self.assertEqual(prefix["seq_num"], 9)
+        self.assertEqual(prefix["elem_count"], 1)
+        self.assertEqual(parsed.function, RPC_FN_ECHO)
+        self.assertEqual(parsed.params, params)
+
+    def test_queue_element_checksum_verifies_and_detects_tamper(self):
+        rpc = build_rpc_message(RPC_FN_ECHO, bytes(range(32)), sequence=1)
+        element = build_queue_element(rpc, sequence=1)
+        self.assertEqual(compute_message_checksum(element), 0)
+        tampered = bytearray(element)
+        tampered[0x50] ^= 0x01
+        self.assertNotEqual(compute_message_checksum(bytes(tampered)), 0)
+        with self.assertRaises(HostTrafficError):
+            parse_queue_element(bytes(tampered))
+
+    def test_multi_element_message_roundtrip(self):
+        params = bytes((i * 7) & 0xFF for i in range(5000))
+        rpc = build_rpc_message(RPC_FN_ECHO, params, sequence=1)
+        element = build_queue_element(rpc, sequence=1)
+        self.assertEqual(element[0x28:0x2C], (2).to_bytes(4, "little"))
+        prefix, parsed = parse_queue_element(element)
+        self.assertEqual(prefix["elem_count"], 2)
+        self.assertEqual(parsed.params, params)
+        host = HostTraffic()
+        host.submit_rpc(RPC_FN_ECHO, params, sequence=1)
+        self.assertEqual(host.poll(), 1)
+        reply = host.read_reply()
+        self.assertEqual(reply["params"], params)
+        self.assertEqual(host.queues.status.header().write_ptr, 2)
+        host.close()
+
+
+class MsgqTests(unittest.TestCase):
+    def test_queue_header_layout_matches_msgq_arithmetic(self):
+        mem = RegionMemory(SHARED_MEM_BASE, 0x40000)
+        queue = MsgqQueue(mem, SHARED_MEM_BASE, 0x40000)
+        queue.format()
+        header = queue.header()
+        self.assertEqual(header.version, 0)
+        self.assertEqual(header.size, 0x40000)
+        self.assertEqual(header.msg_size, GSP_MSG_QUEUE_ELEMENT_SIZE_MIN)
+        self.assertEqual(header.msg_count, (0x40000 - MSGQ_ENTRY_OFF) // 0x1000)
+        self.assertEqual(header.write_ptr, 0)
+        self.assertEqual(header.flags, MSGQ_FLAGS_SWAP_RX)
+        self.assertEqual(header.rx_hdr_off, MSGQ_RX_HDR_OFF)
+        self.assertEqual(header.entry_off, MSGQ_ENTRY_OFF)
+        self.assertEqual(MSGQ_RX_HDR_OFF, 32)
+        self.assertEqual(MSGQ_ENTRY_OFF, 48)
+
+    def test_write_publish_and_read_back_entry(self):
+        mem = RegionMemory(SHARED_MEM_BASE, 0x40000)
+        queue = MsgqQueue(mem, SHARED_MEM_BASE, 0x40000)
+        queue.format()
+        payload = bytes(range(256))
+        queue.write_entry(0, payload)
+        self.assertEqual(queue.read_entry(0, 256), payload)
+        queue.publish(1)
+        self.assertEqual(queue.write_ptr, 1)
+        queue.set_read(1)
+        self.assertEqual(queue.read_ptr, 1)
+
+    def test_shared_queues_cross_link_rx_headers(self):
+        queues = SharedQueues(RegionMemory(SHARED_MEM_BASE, 0x80000))
+        queues.format()
+        self.assertEqual(queues.command.rx_hdr_addr, queues.status.tx_base + MSGQ_RX_HDR_OFF)
+        self.assertEqual(queues.status.rx_hdr_addr, queues.command.tx_base + MSGQ_RX_HDR_OFF)
+        queues.command.set_read(5)
+        raw = queues.mem.read(queues.status.tx_base + MSGQ_RX_HDR_OFF, 4)
+        self.assertEqual(int.from_bytes(raw, "little"), 5)
+        queues.status.set_read(2)
+        raw = queues.mem.read(queues.command.tx_base + MSGQ_RX_HDR_OFF, 4)
+        self.assertEqual(int.from_bytes(raw, "little"), 2)
+
+    def test_submit_and_take_roundtrip(self):
+        queues = SharedQueues(RegionMemory(SHARED_MEM_BASE, 0x80000))
+        queues.format()
+        element = build_queue_element(
+            build_rpc_message(RPC_FN_ECHO, b"\xaa\xbb", sequence=4), sequence=0
+        )
+        first = queues.submit_element(element)
+        self.assertEqual(first["index"], 0)
+        self.assertEqual(queues.command.write_ptr, 1)
+        taken = queues.take_command()
+        self.assertIsNotNone(taken)
+        self.assertEqual(taken["element"], element)
+        self.assertEqual(queues.take_command(), None)
+        published = queues.publish_element(element)
+        self.assertEqual(published["index"], 0)
+        status = queues.take_status()
+        self.assertEqual(status["element"], element)
+        self.assertEqual(queues.take_status(), None)
+
+
+class DoorbellTests(unittest.TestCase):
+    def test_model_records_writes(self):
+        host = HostTraffic()
+        host.doorbell.ring(0, 0)
+        host.doorbell.ring(0, 0x11)
+        host.doorbell.ring(2, 0x22)
+        self.assertEqual(host.doorbell.head(0), 0x11)
+        self.assertEqual(host.doorbell.head(2), 0x22)
+        self.assertEqual(host.doorbell.kicks, 3)
+        state = host.doorbell.state()
+        self.assertEqual(len(state["writes"]), 3)
+        self.assertEqual(state["writes"][-1]["queue"], 2)
+        with self.assertRaises(ValueError):
+            host.doorbell.ring(8, 0)
+        host.close()
+
+    def test_block_serves_guest_read(self):
+        host = HostTraffic()
+        host.doorbell.ring(1, 0xCAFEF00D)
+        block = host.doorbell_block()
+        self.assertEqual(block.read(0, 4), 0)
+        self.assertEqual(block.read(8, 4), 0xCAFEF00D)
+        self.assertIsNone(block.read(8 * 8, 4))
+        self.assertEqual(block.base, DOORBELL_BASE)
+        self.assertEqual(block.size, DOORBELL_SIZE)
+        host.close()
+
+    def test_registry_overlay_beats_wider_falcon_window(self):
+        registry = DeviceRegistry()
+        registry.add(FalconBlock(), FALCON_BASE, 0x1000)
+        host = HostTraffic()
+        host.doorbell.ring(0, 0x1234)
+        registry.add(host.doorbell_block())
+        handled, value = registry.read(DOORBELL_BASE, 4)
+        self.assertTrue(handled)
+        self.assertEqual(value, 0x1234)
+        handled, value = registry.read(FALCON_BASE + FALCON_DEBUGINFO, 4)
+        self.assertTrue(handled)
+        self.assertEqual(value, 0)
+        host.close()
+
+    def test_guest_load_reads_doorbell(self):
+        machine = Machine(memory_mb=8, ram_base=0)
+        machine.load_segments([(0, PROGRAM_READ_DOORBELL)], entry=0)
+        host = HostTraffic()
+        host.doorbell.ring(0, 0x5A5A5A5A)
+        registry = DeviceRegistry()
+        registry.add(host.doorbell_block())
+        machine.devices = registry
+        reason, _pc, _a, _b = machine.run(100)
+        self.assertEqual(reason, R_EBREAK)
+        self.assertEqual(machine.x[10], 0x5A5A5A5A)
+        host.close()
+
+    def test_guest_store_is_recorded(self):
+        machine = Machine(memory_mb=8, ram_base=0)
+        machine.load_segments([(0, PROGRAM_WRITE_DOORBELL)], entry=0)
+        machine.apply_reset_regs({"a1": 0x77})
+        host = HostTraffic()
+        registry = DeviceRegistry()
+        registry.add(host.doorbell_block())
+        machine.devices = registry
+        reason, _pc, _a, _b = machine.run(100)
+        self.assertEqual(reason, R_EBREAK)
+        self.assertEqual(host.doorbell.head(2), 0x77)
+        host.close()
+
+
+class TrafficTests(unittest.TestCase):
+    def test_submit_writes_ring_and_rings_doorbell(self):
+        host = HostTraffic()
+        raw = build_rpc_message(RPC_FN_ECHO, b"\x01\x02\x03", sequence=5)
+        result = host.submit(raw)
+        self.assertEqual(result["function"], RPC_FN_ECHO)
+        self.assertEqual(result["sequence"], 5)
+        self.assertEqual(result["index"], 0)
+        self.assertEqual(result["write_ptr"], 1)
+        self.assertEqual(host.queues.command.write_ptr, 1)
+        self.assertEqual(host.doorbell.kicks, 1)
+        self.assertEqual(host.doorbell.head(0), 0)
+        taken = host.queues.take_command()
+        self.assertIsNotNone(taken)
+        _prefix, parsed = parse_queue_element(taken["element"])
+        self.assertEqual(parsed.params, b"\x01\x02\x03")
+        host.close()
+
+    def test_poll_dispatches_echo(self):
+        host = HostTraffic()
+        host.submit_rpc(RPC_FN_ECHO, b"\xaa\xbb\xcc", sequence=1)
+        self.assertEqual(host.poll(), 1)
+        self.assertEqual(host.poll(), 0)
+        reply = host.read_reply()
+        self.assertIsNotNone(reply)
+        self.assertEqual(reply["function"], RPC_FN_ECHO)
+        self.assertEqual(reply["sequence"], 1)
+        self.assertEqual(reply["result"], NV_OK)
+        self.assertEqual(reply["params"], b"\xaa\xbb\xcc")
+        self.assertEqual(host.read_reply(), None)
+        host.close()
+
+    def test_unknown_function_gets_not_supported(self):
+        host = HostTraffic()
+        host.submit_rpc(0x999, b"\x01", sequence=2)
+        self.assertEqual(host.poll(), 1)
+        reply = host.read_reply()
+        self.assertEqual(reply["function"], 0x999)
+        self.assertEqual(reply["result"], NV_ERR_NOT_SUPPORTED)
+        self.assertEqual(reply["params"], b"")
+        dispatched = [
+            event for event in host.transcript()["events"]
+            if event["type"] == "dispatch"
+        ]
+        self.assertEqual(dispatched[0]["handler"], "stub:unsupported")
+        host.close()
+
+    def test_status_query_reply_is_deterministic(self):
+        replies = []
+        for _ in range(2):
+            host = HostTraffic()
+            host.submit_rpc(RPC_FN_ECHO, b"\x01", sequence=1)
+            host.submit_rpc(RPC_FN_STATUS_QUERY, b"", sequence=2)
+            host.poll()
+            first = host.read_reply()
+            second = host.read_reply()
+            replies.append((first["params"], second["params"]))
+            host.close()
+        self.assertEqual(replies[0], replies[1])
+        self.assertEqual(replies[0][1][:4], STATUS_MAGIC)
+
+    def test_empty_poll_and_read_reply(self):
+        host = HostTraffic()
+        self.assertEqual(host.poll(), 0)
+        self.assertIsNone(host.read_reply())
+        host.close()
+
+    def test_run_cycles_is_bounded(self):
+        machine = Machine(memory_mb=32)
+        host = HostTraffic(machine=machine)
+        host.submit_rpc(RPC_FN_ECHO, b"\x01", sequence=1)
+        result = host.run_cycles(64)
+        self.assertEqual(result["budget"], 64)
+        self.assertEqual(result["insns"], machine.insns)
+        self.assertEqual(result["dispatched"], 1)
+        self.assertEqual(result["replies"], 1)
+        with self.assertRaises(ValueError):
+            host.run_cycles(0)
+        with self.assertRaises(ValueError):
+            host.run_cycles(-1)
+        host.close()
+
+    def test_oversized_message_is_refused(self):
+        host = HostTraffic()
+        with self.assertRaises(HostTrafficError):
+            host.submit_rpc(RPC_FN_ECHO, bytes(GSP_MSG_QUEUE_ELEMENT_SIZE_MIN * 16),
+                            sequence=1)
+        host.close()
+
+    def test_transcript_is_deterministic(self):
+        def run():
+            host = HostTraffic()
+            host.submit_rpc(RPC_FN_ECHO, b"\x10\x20", sequence=1)
+            host.submit_rpc(RPC_FN_STATUS_QUERY, b"", sequence=2)
+            host.submit_rpc(0x999, b"", sequence=3)
+            host.poll()
+            while host.read_reply() is not None:
+                pass
+            transcript = host.transcript()
+            host.close()
+            return transcript
+
+        first, second = run(), run()
+        self.assertEqual(first, second)
+        self.assertEqual(first["format"], "greeneye-host-traffic-log/1")
+        self.assertEqual(
+            [event["type"] for event in first["events"]],
+            ["submit", "submit", "submit", "dispatch", "dispatch", "dispatch",
+             "reply", "reply", "reply"],
+        )
+
+
+class ReplayTests(unittest.TestCase):
+    def test_replay_stimulus_in_order(self):
+        stimulus = {
+            "format": "greeneye-rm-rpc-stimulus/1",
+            "messages": [
+                {
+                    "seq": 0,
+                    "kind": "send",
+                    "fn": RPC_FN_ECHO,
+                    "params_hex": "0102",
+                    "truncated": False,
+                },
+                {
+                    "seq": 1,
+                    "kind": "send",
+                    "fn": RPC_FN_ECHO,
+                    "params_hex": "0304",
+                    "truncated": False,
+                },
+            ],
+        }
+        host = HostTraffic()
+        report = replay_stimulus(host, stimulus)
+        self.assertEqual(report["format"], REPLAY_FORMAT)
+        self.assertEqual(report["submitted"], 2)
+        self.assertEqual(report["skipped"], [])
+        self.assertEqual([pair["reply"]["params"] for pair in report["pairs"]],
+                         [b"\x01\x02", b"\x03\x04"])
+        host.close()
+
+    def test_replay_strict_rejects_truncated(self):
+        stimulus = {
+            "format": "greeneye-rm-rpc-stimulus/1",
+            "messages": [
+                {
+                    "seq": 0,
+                    "kind": "control",
+                    "params_hex": "00",
+                    "truncated": True,
+                }
+            ],
+        }
+        host = HostTraffic()
+        with self.assertRaises(HostTrafficError):
+            replay_stimulus(host, stimulus, strict=True)
+        host.close()
+
+    def test_replay_lenient_skips_truncated(self):
+        stimulus = {
+            "format": "greeneye-rm-rpc-stimulus/1",
+            "messages": [
+                {
+                    "seq": 0,
+                    "kind": "send",
+                    "fn": RPC_FN_ECHO,
+                    "params_hex": "01",
+                    "truncated": False,
+                },
+                {
+                    "seq": 1,
+                    "kind": "control",
+                    "params_hex": "00",
+                    "truncated": True,
+                },
+            ],
+        }
+        host = HostTraffic()
+        report = replay_stimulus(host, stimulus, strict=False)
+        self.assertEqual(report["submitted"], 1)
+        self.assertEqual(len(report["skipped"]), 1)
+        self.assertIn("truncated", report["skipped"][0]["reason"])
+        host.close()
+
+    def test_control_stimulus_maps_to_control_function(self):
+        stimulus = {
+            "format": "greeneye-rm-rpc-stimulus/1",
+            "messages": [
+                {
+                    "seq": 0,
+                    "kind": "control",
+                    "params_hex": "01020304",
+                    "truncated": False,
+                }
+            ],
+        }
+        host = HostTraffic()
+        report = replay_stimulus(host, stimulus)
+        self.assertEqual(report["pairs"][0]["reply"]["function"], GSP_RM_CONTROL)
+        self.assertEqual(report["pairs"][0]["reply"]["result"], NV_ERR_NOT_SUPPORTED)
+        host.close()
+
+    def test_load_stimulus_from_capture_document(self):
+        from harness.rm_capture import (
+            capture_to_dict,
+            decode_text,
+            stimulus_from_capture,
+        )
+
+        capture = decode_text(_sample_capture_log(), source="fixture")
+        self.assertIsNotNone(stimulus_from_capture(capture))
+        with tempfile.TemporaryDirectory() as tmp:
+            capture_path = os.path.join(tmp, "capture.json")
+            with open(capture_path, "w", encoding="utf-8") as fh:
+                json.dump(capture_to_dict(capture), fh)
+            loaded = load_stimulus(capture_path)
+        self.assertEqual(loaded["format"], "greeneye-rm-rpc-stimulus/1")
+        self.assertEqual(len(loaded["messages"]), 2)
+
+    def test_load_stimulus_accepts_inline_document(self):
+        stimulus = {"format": "greeneye-rm-rpc-stimulus/1", "messages": []}
+        self.assertEqual(load_stimulus(stimulus), stimulus)
+        with self.assertRaises(HostTrafficError):
+            load_stimulus({"format": "not-a-stimulus"})
+
+
+class SessionTests(unittest.TestCase):
+    def test_recorder_events_and_blob(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            recorder = Recorder(tmp, name="host-traffic-test", meta={"kind": "test"})
+            host = HostTraffic(recorder=recorder)
+            host.submit_rpc(RPC_FN_ECHO, b"\x01", sequence=1)
+            host.poll()
+            host.read_reply()
+            host.close()
+            recorder.close()
+            events = load_events(tmp)
+            types = [event["type"] for event in events]
+            self.assertIn("host_traffic_config", types)
+            self.assertIn("doorbell", types)
+            self.assertIn("host_submit", types)
+            self.assertIn("host_dispatch", types)
+            self.assertIn("host_reply_read", types)
+            blob_path = os.path.join(tmp, "blobs", "host-traffic.json")
+            self.assertTrue(os.path.isfile(blob_path))
+            with open(blob_path, "r", encoding="utf-8") as fh:
+                payload = json.load(fh)
+            self.assertEqual(payload["format"], "greeneye-host-traffic-log/1")
+
+
+class DemoTests(unittest.TestCase):
+    def test_demo_end_to_end_and_deterministic(self):
+        first, second = run_demo(), run_demo()
+        self.assertTrue(first["ok"], first.get("checks"))
+        self.assertEqual(first["format"], DEMO_FORMAT)
+        self.assertEqual(first["replies"], second["replies"])
+        self.assertEqual(first["transcript"], second["transcript"])
+        self.assertEqual(first["control"]["bytes"], 512)
+        self.assertEqual(first["control"]["reply_result"], NV_ERR_NOT_SUPPORTED)
+        self.assertTrue(all(reply["read"] for reply in first["replies"]))
+
+    def test_demo_writes_artifacts(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            result = run_demo(out_dir=tmp)
+            self.assertTrue(result["ok"], result.get("checks"))
+            self.assertTrue(os.path.isfile(os.path.join(tmp, "blobs", "host-traffic.json")))
+            self.assertTrue(os.path.isfile(os.path.join(tmp, "host-traffic-demo.json")))
+            self.assertTrue(os.path.isfile(os.path.join(tmp, "session.json")))
+
+    def test_cli_demo(self):
+        import contextlib
+        import io
+
+        with tempfile.TemporaryDirectory() as tmp:
+            out = io.StringIO()
+            with contextlib.redirect_stdout(out):
+                code = main(["demo", "--out", tmp])
+            self.assertEqual(code, 0)
+            payload = json.loads(out.getvalue())
+            self.assertEqual(payload["format"], DEMO_FORMAT)
+            self.assertTrue(payload["ok"])
+
+    def test_cli_replay(self):
+        import contextlib
+        import io
+
+        stimulus = {
+            "format": "greeneye-rm-rpc-stimulus/1",
+            "messages": [
+                {
+                    "seq": 0,
+                    "kind": "send",
+                    "fn": RPC_FN_ECHO,
+                    "params_hex": "0102",
+                    "truncated": False,
+                }
+            ],
+        }
+        with tempfile.TemporaryDirectory() as tmp:
+            path = os.path.join(tmp, "stimulus.json")
+            with open(path, "w", encoding="utf-8") as fh:
+                json.dump(stimulus, fh)
+            out = io.StringIO()
+            with contextlib.redirect_stdout(out):
+                code = main(["replay", "--stimulus", path, "--out", tmp])
+            self.assertEqual(code, 0)
+            payload = json.loads(out.getvalue())
+            self.assertEqual(payload["format"], REPLAY_FORMAT)
+
+    def test_cli_requires_a_subcommand(self):
+        import contextlib
+        import io
+
+        err = io.StringIO()
+        with contextlib.redirect_stderr(err):
+            code = main([])
+        self.assertEqual(code, 2)
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/debug/risc-v-gsp-rm-vm/tests/test_launch_audit.py b/debug/risc-v-gsp-rm-vm/tests/test_launch_audit.py
new file mode 100644
index 0000000..55dce3e
--- /dev/null
+++ b/debug/risc-v-gsp-rm-vm/tests/test_launch_audit.py
@@ -0,0 +1,477 @@
+"""Tests for the launch memory-contract audit (harness/launch_audit.py, docs/48).
+
+Offline only: synthetic page-table trees plus the captured channel-PDB vector.
+
+Run::
+
+    python3 -m unittest tests.test_launch_audit -v
+"""
+from __future__ import annotations
+
+import contextlib
+import io
+import json
+import os
+import struct
+import sys
+import tempfile
+import unittest
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+TOOL_ROOT = os.path.abspath(os.path.join(HERE, ".."))
+if TOOL_ROOT not in sys.path:
+    sys.path.insert(0, TOOL_ROOT)
+
+from harness.launch_audit import (  # noqa: E402
+    AUDIT_FORMAT,
+    CHECK_FORMAT,
+    SPEC_FORMAT,
+    VERDICT_ACCESS_DENIED,
+    VERDICT_OK,
+    VERDICT_PA_MISMATCH,
+    VERDICT_UNMAPPED,
+    AuditError,
+    _tree_for_va,
+    audit,
+    check,
+    main,
+    run_demo,
+)
+from sim.gmmu_walk import (  # noqa: E402
+    MemoryView,
+    encode_dual_slot,
+    encode_pde_pointer,
+    encode_pte,
+)
+
+
+def _spec(root_pa: int, mem: MemoryView, contract: list, **extra) -> dict:
+    spec = {
+        "format": SPEC_FORMAT,
+        "name": "test",
+        "gmmu": "ver2",
+        "page_mode": "4k",
+        "pdb_pa": f"{root_pa:#x}",
+        "memory": [{"pa": f"{pa:#x}", "hex": b.hex()} for pa, b in _blobs(mem)],
+        "contract": contract,
+    }
+    spec.update(extra)
+    return spec
+
+
+def _blobs(mem: MemoryView):
+    return [(int(item["pa"], 16), mem.read(int(item["pa"], 16), item["size"]))
+            for item in mem.spans()]
+
+
+def _drop(mem: MemoryView, drop_pas) -> MemoryView:
+    """Copy a memory view without the given table pages (a partial dump)."""
+    out = MemoryView()
+    for pa, blob in _blobs(mem):
+        if pa not in drop_pas:
+            out.add(pa, blob)
+    return out
+
+
+def _run_main(argv):
+    out, err = io.StringIO(), io.StringIO()
+    with contextlib.redirect_stdout(out), contextlib.redirect_stderr(err):
+        rc = main(argv)
+    return rc, out.getvalue(), err.getvalue()
+
+
+class AuditTests(unittest.TestCase):
+    def test_shadow_maps_to_expected_pa_is_ok(self):
+        va, shadow = 0x204608000, 0x15E8FA000
+        root = 0x60000000
+        report = audit(_spec(root, _tree_for_va(root, va, shadow), [
+            {"label": "qmd-dest", "va": f"{va:#x}", "access": "write",
+             "expect_pa": f"{shadow:#x}"},
+        ]))
+        self.assertEqual(report["format"], AUDIT_FORMAT)
+        row = report["contract"][0]
+        self.assertEqual(row["verdict"], VERDICT_OK)
+        self.assertTrue(report["ok"])
+
+    def test_host_checks_guest_page_is_pa_mismatch(self):
+        va, shadow = 0x204608000, 0x15E8FA000
+        root = 0x60000000
+        report = audit(_spec(root, _tree_for_va(root, va, shadow), [
+            {"label": "qmd-dest", "va": f"{va:#x}", "access": "write",
+             "expect_pa": f"{va:#x}"},
+        ]))
+        row = report["contract"][0]
+        self.assertEqual(row["verdict"], VERDICT_PA_MISMATCH)
+        self.assertFalse(report["ok"])
+        self.assertIn("0x15e8fa000", row["detail"])
+
+    def test_unmapped_leaf_is_unmapped(self):
+        va = 0x204608000
+        root = 0x60000000
+        report = audit(_spec(root, _tree_for_va(root, va, None), [
+            {"label": "qmd-dest", "va": f"{va:#x}", "access": "write"},
+        ]))
+        self.assertEqual(report["contract"][0]["verdict"], VERDICT_UNMAPPED)
+        self.assertFalse(report["ok"])
+
+    def test_read_only_leaf_for_write_is_access_denied(self):
+        va, target = 0x204608000, 0x15E8FA000
+        # build the same tree as _tree_for_va but with a read-only leaf
+        i0, i1 = (va >> 47) & 0x3, (va >> 38) & 0x1FF
+        i2, i3 = (va >> 29) & 0x1FF, (va >> 21) & 0xFF
+        i4 = (va >> 12) & 0x1FF
+        base = 0x50000000
+        pd2, pd1, pd0, pt = base + 0x1000, base + 0x2000, base + 0x3000, base + 0x4000
+        mem = MemoryView()
+        page = bytearray(0x1000)
+        struct.pack_into("<Q", page, i0 * 8, encode_pde_pointer(pd2))
+        mem.add(base, bytes(page))
+        page = bytearray(0x1000)
+        struct.pack_into("<Q", page, i1 * 8, encode_pde_pointer(pd1))
+        mem.add(pd2, bytes(page))
+        page = bytearray(0x1000)
+        struct.pack_into("<Q", page, i2 * 8, encode_pde_pointer(pd0))
+        mem.add(pd1, bytes(page))
+        page = bytearray(0x1000)
+        struct.pack_into("<16s", page, i3 * 16, encode_dual_slot(None, pt))
+        mem.add(pd0, bytes(page))
+        page = bytearray(0x1000)
+        struct.pack_into("<Q", page, i4 * 8, encode_pte(target, read_only=True))
+        mem.add(pt, bytes(page))
+        root = 0x60000000  # separate root alias is not needed; spec uses 'base'
+        report = audit(_spec(base, mem, [
+            {"label": "release-slot", "va": f"{va:#x}", "access": "write"},
+        ]))
+        self.assertEqual(report["contract"][0]["verdict"], VERDICT_ACCESS_DENIED)
+        self.assertFalse(report["ok"])
+        self.assertNotEqual(root, base)
+
+    def test_spec_errors(self):
+        with self.assertRaises(AuditError):
+            audit({"format": "nope"})
+        with self.assertRaises(AuditError):
+            audit({"format": SPEC_FORMAT})
+        with self.assertRaises(AuditError):
+            audit(_spec(0x60000000, MemoryView(), [], gmmu="ver9"))
+        with self.assertRaises(AuditError):
+            audit(_spec(0x60000000, MemoryView(), [{"label": "x"}]))
+
+    def test_multiple_contract_legs_counted(self):
+        va, shadow = 0x204608000, 0x15E8FA000
+        root = 0x60000000
+        report = audit(_spec(root, _tree_for_va(root, va, shadow), [
+            {"label": "qmd", "va": f"{va:#x}", "access": "write",
+             "expect_pa": f"{shadow:#x}"},
+            {"label": "missing", "va": f"{va + 0x1000:#x}", "access": "read"},
+        ]))
+        self.assertEqual(report["summary"][VERDICT_OK], 1)
+        self.assertEqual(report["summary"][VERDICT_UNMAPPED], 1)
+        self.assertFalse(report["ok"])
+
+
+class CheckTests(unittest.TestCase):
+    """The `check` mode: which table pages does a partial dump still need?"""
+
+    def test_check_empty_memory_reports_root_per_leg(self):
+        spec = {
+            "format": SPEC_FORMAT,
+            "name": "empty-memory",
+            "gmmu": "ver2",
+            "page_mode": "4k",
+            "pdb_pa": "0x12f16e000",
+            "contract": [
+                {"label": "qmd", "va": "0x204608000", "access": "write"},
+                {"label": "ring", "va": "0x100000", "access": "read"},
+            ],
+        }
+        report = check(spec)
+        self.assertEqual(report["format"], CHECK_FORMAT)
+        self.assertEqual(
+            set(report),
+            {"format", "name", "firmware", "gmmu", "page_mode", "pdb_pa",
+             "memory_pages", "contract", "needed_pages", "complete", "limits"})
+        self.assertEqual(report["memory_pages"], 0)
+        self.assertFalse(report["complete"])
+        for row in report["contract"]:
+            self.assertEqual(row["status"], "missing")
+            self.assertEqual(row["first_missing"]["pa"], "0x12f16e000")
+            self.assertEqual(row["first_missing"]["table_pa"], "0x12f16e000")
+            self.assertEqual(row["first_missing"]["level"], "root")
+            self.assertEqual(row["first_missing"]["depth"], 0)
+            self.assertFalse(row["first_missing"]["present"])
+            self.assertEqual(row["path"][-1]["pa"], "0x12f16e000")
+            self.assertFalse(row["path"][-1]["present"])
+        needs = report["needed_pages"]
+        self.assertEqual([page["pa"] for page in needs], ["0x12f16e000"])
+        self.assertEqual(needs[0]["levels"], ["root"])
+        self.assertEqual(needs[0]["legs"], ["qmd", "ring"])
+
+    def test_check_partial_memory_reports_first_missing_with_path(self):
+        va = 0x204608000
+        root = 0x60000000
+        pde1 = 0x50000000 + 0x2000
+        full = _tree_for_va(root, va, 0x12345000)
+        partial = _drop(full, {pde1, 0x50000000 + 0x4000})
+        report = check(_spec(root, partial, [
+            {"label": "qmd", "va": f"{va:#x}", "access": "write"},
+        ]))
+        row = report["contract"][0]
+        self.assertEqual(row["status"], "missing")
+        missing = row["first_missing"]
+        index = (va >> 29) & 0x1FF
+        self.assertEqual(missing["pa"], f"{pde1:#x}")
+        self.assertEqual(missing["table_pa"], f"{pde1:#x}")
+        self.assertEqual(missing["level"], "pde1")
+        self.assertEqual(missing["depth"], 2)
+        self.assertEqual(missing["index"], index)
+        self.assertEqual(missing["entry_pa"], f"{pde1 + index * 8:#x}")
+        self.assertEqual([hop["level"] for hop in row["path"]],
+                         ["root", "pde2", "pde1"])
+        self.assertEqual(row["path"][0]["table_pa"], f"{root:#x}")
+        self.assertTrue(row["path"][0]["present"])
+        self.assertFalse(row["path"][-1]["present"])
+        self.assertEqual(report["needed_pages"][0]["pa"], f"{pde1:#x}")
+        self.assertEqual(report["needed_pages"][0]["legs"], ["qmd"])
+
+    def test_check_partial_memory_marks_complete_and_missing_legs(self):
+        root = 0x60000000
+        base = 0x51000000
+        pd2, pd1 = base + 0x1000, base + 0x2000
+        pd0_done, pt_done = base + 0x3000, base + 0x4000
+        pd0_missing = base + 0x5000
+        target = 0x12345000
+        va_done, va_miss = 0x100000, 0x204608000
+
+        def idx(va):
+            return ((va >> 47) & 3, (va >> 38) & 0x1FF, (va >> 29) & 0x1FF,
+                    (va >> 21) & 0xFF, (va >> 12) & 0x1FF)
+
+        i0d, i1d, i2d, i3d, i4d = idx(va_done)
+        i0m, i1m, i2m, _i3m, _i4m = idx(va_miss)
+        self.assertEqual((i0d, i1d), (i0m, i1m))  # share root[i0] and pd2[i1]
+        self.assertNotEqual(i2d, i2m)
+
+        mem = MemoryView()
+        page = bytearray(0x1000)
+        struct.pack_into("<Q", page, i0d * 8, encode_pde_pointer(pd2))
+        mem.add(root, bytes(page))
+        page = bytearray(0x1000)
+        struct.pack_into("<Q", page, i1d * 8, encode_pde_pointer(pd1))
+        mem.add(pd2, bytes(page))
+        page = bytearray(0x1000)
+        struct.pack_into("<Q", page, i2d * 8, encode_pde_pointer(pd0_done))
+        struct.pack_into("<Q", page, i2m * 8, encode_pde_pointer(pd0_missing))
+        mem.add(pd1, bytes(page))
+        page = bytearray(0x1000)
+        struct.pack_into("<16s", page, i3d * 16, encode_dual_slot(None, pt_done))
+        mem.add(pd0_done, bytes(page))
+        page = bytearray(0x1000)
+        struct.pack_into("<Q", page, i4d * 8, encode_pte(target))
+        mem.add(pt_done, bytes(page))
+
+        report = check(_spec(root, mem, [
+            {"label": "done", "va": f"{va_done:#x}", "access": "read"},
+            {"label": "miss", "va": f"{va_miss:#x}", "access": "write"},
+        ]))
+        rows = {row["label"]: row for row in report["contract"]}
+        self.assertEqual(rows["done"]["status"], "complete")
+        self.assertEqual(rows["done"]["resolved_pa"], f"{target:#x}")
+        self.assertEqual(rows["miss"]["status"], "missing")
+        self.assertEqual(rows["miss"]["first_missing"]["pa"], f"{pd0_missing:#x}")
+        self.assertEqual(rows["miss"]["first_missing"]["level"], "pde0")
+        self.assertEqual(rows["miss"]["first_missing"]["index"], idx(va_miss)[3])
+        self.assertEqual([page["pa"] for page in report["needed_pages"]],
+                         [f"{pd0_missing:#x}"])
+        self.assertFalse(report["complete"])
+
+    def test_check_needed_pages_dedup_and_ordering(self):
+        root = 0x60000000
+        pd2_a, pd2_b = 0x70001000, 0x70000000
+        va_a1, va_a2 = 0x204608000, 0x100000
+        va_b = 0x800000000000
+        self.assertEqual((va_a1 >> 47) & 3, 0)
+        self.assertEqual((va_a2 >> 47) & 3, 0)
+        self.assertEqual((va_b >> 47) & 3, 1)
+        mem = MemoryView()
+        page = bytearray(0x1000)
+        struct.pack_into("<Q", page, 0, encode_pde_pointer(pd2_a))
+        struct.pack_into("<Q", page, 8, encode_pde_pointer(pd2_b))
+        mem.add(root, bytes(page))
+        report = check(_spec(root, mem, [
+            {"label": "b-leg", "va": f"{va_b:#x}", "access": "read"},
+            {"label": "a1-leg", "va": f"{va_a1:#x}", "access": "write"},
+            {"label": "a2-leg", "va": f"{va_a2:#x}", "access": "read"},
+        ]))
+        pages = report["needed_pages"]
+        self.assertEqual([page["pa"] for page in pages],
+                         [f"{pd2_b:#x}", f"{pd2_a:#x}"])
+        self.assertEqual(pages[0]["levels"], ["pde2"])
+        self.assertEqual(pages[0]["legs"], ["b-leg"])
+        self.assertEqual(pages[1]["levels"], ["pde2"])
+        self.assertEqual(pages[1]["legs"], ["a1-leg", "a2-leg"])
+        self.assertFalse(report["complete"])
+
+    def test_check_fault_leg_needs_no_pages(self):
+        va = 0x204608000
+        root = 0x60000000
+        mem = _tree_for_va(root, va, None)  # zero leaf entry -> pte_invalid
+        report = check(_spec(root, mem, [
+            {"label": "qmd", "va": f"{va:#x}", "access": "write"},
+        ]))
+        row = report["contract"][0]
+        self.assertEqual(row["status"], "fault")
+        self.assertEqual(row["fault"], "pte_invalid")
+        self.assertIsNone(row["first_missing"])
+        self.assertEqual(report["needed_pages"], [])
+        self.assertTrue(report["complete"])
+
+    def test_check_64k_dual_missing_big_table(self):
+        va = 0x204600000
+        root, pd2, pd1, pd0 = 0x60000000, 0x50001000, 0x50002000, 0x50003000
+        big_pt, small_pt = 0x80000000, 0x80001000
+        i0, i1, i2 = (va >> 47) & 3, (va >> 38) & 0x1FF, (va >> 29) & 0x1FF
+        i3 = (va >> 21) & 0xFF
+        mem = MemoryView()
+        page = bytearray(0x1000)
+        struct.pack_into("<Q", page, i0 * 8, encode_pde_pointer(pd2))
+        mem.add(root, bytes(page))
+        page = bytearray(0x1000)
+        struct.pack_into("<Q", page, i1 * 8, encode_pde_pointer(pd1))
+        mem.add(pd2, bytes(page))
+        page = bytearray(0x1000)
+        struct.pack_into("<Q", page, i2 * 8, encode_pde_pointer(pd0))
+        mem.add(pd1, bytes(page))
+        page = bytearray(0x1000)
+        struct.pack_into("<16s", page, i3 * 16,
+                         encode_dual_slot(big_pt, small_pt))
+        mem.add(pd0, bytes(page))
+        report = check(_spec(root, mem, [
+            {"label": "prog", "va": f"{va:#x}", "access": "read"},
+        ], page_mode="64k"))
+        row = report["contract"][0]
+        self.assertEqual(row["status"], "missing")
+        missing = row["first_missing"]
+        self.assertEqual(missing["pa"], f"{big_pt:#x}")
+        self.assertEqual(missing["level"], "pte")
+        self.assertEqual(missing["depth"], 4)
+        self.assertEqual(missing["index"], (va >> 16) & 0x1FF)
+
+    def test_check_is_deterministic(self):
+        va = 0x204608000
+        root = 0x60000000
+        full = _tree_for_va(root, va, 0x12345000)
+        partial = _drop(full, {0x50002000})
+        spec = _spec(root, partial, [
+            {"label": "qmd", "va": f"{va:#x}", "access": "write"},
+        ])
+        self.assertEqual(json.dumps(check(spec), sort_keys=True),
+                         json.dumps(check(spec), sort_keys=True))
+
+    def test_check_spec_errors(self):
+        with self.assertRaises(AuditError):
+            check({"format": SPEC_FORMAT})
+        with self.assertRaises(AuditError):
+            check({"format": "nope", "pdb_pa": "0x1000", "contract": []})
+        with self.assertRaises(AuditError):
+            check(_spec(0x60000000, MemoryView(), [], gmmu="ver9"))
+        with self.assertRaises(AuditError):
+            check(_spec(0x60000000, MemoryView(), [{"label": "x"}]))
+
+    def test_cli_check_exit_codes_and_artifact(self):
+        va = 0x204608000
+        root = 0x60000000
+        full = _tree_for_va(root, va, 0x12345000)
+        partial = _drop(full, {0x50002000})
+        spec = _spec(root, partial, [
+            {"label": "qmd", "va": f"{va:#x}", "access": "write"},
+        ])
+        with tempfile.TemporaryDirectory() as tmp:
+            spec_path = os.path.join(tmp, "spec.json")
+            with open(spec_path, "w", encoding="utf-8") as fh:
+                json.dump(spec, fh)
+            rc, out, _ = _run_main(["check", "--spec", spec_path, "--out", tmp])
+            self.assertEqual(rc, 3)
+            self.assertIn("need 0x50002000", out)
+            self.assertIn("dump next:", out)
+            report_path = os.path.join(tmp, "launch-audit-check.json")
+            self.assertTrue(os.path.isfile(report_path))
+            with open(report_path, encoding="utf-8") as fh:
+                report = json.load(fh)
+            self.assertEqual(report["format"], CHECK_FORMAT)
+            self.assertEqual(report["needed_pages"][0]["pa"], "0x50002000")
+            # a complete view -> exit 0
+            spec["memory"] = [{"pa": f"{pa:#x}", "hex": b.hex()}
+                              for pa, b in _blobs(full)]
+            with open(spec_path, "w", encoding="utf-8") as fh:
+                json.dump(spec, fh)
+            rc, out, _ = _run_main(["check", "--spec", spec_path])
+            self.assertEqual(rc, 0)
+            self.assertIn("complete", out)
+
+
+class DemoAndCliTests(unittest.TestCase):
+    def test_demo_is_green_and_deterministic(self):
+        first = run_demo()
+        second = run_demo()
+        self.assertEqual(first["format"], "greeneye-launch-audit-demo/1")
+        self.assertTrue(first["ok"])
+        self.assertEqual(json.dumps(first, sort_keys=True),
+                         json.dumps(second, sort_keys=True))
+
+    def test_audit_is_deterministic(self):
+        va, shadow = 0x204608000, 0x15E8FA000
+        root = 0x60000000
+        spec = _spec(root, _tree_for_va(root, va, shadow), [
+            {"label": "qmd", "va": f"{va:#x}", "access": "write"},
+        ])
+        self.assertEqual(json.dumps(audit(spec), sort_keys=True),
+                         json.dumps(audit(spec), sort_keys=True))
+
+    def test_cli_demo_and_hints(self):
+        rc, out, _ = _run_main(["demo"])
+        self.assertEqual(rc, 0)
+        self.assertIn("captured channel contract: ok=True", out)
+        self.assertIn("engine-write-shadow-host-checks-guest", out)
+        rc, out, _ = _run_main(["capture-hints"])
+        self.assertEqual(rc, 0)
+        self.assertIn("CAPTURE HINTS", out)
+
+    def test_engine_write_shadow_demo_verdicts(self):
+        demo = run_demo()
+        scenarios = {s["name"]: s for s in demo["engine_write_shadow"]}
+        mismatch = scenarios["engine-write-shadow-host-checks-guest"]
+        aligned = scenarios["engine-write-shadow-bookkeeping-aligned"]
+        self.assertEqual(mismatch["expected_verdict"], VERDICT_PA_MISMATCH)
+        self.assertEqual(
+            mismatch["report"]["contract"][0]["verdict"], VERDICT_PA_MISMATCH)
+        self.assertEqual(aligned["expected_verdict"], VERDICT_OK)
+        self.assertEqual(
+            aligned["report"]["contract"][0]["verdict"], VERDICT_OK)
+        self.assertTrue(demo["ok"])
+        # the original demo sections are still present and unchanged in shape
+        self.assertTrue(demo["captured_channel"]["ok"])
+        self.assertEqual(len(demo["qmd_hypotheses"]), 3)
+
+    def test_cli_run_writes_report_and_exit_codes(self):
+        va, shadow = 0x204608000, 0x15E8FA000
+        root = 0x60000000
+        spec = _spec(root, _tree_for_va(root, va, shadow), [
+            {"label": "qmd", "va": f"{va:#x}", "access": "write"},
+        ])
+        with tempfile.TemporaryDirectory() as tmp:
+            spec_path = os.path.join(tmp, "spec.json")
+            with open(spec_path, "w", encoding="utf-8") as fh:
+                json.dump(spec, fh)
+            rc, out, _ = _run_main(["run", "--spec", spec_path, "--out", tmp])
+            self.assertEqual(rc, 0)
+            self.assertTrue(os.path.isfile(os.path.join(tmp, "launch-audit.json")))
+            # mismatch -> exit 3
+            spec["contract"][0]["expect_pa"] = f"{va:#x}"
+            with open(spec_path, "w", encoding="utf-8") as fh:
+                json.dump(spec, fh)
+            rc, _, _ = _run_main(["run", "--spec", spec_path])
+            self.assertEqual(rc, 3)
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/debug/risc-v-gsp-rm-vm/tests/test_libos_port.py b/debug/risc-v-gsp-rm-vm/tests/test_libos_port.py
new file mode 100644
index 0000000..0f01c01
--- /dev/null
+++ b/debug/risc-v-gsp-rm-vm/tests/test_libos_port.py
@@ -0,0 +1,291 @@
+"""libos port receive-path model tests (sim/libos_port.py, docs/51 section 7).
+
+Hermetic: a fake machine (register file only) and a fake base syscall model;
+no image, no boot, no machine.  The tests pin the port-syscall ABI observed on
+the real 570.144 image (send args a0-a3, receive args a4-a7, wait id t1,
+timeout t2, return a0/a4/a5/a6), the async receive post, the boot handshake
+send, the host doorbell wake's external-interrupt handoff, and the strict
+"never complete an uncompleted receive" rule.
+
+Offline only: no hardware, kernel module, mdev, guest, network or QEMU.
+
+Run::
+
+    cd open-iov/debug/risc-v-gsp-rm-vm
+    python3 -m unittest tests.test_libos_port -v
+"""
+from __future__ import annotations
+
+import json
+import os
+import sys
+import unittest
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+TOOL_ROOT = os.path.abspath(os.path.join(HERE, ".."))
+if TOOL_ROOT not in sys.path:
+    sys.path.insert(0, TOOL_ROOT)
+
+from sim.libos import INTERRUPT_ARM_EXTERNAL, TASK_PORT_SEND_RECV_WAIT  # noqa: E402
+from sim.libos_port import (  # noqa: E402
+    EXTINTR_WAKE_CONTRACT,
+    HANDSHAKE_CONTRACT,
+    REPORT_FORMAT,
+    SHUTTLE_ANY,
+    TIMEOUT_INFINITE,
+    LibosPortModel,
+    role_for,
+)
+
+
+class FakeMachine:
+    """Register file only: the port model never touches memory."""
+
+    def __init__(self):
+        self.x = [0] * 32
+        self.insns = 0
+        self.pc = 0
+
+    def set(self, index: int, value: int) -> None:
+        self.x[index] = value & 0xFFFFFFFFFFFFFFFF
+
+
+class FakeBase:
+    """Minimal base syscall model: handles arm, records and stops the rest."""
+
+    name = "fake-libos"
+
+    def __init__(self):
+        self.calls: list[int] = []
+        self.armed_mask = 0
+        self.stopped: dict | None = None
+
+    def handle(self, machine, pc: int = 0):
+        number = int(machine.x[5])
+        if number == INTERRUPT_ARM_EXTERNAL:
+            self.armed_mask |= int(machine.x[10])
+            return True, 0
+        self.calls.append(number)
+        self.stopped = {"num": number, "name": f"sys_{number}", "pc": int(pc)}
+        return False, None
+
+
+def _port_call(machine, *, send=None, recv=None, wait=None,
+               timeout=TIMEOUT_INFINITE) -> None:
+    """Write one ``TASK_PORT_SEND_RECV_WAIT`` argument set (explicit fields)."""
+    machine.set(5, TASK_PORT_SEND_RECV_WAIT)  # t0
+    for index, value in enumerate(send or (0, 0, 0, 0), start=10):
+        machine.set(index, value)
+    for index, value in enumerate(recv or (0, 0, 0, 0), start=14):
+        machine.set(index, value)
+    machine.set(6, 0 if wait is None else wait)  # t1
+    machine.set(7, timeout)  # t2
+
+
+class PortSurfaceTests(unittest.TestCase):
+    """Role table and argument contract."""
+
+    def test_observed_roles(self):
+        self.assertEqual(role_for(6), "extintr")
+        self.assertEqual(role_for(7), "notification")
+        self.assertEqual(role_for(8), "request")
+        self.assertEqual(role_for(12), "handshake")
+        self.assertEqual(role_for(13), "event")
+        self.assertEqual(role_for(99), "data")
+
+    def test_async_post_keeps_receive_args_and_returns_ok(self):
+        model = LibosPortModel(base=FakeBase())
+        machine = FakeMachine()
+        _port_call(machine, recv=(10, 6, 0, 0))
+        handled, result = model.handle(machine, pc=0x4018038)
+        self.assertTrue(handled)
+        self.assertEqual(result, 0)
+        self.assertEqual(len(model.posts), 1)
+        self.assertEqual(model.posts[0]["shuttle"], 10)
+        self.assertEqual(model.posts[0]["role"], "extintr")
+        # The kernel returns LIBOS_OK through a0; the wrapper masks a0 only.
+        self.assertEqual(machine.x[14], 10)
+        self.assertEqual(machine.x[15], 6)
+
+    def test_send_completes_with_the_transcribe_contract(self):
+        model = LibosPortModel(base=FakeBase())
+        machine = FakeMachine()
+        _port_call(machine, send=(1, 12, 0, 0, ), wait=1)
+        handled, result = model.handle(machine, pc=0x4019CD4)
+        self.assertTrue(handled)
+        self.assertEqual(result, 0)
+        self.assertEqual(len(model.sends), 1)
+        self.assertEqual(model.sends[0]["contract"], HANDSHAKE_CONTRACT)
+        self.assertEqual(model.sends[0]["role"], "handshake")
+        self.assertEqual(machine.x[14], 0)  # completed size
+        self.assertEqual(machine.x[15], 0)  # originating task
+        self.assertEqual(machine.x[16], 1)  # completed shuttle
+        self.assertEqual(machine.x[17], 0)
+
+    def test_blocking_wait_stops_without_completion(self):
+        model = LibosPortModel(base=FakeBase())
+        machine = FakeMachine()
+        _port_call(machine, recv=(9, 8, 0x4015070, 0x808))
+        model.handle(machine, pc=0x4018038)
+        _port_call(machine, wait=SHUTTLE_ANY)
+        handled, result = model.handle(machine, pc=0x4019DF4)
+        self.assertFalse(handled)
+        self.assertIsNone(result)
+        self.assertIsNotNone(model.stopped)
+        self.assertEqual(model.stopped["reason"], "port-wait-blocks")
+        self.assertEqual(model.stopped["num"], TASK_PORT_SEND_RECV_WAIT)
+        self.assertEqual(model.waits[-1]["reason"], "port-wait-blocks")
+
+    def test_unmodelled_syscall_is_recorded_and_stopped(self):
+        base = FakeBase()
+        model = LibosPortModel(base=base)
+        machine = FakeMachine()
+        machine.set(5, 6)  # task_state_query
+        machine.set(10, 0)
+        handled, result = model.handle(machine, pc=0x401A46C)
+        self.assertFalse(handled)
+        self.assertIsNone(result)
+        self.assertEqual(model.stopped["num"], 6)
+        self.assertEqual(model.stopped["reason"], "unmodelled-syscall")
+        self.assertEqual(base.calls, [6])
+
+
+class DoorbellWakeTests(unittest.TestCase):
+    """The host wake's external-interrupt handoff (RTOS report contract)."""
+
+    def test_kick_before_post_and_arm_is_delivered_at_post(self):
+        base = FakeBase()
+        model = LibosPortModel(base=base)
+        machine = FakeMachine()
+        event = model.host_kick(queue=0, value=0)
+        self.assertIsNone(event["delivered"])
+        self.assertTrue(event["pending"])
+        # Arm first, then post the extintr receive: the pending wake fills it.
+        machine.set(5, INTERRUPT_ARM_EXTERNAL)
+        machine.set(10, 0xFF)
+        model.handle(machine, pc=0x4018054)
+        _port_call(machine, recv=(10, 6, 0, 0))
+        model.handle(machine, pc=0x4018038)
+        self.assertEqual(model.kicks_pending, 0)
+        self.assertEqual([entry["shuttle"] for entry in model.completed], [10])
+        self.assertEqual(model.completed[0]["payload_size"], 0)
+        self.assertEqual(model.completed[0]["origin_task"], 0)
+        self.assertEqual(model.completed[0]["contract"], EXTINTR_WAKE_CONTRACT)
+
+    def test_wake_requires_an_arm(self):
+        model = LibosPortModel(base=FakeBase())
+        machine = FakeMachine()
+        _port_call(machine, recv=(10, 6, 0, 0))
+        model.handle(machine, pc=0x4018038)
+        event = model.host_kick()
+        self.assertIsNone(event["delivered"])
+        self.assertEqual(model.kicks_pending, 1)
+        self.assertEqual(model.completed, [])
+        # Without interrupts armed the wait still blocks (no fabricated wake).
+        _port_call(machine, wait=SHUTTLE_ANY)
+        handled, _ = model.handle(machine, pc=0x4019DF4)
+        self.assertFalse(handled)
+
+    def test_arm_seeds_from_the_attached_base_model(self):
+        base = FakeBase()
+        base.armed_mask = 0xFF
+        model = LibosPortModel(base=base)
+        self.assertEqual(model.armed_mask, 0xFF)
+        machine = FakeMachine()
+        model.host_kick()
+        _port_call(machine, recv=(10, 6, 0, 0))
+        model.handle(machine, pc=0x4018038)
+        self.assertEqual(model.kicks_pending, 0)
+        self.assertEqual([entry["shuttle"] for entry in model.completed], [10])
+
+    def test_wait_returns_the_wake_with_the_argument_contract(self):
+        model = LibosPortModel(base=FakeBase())
+        machine = FakeMachine()
+        model.host_kick()
+        _port_call(machine, recv=(10, 6, 0, 0))
+        model.handle(machine, pc=0x4018038)
+        machine.set(5, INTERRUPT_ARM_EXTERNAL)
+        machine.set(10, 0xFF)
+        model.handle(machine, pc=0x4018054)
+        _port_call(machine, wait=SHUTTLE_ANY)
+        handled, result = model.handle(machine, pc=0x4019DF4)
+        self.assertTrue(handled)
+        self.assertEqual(result, 0)
+        self.assertEqual(machine.x[14], 0)   # completed size
+        self.assertEqual(machine.x[15], 0)   # originating task
+        self.assertEqual(machine.x[16], 10)  # completed shuttle
+        self.assertEqual(model.waits[-1]["completed"], 10)
+        self.assertEqual(model.waits[-1]["role"], "extintr")
+
+    def test_one_wake_completes_one_receive_only(self):
+        model = LibosPortModel(base=FakeBase())
+        machine = FakeMachine()
+        model.host_kick()
+        _port_call(machine, recv=(10, 6, 0, 0))
+        model.handle(machine, pc=0x4018038)
+        machine.set(5, INTERRUPT_ARM_EXTERNAL)
+        machine.set(10, 0xFF)
+        model.handle(machine, pc=0x4018054)
+        self.assertEqual(len(model.completed), 1)
+        # Consume the wake, re-post and wait again: nothing completes.
+        _port_call(machine, wait=SHUTTLE_ANY)
+        handled, _ = model.handle(machine, pc=0x4019DF4)
+        self.assertTrue(handled)
+        _port_call(machine, recv=(10, 6, 0, 0))
+        model.handle(machine, pc=0x4018C8)
+        _port_call(machine, wait=SHUTTLE_ANY)
+        handled, _ = model.handle(machine, pc=0x4019DF4)
+        self.assertFalse(handled)
+        self.assertEqual(model.completed, [])
+
+
+class StateTests(unittest.TestCase):
+    """Deterministic, serialisable report state."""
+
+    def _run(self) -> dict:
+        model = LibosPortModel(base=FakeBase())
+        machine = FakeMachine()
+        model.host_kick(queue=0, value=0)
+        _port_call(machine, recv=(10, 6, 0, 0))
+        model.handle(machine, pc=0x4018038)
+        machine.set(5, INTERRUPT_ARM_EXTERNAL)
+        machine.set(10, 0xFF)
+        model.handle(machine, pc=0x4018054)
+        _port_call(machine, recv=(9, 8, 0x4015070, 0x808))
+        model.handle(machine, pc=0x4018038)
+        _port_call(machine, wait=SHUTTLE_ANY)
+        model.handle(machine, pc=0x4019DF4)
+        return model.state()
+
+    def test_state_is_byte_identical_across_runs(self):
+        first = self._run()
+        second = self._run()
+        self.assertEqual(first["format"], REPORT_FORMAT)
+        self.assertEqual(json.dumps(first, sort_keys=True),
+                         json.dumps(second, sort_keys=True))
+        self.assertEqual(first["armed_mask"], 0xFF)
+        self.assertEqual(first["host_kicks"], 1)
+        self.assertEqual(first["kicks_pending"], 0)
+        self.assertEqual([post["role"] for post in first["posts"]],
+                         ["extintr", "request"])
+        self.assertEqual(first["completed"], [])
+        self.assertEqual(first["waits"][-1]["completed"], 10)
+
+    def test_state_has_no_host_timestamps(self):
+        state = self._run()
+
+        def keys(value):
+            if isinstance(value, dict):
+                for key, item in value.items():
+                    yield key
+                    yield from keys(item)
+            elif isinstance(value, list):
+                for item in value:
+                    yield from keys(item)
+
+        for key in keys(state):
+            self.assertNotIn(str(key), ("t", "ts", "timestamp", "time"))
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/debug/risc-v-gsp-rm-vm/tests/test_log_decode.py b/debug/risc-v-gsp-rm-vm/tests/test_log_decode.py
new file mode 100644
index 0000000..b43dc00
--- /dev/null
+++ b/debug/risc-v-gsp-rm-vm/tests/test_log_decode.py
@@ -0,0 +1,335 @@
+"""Tests for the offline GSP-RM log decoder (harness/log_decode.py).
+
+Crafted inputs cover both real dump forms: the raw ring hexdump
+(``LOGINIT-raw:`` + rows) and the dmesg/evidence record lines
+(``rec seq=.. recType=.. errorCode=.. source=..`` plus ``assertCallStack``
+frames, ``entry[k]:`` aggregates and the liblogdecode ``rec NNN ts=..``
+text form).  A real-file test runs only when the campaign dump is present.
+
+Run::
+
+    cd TOOL_ROOT && python3 -m unittest tests.test_log_decode -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, ".."))
+if TOOL_ROOT not in sys.path:
+    sys.path.insert(0, TOOL_ROOT)
+
+from harness.log_decode import (  # noqa: E402
+    SCHEMA,
+    LogDecodeError,
+    apply_filters,
+    decode_path,
+    decode_text,
+    load_symbols,
+    main,
+    run_demo,
+)
+
+REAL_LOGDUMP = (
+    "/home/rmcore/Git/integdev_gpu_drv/open-iov/backups/"
+    "evidence-20261002-gr/21-logdump-grctx4.txt"
+)
+
+RAW_TEXT = """\
+vgsp-boot: libos task log buffers (read-only)
+LOGINIT-raw: 64 bytes (allocated 64) VA=0xffff0000 dma=0x1000
+00000000: 00000005 00000000 0000000a deadbeef
+00000010: 01020304 05060708 090a0b0c 0d0e0f10
+"""
+
+RECORDS_TEXT = """\
+[Tue Oct  6 12:40:52 2026] vgsp-nocat: rec seq=1 recType=5 errorCode=0x57f1344 subsystem=0x0 bugcheck=0x0 payload=1212 source="ASSERT" faultingEngine=""
+[Tue Oct  6 12:40:52 2026] vgsp-nocat: rec seq=1 assertCallStack count=1 [0x401000,0x402000,0x0] (offsets from load base 0x4000000)
+[Tue Oct  6 12:40:53 2026] vgsp-nocat: entry[0]: count=1 recType=5 errorCode=0x57f1344 firstSeq=1 lastSeq=1 source="ASSERT"
+an unmatched filler line
+rec   7 ts=0x18dbb29118ec6200 meta=0x0000000005bb9c48 args=0x401000,0x402000
+rec   9 ts=0x18dbb29118ec7280 meta=0x0000000005bb9bd0 args=[0x10, 0x20]
+"""
+
+SYMBOLS = {
+    "symbols": [
+        {"name": "gr_init", "addr": "0x401000"},
+        {"name": "gr_done", "addr": 0x403000},
+    ]
+}
+
+
+def _write(tmp: str, name: str, text: str) -> str:
+    path = os.path.join(tmp, name)
+    with open(path, "w", encoding="utf-8") as fh:
+        fh.write(text)
+    return path
+
+
+def _records(report: dict, kind: str = "record") -> list[dict]:
+    return [r for r in report["records"] if r["kind"] == kind]
+
+
+class RawHexTests(unittest.TestCase):
+    def test_header_and_rows_decode_to_words(self):
+        report = decode_text(RAW_TEXT, source="raw.txt")
+        self.assertEqual(report["schema"], SCHEMA)
+        self.assertEqual(report["source"], "raw.txt")
+        self.assertEqual(report["count"], 1)
+        self.assertEqual(report["unparsed"], 1)  # the intro line is not a record
+        raw = report["records"][0]
+        self.assertEqual(raw["kind"], "raw-hex")
+        self.assertEqual(raw["label"], "LOGINIT")
+        self.assertEqual(raw["source_name"], "LOGINIT")
+        self.assertEqual(len(raw["frames"]), 8)
+        self.assertEqual(raw["frames"][0], "0x5")
+        self.assertEqual(raw["frames"][3], "0xdeadbeef")
+        self.assertEqual(raw["frames"][7], "0xd0e0f10")
+        # raw is the decoded little-endian byte buffer of the rows
+        self.assertEqual(len(raw["raw"]), 2 * 8 * 8)
+        self.assertEqual(bytes.fromhex(raw["raw"])[:8].hex(), "0500000000000000")
+        self.assertEqual(raw["raw"].endswith("100f0e0d00000000"), True)
+
+    def test_two_sections_stay_separate(self):
+        text = RAW_TEXT + (
+            "LOGRM-raw: 16 bytes (allocated 16) VA=0x1 dma=0x2\n"
+            "00000000: 11111111 22222222 33333333 44444444\n"
+        )
+        report = decode_text(text)
+        raw = _records(report, "raw-hex")
+        self.assertEqual(len(raw), 2)
+        self.assertEqual(raw[1]["label"], "LOGRM")
+        self.assertEqual(raw[1]["frames"], ["0x11111111", "0x22222222",
+                                            "0x33333333", "0x44444444"])
+
+    def test_blank_and_comment_lines_are_structure(self):
+        report = decode_text("\n# LOGINIT\n" + RAW_TEXT)
+        self.assertEqual(report["unparsed"], 1)  # only the intro line
+
+
+class RecordLineTests(unittest.TestCase):
+    def setUp(self):
+        self.report = decode_text(RECORDS_TEXT)
+        self.rec = self.report["records"][0]
+
+    def test_rec_line_fields(self):
+        self.assertEqual(self.rec["kind"], "record")
+        self.assertEqual(self.rec["label"], "vgsp-nocat")
+        self.assertEqual(self.rec["seq"], 1)
+        self.assertEqual(self.rec["rec_type"], 5)
+        self.assertEqual(self.rec["error_code"], "0x57f1344")
+        self.assertEqual(self.rec["source_name"], "ASSERT")
+
+    def test_assert_callstack_merges_frames(self):
+        self.assertEqual(self.rec["frames"], ["0x401000", "0x402000", "0x0"])
+        # no duplicate standalone continuation record was emitted
+        self.assertEqual(len(_records(self.report)), 4)
+
+    def test_entry_aggregate_line(self):
+        entry = self.report["records"][1]
+        self.assertEqual(entry["rec_type"], 5)
+        self.assertEqual(entry["error_code"], "0x57f1344")
+        self.assertEqual(entry["source_name"], "ASSERT")
+        self.assertIn("entry[0]:", entry["raw"])
+        self.assertIn("vgsp-nocat:", entry["raw"])
+
+    def test_decoded_rec_shapes_with_and_without_brackets(self):
+        records = _records(self.report)
+        plain, bracketed = records[2], records[3]
+        self.assertEqual(plain["seq"], 7)
+        self.assertEqual(plain["frames"], ["0x401000", "0x402000"])
+        self.assertEqual(bracketed["seq"], 9)
+        self.assertEqual(bracketed["frames"], ["0x10", "0x20"])
+
+    def test_unparsed_accounting(self):
+        self.assertEqual(self.report["unparsed"], 1)
+        self.assertEqual(len(self.report["records"]), 4)
+
+    def test_histogram(self):
+        hist = self.report["histogram"]
+        self.assertEqual(hist["rec_type"], {"5": 2})
+        self.assertEqual(hist["error_code"], {"0x57f1344": 2})
+        self.assertEqual(hist["source"], {"ASSERT": 2})
+
+    def test_standalone_continuation_becomes_record(self):
+        text = ("vgsp-nocat: rec seq=4 assertCallStack count=1 "
+                "[0x1000,0x2000] (offsets from load base 0x4000000)\n")
+        report = decode_text(text)
+        self.assertEqual(report["count"], 1)
+        self.assertEqual(report["records"][0]["seq"], 4)
+        self.assertEqual(report["records"][0]["frames"], ["0x1000", "0x2000"])
+
+
+class SymbolTests(unittest.TestCase):
+    def test_nearest_below_with_offset(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            path = _write(tmp, "symbols.json", json.dumps(SYMBOLS))
+            table = load_symbols(path)
+        self.assertEqual(table.resolve(0x401000), ("gr_init", 0x0))
+        self.assertEqual(table.resolve(0x402fff), ("gr_init", 0x1FFF))
+        self.assertEqual(table.resolve(0x403000), ("gr_done", 0x0))
+        self.assertIsNone(table.resolve(0x400fff))  # below the first symbol
+        self.assertIsNone(table.resolve(0x0))
+
+    def test_top_level_list_shape_accepted(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            path = _write(tmp, "symbols.json",
+                          json.dumps([{"name": "a", "addr": "0x10"}]))
+            table = load_symbols(path)
+        self.assertEqual(table.resolve(0x1f), ("a", 0xF))
+
+    def test_resolved_entries_align_with_frames(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            path = _write(tmp, "symbols.json", json.dumps(SYMBOLS))
+            report = decode_text(RECORDS_TEXT, symbols=load_symbols(path))
+        resolved = report["records"][0]["resolved"]
+        self.assertEqual(resolved[0],
+                         {"frame": "0x401000", "symbol": "gr_init",
+                          "offset": "0x0"})
+        self.assertEqual(resolved[1],
+                         {"frame": "0x402000", "symbol": "gr_init",
+                          "offset": "0x1000"})
+        self.assertEqual(resolved[2],
+                         {"frame": "0x0", "symbol": None, "offset": None})
+
+    def test_malformed_symbols_fail_cleanly(self):
+        cases = [
+            "{not json",
+            json.dumps([{"name": "no-addr"}]),
+            json.dumps({"symbols": "nope"}),
+            json.dumps({"nope": []}),
+            json.dumps([42]),
+            json.dumps([{"name": "x", "addr": "zzz"}]),
+        ]
+        with tempfile.TemporaryDirectory() as tmp:
+            for index, payload in enumerate(cases):
+                path = _write(tmp, f"bad{index}.json", payload)
+                with self.assertRaises(LogDecodeError):
+                    load_symbols(path)
+            self.assertRaises(LogDecodeError, load_symbols,
+                              os.path.join(tmp, "missing.json"))
+            # the CLI exits non-zero with a clear message, not a traceback
+            good = _write(tmp, "good.txt", RECORDS_TEXT)
+            bad = _write(tmp, "bad.json", cases[0])
+            stderr = io.StringIO()
+            with contextlib.redirect_stderr(stderr):
+                code = main(["decode", good, "--symbols", bad])
+            self.assertEqual(code, 2)
+            self.assertIn("symbols JSON", stderr.getvalue())
+
+
+class FilterTests(unittest.TestCase):
+    def setUp(self):
+        self.report = decode_text(RECORDS_TEXT)
+        self.raw_report = decode_text(RAW_TEXT)
+
+    def test_kind_filter(self):
+        self.assertEqual(apply_filters(self.report, kind="record")["count"], 4)
+        self.assertEqual(apply_filters(self.report, kind="raw-hex")["count"], 0)
+        self.assertEqual(apply_filters(self.raw_report, kind="raw-hex")["count"], 1)
+
+    def test_rec_type_filter_accepts_hex_and_decimal(self):
+        for value in ("5", "0x5"):
+            filtered = apply_filters(self.report, rec_type=value)
+            self.assertEqual(filtered["count"], 2)
+            self.assertTrue(all(r["rec_type"] == 5 for r in filtered["records"]))
+
+    def test_error_code_filter(self):
+        filtered = apply_filters(self.report, error_code="0x57f1344")
+        self.assertEqual(filtered["count"], 2)
+        zeros = apply_filters(self.report, error_code="0x1")
+        self.assertEqual(zeros["count"], 0)
+
+    def test_source_filter_is_case_insensitive_substring(self):
+        self.assertEqual(apply_filters(self.report, source="assert")["count"], 2)
+        self.assertEqual(apply_filters(self.report, source="Display")["count"], 0)
+
+    def test_limit_and_combined_filters(self):
+        self.assertEqual(apply_filters(self.report, limit=1)["count"], 1)
+        combined = apply_filters(self.report, kind="record", source="ASSERT",
+                                 limit=1)
+        self.assertEqual(combined["count"], 1)
+        self.assertEqual(combined["records"][0]["seq"], 1)
+        self.assertEqual(apply_filters(self.report, limit=0)["count"], 0)
+
+    def test_bad_filter_value_raises(self):
+        with self.assertRaises(LogDecodeError):
+            apply_filters(self.report, rec_type="zzz")
+        with self.assertRaises(LogDecodeError):
+            apply_filters(self.report, kind="bogus")
+        with self.assertRaises(LogDecodeError):
+            apply_filters(self.report, limit=-1)
+
+    def test_histogram_follows_filter(self):
+        filtered = apply_filters(self.report, kind="record", limit=1)
+        self.assertEqual(filtered["histogram"]["rec_type"], {"5": 1})
+        self.assertEqual(filtered["unparsed"], self.report["unparsed"])
+
+
+class CliTests(unittest.TestCase):
+    def test_decode_writes_json_report(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            path = _write(tmp, "records.txt", RECORDS_TEXT)
+            out = os.path.join(tmp, "out", "report.json")
+            stdout = io.StringIO()
+            with contextlib.redirect_stdout(stdout):
+                code = main(["decode", path, "--json", out,
+                             "--source", "ASSERT", "--limit", "1"])
+            self.assertEqual(code, 0)
+            with open(out, "r", encoding="utf-8") as fh:
+                report = json.load(fh)
+        self.assertEqual(report["schema"], SCHEMA)
+        self.assertEqual(report["source"], "records.txt")
+        self.assertEqual(report["count"], 1)
+        self.assertIn("report_json", stdout.getvalue())
+
+    def test_missing_input_fails_cleanly(self):
+        stderr = io.StringIO()
+        with contextlib.redirect_stderr(stderr):
+            code = main(["decode", "/nonexistent/log.txt"])
+        self.assertEqual(code, 2)
+        self.assertIn("input not found", stderr.getvalue())
+
+    def test_demo_runs_deterministic(self):
+        first = run_demo()
+        second = run_demo()
+        self.assertTrue(first["ok"], first)
+        self.assertEqual(first["raw"]["count"], second["raw"]["count"])
+        self.assertEqual(first["records"]["records"][0]["frames"],
+                         second["records"]["records"][0]["frames"])
+        stdout = io.StringIO()
+        with contextlib.redirect_stdout(stdout):
+            code = main(["demo"])
+        self.assertEqual(code, 0)
+        self.assertIn('"ok": true', stdout.getvalue())
+
+
+@unittest.skipUnless(os.path.isfile(REAL_LOGDUMP),
+                     f"real logdump not present: {REAL_LOGDUMP}")
+class RealLogdumpTests(unittest.TestCase):
+    """Decode the campaign dump when it is present; skipped otherwise."""
+
+    def test_real_logdump_decodes(self):
+        report = decode_path(REAL_LOGDUMP)
+        self.assertEqual(report["schema"], SCHEMA)
+        self.assertEqual(report["source"], os.path.basename(REAL_LOGDUMP))
+        self.assertGreater(report["count"], 0)
+        self.assertTrue(any(report["histogram"].values()),
+                        report["histogram"])
+        raw_labels = {r["label"] for r in _records(report, "raw-hex")}
+        self.assertEqual(raw_labels, {"LOGINIT", "LOGRM"})
+        framed = [r for r in _records(report)
+                  if r.get("seq") is not None and r.get("frames")]
+        self.assertTrue(framed)
+        self.assertEqual(report["count"],
+                         len(report["records"]))
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/debug/risc-v-gsp-rm-vm/tests/test_msgq_attach.py b/debug/risc-v-gsp-rm-vm/tests/test_msgq_attach.py
new file mode 100644
index 0000000..749157e
--- /dev/null
+++ b/debug/risc-v-gsp-rm-vm/tests/test_msgq_attach.py
@@ -0,0 +1,342 @@
+"""Message-queue attach tests (sim/msgq_attach.py, docs/51 section 7).
+
+Hermetic tests drive the pure classification/report functions with crafted
+inputs -- no image, no boot, no machine.  One image test runs the real 570.144
+firmware with a small bounded budget when the image is present and asserts
+honest, deterministic outcomes only: it never requires the message loop to be
+reached, because the tool must report a first divergence rather than fake one.
+
+Offline only: no hardware, kernel module, mdev, guest, network or QEMU.
+
+Run::
+
+    cd open-iov/debug/risc-v-gsp-rm-vm
+    python3 -m unittest tests.test_msgq_attach -v
+"""
+from __future__ import annotations
+
+import os
+import sys
+import unittest
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+TOOL_ROOT = os.path.abspath(os.path.join(HERE, ".."))
+if TOOL_ROOT not in sys.path:
+    sys.path.insert(0, TOOL_ROOT)
+
+from sim.host_traffic import (  # noqa: E402
+    DOORBELL_BASE,
+    DOORBELL_SIZE,
+    GSP_MSG_QUEUE_ELEMENT_HDR_SIZE,
+    GSP_MSG_QUEUE_ELEMENT_SIZE_MIN,
+    MSGQ_ENTRY_OFF,
+    MSGQ_RX_HDR_OFF,
+    MSGQ_TX_HEADER_BYTES,
+    SHARED_MEM_BASE,
+    HostTraffic,
+)
+from sim.msgq_attach import (  # noqa: E402
+    DEFAULT_IMAGE,
+    MAX_RUN_CYCLES,
+    REPORT_FORMAT,
+    MsgqAttachError,
+    RingGeometry,
+    assemble_report,
+    classify_ring_access,
+    missing_semantic_after_handoff,
+    ring_geometry_from_host,
+    run_attach,
+)
+
+#: Small total budget: boot-to-B3 needs ~474k instructions; the rest is a
+#: bounded continuation tail.  Two runs stay well under the test time cap.
+SMALL_BUDGET = 600_000
+
+
+def _geometry() -> RingGeometry:
+    return RingGeometry(
+        shared_base=SHARED_MEM_BASE,
+        cmd_base=SHARED_MEM_BASE,
+        stat_base=SHARED_MEM_BASE + 0x40000,
+        cmd_size=0x40000,
+        stat_size=0x40000,
+        msg_size=GSP_MSG_QUEUE_ELEMENT_SIZE_MIN,
+        entry_off=MSGQ_ENTRY_OFF,
+        rx_hdr_off=MSGQ_RX_HDR_OFF,
+        doorbell_base=DOORBELL_BASE,
+        doorbell_size=DOORBELL_SIZE,
+    )
+
+
+def _image() -> dict:
+    return {"path": "/nonexistent/image.bin", "sha256": "0" * 64, "size": 1}
+
+
+def _profile() -> dict:
+    return {"name": "crafted", "path": "/nonexistent/profile.json"}
+
+
+def _boot(reached: bool) -> dict:
+    return {
+        "ran": True,
+        "skipped": False,
+        "rung": "B3" if reached else "B1",
+        "message_loop_wait_reached": reached,
+        "instructions_executed": 474043 if reached else 1000,
+        "stop": {"reason": "ecall" if reached else "budget",
+                 "pc": 0x4018038 if reached else 0x4009000},
+    }
+
+
+def _assemble(reached: bool, polls, divergence) -> dict:
+    return assemble_report(
+        image=_image(),
+        profile=_profile(),
+        boot=_boot(reached),
+        attach={"cmd_base": SHARED_MEM_BASE},
+        doorbell={"requested": True},
+        polls=polls,
+        divergence=divergence,
+        missing_mmio={"count": 0, "first": None},
+        determinism={"instructions": 474043 if reached else 1000},
+    )
+
+
+class ClassifierTests(unittest.TestCase):
+    """Pure ring-access classification (no machine, no boot)."""
+
+    def test_geometry_from_host_matches_the_attached_layout(self):
+        host = HostTraffic()
+        try:
+            geometry = ring_geometry_from_host(host)
+        finally:
+            host.close()
+        self.assertEqual(geometry.cmd_base, SHARED_MEM_BASE)
+        self.assertEqual(geometry.stat_base, SHARED_MEM_BASE + 0x40000)
+        self.assertEqual(geometry.msg_size, GSP_MSG_QUEUE_ELEMENT_SIZE_MIN)
+        self.assertEqual(geometry.entry_off, MSGQ_ENTRY_OFF)
+        self.assertEqual(geometry.rx_hdr_off, MSGQ_RX_HDR_OFF)
+        self.assertEqual(geometry.doorbell_base, DOORBELL_BASE)
+        self.assertEqual(geometry.doorbell_size, DOORBELL_SIZE)
+
+    def test_queue_head_classification(self):
+        geometry = _geometry()
+        poll = classify_ring_access(geometry.doorbell_base + 8, 4, 0x11, geometry)
+        self.assertTrue(poll["in_ring"])
+        self.assertEqual(poll["kind"], "queue-head")
+        self.assertEqual(poll["queue"], 1)
+        self.assertEqual(poll["register"], "NV_PGSP_QUEUE_HEAD(1)")
+        self.assertEqual(poll["window"], "PRIV+0x110c00")
+        self.assertEqual(poll["value"], 0x11)
+
+    def test_tx_header_classification(self):
+        geometry = _geometry()
+        poll = classify_ring_access(geometry.cmd_base + 16, 4, 0, geometry)
+        self.assertTrue(poll["in_ring"])
+        self.assertEqual(poll["kind"], "tx-header")
+        self.assertEqual(poll["buffer"], "command")
+        self.assertEqual(poll["field"], "writePtr")
+        self.assertEqual(poll["field_offset"], 16)
+        self.assertEqual(MSGQ_TX_HEADER_BYTES, 32)
+
+    def test_rx_header_classification_carries_the_cross_link(self):
+        geometry = _geometry()
+        poll = classify_ring_access(geometry.stat_base + MSGQ_RX_HDR_OFF, 4, 0,
+                                    geometry)
+        self.assertEqual(poll["kind"], "rx-header")
+        self.assertEqual(poll["buffer"], "status")
+        self.assertEqual(poll["cross_linked_owner"], "command")
+        self.assertEqual(poll["semantics"], "readPtr")
+
+    def test_entry_classification_and_slot(self):
+        geometry = _geometry()
+        addr = (geometry.cmd_base + MSGQ_ENTRY_OFF
+                + 3 * GSP_MSG_QUEUE_ELEMENT_SIZE_MIN + 0x40)
+        poll = classify_ring_access(addr, 4, 0xAB, geometry)
+        self.assertEqual(poll["kind"], "entry")
+        self.assertEqual(poll["buffer"], "command")
+        self.assertEqual(poll["entry_index"], 3)
+        self.assertEqual(poll["slot_offset"], 0x40)
+        self.assertEqual(poll["region"], "entry-payload")
+        header = classify_ring_access(
+            geometry.cmd_base + MSGQ_ENTRY_OFF + 0x28, 4, 0, geometry)
+        self.assertEqual(header["region"], "entry-header")
+        self.assertEqual(GSP_MSG_QUEUE_ELEMENT_HDR_SIZE, 0x30)
+
+    def test_non_ring_access_is_other(self):
+        geometry = _geometry()
+        poll = classify_ring_access(0xDEAD, 4, 0, geometry)
+        self.assertFalse(poll["in_ring"])
+        self.assertEqual(poll["kind"], "other")
+
+
+class ReportTests(unittest.TestCase):
+    """Pure report assembly: schema, outcomes and honest divergence."""
+
+    def test_not_reached_keeps_divergence_and_no_poll(self):
+        geometry = _geometry()
+        divergence = {"kind": "budget",
+                      "reason": "message-loop port wait not reached within budget"}
+        report = _assemble(False, [], divergence)
+        self.assertEqual(report["format"], REPORT_FORMAT)
+        self.assertEqual(report["outcome"], "not-reached")
+        self.assertFalse(report["ok"])
+        self.assertIsNone(report["first_poll"])
+        self.assertEqual(report["subsequent_polls"], [])
+        self.assertEqual(report["poll_count"], 0)
+        self.assertTrue(report["divergence"]["reason"])
+        self.assertEqual(report["determinism"]["report_sha256"],
+                         _assemble(False, [], divergence)["determinism"]["report_sha256"])
+        self.assertEqual(geometry.doorbell_base, DOORBELL_BASE)
+
+    def test_reached_with_poll_is_classified(self):
+        geometry = _geometry()
+        poll = classify_ring_access(geometry.doorbell_base, 4, 0, geometry)
+        poll["pc"] = 0x4019000
+        report = _assemble(True, [poll], None)
+        self.assertEqual(report["outcome"], "wait-reached-poll-classified")
+        self.assertTrue(report["ok"])
+        self.assertEqual(report["first_poll"]["kind"], "queue-head")
+        self.assertEqual(report["first_poll"]["pc"], 0x4019000)
+        self.assertEqual(report["subsequent_polls"], [])
+        self.assertIsNone(report["divergence"])
+
+    def test_reached_without_poll_is_honest(self):
+        report = _assemble(True, [], None)
+        self.assertEqual(report["outcome"], "wait-reached-no-poll")
+        self.assertTrue(report["ok"])
+        self.assertIsNone(report["first_poll"])
+
+    def test_subsequent_polls_are_bounded_to_sixteen(self):
+        geometry = _geometry()
+        polls = []
+        for index in range(20):
+            poll = classify_ring_access(geometry.cmd_base + 8, 4, index, geometry)
+            poll["pc"] = 0x4018000 + index
+            polls.append(poll)
+        report = _assemble(True, polls, None)
+        self.assertEqual(report["poll_count"], 20)
+        self.assertEqual(report["first_poll"]["value"], 0)
+        self.assertEqual(len(report["subsequent_polls"]), 16)
+        self.assertEqual(report["subsequent_polls"][-1]["value"], 16)
+
+    def test_absent_image_is_a_clean_skip(self):
+        report = run_attach(image="/nonexistent/gsp-image.bin",
+                            max_cycles=SMALL_BUDGET)
+        self.assertEqual(report["format"], REPORT_FORMAT)
+        self.assertEqual(report["outcome"], "skipped")
+        self.assertFalse(report["ok"])
+        self.assertTrue(report["boot"]["reason"])
+
+    def test_bad_budget_is_refused(self):
+        with self.assertRaises(MsgqAttachError):
+            run_attach(image="/nonexistent/gsp-image.bin", max_cycles=0)
+        with self.assertRaises(MsgqAttachError):
+            run_attach(image="/nonexistent/gsp-image.bin",
+                       max_cycles=MAX_RUN_CYCLES + 1)
+        with self.assertRaises(MsgqAttachError):
+            run_attach(image="/nonexistent/gsp-image.bin", max_cycles=True)
+
+
+class PortModelReportTests(unittest.TestCase):
+    """The additive port_model / missing_semantic report fields."""
+
+    def test_defaults_are_additive_and_honest(self):
+        report = _assemble(True, [], None)
+        self.assertEqual(report["port_model"], {"enabled": False})
+        self.assertIsNone(report["missing_semantic"])
+        self.assertEqual(report["format"], REPORT_FORMAT)
+
+    def test_port_model_state_is_kept_verbatim(self):
+        state = {"enabled": True, "format": "greeneye-msgq-port/1",
+                 "armed_mask": 0xFF, "host_kicks": 1}
+        report = assemble_report(
+            image=_image(),
+            profile=_profile(),
+            boot=_boot(True),
+            attach={},
+            doorbell={"requested": True},
+            polls=[],
+            divergence=None,
+            missing_mmio={"count": 0, "first": None},
+            determinism={"instructions": 1},
+            port_model=state,
+            missing_semantic={"stage": "after-extintr-handoff"},
+        )
+        self.assertTrue(report["port_model"]["enabled"])
+        self.assertEqual(report["port_model"]["armed_mask"], 0xFF)
+        self.assertEqual(report["missing_semantic"]["stage"],
+                         "after-extintr-handoff")
+
+    def test_missing_semantic_after_handoff_is_precise(self):
+        stopped = {"num": 6, "name": "task_state_query", "pc": 0x401A46C,
+                   "reason": "unmodelled-syscall"}
+        missing = missing_semantic_after_handoff(stopped=stopped, polls=[])
+        self.assertEqual(missing["stage"], "after-extintr-handoff")
+        self.assertEqual(missing["stop"]["num"], 6)
+        self.assertEqual(missing["polls"], 0)
+        self.assertGreaterEqual(len(missing["missing"]), 3)
+        self.assertTrue(missing["next_step"])
+        self.assertTrue(all(isinstance(item, str) for item in missing["observed"]))
+
+
+@unittest.skipUnless(os.path.isfile(DEFAULT_IMAGE), "real GSP firmware absent")
+class ImageTests(unittest.TestCase):
+    """Bounded real-firmware run: deterministic, honest outcome only.
+
+    Limitation: the budget is deliberately small (600k instructions; boot-to-B3
+    needs ~474k).  If the firmware needs more, the report is expected to take
+    the honest ``not-reached`` path with a non-empty divergence reason -- this
+    test accepts either outcome and asserts only determinism, never success.
+    With the default port model the run advances past the host wake to the
+    external-interrupt dispatch and reports the exact unmodelled semantic
+    (``task-state-query``); a fabricated poll is never accepted.
+    """
+
+    def test_bounded_run_is_deterministic_and_honest(self):
+        first = run_attach(image=DEFAULT_IMAGE, max_cycles=SMALL_BUDGET,
+                           doorbell=True)
+        second = run_attach(image=DEFAULT_IMAGE, max_cycles=SMALL_BUDGET,
+                            doorbell=True)
+        allowed = ("not-reached", "wait-reached-no-poll",
+                   "wait-reached-poll-classified")
+        for report in (first, second):
+            self.assertEqual(report["format"], REPORT_FORMAT)
+            self.assertIn(report["outcome"], allowed)
+            self.assertGreater(report["image"]["size"], 0)
+            self.assertEqual(len(report["image"]["sha256"]), 64)
+            self.assertTrue(report["port_model"]["enabled"])
+            if not report["boot"]["message_loop_wait_reached"]:
+                self.assertIsNotNone(report["divergence"])
+                self.assertTrue(report["divergence"]["reason"])
+            if report["first_poll"] is not None:
+                self.assertTrue(report["first_poll"]["in_ring"])
+                self.assertTrue(report["first_poll"]["kind"] != "other")
+            divergence = report["divergence"]
+            if divergence and divergence.get("kind") == "task-state-query":
+                # The port model delivered the doorbell wake (extintr shuttle
+                # 10) and stopped at the libos trap/report path; the missing
+                # semantic is named, and no poll is claimed.
+                self.assertIsNone(report["first_poll"])
+                self.assertIsNotNone(report["missing_semantic"])
+                self.assertEqual(report["missing_semantic"]["stage"],
+                                 "after-extintr-handoff")
+                completed = [wait.get("completed")
+                             for wait in report["port_model"].get("waits", [])]
+                self.assertIn(10, completed)
+        self.assertEqual(first["boot"]["message_loop_wait_reached"],
+                         second["boot"]["message_loop_wait_reached"])
+        self.assertEqual(first["boot"]["instructions_executed"],
+                         second["boot"]["instructions_executed"])
+        self.assertEqual(first["outcome"], second["outcome"])
+        self.assertEqual(first["first_poll"], second["first_poll"])
+        self.assertEqual(first["divergence"], second["divergence"])
+        self.assertEqual(first["continuation"], second["continuation"])
+        self.assertEqual(first["port_model"], second["port_model"])
+        self.assertEqual(first["missing_semantic"], second["missing_semantic"])
+        self.assertEqual(first["determinism"]["report_sha256"],
+                         second["determinism"]["report_sha256"])
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/debug/risc-v-gsp-rm-vm/tests/test_rehost.py b/debug/risc-v-gsp-rm-vm/tests/test_rehost.py
new file mode 100644
index 0000000..3c88355
--- /dev/null
+++ b/debug/risc-v-gsp-rm-vm/tests/test_rehost.py
@@ -0,0 +1,135 @@
+"""Tests for the surgical-rehost harness (harness/rehost.py, docs/46).
+
+Fixtures: the demo firmware (built by ``make demo``) is rehosted offline.  The
+real-image probes are skipped when the firmware is absent, and are bounded to
+one short call each.  No hardware, no module, no guest, no network.
+
+Run::
+
+    python3 -m unittest tests.test_rehost -v
+"""
+from __future__ import annotations
+
+import contextlib
+import io
+import json
+import os
+import sys
+import unittest
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+TOOL_ROOT = os.path.abspath(os.path.join(HERE, ".."))
+if TOOL_ROOT not in sys.path:
+    sys.path.insert(0, TOOL_ROOT)
+
+from harness.rehost import (  # noqa: E402
+    DEFAULT_IMAGE,
+    DEMO_ELF,
+    DEMO_FORMAT,
+    REHOST_FORMAT,
+    RehostError,
+    main,
+    parse_hex_bytes,
+    parse_mem_spec,
+    parse_snap_spec,
+    rehost,
+    run_demo,
+)
+
+
+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()
+
+
+class SpecTests(unittest.TestCase):
+    def test_parse_hex_bytes_and_mem_spec(self):
+        self.assertEqual(parse_hex_bytes("aa bb cc"), b"\xaa\xbb\xcc")
+        addr, data = parse_mem_spec("0x1000=0102")
+        self.assertEqual((addr, data), (0x1000, b"\x01\x02"))
+
+    def test_bad_specs_are_clean_errors(self):
+        with self.assertRaises(RehostError):
+            parse_hex_bytes("abc")
+        with self.assertRaises(RehostError):
+            parse_mem_spec("0x1000")
+        with self.assertRaises(RehostError):
+            parse_snap_spec("0x1000")
+
+    def test_missing_image_is_clean_error(self):
+        with self.assertRaises(RehostError):
+            rehost(image="/nonexistent/fw.bin", entry=0x1000)
+
+
+@unittest.skipUnless(os.path.isfile(DEMO_ELF), "demo firmware not built")
+class DemoRehostTests(unittest.TestCase):
+    def test_demo_rehost_is_deterministic_and_correct(self):
+        payload = run_demo()
+        self.assertEqual(payload["format"], DEMO_FORMAT)
+        self.assertTrue(payload["ok"])
+        self.assertEqual(payload["first"]["status"], "return")
+        self.assertEqual(payload["first"]["retval"], payload["expected"])
+        self.assertEqual(payload["first"]["retval"], payload["second"]["retval"])
+
+    def test_rehost_writes_report_and_session(self):
+        import tempfile
+
+        with tempfile.TemporaryDirectory() as tmp:
+            report = rehost(
+                image=DEMO_ELF,
+                entry="rpc_sum",
+                args=(0x80012000, 4),
+                mem=((0x80012000, b"\x01\x02\x03\x04"),),
+                snap=((0x80012000, 4),),
+                budget=50_000,
+                timeout=20.0,
+                out=tmp,
+                record=True,
+                name="rehost-test",
+            )
+            self.assertEqual(report["format"], REHOST_FORMAT)
+            self.assertEqual(report["status"], "return")
+            self.assertEqual(report["retval"], 10)
+            self.assertTrue(os.path.isfile(report["report_json"]))
+            self.assertEqual(len(report["snapshots"]), 1)
+            self.assertEqual(report["snapshots"][0]["changed_bytes"], 0)
+            self.assertTrue(report["snapshots"][0]["before_sha256"])
+            self.assertIn("fidelity", " ".join(report["limits"]).lower())
+
+    def test_cli_demo_returns_zero(self):
+        rc, out, _ = _run_main(["demo"])
+        self.assertEqual(rc, 0)
+        self.assertTrue(json.loads(out)["ok"])
+
+    def test_cli_missing_image_returns_two(self):
+        rc, _, err = _run_main(["run", "--image", "/nonexistent/fw.bin", "--entry", "0x1000"])
+        self.assertEqual(rc, 2)
+        self.assertIn("error:", err)
+
+
+class RealImageProbeTests(unittest.TestCase):
+    """Bounded probes on the shipped 570.144 image; skipped without it."""
+
+    @unittest.skipUnless(os.path.isfile(DEFAULT_IMAGE), "570.144 firmware absent")
+    def test_gate_probe_reproduces_invalid_state_without_state(self):
+        report = rehost(
+            image=DEFAULT_IMAGE,
+            profile_path=os.path.join(TOOL_ROOT, "sim", "profiles", "tu104-570.144.json"),
+            entry=0x5B01210,  # per-buffer commit gate (from live analysis; numeric only)
+            args=(0,),
+            budget=50_000,
+            timeout=20.0,
+            record=False,
+            name="probe-gate-test",
+        )
+        self.assertEqual(report["format"], REHOST_FORMAT)
+        self.assertIn(report["status"], ("return", "fault", "budget", "ecall"))
+        # With no engine-context state the gate family returns NV_ERR_INVALID_STATE.
+        self.assertEqual(report["retval"], 0x40)
+        self.assertIn("Not a boot", report["limits"][0])
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/debug/risc-v-gsp-rm-vm/tests/test_rm_capture.py b/debug/risc-v-gsp-rm-vm/tests/test_rm_capture.py
new file mode 100644
index 0000000..cb62bea
--- /dev/null
+++ b/debug/risc-v-gsp-rm-vm/tests/test_rm_capture.py
@@ -0,0 +1,409 @@
+"""Tests for the GSP RPC capture ingestion harness (harness/rm_capture.py, docs/45).
+
+All fixtures are synthetic text written in-process or to temporary
+directories: no capture, no hardware, no driver, no simulator run, no QEMU,
+no network.  The decode/compare/stimulus loop is offline file-in/file-out.
+Tests that would need a real capture assert the clean error path instead.
+
+Run::
+
+    python3 -m unittest tests.test_rm_capture -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, ".."))
+if TOOL_ROOT not in sys.path:
+    sys.path.insert(0, TOOL_ROOT)
+
+from harness.rm_capture import (  # noqa: E402
+    CAPTURE_FORMAT,
+    COMPARE_FORMAT,
+    DEMO_FORMAT,
+    GSP_RM_ALLOC,
+    GSP_RM_CONTROL,
+    MANUAL_DECODE_BASELINE_SECONDS,
+    RmCaptureError,
+    STIMULUS_FORMAT,
+    WORKFLOW_THRESHOLD_FRACTION,
+    canonical_json,
+    compare,
+    decode_file,
+    decode_text,
+    load_capture,
+    main,
+    print_capture_hints,
+    run_compare,
+    run_decode,
+    run_demo,
+    run_stimulus,
+    stimulus_from_capture,
+)
+
+
+def _hex_lines(tag: str, payload: bytes) -> list[str]:
+    return [f"GE-{tag} 000:" + "".join(f" {b:02x}" for b in payload)]
+
+
+def _sample_log() -> str:
+    """A small log exercising every tag; params are contiguous single chunks."""
+    return "\n".join(
+        [
+            "[ 100.000000] NVRM: GE-GOLDEN-TRIGGER gspclient=1 mig=0",
+            "[ 100.000010] NVRM: GE-GOLDEN-BEGIN",
+            "[ 100.000020] NVRM: GE-ALLOC cls=0x90f1 hClient=0xbaba0001 hParent=0x0 "
+            "hObject=0xbaba1001 size=0x10 paramp=0xffffffffc0a00000 ret=0xffffffffc0b00000",
+            "[ 100.000020] NVRM: " + _hex_lines("ALLOC", bytes(range(16)))[0],
+            "[ 100.000030] NVRM: GE-CTRL cmd=0x2080012b hClient=0xc2000006 "
+            "hObject=0xabcd0080 size=0x8 paramp=0xffffffffc0a10000 ret=0xffffffffc0b00000",
+            "[ 100.000030] NVRM: " + _hex_lines("CTRL", bytes([1, 2, 3, 4, 5, 6, 7, 8]))[0],
+            "[ 100.000040] NVRM: GE-SEND fn=0x0000004c len=0x00000028 ret=0xffffffffc0b00000",
+            "[ 100.000040] NVRM: "
+            + _hex_lines("SEND", (0xC2000006).to_bytes(4, "little")
+                         + (0xABCD0080).to_bytes(4, "little")
+                         + (0x2080012B).to_bytes(4, "little")
+                         + bytes(4))[0],
+            "[ 100.000050] NVRM: GE-CTRL-DONE cmd=0x2080012b rpc_status=0x0 params_status=0x0",
+            "[ 100.000060] NVRM: GE-GOLDEN-END status=0x0",
+        ]
+    ) + "\n"
+
+
+def _follow(text: str, header_fragment: str) -> str:
+    """Return the hex block emitted immediately after a header line."""
+    lines = text.splitlines()
+    for index, line in enumerate(lines):
+        if header_fragment in line:
+            return lines[index + 1]
+    raise AssertionError(f"fragment not found: {header_fragment}")
+
+
+def _chunked_control_log(size: int) -> str:
+    payload = bytes((i * 5) & 0xFF for i in range(size))
+    lines = ["[ 1.0] NVRM: GE-GOLDEN-BEGIN"]
+    lines.append(
+        f"[ 1.1] NVRM: GE-CTRL cmd=0x2080012b hClient=0xc2000006 "
+        f"hObject=0xabcd0080 size={size:#x} paramp=0x0 ret=0x0"
+    )
+    for start in range(0, len(payload), 192):
+        piece = payload[start : start + 192]
+        lines.append(
+            f"[ 1.1] NVRM: GE-CTRL {start:03x}:"
+            + "".join(f" {b:02x}" for b in piece)
+        )
+    lines.append("[ 1.2] NVRM: GE-GOLDEN-END status=0x0")
+    return "\n".join(lines) + "\n"
+
+
+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()
+
+
+class DecodeTests(unittest.TestCase):
+    def test_decodes_every_record_kind_in_order(self):
+        capture = decode_text(_sample_log(), source="sample")
+        kinds = [record.kind for record in capture.records]
+        self.assertEqual(
+            kinds,
+            ["golden", "golden", "alloc", "control", "send", "ctrl-done", "golden"],
+        )
+        trigger, begin = capture.records[0], capture.records[1]
+        self.assertEqual(trigger.marker, "trigger")
+        self.assertEqual(trigger.gspclient, 1)
+        self.assertEqual(trigger.mig, 0)
+        self.assertEqual(begin.marker, "begin")
+        alloc = capture.records[2]
+        self.assertEqual(alloc.hclass, 0x90F1)
+        self.assertEqual(alloc.hclient, 0xBABA0001)
+        self.assertEqual(alloc.hobject, 0xBABA1001)
+        self.assertEqual(alloc.size, 0x10)
+        self.assertEqual(alloc.caller, 0xFFFFFFFFC0B00000)
+        control = capture.records[3]
+        self.assertEqual(control.cmd, 0x2080012B)
+        self.assertEqual(control.params, bytes([1, 2, 3, 4, 5, 6, 7, 8]))
+        send = capture.records[4]
+        self.assertEqual(send.fn, 0x4C)
+        self.assertEqual(send.length, 0x28)
+        self.assertEqual(send.expected_params_size, 0x28 - 32)
+        self.assertEqual(capture.records[5].rpc_status, 0)
+        self.assertEqual(capture.records[6].status, 0)
+        self.assertEqual(capture.golden_status(), 0)
+        self.assertFalse(capture.warnings)
+
+    def test_dmesg_timestamp_and_prefix_are_stripped(self):
+        capture = decode_text(_sample_log(), source="sample")
+        self.assertEqual(capture.records[0].ts, " 100.000000")
+
+    def test_multichunk_payload_is_assembled(self):
+        capture = decode_text(_chunked_control_log(560), source="chunked")
+        control = capture.records[1]
+        self.assertEqual(len(control.params), 560)
+        self.assertFalse(control.truncated)
+        self.assertEqual(control.chunks[0].__len__(), 192)
+        self.assertIn(0xC0, control.chunks)
+        self.assertIn(0x180, control.chunks)
+        self.assertEqual(control.params[192], (192 * 5) & 0xFF)
+
+    def test_truncated_payload_is_flagged(self):
+        # The runbook helper caps at 192 bytes: declared 560, captured 192.
+        capture = decode_text(_chunked_control_log(192), source="truncated")
+        control = capture.records[1]
+        control.size = 560
+        self.assertTrue(control.truncated)
+        self.assertEqual(control.expected_params_size, 560)
+        self.assertEqual(len(control.params), 192)
+
+    def test_orphan_hex_warns_and_is_not_attached(self):
+        capture = decode_text("[ 1.0] NVRM: GE-CTRL 000: 01 02 03 04\n", source="orphan")
+        self.assertEqual(capture.records, [])
+        self.assertEqual(len(capture.warnings), 1)
+        self.assertIn("orphan CTRL hex", capture.warnings[0])
+
+    def test_wire_id_derived_from_send_payload(self):
+        control_send = decode_text(_sample_log(), source="s").records[4]
+        self.assertEqual(control_send.wire_id, 0x2080012B)
+        alloc_payload = (
+            (0xBABA0001).to_bytes(4, "little")
+            + (0x0).to_bytes(4, "little")
+            + (0xBABA1002).to_bytes(4, "little")
+            + (0x0040).to_bytes(4, "little")
+        )
+        log = (
+            "[ 1.0] NVRM: GE-SEND fn=0x00000067 len=0x30 ret=0x0\n"
+            "[ 1.0] NVRM: GE-SEND 000:" + "".join(f" {b:02x}" for b in alloc_payload) + "\n"
+        )
+        record = decode_text(log, source="a").records[0]
+        self.assertEqual(record.fn, GSP_RM_ALLOC)
+        self.assertEqual(record.wire_id, 0x0040)
+        self.assertNotEqual(GSP_RM_CONTROL, GSP_RM_ALLOC)
+
+    def test_overrun_is_an_anomaly(self):
+        log = (
+            "[ 1.0] NVRM: GE-CTRL cmd=0x1 hClient=0x0 hObject=0x0 size=0x4 "
+            "paramp=0x0 ret=0x0\n"
+            "[ 1.0] NVRM: GE-CTRL 000: 01 02 03 04 05 06 07 08\n"
+        )
+        record = decode_text(log, source="overrun").records[0]
+        self.assertTrue(record.overrun)
+        self.assertTrue(any("> declared" in item for item in record.anomalies))
+
+    def test_golden_status_none_without_end_record(self):
+        log = "[ 1.0] NVRM: GE-GOLDEN-BEGIN\n"
+        self.assertIsNone(decode_text(log, source="x").golden_status())
+
+
+class CaptureIOTests(unittest.TestCase):
+    def test_decode_write_load_roundtrip(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            log_path = os.path.join(tmp, "capture.log")
+            with open(log_path, "w", encoding="utf-8") as fh:
+                fh.write(_sample_log())
+            out_dir = os.path.join(tmp, "out")
+            payload = run_decode(log_path, out_dir)
+            self.assertEqual(payload["format"], CAPTURE_FORMAT)
+            self.assertTrue(os.path.isfile(payload["artifacts"]["capture_json"]))
+            self.assertTrue(os.path.isfile(payload["artifacts"]["capture_md"]))
+            loaded = load_capture(payload["artifacts"]["capture_json"])
+            self.assertEqual(len(loaded.records), len(decode_file(log_path).records))
+            self.assertEqual(loaded.records[3].cmd, 0x2080012B)
+
+    def test_load_capture_autodetects_raw_log(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            log_path = os.path.join(tmp, "raw.log")
+            with open(log_path, "w", encoding="utf-8") as fh:
+                fh.write(_sample_log())
+            loaded = load_capture(log_path)
+            self.assertEqual(loaded.records[0].kind, "golden")
+
+    def test_load_capture_rejects_unknown_format(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            path = os.path.join(tmp, "bad.json")
+            with open(path, "w", encoding="utf-8") as fh:
+                json.dump({"format": "something-else", "records": []}, fh)
+            with self.assertRaises(RmCaptureError):
+                load_capture(path)
+
+    def test_missing_file_is_clean_error(self):
+        with self.assertRaises(RmCaptureError):
+            decode_file("/nonexistent/capture.log")
+
+
+class CompareTests(unittest.TestCase):
+    def _captures(self):
+        capture_a = decode_text(_sample_log(), source="a")
+        capture_b = decode_text(_sample_log(), source="b")
+        return capture_a, capture_b
+
+    def test_identical_captures_fully_agree(self):
+        capture_a, capture_b = self._captures()
+        report = compare(capture_a, capture_b)
+        self.assertEqual(report["format"], COMPARE_FORMAT)
+        summary = report["summary"]
+        self.assertEqual(summary["mismatch"], 0)
+        self.assertEqual(summary["a_only"], 0)
+        self.assertEqual(summary["b_only"], 0)
+        self.assertEqual(summary["agree"], len(capture_a.records))
+        self.assertEqual(summary["agreement_pct"], 100.0)
+        self.assertEqual(summary["coverage_pct"], 100.0)
+
+    def test_dropped_record_is_a_only(self):
+        capture_a, capture_b = self._captures()
+        capture_b.records = [
+            record for record in capture_b.records if record.kind != "alloc"
+        ]
+        report = compare(capture_a, capture_b)
+        self.assertEqual(report["summary"]["a_only"], 1)
+        self.assertEqual(report["a_only"][0]["label"].startswith("ALLOC"), True)
+
+    def test_param_byte_flip_is_shape_agree(self):
+        capture_a, capture_b = self._captures()
+        control = next(r for r in capture_b.records if r.kind == "control")
+        control.chunks[0] = bytes([9]) + control.chunks[0][1:]
+        report = compare(capture_a, capture_b)
+        self.assertEqual(report["summary"]["shape_agree"], 1)
+        row = next(r for r in report["rows"] if r["status"] == "shape-agree")
+        self.assertEqual(row["detail"]["first_diff_offset"], 0)
+        self.assertLess(row["detail"]["byte_agreement_pct"], 100.0)
+
+    def test_truncated_pair_is_partial(self):
+        capture_a = decode_text(_chunked_control_log(192), source="a")
+        capture_b = decode_text(_chunked_control_log(192), source="b")
+        for capture in (capture_a, capture_b):
+            capture.records[1].size = 560
+        report = compare(capture_a, capture_b)
+        self.assertEqual(report["summary"]["partial"], 1)
+        self.assertTrue(report["rows"][1]["a_truncated"])
+
+    def test_compare_is_deterministic_and_self_describing(self):
+        capture_a, capture_b = self._captures()
+        report_one = compare(capture_a, capture_b)
+        report_two = compare(capture_a, capture_b)
+        self.assertEqual(canonical_json(report_one), canonical_json(report_two))
+        self.assertEqual(report_one["format"], COMPARE_FORMAT)
+        self.assertIn("L2", report_one["fidelity"]["levels"])
+        self.assertTrue(report_one["limits"])
+
+    def test_run_compare_writes_json_and_markdown(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            log_path = os.path.join(tmp, "capture.log")
+            with open(log_path, "w", encoding="utf-8") as fh:
+                fh.write(_sample_log())
+            report = run_compare(log_path, log_path, os.path.join(tmp, "out"))
+            self.assertTrue(os.path.isfile(report["artifacts"]["report_json"]))
+            self.assertTrue(os.path.isfile(report["artifacts"]["report_md"]))
+            with open(report["artifacts"]["report_md"], "r", encoding="utf-8") as fh:
+                self.assertIn("# GSP RPC capture comparison", fh.read())
+
+
+class StimulusTests(unittest.TestCase):
+    def test_stimulus_lists_rpcs_in_order_without_handles(self):
+        capture = decode_text(_sample_log(), source="s")
+        stimulus = stimulus_from_capture(capture)
+        self.assertEqual(stimulus["format"], STIMULUS_FORMAT)
+        kinds = [message["kind"] for message in stimulus["messages"]]
+        self.assertEqual(kinds, ["alloc", "control", "send"])
+        self.assertEqual(stimulus["messages"][1]["cmd"], 0x2080012B)
+        self.assertEqual(stimulus["messages"][2]["fn"], GSP_RM_CONTROL)
+        self.assertEqual(stimulus["messages"][2]["wire_id"], 0x2080012B)
+        self.assertNotIn("hclient", stimulus["messages"][1])
+        self.assertEqual(len(stimulus["markers"]), 3)
+        self.assertTrue(stimulus["limits"])
+
+    def test_stimulus_roundtrips_params_and_flags_truncation(self):
+        capture = decode_text(_chunked_control_log(192), source="s")
+        capture.records[1].size = 560
+        stimulus = stimulus_from_capture(capture)
+        message = stimulus["messages"][0]
+        self.assertTrue(message["truncated"])
+        self.assertEqual(message["params_size_expected"], 560)
+        self.assertEqual(bytes.fromhex(message["params_hex"]), capture.records[1].params)
+
+    def test_run_stimulus_writes_file(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            out_path = os.path.join(tmp, "nested", "stimulus.json")
+            log_path = os.path.join(tmp, "capture.log")
+            with open(log_path, "w", encoding="utf-8") as fh:
+                fh.write(_sample_log())
+            payload = run_stimulus(log_path, out_path)
+            self.assertTrue(os.path.isfile(out_path))
+            self.assertEqual(payload["format"], STIMULUS_FORMAT)
+            with open(out_path, "r", encoding="utf-8") as fh:
+                written = json.load(fh)
+            self.assertEqual(written["format"], STIMULUS_FORMAT)
+
+
+class DemoAndCliTests(unittest.TestCase):
+    def test_demo_is_ok_and_within_threshold(self):
+        payload = run_demo()
+        self.assertEqual(payload["format"], DEMO_FORMAT)
+        self.assertTrue(payload["ok"])
+        self.assertLessEqual(
+            payload["wall_seconds"],
+            MANUAL_DECODE_BASELINE_SECONDS * WORKFLOW_THRESHOLD_FRACTION,
+        )
+        self.assertGreaterEqual(payload["summary"]["agree"], 1)
+        self.assertEqual(payload["summary"]["shape_agree"], 1)
+        self.assertEqual(payload["summary"]["a_only"], 1)
+
+    def test_demo_is_deterministic_apart_from_timing(self):
+        first = run_demo()
+        second = run_demo()
+        for payload in (first, second):
+            payload.pop("wall_seconds", None)
+            payload.pop("artifacts", None)
+        self.assertEqual(canonical_json(first), canonical_json(second))
+
+    def test_demo_writes_artifacts(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            payload = run_demo(tmp)
+            for key in ("capture_a", "replay_b", "compare", "stimulus"):
+                self.assertTrue(os.path.isfile(payload["artifacts"][key]))
+
+    def test_cli_decode_compare_stimulus_return_zero(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            log_path = os.path.join(tmp, "capture.log")
+            with open(log_path, "w", encoding="utf-8") as fh:
+                fh.write(_sample_log())
+            rc, out, _ = _run_main(["decode", "--log", log_path, "--out", tmp])
+            self.assertEqual(rc, 0)
+            self.assertEqual(json.loads(out)["format"], CAPTURE_FORMAT)
+            rc, out, _ = _run_main(
+                ["compare", "--a", log_path, "--b", log_path, "--out", tmp]
+            )
+            self.assertEqual(rc, 0)
+            self.assertEqual(json.loads(out)["format"], COMPARE_FORMAT)
+            rc, out, _ = _run_main(
+                ["stimulus", "--capture", log_path, "--out", os.path.join(tmp, "s.json")]
+            )
+            self.assertEqual(rc, 0)
+            self.assertEqual(json.loads(out)["format"], STIMULUS_FORMAT)
+
+    def test_cli_missing_input_returns_two(self):
+        rc, _, err = _run_main(["decode", "--log", "/nonexistent/capture.log"])
+        self.assertEqual(rc, 2)
+        self.assertIn("error:", err)
+
+    def test_hints_execute_nothing_and_name_the_rules(self):
+        out = io.StringIO()
+        with contextlib.redirect_stdout(out):
+            print_capture_hints()
+        text = out.getvalue()
+        self.assertIn("executes", text)
+        self.assertIn("NOTHING", text)
+        self.assertIn("GE-SEND", text)
+        self.assertIn("192 bytes", text)
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/debug/risc-v-gsp-rm-vm/tests/test_server_snapshot.py b/debug/risc-v-gsp-rm-vm/tests/test_server_snapshot.py
new file mode 100644
index 0000000..b82d09c
--- /dev/null
+++ b/debug/risc-v-gsp-rm-vm/tests/test_server_snapshot.py
@@ -0,0 +1,364 @@
+"""Tests for the server-state snapshot collector (harness/server_snapshot.py).
+
+Crafted evidence covers every supported shape (``LABEL[offset]=0xVALUE``,
+labelled hexdump blocks, ``LOGINIT-raw`` headers, record lines, duplicate
+labels and unparseable dump-looking lines).  Optional tests run against the
+real 2026-10-02 campaign backups when they are present; they are skipped
+otherwise.  All tests are offline, hermetic and touch no hardware.
+
+Run::
+
+    python3 -m unittest tests.test_server_snapshot -v
+"""
+from __future__ import annotations
+
+import contextlib
+import io
+import json
+import os
+import sys
+import tempfile
+import unittest
+from unittest import mock
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+TOOL_ROOT = os.path.abspath(os.path.join(HERE, ".."))
+if TOOL_ROOT not in sys.path:
+    sys.path.insert(0, TOOL_ROOT)
+
+from harness.server_snapshot import (  # noqa: E402
+    DIFF_FORMAT,
+    SNAPSHOT_FORMAT,
+    SnapshotError,
+    collect,
+    diff_snapshots,
+    main,
+)
+
+EVIDENCE_VICTORY = ("/home/rmcore/Git/integdev_gpu_drv/open-iov/backups/"
+                    "evidence-20261002-victory")
+LADDER_M12 = os.path.join(EVIDENCE_VICTORY, "06-ladder-m12.txt")
+PHYS_STATE_M26 = os.path.join(EVIDENCE_VICTORY, "20-gsp-phys-state-m26.txt")
+EVIDENCE_GR = ("/home/rmcore/Git/integdev_gpu_drv/open-iov/backups/"
+               "evidence-20261002-gr")
+LOGDUMP_GRCTX4 = os.path.join(EVIDENCE_GR, "21-logdump-grctx4.txt")
+
+FIXED_CREATED = "2026-01-01T00:00:00Z"
+
+
+def _run_main(argv):
+    out, err = io.StringIO(), io.StringIO()
+    with contextlib.redirect_stdout(out), contextlib.redirect_stderr(err):
+        rc = main(argv)
+    return rc, out.getvalue(), err.getvalue()
+
+
+class ServerSnapshotTests(unittest.TestCase):
+    """Collector normalization, collision, diff, CLI and determinism."""
+
+    def setUp(self):
+        holder = tempfile.TemporaryDirectory(prefix="server-snapshot-test-")
+        self.addCleanup(holder.cleanup)
+        self.tmp = holder.name
+
+    def _write(self, name, text):
+        path = os.path.join(self.tmp, name)
+        with open(path, "w", encoding="utf-8") as handle:
+            handle.write(text)
+        return path
+
+    def _collect_text(self, name, text, **kwargs):
+        path = self._write(name, text)
+        kwargs.setdefault("source", f"test:{name}")
+        kwargs.setdefault("created", FIXED_CREATED)
+        snapshot, stats = collect(files=[path], **kwargs)
+        return snapshot, stats, path
+
+    @staticmethod
+    def _chunks(snapshot):
+        return {chunk["label"]: chunk for chunk in snapshot["chunks"]}
+
+    # -- crafted shapes ----------------------------------------------------
+
+    def test_labeled_value_lines(self):
+        snapshot, stats, _ = self._collect_text(
+            "values.txt",
+            "alpha[0]=0x00000011\n"
+            "alpha[4]=0x00000022\n"
+            "beta[8]=0x0000000112345678\n",
+        )
+        chunks = self._chunks(snapshot)
+        self.assertEqual(chunks["alpha"]["base"], "0x0")
+        self.assertEqual(chunks["alpha"]["words"],
+                         ["0x00000011", "0x00000022"])
+        self.assertEqual(chunks["beta"]["base"], "0x8")
+        self.assertEqual(chunks["beta"]["words"],
+                         ["0x12345678", "0x00000001"])
+        self.assertEqual(stats["unparsed"], 0)
+        self.assertEqual(stats["conflicts"], 0)
+
+    def test_labeled_hexdump_block(self):
+        snapshot, stats, _ = self._collect_text(
+            "block.txt",
+            "preamble text that is not a header\n"
+            "SOME_BUF: 64 bytes of server state\n"
+            "00000000: 00000001 00000002 00000003 00000004\n"
+            "00000010: 00000005 00000006 00000007 00000008\n"
+            "RAMFC: a second block with a nonzero base\n"
+            "00000020: 0000000a 0000000b\n",
+        )
+        chunks = self._chunks(snapshot)
+        self.assertEqual(chunks["SOME_BUF"]["base"], "0x0")
+        self.assertEqual(
+            chunks["SOME_BUF"]["words"],
+            ["0x00000001", "0x00000002", "0x00000003", "0x00000004",
+             "0x00000005", "0x00000006", "0x00000007", "0x00000008"])
+        self.assertEqual(chunks["RAMFC"]["base"], "0x20")
+        self.assertEqual(chunks["RAMFC"]["words"],
+                         ["0x0000000a", "0x0000000b"])
+        self.assertEqual(stats["unparsed"], 0)
+        self.assertEqual(stats["chunks"], 2)
+
+    def test_logininit_raw_block(self):
+        snapshot, stats, _ = self._collect_text(
+            "loginit.txt",
+            "vgsp-boot: libos task log buffers (read-only)\n"
+            "LOGINIT-raw: 65536 bytes (allocated 65536) "
+            "VA=0xffff88a673af0000 dma=0x4f3af0000\n"
+            "00000000: 000002d2 00000000 05bb9e10 00000000\n"
+            "00000010: f7660780 18dbb28f fe2c0000 61800001\n",
+        )
+        chunk = self._chunks(snapshot)["LOGINIT-raw"]
+        self.assertEqual(chunk["base"], "0x0")
+        self.assertEqual(
+            chunk["words"],
+            ["0x000002d2", "0x00000000", "0x05bb9e10", "0x00000000",
+             "0xf7660780", "0x18dbb28f", "0xfe2c0000", "0x61800001"])
+        self.assertEqual(stats["unparsed"], 0)
+
+    def test_inline_labeled_rows_group_into_one_chunk(self):
+        snapshot, _, _ = self._collect_text(
+            "inline.txt",
+            "[dmesg] vgsp-boot: rm-call: INSTANCE[0000]: "
+            "00000000 00000000 0a545003 00000001\n"
+            "[dmesg] vgsp-boot: rm-call: INSTANCE[0020]: "
+            "0000face 00000000 00000000 00000000\n",
+        )
+        chunk = self._chunks(snapshot)["INSTANCE"]
+        self.assertEqual(chunk["base"], "0x0")
+        self.assertEqual(chunk["words"],
+                         ["0x00000000", "0x00000000", "0x0a545003",
+                          "0x00000001", "0x0000face", "0x00000000",
+                          "0x00000000", "0x00000000"])
+
+    def test_record_lines(self):
+        snapshot, stats, _ = self._collect_text(
+            "records.txt",
+            'vgsp-nocat: rec seq=1 recType=5 errorCode=0x57f1344 '
+            'subsystem=0x0 bugcheck=0x0 payload=1212 source="ASSERT" '
+            'faultingEngine=""\n',
+        )
+        self.assertEqual(len(snapshot["records"]), 1)
+        record = snapshot["records"][0]
+        self.assertEqual(record["kind"], "record")
+        self.assertEqual(record["rec_type"], 5)
+        self.assertEqual(record["error_code"], 0x57F1344)
+        self.assertEqual(record["source_name"], "ASSERT")
+        self.assertEqual(stats["records"], 1)
+        self.assertEqual(snapshot["chunks"], [])
+
+    # -- duplicates, collisions, unparsed ---------------------------------
+
+    def test_identical_duplicate_merged(self):
+        snapshot, stats, _ = self._collect_text(
+            "merged.txt",
+            "INSTANCE[0000]: 00000001 00000002\n"
+            "plain prose closes the first block\n"
+            "INSTANCE[0000]: 00000001 00000002\n",
+        )
+        self.assertEqual(list(self._chunks(snapshot)), ["INSTANCE"])
+        self.assertEqual(stats["merged"], 1)
+        self.assertEqual(stats["collisions"], 0)
+
+    def test_conflicting_duplicate_renamed(self):
+        snapshot, stats, _ = self._collect_text(
+            "collision.txt",
+            "INSTANCE[0000]: 00000001 00000002\n"
+            "plain prose closes the first block\n"
+            "INSTANCE[0000]: 00000003 00000004\n",
+        )
+        self.assertEqual(list(self._chunks(snapshot)),
+                         ["INSTANCE", "INSTANCE#2"])
+        self.assertEqual(stats["collisions"], 1)
+        self.assertEqual(stats["collision_list"],
+                         [{"label": "INSTANCE", "renamed": "INSTANCE#2",
+                           "base": "0x0", "words": 2}])
+
+    def test_unparsed_candidates_counted(self):
+        snapshot, stats, _ = self._collect_text(
+            "unparsed.txt",
+            "BROKEN[0x10]: zzzz\n"
+            "00000000: zz00\n"
+            "ordinary prose is not a candidate\n",
+        )
+        self.assertEqual(snapshot["chunks"], [])
+        self.assertEqual(stats["unparsed"], 2)
+
+    # -- diff --------------------------------------------------------------
+
+    def test_diff_changed_added_removed(self):
+        snap_a, _, _ = self._collect_text(
+            "run-a.txt",
+            "MOCK[0000]: 00000001 00000002\n"
+            "shared[0]=0x00000007\n"
+            "only_a[0]=0x00000009\n",
+        )
+        snap_b, _, _ = self._collect_text(
+            "run-b.txt",
+            "MOCK[0000]: 00000001 00000003 00000004\n"
+            "shared[0]=0x00000007\n"
+            "only_b[0]=0x0000000a\n",
+        )
+        report = diff_snapshots(snap_a, snap_b)
+        self.assertEqual(report["schema"], DIFF_FORMAT)
+        rows = {row["label"]: row for row in report["chunks"]}
+        self.assertEqual(rows["MOCK"]["status"], "changed")
+        self.assertEqual(rows["MOCK"]["changed"],
+                         [{"index": 1, "a": "0x00000002", "b": "0x00000003"}])
+        self.assertEqual(rows["MOCK"]["added"],
+                         [{"index": 2, "value": "0x00000004"}])
+        self.assertEqual(rows["MOCK"]["removed"], [])
+        self.assertEqual(rows["shared"]["status"], "unchanged")
+        self.assertEqual(rows["only_a"]["status"], "removed")
+        self.assertEqual(rows["only_a"]["removed"],
+                         [{"index": 0, "value": "0x00000009"}])
+        self.assertEqual(rows["only_b"]["status"], "added")
+        self.assertEqual(rows["only_b"]["added"],
+                         [{"index": 0, "value": "0x0000000a"}])
+        summary = report["summary"]
+        self.assertFalse(summary["identical"])
+        self.assertEqual(summary["chunks_added"], 1)
+        self.assertEqual(summary["chunks_removed"], 1)
+        self.assertEqual(summary["chunks_changed"], 1)
+        self.assertEqual(summary["chunks_unchanged"], 1)
+        self.assertEqual(summary["words_added"], 2)
+        self.assertEqual(summary["words_removed"], 1)
+        self.assertEqual(summary["words_changed"], 1)
+
+    def test_diff_identical(self):
+        snapshot, _, _ = self._collect_text(
+            "same.txt", "INSTANCE[0000]: 00000001 00000002\n")
+        report = diff_snapshots(snapshot, snapshot)
+        self.assertTrue(report["summary"]["identical"])
+        self.assertEqual(report["summary"]["words_changed"], 0)
+        self.assertEqual([row["status"] for row in report["chunks"]],
+                         ["unchanged"])
+
+    def test_diff_rejects_wrong_schema(self):
+        with self.assertRaises(SnapshotError):
+            diff_snapshots({"schema": "other"}, {"schema": SNAPSHOT_FORMAT})
+
+    # -- determinism and CLI ----------------------------------------------
+
+    def test_determinism_with_source_date_epoch(self):
+        path = self._write(
+            "det.txt",
+            "INSTANCE[0000]: 00000001 00000002\n"
+            'vgsp-nocat: rec seq=1 recType=5 errorCode=0x57f1344 '
+            'source="ASSERT"\n',
+        )
+        with mock.patch.dict(os.environ, {"SOURCE_DATE_EPOCH": "1700000000"}):
+            first, _ = collect(files=[path], source="det")
+            second, _ = collect(files=[path], source="det")
+        self.assertEqual(first["created"], "2023-11-14T22:13:20Z")
+        self.assertEqual(first, second)
+        self.assertEqual(json.dumps(first, sort_keys=True),
+                         json.dumps(second, sort_keys=True))
+
+    def test_cli_collect_stdout_and_json(self):
+        path = self._write("cli.txt", "INSTANCE[0000]: 00000001 00000002\n")
+        rc, out, err = _run_main(
+            ["collect", "--file", path, "--source", "cli",
+             "--created", FIXED_CREATED])
+        self.assertEqual(rc, 0)
+        snapshot = json.loads(out)
+        self.assertEqual(set(snapshot),
+                         {"schema", "source", "created", "chunks", "records"})
+        self.assertEqual(snapshot["schema"], SNAPSHOT_FORMAT)
+        self.assertEqual(snapshot["source"], "cli")
+        self.assertIn("1 chunk(s)", err)
+
+        json_path = os.path.join(self.tmp, "snapshot.json")
+        rc, out, _ = _run_main(
+            ["collect", "--file", path, "--json", json_path,
+             "--created", FIXED_CREATED])
+        self.assertEqual(rc, 0)
+        with open(json_path, "r", encoding="utf-8") as handle:
+            written = json.load(handle)
+        self.assertEqual(written["schema"], SNAPSHOT_FORMAT)
+        self.assertEqual(json.loads(out)["chunks"], len(written["chunks"]))
+
+    def test_cli_errors(self):
+        rc, _, err = _run_main(["collect"])
+        self.assertEqual(rc, 2)
+        self.assertIn("error:", err)
+        missing = os.path.join(self.tmp, "missing.txt")
+        rc, _, _ = _run_main(["collect", "--file", missing])
+        self.assertEqual(rc, 2)
+
+    def test_discover_include_globs(self):
+        self._write("keep.txt", "AAA[0]=0x00000001\n")
+        self._write("skip.bin", "BBB[0]=0x00000002\n")
+        snapshot, _ = collect(dirs=[self.tmp], source="glob",
+                              created=FIXED_CREATED)
+        self.assertEqual(list(self._chunks(snapshot)), ["AAA"])
+        snapshot, _ = collect(dirs=[self.tmp], includes=["*.bin"],
+                              source="glob", created=FIXED_CREATED)
+        self.assertEqual(list(self._chunks(snapshot)), ["AAA", "BBB"])
+
+    def test_demo_cli(self):
+        rc, out, _ = _run_main(["demo"])
+        self.assertEqual(rc, 0)
+        self.assertIn("MOCK", out)
+        summary = json.loads(out[out.index("{"):])
+        self.assertFalse(summary["identical"])
+        self.assertEqual(summary["words_changed"], 2)
+
+    # -- real campaign evidence (optional) --------------------------------
+
+    @unittest.skipUnless(os.path.isdir(EVIDENCE_VICTORY),
+                         "victory evidence backup not present")
+    def test_real_victory_evidence(self):
+        files = [path for path in (LADDER_M12, PHYS_STATE_M26)
+                 if os.path.isfile(path)]
+        if not files:
+            self.skipTest("expected victory evidence files missing")
+        snapshot, stats = collect(files=files, source="evidence-victory",
+                                  created=FIXED_CREATED)
+        self.assertTrue(snapshot["chunks"])
+        chunks = self._chunks(snapshot)
+        self.assertIn("INSTANCE", chunks)
+        self.assertEqual(chunks["INSTANCE"]["base"], "0x0")
+        self.assertEqual(len(chunks["INSTANCE"]["words"]), 1024)
+        self.assertIn("pt", chunks)
+        self.assertEqual(len(chunks["pt"]["words"]), 5)
+        self.assertGreaterEqual(stats["records"], 1)
+        self.assertEqual(stats["files"], len(files))
+
+    @unittest.skipUnless(os.path.isfile(LOGDUMP_GRCTX4),
+                         "gr logdump evidence not present")
+    def test_real_loginit_raw_block(self):
+        snapshot, stats = collect(files=[LOGDUMP_GRCTX4],
+                                  source="evidence-gr",
+                                  created=FIXED_CREATED)
+        chunks = self._chunks(snapshot)
+        self.assertIn("LOGINIT-raw", chunks)
+        self.assertIn("LOGRM-raw", chunks)
+        self.assertEqual(chunks["LOGINIT-raw"]["base"], "0x0")
+        self.assertEqual(len(chunks["LOGINIT-raw"]["words"]), 65536 // 4)
+        self.assertEqual(len(chunks["LOGRM-raw"]["words"]), 65536 // 4)
+        self.assertEqual(stats["unparsed"], 0)
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/debug/risc-v-gsp-rm-vm/tests/test_state_decode.py b/debug/risc-v-gsp-rm-vm/tests/test_state_decode.py
new file mode 100644
index 0000000..5c05ccd
--- /dev/null
+++ b/debug/risc-v-gsp-rm-vm/tests/test_state_decode.py
@@ -0,0 +1,518 @@
+"""Tests for the engine-visible state decoder (harness/state_decode.py).
+
+Offline only: text/hexdump/binary/snapshot-JSON parsing, the bundled
+INSTANCE/RAMFC field table, map override/extension, decode/diff reports and
+the CLI.  The real-evidence test parses a frozen live ladder trace and skips
+when the backup is absent.  No hardware, no module, no guest, no network.
+
+Run::
+
+    python3 -m unittest tests.test_state_decode -v
+"""
+from __future__ import annotations
+
+import contextlib
+import io
+import json
+import os
+import struct
+import sys
+import tempfile
+import unittest
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+TOOL_ROOT = os.path.abspath(os.path.join(HERE, ".."))
+if TOOL_ROOT not in sys.path:
+    sys.path.insert(0, TOOL_ROOT)
+
+from harness.state_decode import (  # noqa: E402
+    DECODE_FORMAT,
+    DEMO_FORMAT,
+    DIFF_FORMAT,
+    MAP_FORMAT,
+    SNAPSHOT_FORMAT,
+    StateDecodeError,
+    decode_chunk,
+    decode_specs,
+    diff_reports,
+    load_input,
+    load_map,
+    main,
+    run_demo,
+)
+
+EVIDENCE = os.environ.get("GREENEYE_STATE_DECODE_EVIDENCE") or os.path.normpath(
+    os.path.join(TOOL_ROOT, "..", "..", "backups",
+                 "evidence-20261002-victory", "06-ladder-m12.txt"))
+
+
+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 _write(path: str, text: str) -> str:
+    with open(path, "w", encoding="utf-8") as handle:
+        handle.write(text)
+    return path
+
+
+def _write_snapshot(tmp: str, name: str, chunks: list[dict]) -> str:
+    path = os.path.join(tmp, name)
+    with open(path, "w", encoding="utf-8") as handle:
+        json.dump({"schema": SNAPSHOT_FORMAT, "source": name,
+                   "chunks": chunks}, handle)
+    return path
+
+
+def _write_map(tmp: str, name: str, doc) -> str:
+    path = os.path.join(tmp, name)
+    with open(path, "w", encoding="utf-8") as handle:
+        json.dump(doc, handle)
+    return path
+
+
+def _report_from_chunks(chunks: list[dict], map_path: str | None = None) -> dict:
+    entries, source = load_map(map_path)
+    return {
+        "format": DECODE_FORMAT,
+        "map_source": source,
+        "sources": [],
+        "chunks": [decode_chunk(chunk, entries) for chunk in chunks],
+        "limits": [],
+    }
+
+
+def _chunk_field(report: dict, label: str, name: str) -> dict:
+    for chunk in report["chunks"]:
+        if chunk["label"] != label:
+            continue
+        for field in chunk["fields"]:
+            if field["name"] == name:
+                return field
+    raise AssertionError(f"field {name} not found in chunk {label}")
+
+
+def _spec(path: str, label: str | None = None, base=None,
+          binary: bool = False) -> dict:
+    return {"path": path, "label": label, "base": base, "binary": binary}
+
+
+class ParsingTests(unittest.TestCase):
+    def test_text_lines_parse_labels_offsets_and_values(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            path = _write(os.path.join(tmp, "lines.txt"), "\n".join([
+                "banner line that must be ignored",
+                "INSTANCE[0x10]=0xface",
+                "mid-text noise",
+                "Instance[00ac]=0x00030000",
+                "INSTANCE[72]=0x1",      # decimal offset 72 == 0x48
+                "INSTANCE[0x4c]=64206",  # decimal value 64206 == 0xface
+                "",
+            ]))
+            document = load_input(_spec(path))
+            self.assertEqual(document["format"], "text")
+            self.assertEqual(len(document["chunks"]), 1)
+            chunk = document["chunks"][0]
+            self.assertEqual(chunk["label"], "INSTANCE")  # first spelling wins
+            self.assertEqual(chunk["words"][0x10], 0xFACE)
+            self.assertEqual(chunk["words"][0xAC], 0x00030000)
+            self.assertEqual(chunk["words"][0x48], 0x1)
+            self.assertEqual(chunk["words"][0x4C], 64206)
+
+    def test_hexdump_rows_parse_with_label_and_base(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            path = _write(os.path.join(tmp, "dump.txt"), "\n".join([
+                "00000000: 000002d2 00000000 00000000 00000000",
+                "00000010: 0000face 00000000 00000000 00000000",
+            ]))
+            document = load_input(_spec(path, label="INSTANCE",
+                                       base="0x1f4000000"))
+            chunk = document["chunks"][0]
+            self.assertEqual(chunk["label"], "INSTANCE")
+            self.assertEqual(chunk["base"], 0x1F4000000)
+            self.assertEqual(chunk["words"][0x0], 0x2D2)
+            self.assertEqual(chunk["words"][0x4], 0x0)
+            self.assertEqual(chunk["words"][0x10], 0xFACE)
+
+    def test_raw_binary_parses_little_endian_words(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            path = os.path.join(tmp, "ramfc.bin")
+            with open(path, "wb") as handle:
+                handle.write(struct.pack("<4I", 0x2D2, 0, 0x30000, 0))
+            document = load_input(_spec(path, label="RAMFC", binary=True))
+            self.assertEqual(document["format"], "binary")
+            chunk = document["chunks"][0]
+            self.assertEqual(chunk["label"], "RAMFC")
+            self.assertEqual(chunk["words"][0x0], 0x2D2)
+            self.assertEqual(chunk["words"][0x8], 0x30000)
+
+    def test_raw_binary_is_auto_detected(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            path = os.path.join(tmp, "auto.bin")
+            with open(path, "wb") as handle:
+                handle.write(struct.pack("<8I", 1, 2, 3, 4, 5, 6, 7, 8))
+            document = load_input(_spec(path, label="WINDOW"))
+            self.assertEqual(document["format"], "binary")
+            chunk = document["chunks"][0]
+            self.assertEqual(chunk["words"][0x1C], 8)
+
+    def test_snapshot_json_parses_exactly(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            path = _write_snapshot(tmp, "snap.json", [{
+                "label": "RAMFC", "base": "0x1000",
+                "words": ["0x2d2", 0, "0x30000", "0x0"],
+            }])
+            document = load_input(_spec(path))
+            self.assertEqual(document["format"], "snapshot-json")
+            chunk = document["chunks"][0]
+            self.assertEqual(chunk["label"], "RAMFC")
+            self.assertEqual(chunk["base"], 0x1000)
+            self.assertEqual(chunk["words"][0x0], 0x2D2)
+            self.assertEqual(chunk["words"][0x8], 0x30000)
+
+    def test_snapshot_schema_rejected(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            path = os.path.join(tmp, "other.json")
+            with open(path, "w", encoding="utf-8") as handle:
+                json.dump({"schema": "something-else/1", "chunks": []}, handle)
+            with self.assertRaises(StateDecodeError):
+                load_input(_spec(path))
+
+    def test_unparseable_text_rejected(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            path = _write(os.path.join(tmp, "nothing.txt"),
+                          "just prose\nwith no fields at all\n")
+            with self.assertRaises(StateDecodeError):
+                load_input(_spec(path))
+
+    def test_bad_binary_size_rejected(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            path = os.path.join(tmp, "odd.bin")
+            with open(path, "wb") as handle:
+                handle.write(b"\x01\x02\x03")
+            with self.assertRaises(StateDecodeError):
+                load_input(_spec(path, binary=True))
+
+
+class DecodeTests(unittest.TestCase):
+    def _ramfc_target(self, word48: int) -> dict:
+        report = _report_from_chunks([
+            {"label": "RAMFC", "base": 0, "words": {0x48: word48}},
+        ])
+        return _chunk_field(report, "RAMFC", "RAMFC_TARGET")
+
+    def test_ramfc_target_both_bits_set(self):
+        field = self._ramfc_target(0x00030000)
+        self.assertEqual(field["offset"], "0x48")
+        self.assertEqual(field["raw"], "0x00030000")
+        self.assertEqual(field["kind"], "bits")
+        bits = {bit["name"]: bit for bit in field["bitfields"]}
+        self.assertEqual(bits["ENG_CTX_VALID"]["value"], 1)
+        self.assertEqual(bits["ENG_CTX_VALID"]["state"], "set")
+        self.assertEqual(bits["CE_CTX_VALID"]["value"], 1)
+        self.assertEqual(bits["CE_CTX_VALID"]["state"], "set")
+
+    def test_ramfc_target_only_eng_bit16(self):
+        field = self._ramfc_target(0x00010000)
+        bits = {bit["name"]: bit for bit in field["bitfields"]}
+        self.assertEqual(bits["ENG_CTX_VALID"]["state"], "set")
+        self.assertEqual(bits["CE_CTX_VALID"]["state"], "clear")
+
+    def test_ramfc_target_only_ce_bit17(self):
+        field = self._ramfc_target(0x00020000)
+        bits = {bit["name"]: bit for bit in field["bitfields"]}
+        self.assertEqual(bits["ENG_CTX_VALID"]["state"], "clear")
+        self.assertEqual(bits["CE_CTX_VALID"]["state"], "set")
+
+    def test_instance_fields_and_unmapped_words(self):
+        report = _report_from_chunks([{
+            "label": "INSTANCE", "base": 0,
+            "words": {0x0: 0, 0x10: 0xFACE, 0x48: 0x100000, 0x4C: 0x30000},
+        }])
+        chunk = report["chunks"][0]
+        names = {field["name"] for field in chunk["fields"]}
+        self.assertEqual(names, {"MAGIC_0FACE", "GPFIFO_VA"})
+        self.assertEqual(chunk["word_count"], 4)
+        self.assertEqual(chunk["unmapped_count"], 2)
+        self.assertEqual({word["offset"] for word in chunk["unmapped"]},
+                         {"0x0", "0x4c"})
+        self.assertEqual(chunk["words"]["0x48"], "0x00100000")
+
+    def test_decode_report_format_and_determinism(self):
+        report = _report_from_chunks([{
+            "label": "INSTANCE", "base": 0, "words": {0x48: 0x100000},
+        }])
+        self.assertEqual(report["format"], DECODE_FORMAT)
+        again = _report_from_chunks([{
+            "label": "INSTANCE", "base": 0, "words": {0x48: 0x100000},
+        }])
+        self.assertEqual(json.dumps(report, sort_keys=True),
+                         json.dumps(again, sort_keys=True))
+
+
+class MapTests(unittest.TestCase):
+    def test_override_replaces_bundled_and_extends(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            path = _write_map(tmp, "map.json", {
+                "format": MAP_FORMAT,
+                "entries": [
+                    {"chunk": "INSTANCE", "label": "GPFIFO_VA_ALT",
+                     "offset": "0x48", "width": 4, "note": "override"},
+                    {"chunk": "INSTANCE", "label": "CTX_MARK",
+                     "offset": "0x4c", "width": 4,
+                     "bits": [
+                         {"name": "ENG", "lo": 16, "hi": 16},
+                         {"name": "CE", "lo": 17, "hi": 17},
+                         {"name": "MODE", "lo": 0, "hi": 7,
+                          "values": {"0": "zero", "1": "one"}},
+                     ]},
+                ],
+            })
+            report = _report_from_chunks([{
+                "label": "INSTANCE", "base": 0,
+                "words": {0x10: 0xFACE, 0x48: 0x100000, 0x4C: 0x00030001},
+            }], map_path=path)
+            chunk = report["chunks"][0]
+            names = [field["name"] for field in chunk["fields"]]
+            self.assertIn("GPFIFO_VA_ALT", names)
+            self.assertNotIn("GPFIFO_VA", names)
+            self.assertIn("MAGIC_0FACE", names)  # bundled table still extends
+            mark = _chunk_field(report, "INSTANCE", "CTX_MARK")
+            bits = {bit["name"]: bit for bit in mark["bitfields"]}
+            self.assertEqual(bits["MODE"]["state"], "one")
+            self.assertEqual(bits["ENG"]["state"], "set")
+            self.assertEqual(bits["CE"]["state"], "set")
+            self.assertEqual(report["map_source"], os.path.abspath(path))
+
+    def test_replace_mode_drops_bundled_table(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            path = _write_map(tmp, "replace.json", {
+                "format": MAP_FORMAT,
+                "mode": "replace",
+                "entries": [
+                    {"chunk": "RAMFC", "label": "TARGET_ALT",
+                     "offset": "0x48", "width": 4,
+                     "bits": [{"name": "VALID", "lo": 16, "hi": 16}]},
+                ],
+            })
+            report = _report_from_chunks([{
+                "label": "INSTANCE", "base": 0, "words": {0x10: 0xFACE},
+            }, {
+                "label": "RAMFC", "base": 0, "words": {0x48: 0x10000},
+            }], map_path=path)
+            self.assertEqual(report["chunks"][0]["fields"], [])
+            field = _chunk_field(report, "RAMFC", "TARGET_ALT")
+            self.assertEqual(field["bitfields"][0]["state"], "set")
+
+    def test_bad_map_rejected(self):
+        cases = [
+            {"entries": [{"chunk": "X", "label": "Y", "offset": "0x48",
+                          "width": 3}]},
+            {"entries": [{"chunk": "X", "label": "Y", "offset": "0x48",
+                          "width": 4,
+                          "bits": [{"name": "B", "lo": 0, "hi": 99}]}]},
+            {"entries": [{"chunk": "X", "label": "Y", "offset": "0x0",
+                          "width": 4}], "mode": "sideways"},
+        ]
+        for doc in cases:
+            with self.subTest(doc=doc):
+                with tempfile.TemporaryDirectory() as tmp:
+                    path = _write_map(tmp, "bad.json", doc)
+                    with self.assertRaises(StateDecodeError):
+                        load_map(path)
+
+
+class DiffTests(unittest.TestCase):
+    def test_diff_detects_changed_word_and_named_field(self):
+        left = _report_from_chunks([{
+            "label": "INSTANCE", "base": 0,
+            "words": {0x10: 0xFACE, 0x48: 0x100000},
+        }])
+        right = _report_from_chunks([{
+            "label": "INSTANCE", "base": 0,
+            "words": {0x10: 0xFACE, 0x48: 0x200000},
+        }])
+        report = diff_reports(left, right)
+        self.assertEqual(report["format"], DIFF_FORMAT)
+        self.assertFalse(report["identical"])
+        chunk = report["chunks"][0]
+        self.assertEqual(chunk["status"], "changed")
+        self.assertEqual(chunk["changed"],
+                         [{"offset": "0x48", "old": "0x00100000",
+                           "new": "0x00200000"}])
+        self.assertEqual(len(chunk["changed_fields"]), 1)
+        field = chunk["changed_fields"][0]
+        self.assertEqual(field["name"], "GPFIFO_VA")
+        self.assertEqual(field["old"], "0x00100000")
+        self.assertEqual(field["new"], "0x00200000")
+        self.assertEqual(report["summary"]["changed_words"], 1)
+        self.assertEqual(report["summary"]["changed_fields"], 1)
+
+    def test_diff_reports_bitfield_state_change(self):
+        left = _report_from_chunks([
+            {"label": "RAMFC", "base": 0, "words": {0x48: 0x00030000}}])
+        right = _report_from_chunks([
+            {"label": "RAMFC", "base": 0, "words": {0x48: 0x00010000}}])
+        report = diff_reports(left, right)
+        field = report["chunks"][0]["changed_fields"][0]
+        self.assertEqual(field["name"], "RAMFC_TARGET")
+        bits = {bit["name"]: bit for bit in field["bits"]}
+        self.assertEqual(bits["ENG_CTX_VALID"]["old"], "set")
+        self.assertEqual(bits["ENG_CTX_VALID"]["new"], "set")
+        self.assertEqual(bits["CE_CTX_VALID"]["old"], "set")
+        self.assertEqual(bits["CE_CTX_VALID"]["new"], "clear")
+
+    def test_diff_identical_is_empty_and_deterministic(self):
+        left = _report_from_chunks([{
+            "label": "INSTANCE", "base": 0, "words": {0x10: 0xFACE},
+        }])
+        right = _report_from_chunks([{
+            "label": "INSTANCE", "base": 0, "words": {0x10: 0xFACE},
+        }])
+        report = diff_reports(left, right)
+        self.assertTrue(report["identical"])
+        self.assertEqual(report["summary"]["changed_words"], 0)
+        again = diff_reports(left, right)
+        self.assertEqual(json.dumps(report, sort_keys=True),
+                         json.dumps(again, sort_keys=True))
+
+    def test_diff_added_and_removed_words(self):
+        left = _report_from_chunks([{
+            "label": "INSTANCE", "base": 0, "words": {0x0: 1, 0x4: 2},
+        }])
+        right = _report_from_chunks([{
+            "label": "INSTANCE", "base": 0, "words": {0x0: 1, 0x8: 3},
+        }])
+        report = diff_reports(left, right)
+        chunk = report["chunks"][0]
+        self.assertEqual([row["offset"] for row in chunk["removed"]], ["0x4"])
+        self.assertEqual([row["offset"] for row in chunk["added"]], ["0x8"])
+        self.assertEqual(report["summary"]["added"], 1)
+        self.assertEqual(report["summary"]["removed"], 1)
+
+
+class CliAndDemoTests(unittest.TestCase):
+    def test_cli_decode_prints_table_and_writes_json(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            path = _write(os.path.join(tmp, "dump.txt"), "\n".join([
+                "00000000: 00000000 00000000 00000000 00000000",
+                "00000010: 0000face 00000000 00000000 00000000",
+                "00000020: 00000000 00000000 00000000 00000000",
+                "00000030: 00000000 00000000 00000000 00000000",
+                "00000040: 00000000 00000000 00100000 00030000",
+            ]))
+            out_path = os.path.join(tmp, "decoded.json")
+            rc, out, err = _run_main(["decode", "--input", path,
+                                      "--label", "INSTANCE",
+                                      "--json", out_path])
+            self.assertEqual(rc, 0, err)
+            self.assertIn("GPFIFO_VA", out)
+            self.assertIn("0x00100000", out)
+            with open(out_path, encoding="utf-8") as handle:
+                report = json.load(handle)
+            self.assertEqual(report["format"], DECODE_FORMAT)
+            self.assertEqual(len(report["chunks"]), 1)
+            self.assertEqual(_chunk_field(report, "INSTANCE",
+                                         "GPFIFO_VA")["raw"], "0x00100000")
+
+    def test_cli_multiple_inputs_with_per_input_labels(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            left = _write(os.path.join(tmp, "left.txt"),
+                          "00000000: 00000001 00000000 00000000 00000000\n")
+            right = _write(os.path.join(tmp, "right.txt"),
+                           "00000000: 00000002 00000000 00000000 00000000\n")
+            rc, out, err = _run_main([
+                "decode", "--input", left, "--label", "LEFT",
+                "--input", right, "--label", "RIGHT"])
+            self.assertEqual(rc, 0, err)
+            self.assertIn("chunk LEFT", out)
+            self.assertIn("chunk RIGHT", out)
+
+    def test_cli_decode_text_lines_reports_canonical_words(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            path = _write(os.path.join(tmp, "lines.txt"),
+                          "INSTANCE[0x48]=0x100000\n")
+            out_path = os.path.join(tmp, "decoded.json")
+            rc, _, err = _run_main(["decode", "--input", path,
+                                    "--json", out_path])
+            self.assertEqual(rc, 0, err)
+            with open(out_path, encoding="utf-8") as handle:
+                report = json.load(handle)
+            chunk = report["chunks"][0]
+            self.assertEqual(chunk["label"], "INSTANCE")
+            self.assertEqual(chunk["words"]["0x48"], "0x00100000")
+            self.assertEqual(chunk["unmapped_count"], 0)
+
+    def test_cli_diff_exit_codes(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            left = _write_snapshot(tmp, "left.json", [{
+                "label": "RAMFC", "base": "0x0",
+                "words": ["0x0", "0x0", "0x0", "0x0", "0x0", "0x0", "0x0",
+                          "0x0", "0x0", "0x0", "0x0", "0x0", "0x0", "0x0",
+                          "0x0", "0x0", "0x0", "0x0", "0x30000", "0x0"],
+            }])
+            right = _write_snapshot(tmp, "right.json", [{
+                "label": "RAMFC", "base": "0x0",
+                "words": ["0x0", "0x0", "0x0", "0x0", "0x0", "0x0", "0x0",
+                          "0x0", "0x0", "0x0", "0x0", "0x0", "0x0", "0x0",
+                          "0x0", "0x0", "0x0", "0x0", "0x10000", "0x0"],
+            }])
+            rc, out, err = _run_main(["diff", left, right])
+            self.assertEqual(rc, 3, err)
+            self.assertIn("CE_CTX_VALID: set -> clear", out)
+            rc, out, err = _run_main(["diff", left, left])
+            self.assertEqual(rc, 0, err)
+            self.assertIn("identical=True", out)
+
+    def test_cli_diff_accepts_decoded_reports(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            path = _write_snapshot(tmp, "snap.json", [{
+                "label": "INSTANCE", "base": "0x0",
+                "words": ["0xface", "0x0", "0x0", "0x0"],
+            }])
+            report_path = os.path.join(tmp, "decoded.json")
+            rc, _, err = _run_main(["decode", "--input", path,
+                                    "--json", report_path])
+            self.assertEqual(rc, 0, err)
+            rc, out, err = _run_main(["diff", report_path, report_path])
+            self.assertEqual(rc, 0, err)
+            self.assertIn("identical=True", out)
+
+    def test_cli_demo_green(self):
+        rc, out, err = _run_main(["demo"])
+        self.assertEqual(rc, 0, err)
+        self.assertIn("ok=True", out)
+        self.assertIn("RAMFC_TARGET", out)
+        self.assertIn("CE_CTX_VALID: set -> clear", out)
+
+    def test_demo_api_is_self_contained_and_deterministic(self):
+        first = run_demo()
+        second = run_demo()
+        self.assertEqual(first["format"], DEMO_FORMAT)
+        self.assertTrue(first["ok"])
+        self.assertEqual(first["diff"]["summary"]["changed_words"], 2)
+        self.assertEqual(first["diff"]["summary"]["changed_fields"], 2)
+        self.assertEqual(json.dumps(first["diff"], sort_keys=True),
+                         json.dumps(second["diff"], sort_keys=True))
+
+
+@unittest.skipUnless(os.path.isfile(EVIDENCE), "ladder evidence absent")
+class EvidenceTests(unittest.TestCase):
+    def test_ladder_trace_decodes_instance_fields(self):
+        entries, source = load_map(None)
+        report = decode_specs([_spec(EVIDENCE)], entries, source)
+        self.assertGreaterEqual(len(report["chunks"]), 1)
+        chunks = {chunk["label"]: chunk for chunk in report["chunks"]}
+        self.assertIn("INSTANCE", chunks)
+        instance = chunks["INSTANCE"]
+        self.assertGreater(instance["word_count"], 0)
+        fields = {field["name"]: field for field in instance["fields"]}
+        self.assertEqual(fields["GPFIFO_VA"]["raw"], "0x00100000")
+        self.assertEqual(fields["MAGIC_0FACE"]["raw"], "0x0000face")
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/debug/risc-v-gsp-rm-vm/tests/test_state_seed.py b/debug/risc-v-gsp-rm-vm/tests/test_state_seed.py
new file mode 100644
index 0000000..439f36d
--- /dev/null
+++ b/debug/risc-v-gsp-rm-vm/tests/test_state_seed.py
@@ -0,0 +1,311 @@
+"""Tests for the RM-object state-seeding harness (harness/state_seed.py, docs/52).
+
+Loader tests run against the native machine directly (no firmware image); the
+FECS mechanism proof and the construct-gate attempt run on the shipped 570.144
+image and skip when it is absent.  No hardware, no module, no guest, no
+network.
+
+Run::
+
+    python3 -m unittest tests.test_state_seed -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, ".."))
+if TOOL_ROOT not in sys.path:
+    sys.path.insert(0, TOOL_ROOT)
+
+from sim.machine import Machine  # noqa: E402
+
+from harness.rehost import DEFAULT_IMAGE  # noqa: E402
+from harness.state_seed import (  # noqa: E402
+    DEMO_FORMAT,
+    REPORT_FORMAT,
+    SEED_FORMAT,
+    StateSeedError,
+    apply_seed,
+    fecs_seed,
+    load_seed,
+    main,
+    run_seed,
+    target_seed,
+)
+from harness.valprobe import encode_return_stub  # noqa: E402
+
+PROFILE = os.path.join(TOOL_ROOT, "sim", "profiles", "tu104-570.144.json")
+
+
+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 _machine() -> Machine:
+    return Machine(memory_mb=64, ram_base=0, unknown_mmio="lenient")
+
+
+def _two_object_seed() -> dict:
+    return {
+        "format": SEED_FORMAT,
+        "name": "loader-test",
+        "entry": "0x5290dd4",
+        "allocator": {"base": "0x100000", "align": 0x100},
+        "objects": [
+            {"name": "client", "type": "client", "size": "0x2000", "fields": {
+                "0x1be0": "@ctx",
+                "0x1dd8": {"value": 0, "size": 8},
+                "0x1aa8": "unknown",
+            }},
+            {"name": "ctx", "type": "graphics_context", "size": "0x1000", "fields": {
+                "0x190": "@table",
+                "0x161": {"value": 0, "size": 1},
+            }},
+            {"name": "table", "type": "rm_object", "size": "0x100", "fields": {
+                "0x30": {"value": 0xCAFE, "size": 8},
+            }},
+        ],
+        "registers": {"a0": "@client", "a1": "@ctx", "t1": 0xC},
+        "stack": "0x300000",
+        "memory": [{"pa": "0x200000", "u32": 7}],
+        "stubs": [{"name": "getter", "addr": "0x3f0000", "return": "0x280000"}],
+        "provenance": {"author": "test"},
+    }
+
+
+class SchemaTests(unittest.TestCase):
+    def test_layout_allocates_objects_in_order(self):
+        plan = load_seed(_two_object_seed())
+        by_name = {item["name"]: item for item in plan["layout"]}
+        self.assertEqual(by_name["client"]["pa"], 0x100000)
+        self.assertEqual(by_name["ctx"]["pa"], 0x102000)
+        self.assertEqual(by_name["table"]["pa"], 0x103000)
+
+    def test_pointer_and_value_fields_are_encoded(self):
+        plan = load_seed(_two_object_seed())
+        writes = {item["label"]: item for item in plan["writes"]}
+        self.assertEqual(bytes.fromhex(writes["client+0x1be0"]["hex"]),
+                         (0x102000).to_bytes(8, "little"))
+        self.assertEqual(bytes.fromhex(writes["ctx+0x190"]["hex"]),
+                         (0x103000).to_bytes(8, "little"))
+        self.assertEqual(bytes.fromhex(writes["table+0x30"]["hex"]),
+                         (0xCAFE).to_bytes(8, "little"))
+        self.assertEqual(bytes.fromhex(writes["mem+0x200000"]["hex"]),
+                         (7).to_bytes(4, "little"))
+
+    def test_unknown_marker_is_reported_and_not_written(self):
+        plan = load_seed(_two_object_seed())
+        unknown = {(item["object"], item["off"]) for item in plan["unknown_fields"]}
+        self.assertIn(("client", "0x1aa8"), unknown)
+        labels = {item["label"] for item in plan["writes"]}
+        self.assertNotIn("client+0x1aa8", labels)
+
+    def test_pc_register_becomes_entry(self):
+        seed = _two_object_seed()
+        del seed["entry"]
+        seed["registers"]["pc"] = "0x5290dd4"
+        plan = load_seed(seed)
+        self.assertEqual(plan["entry"], 0x5290DD4)
+        self.assertNotIn("pc", plan["registers"])
+
+    def test_plan_is_deterministic(self):
+        first = load_seed(_two_object_seed())
+        second = load_seed(_two_object_seed())
+        self.assertEqual(first["writes"], second["writes"])
+        self.assertEqual(first["layout"], second["layout"])
+
+    def test_bad_format_rejected(self):
+        seed = _two_object_seed()
+        seed["format"] = "greeneye-state-seed/2"
+        with self.assertRaises(StateSeedError):
+            load_seed(seed)
+
+    def test_missing_entry_rejected(self):
+        seed = _two_object_seed()
+        del seed["entry"]
+        with self.assertRaises(StateSeedError):
+            load_seed(seed)
+
+    def test_unknown_object_type_rejected(self):
+        seed = _two_object_seed()
+        seed["objects"].append({"name": "x", "type": "gpu_of_doom"})
+        with self.assertRaises(StateSeedError):
+            load_seed(seed)
+
+    def test_duplicate_symbol_rejected(self):
+        seed = _two_object_seed()
+        seed["objects"].append({"name": "client", "type": "client"})
+        with self.assertRaises(StateSeedError):
+            load_seed(seed)
+
+    def test_unknown_pointer_rejected(self):
+        seed = _two_object_seed()
+        seed["objects"][0]["fields"]["0x1de8"] = "@does-not-exist"
+        with self.assertRaises(StateSeedError):
+            load_seed(seed)
+
+    def test_bad_memory_item_rejected(self):
+        seed = _two_object_seed()
+        seed["memory"] = [{"pa": "0x200000"}]
+        with self.assertRaises(StateSeedError):
+            load_seed(seed)
+
+    def test_overlapping_objects_rejected(self):
+        seed = _two_object_seed()
+        seed["objects"][1]["pa"] = "0x100800"
+        with self.assertRaises(StateSeedError):
+            load_seed(seed)
+
+    def test_bad_register_rejected(self):
+        seed = _two_object_seed()
+        seed["registers"]["r99"] = 1
+        with self.assertRaises(StateSeedError):
+            load_seed(seed)
+
+
+class LoaderTests(unittest.TestCase):
+    def test_apply_seed_places_bytes_and_registers(self):
+        plan = load_seed(_two_object_seed())
+        machine = _machine()
+        apply_seed(machine, plan)
+        client = 0x100000
+        self.assertEqual(machine.read_mem(client + 0x1be0, 8),
+                         (0x102000).to_bytes(8, "little"))
+        self.assertEqual(machine.read_mem(client + 0x1dd8, 8), b"\x00" * 8)
+        self.assertEqual(machine.read_mem(client + 0x1aa8, 8), b"\x00" * 8)
+        self.assertEqual(machine.read_mem(0x200000, 4), (7).to_bytes(4, "little"))
+        self.assertEqual(machine.x[10], client)
+        self.assertEqual(machine.x[11], 0x102000)
+        self.assertEqual(machine.x[6], 0xC)
+        self.assertEqual(machine.read_mem(0x3F0000, 12), encode_return_stub(0x280000))
+
+    def test_run_seed_missing_image_is_clean_error(self):
+        with self.assertRaises(StateSeedError):
+            run_seed(_two_object_seed(), image="/nonexistent/fw.bin")
+
+
+@unittest.skipUnless(os.path.isfile(DEFAULT_IMAGE), "570.144 firmware absent")
+class FecsMechanismTests(unittest.TestCase):
+    """Prove the seeder on a path that needs no RM objects (docs/47 predicate)."""
+
+    def test_seeded_slot_value_drives_the_verdict(self):
+        nonnull = run_seed(fecs_seed(slot=1), profile_path=PROFILE, record=False,
+                           name="state-seed-test")
+        null = run_seed(fecs_seed(slot=0), profile_path=PROFILE, record=False,
+                        name="state-seed-test")
+        self.assertEqual(nonnull["format"], REPORT_FORMAT)
+        self.assertEqual(nonnull["status"], "outcome")
+        self.assertEqual(nonnull["outcome"], "refuse-fecs-slot-exists-0x1f")
+        self.assertEqual(null["status"], "outcome")
+        self.assertEqual(null["outcome"], "accept-commit-path")
+
+    def test_fecs_run_is_deterministic(self):
+        first = run_seed(fecs_seed(slot=1), profile_path=PROFILE, record=False,
+                         name="state-seed-test")
+        second = run_seed(fecs_seed(slot=1), profile_path=PROFILE, record=False,
+                          name="state-seed-test")
+        self.assertEqual(first["outcome"], second["outcome"])
+        self.assertEqual(first["instructions"], second["instructions"])
+        self.assertEqual(first["state_reads"], second["state_reads"])
+
+    def test_report_records_session_and_json(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            report = run_seed(fecs_seed(slot=1), profile_path=PROFILE, out=tmp,
+                              record=True, name="state-seed-test")
+            self.assertTrue(os.path.isfile(report["report_json"]))
+            with open(report["report_json"], "r", encoding="utf-8") as fh:
+                doc = json.load(fh)
+            self.assertEqual(doc["format"], REPORT_FORMAT)
+            self.assertTrue(report["session"])
+            self.assertTrue(os.path.isdir(report["session"]))
+
+
+@unittest.skipUnless(os.path.isfile(DEFAULT_IMAGE), "570.144 firmware absent")
+class TargetGateTests(unittest.TestCase):
+    """Attempt FUN_ram_052a70ec with a crafted client/channel/context seed."""
+
+    def test_minimal_seed_reports_first_gate(self):
+        report = run_seed(target_seed(advance=False), profile_path=PROFILE,
+                          record=False, name="state-seed-test")
+        self.assertEqual(report["status"], "fault")
+        self.assertIsNotNone(report["gate"])
+        self.assertEqual(report["gate"]["kind"], "null_indirect_call")
+        self.assertEqual(report["gate"]["label"], "engine_node+0x600")
+        self.assertEqual(report["gate"]["value"], 0)
+        labels = [item["label"] for item in report["state_reads"]]
+        self.assertIn("client+0x1de8", labels)
+        self.assertLess(labels.index("client+0x1de8"),
+                        labels.index("engine_node+0x600"))
+        self.assertTrue(report["unknown_fields"])
+
+    def test_stub_advances_to_the_next_gate(self):
+        report = run_seed(target_seed(advance=True), profile_path=PROFILE,
+                          record=False, name="state-seed-test")
+        self.assertEqual(report["status"], "fault")
+        self.assertEqual(report["gate"]["kind"], "null_indirect_call")
+        self.assertEqual(report["gate"]["label"], "channel+0x898")
+
+    def test_target_seed_is_deterministic(self):
+        first = run_seed(target_seed(advance=False), profile_path=PROFILE,
+                         record=False, name="state-seed-test")
+        second = run_seed(target_seed(advance=False), profile_path=PROFILE,
+                          record=False, name="state-seed-test")
+        self.assertEqual(first["gate"], second["gate"])
+        self.assertEqual(first["state_reads"], second["state_reads"])
+
+
+@unittest.skipUnless(os.path.isfile(DEFAULT_IMAGE), "570.144 firmware absent")
+class CliTests(unittest.TestCase):
+    def test_emit_and_validate_round_trip(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            path = os.path.join(tmp, "fecs.json")
+            rc, out, _ = _run_main(["emit", "--kind", "fecs", "--out", path])
+            self.assertEqual(rc, 0)
+            self.assertTrue(os.path.isfile(path))
+            rc, out, _ = _run_main(["validate", "--seed", path])
+            self.assertEqual(rc, 0)
+            doc = json.loads(out)
+            self.assertEqual(doc["format"], SEED_FORMAT)
+            self.assertEqual(doc["entry"], "0x5290dd4")
+
+    def test_run_emitted_seed_with_out(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            seed_path = os.path.join(tmp, "fecs.json")
+            rc, _, _ = _run_main(["emit", "--kind", "fecs", "--out", seed_path])
+            self.assertEqual(rc, 0)
+            out_dir = os.path.join(tmp, "out")
+            rc, out, _ = _run_main(["run", "--seed", seed_path, "--out", out_dir,
+                                    "--no-record"])
+            self.assertEqual(rc, 0)
+            doc = json.loads(out)
+            self.assertEqual(doc["status"], "outcome")
+            self.assertEqual(doc["outcome"], "refuse-fecs-slot-exists-0x1f")
+            self.assertTrue(os.path.isfile(
+                os.path.join(out_dir, "state-seed-report.json")))
+
+    def test_run_with_bad_seed_returns_two(self):
+        rc, _, err = _run_main(["run", "--seed", "/nonexistent/seed.json"])
+        self.assertEqual(rc, 2)
+        self.assertIn("error:", err)
+
+    def test_demo_returns_zero(self):
+        rc, out, _ = _run_main(["demo"])
+        self.assertEqual(rc, 0)
+        doc = json.loads(out)
+        self.assertEqual(doc["format"], DEMO_FORMAT)
+        self.assertTrue(doc["ok"])
+        self.assertIsNotNone(doc["target"]["gate"])
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/debug/risc-v-gsp-rm-vm/tests/test_valprobe.py b/debug/risc-v-gsp-rm-vm/tests/test_valprobe.py
new file mode 100644
index 0000000..cbf530a
--- /dev/null
+++ b/debug/risc-v-gsp-rm-vm/tests/test_valprobe.py
@@ -0,0 +1,200 @@
+"""Tests for the validation-predicate probe (harness/valprobe.py, docs/47).
+
+The bundled FECS-promote matrix runs offline on the shipped 570.144 image
+(skipped when the image is absent).  No hardware, no module, no guest, no
+network.
+
+Run::
+
+    python3 -m unittest tests.test_valprobe -v
+"""
+from __future__ import annotations
+
+import contextlib
+import io
+import json
+import os
+import struct
+import sys
+import tempfile
+import unittest
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+TOOL_ROOT = os.path.abspath(os.path.join(HERE, ".."))
+if TOOL_ROOT not in sys.path:
+    sys.path.insert(0, TOOL_ROOT)
+
+from sim.decode import K_JALR, K_LUI, K_OPI, decode32  # noqa: E402
+
+from harness.rehost import DEFAULT_IMAGE, _parse_u64  # noqa: E402
+from harness.valprobe import (  # noqa: E402
+    MATRIX_FORMAT,
+    PROBE_FORMAT,
+    PROMOTE_VALIDATION,
+    SPEC_FORMAT,
+    ValprobeError,
+    decode_memory_spec,
+    encode_return_stub,
+    fecs_spec,
+    main,
+    reg_index,
+    run_spec,
+)
+
+PROFILE = os.path.join(TOOL_ROOT, "sim", "profiles", "tu104-570.144.json")
+
+
+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()
+
+
+class EncodingTests(unittest.TestCase):
+    def test_stub_decodes_as_lui_addi_ret(self):
+        code = encode_return_stub(0x2FFFF9800)
+        self.assertEqual(len(code), 12)
+        words = struct.unpack("<III", code)
+        lui = decode32(0, words[0])
+        addi = decode32(4, words[1])
+        ret = decode32(8, words[2])
+        self.assertEqual(lui[0], K_LUI)
+        self.assertEqual(lui[2], 10)
+        self.assertEqual(addi[0], K_OPI)
+        self.assertEqual(addi[2], 10)
+        self.assertEqual(ret[0], K_JALR)
+        self.assertEqual(ret[1], 0)  # jalr zero, 0(ra)
+
+    def test_stub_value_roundtrip(self):
+        # The lui+addi pair must reconstruct every value the probe uses
+        # (scratch addresses below 2^31, offset by the 12-bit signed addi).
+        for value in (0, 0x800, 0xFFF, 0x1234, 0x10000000, 0x1FFFFFFF, 0x7FFF_F000):
+            code = encode_return_stub(value)
+            hi = struct.unpack("<I", code[0:4])[0] >> 12
+            lo = struct.unpack("<I", code[4:8])[0] >> 20
+            if lo >= 0x800:
+                lo -= 0x1000
+            self.assertEqual((hi << 12) + lo, value)
+
+    def test_memory_spec_widths(self):
+        self.assertEqual(decode_memory_spec({"addr": "0x10", "hex": "aabb"}),
+                         (0x10, b"\xaa\xbb"))
+        self.assertEqual(decode_memory_spec({"addr": "0x10", "u64": 1}),
+                         (0x10, b"\x01" + b"\x00" * 7))
+        self.assertEqual(decode_memory_spec({"addr": "0x10", "u32": 2}),
+                         (0x10, b"\x02\x00\x00\x00"))
+        self.assertEqual(decode_memory_spec({"addr": "0x10", "u16": 3}),
+                         (0x10, b"\x03\x00"))
+        self.assertEqual(decode_memory_spec({"addr": "0x10", "u8": 4}),
+                         (0x10, b"\x04"))
+        with self.assertRaises(ValprobeError):
+            decode_memory_spec({"addr": "0x10"})
+
+    def test_register_aliases(self):
+        for name, index in (("s2", 18), ("x18", 18), ("a3", 13), ("t1", 6), ("fp", 8)):
+            self.assertEqual(reg_index(name), index)
+        with self.assertRaises(ValprobeError):
+            reg_index("r99")
+
+
+class SpecTests(unittest.TestCase):
+    def test_fecs_spec_shape(self):
+        spec = fecs_spec()
+        self.assertEqual(spec["format"], SPEC_FORMAT)
+        self.assertEqual(_parse_u64(spec["target"]), PROMOTE_VALIDATION)
+        self.assertGreaterEqual(len(spec["variants"]), 15)
+        for variant in spec["variants"]:
+            self.assertTrue(variant["label"])
+            self.assertIn(variant["expect"], set(spec["outcomes"].values()))
+            self.assertTrue(variant["stubs"])
+
+    def test_fecs_spec_layout_stays_inside_ram(self):
+        spec = fecs_spec()
+        profile_ram_mb = 12288
+        ram_hi = profile_ram_mb * 1024 * 1024
+        for variant in spec["variants"]:
+            for item in variant["memory"]:
+                self.assertLess(_parse_u64(item["addr"]) + 8, ram_hi)
+
+    def test_missing_spec_is_clean_error(self):
+        rc, _, err = _run_main(["run", "--spec", "/nonexistent/spec.json"])
+        self.assertEqual(rc, 2)
+        self.assertIn("error:", err)
+
+
+@unittest.skipUnless(os.path.isfile(DEFAULT_IMAGE), "570.144 firmware absent")
+class RealImageMatrixTests(unittest.TestCase):
+    def _reduced_spec(self) -> dict:
+        spec = fecs_spec()
+        keep = {
+            "FECS init vidmem, slot non-NULL",
+            "FECS init vidmem, slot NULL",
+            "FECS init sysmem attr 0x6, slot non-NULL",
+            "FECS no-init mapped, slot non-NULL",
+            "PM init",
+        }
+        spec["variants"] = [v for v in spec["variants"] if v["label"] in keep]
+        return spec
+
+    def test_matrix_reproduces_live_verdicts_deterministically(self):
+        spec = self._reduced_spec()
+        first = run_spec(spec=spec, profile_path=PROFILE, record=False, name="valprobe-test")
+        second = run_spec(spec=spec, profile_path=PROFILE, record=False, name="valprobe-test")
+        self.assertEqual(first["format"], MATRIX_FORMAT)
+        self.assertEqual(first["evaluated"], len(spec["variants"]))
+        self.assertEqual(first["failed"], 0)
+        verdicts_a = [(r["label"], r["outcome"]) for r in first["variants"]]
+        verdicts_b = [(r["label"], r["outcome"]) for r in second["variants"]]
+        self.assertEqual(verdicts_a, verdicts_b)
+        by_label = dict(verdicts_a)
+        self.assertEqual(by_label["FECS init vidmem, slot non-NULL"],
+                         "refuse-fecs-slot-exists-0x1f")
+        self.assertEqual(by_label["FECS init vidmem, slot NULL"], "accept-commit-path")
+        self.assertEqual(by_label["FECS no-init mapped, slot non-NULL"], "accept-skip-path")
+
+    def test_probe_reports_firmware_slot_index_and_stub(self):
+        spec = self._reduced_spec()
+        spec["variants"] = [v for v in spec["variants"]
+                            if v["label"] == "FECS init vidmem, slot non-NULL"]
+        matrix = run_spec(spec=spec, profile_path=PROFILE, record=False, name="valprobe-test")
+        row = matrix["variants"][0]
+        self.assertEqual(row["format"], PROBE_FORMAT)
+        self.assertEqual(row["status"], "outcome")
+        self.assertEqual(row["instructions"], row["instructions"])
+        self.assertTrue(row["stubs"])
+        slot_index = [r for r in row["reads"] if r.get("label") == "firmware-mapped-slot-index"]
+        self.assertEqual(slot_index[0]["u32"], 6)  # firmware maps FECS -> slot 6
+
+    def test_boot_to_b3_then_probe(self):
+        spec = self._reduced_spec()
+        spec["variants"] = [v for v in spec["variants"]
+                            if v["label"] == "FECS init vidmem, slot non-NULL"]
+        spec["boot_to"] = "B3"
+        matrix = run_spec(spec=spec, profile_path=PROFILE, record=False, name="valprobe-test")
+        row = matrix["variants"][0]
+        self.assertIsNotNone(row["boot"])
+        self.assertEqual(row["boot"]["rung"], "B3")
+        self.assertTrue(row["boot"]["reached"])
+        self.assertEqual(row["outcome"], "refuse-fecs-slot-exists-0x1f")
+
+    def test_matrix_writes_report_json(self):
+        spec = self._reduced_spec()
+        spec["variants"] = spec["variants"][:2]
+        with tempfile.TemporaryDirectory() as tmp:
+            matrix = run_spec(spec=spec, profile_path=PROFILE, out=tmp,
+                              record=False, name="valprobe-test")
+            self.assertTrue(os.path.isfile(os.path.join(tmp, "valprobe-matrix.json")))
+            with open(matrix["report_json"], "r", encoding="utf-8") as fh:
+                doc = json.load(fh)
+            self.assertEqual(doc["format"], MATRIX_FORMAT)
+            self.assertEqual(len(doc["variants"]), 2)
+
+    def test_cli_sweep_returns_zero(self):
+        rc, out, _ = _run_main(["sweep", "--no-record"])
+        self.assertEqual(rc, 0)
+        self.assertIn("PASS", out)
+
+
+if __name__ == "__main__":
+    unittest.main()