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.

1 files changed, 342 insertions, 0 deletions

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