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

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