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, 409 insertions, 0 deletions

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