risc-v-gsp-rm-debugger: B5 two-task attach + aggregate-state decode + docs 51/53 status

- sim/libos_port.py: port-role model + two-task network (extintr handoff,
  cross-task send/reply, TASK_STATE_QUERY trap state)
- sim/msgq_attach.py: image task discovery, bounded two-task schedule,
  second-task-frontier divergence fields (first ring poll still blocked on
  post-init resume + GSP_ARGUMENTS_CACHED handoff; no fabricated polls)
- harness/state_decode.py: aggregate 'group: key=value' state parsing
  (20-gsp-phys-state-* dumps) + kv diff classification
- tests: 127 green across the five observability modules
- docs/51/53: status and changelog updates

1 files changed, 327 insertions, 4 deletions

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
index 5c05ccd..3485ced 100644
--- a/debug/risc-v-gsp-rm-vm/tests/test_state_decode.py
+++ b/debug/risc-v-gsp-rm-vm/tests/test_state_decode.py
@@ -1,9 +1,10 @@
 """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.
+Offline only: text/hexdump/binary/snapshot-JSON/aggregate-group parsing, the
+bundled INSTANCE/RAMFC field table, map override/extension, decode/diff
+reports and the CLI.  The real-evidence tests parse frozen live ladder traces
+and an aggregate GSP physical-state dump, and skip when the backups are
+absent.  No hardware, no module, no guest, no network.
 
 Run::
 
@@ -26,6 +27,7 @@ if TOOL_ROOT not in sys.path:
     sys.path.insert(0, TOOL_ROOT)
 
 from harness.state_decode import (  # noqa: E402
+    AGGREGATE_FORMAT,
     DECODE_FORMAT,
     DEMO_FORMAT,
     DIFF_FORMAT,
@@ -38,6 +40,7 @@ from harness.state_decode import (  # noqa: E402
     load_input,
     load_map,
     main,
+    render_diff,
     run_demo,
 )
 
@@ -45,6 +48,17 @@ 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"))
 
+# Our own evidence artifact directory (not a vendor path).  The aggregate
+# dump is one group per line: ``stage: primed=1 kicked=1 tx_seq_num=106``.
+AGGREGATE_EVIDENCE = os.environ.get(
+    "GREENEYE_STATE_DECODE_EVIDENCE_AGG") or os.path.normpath(
+    os.path.join(TOOL_ROOT, "..", "..", "backups",
+                 "evidence-20261002-victory", "20-gsp-phys-state-m26.txt"))
+AGGREGATE_EVIDENCE_PEER = os.environ.get(
+    "GREENEYE_STATE_DECODE_EVIDENCE_AGG_PEER") or os.path.normpath(
+    os.path.join(TOOL_ROOT, "..", "..", "backups",
+                 "evidence-20261002-victory", "20-gsp-phys-state-m29.txt"))
+
 
 def _run_main(argv: list[str]) -> tuple[int, str, str]:
     out, err = io.StringIO(), io.StringIO()
@@ -197,6 +211,147 @@ class ParsingTests(unittest.TestCase):
                 load_input(_spec(path, binary=True))
 
 
+class AggregateParsingTests(unittest.TestCase):
+    def _decode(self, tmp: str, name: str, text: str) -> dict:
+        entries, source = load_map(None)
+        path = _write(os.path.join(tmp, name), text)
+        return decode_specs([_spec(path)], entries, source)
+
+    def test_aggregate_line_decodes_types_and_preserves_raw(self):
+        text = "\n".join([
+            "stage: primed=1 kicked=1 tx_seq_num=106 variant=V6",
+            "ctx: pa=0x1f4000000 size=0x133000 last_ret=-5 "
+            "userd=sysmem (host-writable)",
+        ])
+        with tempfile.TemporaryDirectory() as tmp:
+            path = _write(os.path.join(tmp, "agg.txt"), text)
+            document = load_input(_spec(path))
+            self.assertEqual(document["format"], AGGREGATE_FORMAT)
+            self.assertEqual([chunk["label"] for chunk in document["chunks"]],
+                             ["PHYS_stage", "PHYS_ctx"])
+            stage = document["chunks"][0]
+            self.assertEqual(stage["format"], AGGREGATE_FORMAT)
+            self.assertIn("stage: primed=1", stage["raw"])
+            entries, source = load_map(None)
+            report = decode_specs([_spec(path)], entries, source)
+            chunks = {chunk["label"]: chunk for chunk in report["chunks"]}
+            stage = chunks["PHYS_stage"]
+            self.assertEqual(stage["input_format"], AGGREGATE_FORMAT)
+            self.assertEqual(stage["word_count"], 0)
+            self.assertEqual(stage["kv_count"], 4)
+            self.assertEqual(stage["raw"],
+                             "stage: primed=1 kicked=1 tx_seq_num=106 "
+                             "variant=V6")
+            fields = {field["name"]: field for field in stage["fields"]}
+            self.assertEqual(fields["primed"]["value"], 1)
+            self.assertTrue(fields["primed"]["numeric"])
+            self.assertEqual(fields["tx_seq_num"]["value"], 106)
+            self.assertEqual(fields["variant"]["value"], "V6")
+            self.assertFalse(fields["variant"]["numeric"])
+            self.assertEqual(fields["variant"]["raw"], "V6")
+            self.assertEqual(fields["variant"]["kind"], "kv")
+            self.assertIsNone(fields["variant"]["offset"])
+            ctx = {field["name"]: field for field in chunks["PHYS_ctx"]["fields"]}
+            self.assertEqual(ctx["pa"]["value"], 0x1F4000000)
+            self.assertEqual(ctx["size"]["value"], 0x133000)
+            self.assertEqual(ctx["last_ret"]["value"], -5)
+            self.assertEqual(ctx["userd"]["value"], "sysmem")
+            self.assertNotIn("host-writable", ctx)
+
+    def test_aggregate_missing_value_and_unparsed_lines(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            report = self._decode(tmp, "agg.txt", "\n".join([
+                "banner line that must be ignored",
+                "status_ring: writePtr slot=29; last produced elements "
+                "(max 8):",
+                "peer: host status readPtr @cmd+0x20 = 8",
+                "empty: a= b=1",
+            ]))
+            self.assertEqual([chunk["label"] for chunk in report["chunks"]],
+                             ["PHYS_empty"])
+            fields = {field["name"]: field for field in
+                      report["chunks"][0]["fields"]}
+            self.assertEqual(fields["a"]["value"], "")
+            self.assertEqual(fields["a"]["raw"], "")
+            self.assertFalse(fields["a"]["numeric"])
+            self.assertEqual(fields["b"]["value"], 1)
+
+    def test_aggregate_single_kv_group_is_captured(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            report = self._decode(tmp, "agg.txt", "guest_ram: active=0\n")
+            self.assertEqual([chunk["label"] for chunk in report["chunks"]],
+                             ["PHYS_guest_ram"])
+            field = report["chunks"][0]["fields"][0]
+            self.assertEqual(field["name"], "active")
+            self.assertEqual(field["value"], 0)
+
+    def test_aggregate_pipe_groups_and_duplicate_keys(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            report = self._decode(tmp, "agg.txt", "\n".join([
+                "rm_call_userd: userd=sysmem (host-writable) | "
+                "rm_call_pdb: vaBase=0x100000 vaGpfifo=0x200000",
+                "queue_dma: shared=0x1a86b6000 ptes=129 | "
+                "radix3=0x105370000 fmc=0x142d25000",
+                "guest_ram: active=1 gpa=0x0 gpa=0x100000000 "
+                "size=0x80000000 size=0x80000000",
+            ]))
+            self.assertEqual([chunk["label"] for chunk in report["chunks"]], [
+                "PHYS_guest_ram", "PHYS_queue_dma", "PHYS_rm_call_pdb",
+                "PHYS_rm_call_userd",
+            ])
+            chunks = {chunk["label"]: chunk for chunk in report["chunks"]}
+            userd = {field["name"]: field for field in
+                     chunks["PHYS_rm_call_userd"]["fields"]}
+            self.assertEqual(userd["userd"]["value"], "sysmem")
+            pdb = [field["name"] for field in chunks["PHYS_rm_call_pdb"]["fields"]]
+            self.assertEqual(pdb, ["vaBase", "vaGpfifo"])
+            queue = [field["name"] for field in chunks["PHYS_queue_dma"]["fields"]]
+            self.assertEqual(queue, ["shared", "ptes", "radix3", "fmc"])
+            gpa = [field for field in chunks["PHYS_guest_ram"]["fields"]
+                   if field["name"] == "gpa"]
+            self.assertEqual([field["index"] for field in gpa], [0, 1])
+            self.assertEqual([field["value"] for field in gpa],
+                             [0, 0x100000000])
+
+    def test_mixed_word_and_aggregate_lines(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            path = _write(os.path.join(tmp, "mixed.txt"), "\n".join([
+                "INSTANCE[0x10]=0xface",
+                "stage: a=1 b=2",
+            ]))
+            document = load_input(_spec(path))
+            self.assertEqual(document["format"], "text")
+            self.assertEqual([chunk["label"] for chunk in document["chunks"]],
+                             ["INSTANCE", "PHYS_stage"])
+
+    def test_unstructured_text_still_rejected(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            path = _write(os.path.join(tmp, "none.txt"), "\n".join([
+                "status_ring: writePtr slot=29; last produced elements:",
+                "peer: host status readPtr @cmd+0x20 = 8",
+            ]))
+            with self.assertRaises(StateDecodeError):
+                load_input(_spec(path))
+
+    def test_aggregate_decode_is_deterministic(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            text = "stage: a=1 b=0x10 c=up\n"
+            first = self._decode(tmp, "one.txt", text)
+            second = self._decode(tmp, "two.txt", text)
+            stripped = [
+                {key: value for key, value in chunk.items()
+                 if key != "source"}
+                for chunk in first["chunks"]
+            ]
+            other = [
+                {key: value for key, value in chunk.items()
+                 if key != "source"}
+                for chunk in second["chunks"]
+            ]
+            self.assertEqual(json.dumps(stripped, sort_keys=True),
+                             json.dumps(other, sort_keys=True))
+
+
 class DecodeTests(unittest.TestCase):
     def _ramfc_target(self, word48: int) -> dict:
         report = _report_from_chunks([
@@ -394,6 +549,132 @@ class DiffTests(unittest.TestCase):
         self.assertEqual(report["summary"]["removed"], 1)
 
 
+class AggregateDiffTests(unittest.TestCase):
+    def _pair(self, tmp: str, left_text: str, right_text: str) -> tuple:
+        entries, source = load_map(None)
+        left = _write(os.path.join(tmp, "left.txt"), left_text)
+        right = _write(os.path.join(tmp, "right.txt"), right_text)
+        return (
+            decode_specs([_spec(left)], entries, source),
+            decode_specs([_spec(right)], entries, source),
+        )
+
+    def test_diff_classifies_added_removed_changed_kvs(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            left, right = self._pair(
+                tmp,
+                "stage: a=1 b=2 gone=1 variant=V6\n",
+                "stage: a=1 b=3 variant=V7 extra=4\n")
+            report = diff_reports(left, right)
+            self.assertEqual(report["format"], DIFF_FORMAT)
+            self.assertFalse(report["identical"])
+            chunk = report["chunks"][0]
+            self.assertEqual(chunk["label"], "PHYS_stage")
+            self.assertEqual(chunk["status"], "changed")
+            self.assertEqual(chunk["added"], [])
+            self.assertEqual(chunk["removed"], [])
+            self.assertEqual(chunk["changed"], [])
+            rows = {row["name"]: row for row in chunk["changed_fields"]}
+            self.assertEqual(rows["b"]["status"], "changed")
+            self.assertEqual((rows["b"]["old"], rows["b"]["new"]), (2, 3))
+            self.assertEqual(rows["variant"]["status"], "changed")
+            self.assertEqual((rows["variant"]["old"], rows["variant"]["new"]),
+                             ("V6", "V7"))
+            self.assertEqual(rows["extra"]["status"], "added")
+            self.assertIsNone(rows["extra"]["old"])
+            self.assertEqual(rows["extra"]["new"], 4)
+            self.assertEqual(rows["gone"]["status"], "removed")
+            self.assertEqual(rows["gone"]["old"], 1)
+            self.assertIsNone(rows["gone"]["new"])
+            self.assertEqual([row["name"] for row in chunk["fields_added"]],
+                             ["extra"])
+            self.assertEqual([row["name"] for row in chunk["fields_removed"]],
+                             ["gone"])
+            self.assertEqual([row["name"] for row in chunk["fields_changed"]],
+                             ["b", "variant"])
+            self.assertEqual(report["summary"]["fields_added"], 1)
+            self.assertEqual(report["summary"]["fields_removed"], 1)
+            self.assertEqual(report["summary"]["fields_changed"], 2)
+            self.assertEqual(report["summary"]["changed_fields"], 4)
+            again = diff_reports(left, right)
+            self.assertEqual(json.dumps(report, sort_keys=True),
+                             json.dumps(again, sort_keys=True))
+
+    def test_diff_aggregate_identical_and_one_sided(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            left, right = self._pair(tmp, "stage: a=1 b=up\n",
+                                     "stage: a=1 b=up\n")
+            report = diff_reports(left, right)
+            self.assertTrue(report["identical"])
+            self.assertEqual(report["summary"]["fields_added"], 0)
+            self.assertEqual(report["summary"]["fields_removed"], 0)
+            self.assertEqual(report["summary"]["fields_changed"], 0)
+            only_left, only_right = self._pair(
+                tmp, "stage: a=1 b=2\n", "ctx: x=1\n")
+            report = diff_reports(only_left, only_right)
+            labels = {chunk["label"]: chunk for chunk in report["chunks"]}
+            self.assertEqual(labels["PHYS_stage"]["status"], "left_only")
+            self.assertEqual(
+                [row["name"] for row in labels["PHYS_stage"]["fields_removed"]],
+                ["a", "b"])
+            self.assertEqual(labels["PHYS_ctx"]["status"], "right_only")
+            self.assertEqual(
+                [row["name"] for row in labels["PHYS_ctx"]["fields_added"]],
+                ["x"])
+
+    def test_diff_aggregate_duplicate_key_occurrence(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            left, right = self._pair(
+                tmp, "mem: size=0x10 size=0x20\n",
+                "mem: size=0x10 size=0x30\n")
+            chunk = diff_reports(left, right)["chunks"][0]
+            self.assertEqual(len(chunk["fields_changed"]), 1)
+            row = chunk["fields_changed"][0]
+            self.assertEqual(row["name"], "size")
+            self.assertEqual(row["index"], 1)
+            self.assertEqual((row["old"], row["new"]), (0x20, 0x30))
+
+    def test_cli_diff_aggregate_exit_codes_and_kv_rendering(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            left = _write(os.path.join(tmp, "left.txt"),
+                          "stage: a=1 b=2 state=up\n")
+            right = _write(os.path.join(tmp, "right.txt"),
+                           "stage: a=1 b=3 state=up\n")
+            rc, out, err = _run_main(["diff", left, right])
+            self.assertEqual(rc, 3, err)
+            self.assertIn("PHYS_stage", out)
+            self.assertIn("kv b changed: 2 -> 3", out)
+            rc, out, err = _run_main(["diff", left, left])
+            self.assertEqual(rc, 0, err)
+            self.assertIn("identical=True", out)
+
+    def test_render_diff_aggregate_handles_added_and_removed(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            left, right = self._pair(tmp, "stage: gone=1\n",
+                                     "stage: extra=2\n")
+            text = render_diff(diff_reports(left, right))
+            self.assertIn("kv extra added: - -> 2", text)
+            self.assertIn("kv gone removed: 1 -> -", text)
+
+    def test_cli_decode_aggregate_writes_report(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            path = _write(os.path.join(tmp, "agg.txt"),
+                          "stage: primed=1 variant=V6\n")
+            out_path = os.path.join(tmp, "agg.json")
+            rc, out, err = _run_main(["decode", "--input", path,
+                                      "--json", out_path])
+            self.assertEqual(rc, 0, err)
+            self.assertIn("PHYS_stage", out)
+            self.assertIn("(aggregate)", out)
+            with open(out_path, encoding="utf-8") as handle:
+                report = json.load(handle)
+            self.assertEqual(report["format"], DECODE_FORMAT)
+            self.assertEqual(report["sources"][0]["format"], AGGREGATE_FORMAT)
+            self.assertEqual(report["chunks"][0]["input_format"],
+                             AGGREGATE_FORMAT)
+            self.assertEqual(report["chunks"][0]["fields"][1]["value"], "V6")
+
+
 class CliAndDemoTests(unittest.TestCase):
     def test_cli_decode_prints_table_and_writes_json(self):
         with tempfile.TemporaryDirectory() as tmp:
@@ -514,5 +795,47 @@ class EvidenceTests(unittest.TestCase):
         self.assertEqual(fields["MAGIC_0FACE"]["raw"], "0x0000face")
 
 
+@unittest.skipUnless(os.path.isfile(AGGREGATE_EVIDENCE),
+                     "aggregate GSP physical-state evidence absent")
+class AggregateEvidenceTests(unittest.TestCase):
+    """Frozen GSP physical-state dumps from our own evidence directory.
+
+    Override with GREENEYE_STATE_DECODE_EVIDENCE_AGG (and ..._PEER); skipped
+    when the files are absent.  No hardware, no module, no guest, no network.
+    """
+
+    def test_aggregate_evidence_decodes_groups(self):
+        entries, source = load_map(None)
+        report = decode_specs([_spec(AGGREGATE_EVIDENCE)], entries, source)
+        self.assertGreater(len(report["chunks"]), 0)
+        self.assertEqual(report["sources"][0]["format"], AGGREGATE_FORMAT)
+        chunks = {chunk["label"]: chunk for chunk in report["chunks"]}
+        for label in ("PHYS_stage", "PHYS_rm_call", "PHYS_guest_ram",
+                      "PHYS_rm_call_ctx"):
+            self.assertIn(label, chunks)
+        stage = {field["name"]: field for field in
+                 chunks["PHYS_stage"]["fields"]}
+        self.assertEqual(stage["primed"]["value"], 1)
+        self.assertTrue(stage["tx_seq_num"]["numeric"])
+
+    def test_aggregate_evidence_diff_peer_shows_changed_kvs(self):
+        if not os.path.isfile(AGGREGATE_EVIDENCE_PEER):
+            self.skipTest("aggregate peer evidence absent")
+        entries, source = load_map(None)
+        left = decode_specs([_spec(AGGREGATE_EVIDENCE)], entries, source)
+        right = decode_specs([_spec(AGGREGATE_EVIDENCE_PEER)], entries, source)
+        report = diff_reports(left, right)
+        self.assertFalse(report["identical"])
+        self.assertGreater(report["summary"]["fields_changed"], 0)
+        self.assertEqual(report["summary"]["changed_words"], 0)
+        chunks = {chunk["label"]: chunk for chunk in report["chunks"]}
+        rows = {row["name"]: row for row in
+                chunks["PHYS_guest_ram"]["changed_fields"]}
+        self.assertEqual(rows["active"]["status"], "changed")
+        self.assertEqual((rows["active"]["old"], rows["active"]["new"]),
+                         (1, 0))
+        render_diff(report)
+
+
 if __name__ == "__main__":
     unittest.main()