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, 305 insertions, 12 deletions

diff --git a/debug/risc-v-gsp-rm-vm/harness/state_decode.py b/debug/risc-v-gsp-rm-vm/harness/state_decode.py
index 351515a..d57d3a6 100644
--- a/debug/risc-v-gsp-rm-vm/harness/state_decode.py
+++ b/debug/risc-v-gsp-rm-vm/harness/state_decode.py
@@ -32,13 +32,33 @@ Input formats (auto-detected per ``--input``):
     ``--label``/``--base``;
 (d) raw little-endian u32 words (``--bin``, or auto-detected when a text parse
     finds nothing and the bytes are control-heavy and 4-byte aligned);
-(e) the snapshot JSON above.
+(e) the snapshot JSON above;
+(f) aggregate state groups -- one ``GROUP: key=value ...`` group per line,
+    e.g. ``stage: primed=1 kicked=1 tx_seq_num=106``; the token right after
+    the group colon must itself be a ``key=value`` token, and a line may
+    carry several ``|``-separated groups (``rm_call_userd: userd=sysmem
+    (host-writable) | rm_call_pdb: vaBase=0x100000 ...``).  Each group
+    becomes a chunk named ``PHYS_<group>`` whose fields are the kv pairs:
+    values are kept as integers when the module's number convention parses
+    them (``0x..``, bare hex letters, leading-zero hex, decimal, signed),
+    everything else stays a raw string (``variant=V6``, ``userd=sysmem``),
+    and each group's raw text is preserved on the chunk.  Aggregate chunks
+    carry no words, so the map does not apply to them; the diff compares
+    their kvs and classifies every row as ``added`` / ``removed`` /
+    ``changed``.
 
 Offsets inside a chunk are byte offsets relative to the chunk start (all
 inputs above are u32 word dumps, so offsets are multiples of 4).  ``base`` is
 the engine-visible address of chunk-relative offset 0: it is recorded as
 provenance and for the diff, but it does not shift map lookup.
 
+Additive evolution (2026-10, aggregate inputs; no key renamed or removed):
+aggregate chunks add ``group`` / ``kv_count`` / ``raw`` chunk keys, kv fields
+add ``kind: "kv"`` / ``index`` (duplicate-key occurrence) / ``numeric``, diff
+chunks and ``summary`` add ``fields_added`` / ``fields_removed`` /
+``fields_changed``, and every ``changed_fields`` row now carries ``status``
+(``added`` / ``removed`` / ``changed``).
+
 Bundled map (deliberately small; extend or override it from a map file, never
 by guessing here).  Provenance is live bring-up observation (2026-10 GSP
 ladder runs):
@@ -90,6 +110,10 @@ WORD_BYTES = 4
 WORD_MASK = 0xFFFFFFFF
 WORD_WIDTHS = (4, 8)
 
+AGGREGATE_FORMAT = "aggregate"
+AGGREGATE_PREFIX = "PHYS_"
+KV_KIND = "kv"
+
 
 class StateDecodeError(ValueError):
     """An input, map or request the tool cannot honour."""
@@ -372,15 +396,99 @@ def _document(path: str, fmt: str, label: str | None, base: int,
     }
 
 
+# ---------------------------------------------------------------------------
+# Aggregate state groups (``GROUP: key=value ...``)
+# ---------------------------------------------------------------------------
+# A group head is a name token at line start or after a ``|`` separator, e.g.
+# ``rm_call_pdb: vaBase=0x100000``.  Covers group names like ``vgsp-state``
+# and ``rm_call``.  A kv token starts at a token boundary (whitespace or
+# string start), so annotations such as ``(VAL=0x1f7800)`` stay raw text
+# instead of becoming a second ``VAL`` field.
+_AGG_HEAD_RE = re.compile(
+    r"(?:^|\|)[ \t]*(?P<group>[A-Za-z0-9_.-]+):[ \t]+")
+
+_KV_RE = re.compile(
+    r"(?<!\S)(?P<key>[A-Za-z0-9_.-]+)=(?P<value>\S*)")
+
+
+def parse_aggregate_line(line: str) -> list[dict[str, Any]]:
+    """Parse one aggregate state line into ``{group, raw, kvs}`` dicts.
+
+    A group head is accepted only when the token right after the colon is
+    itself a ``key=value`` token (the shape ``^GROUP:\\s+key=``); structured
+    lines whose first token is not a kv (``status: writePtr slot=29 ...``,
+    ``peer: host readPtr @cmd+0x20 = 8``) stay unparsed.  A line may hold
+    several ``|``-separated groups, in which case each becomes its own group.
+    Non-aggregate lines return ``[]``.
+    """
+    heads = list(_AGG_HEAD_RE.finditer(line))
+    if not heads:
+        return []
+    groups = []
+    for index, head in enumerate(heads):
+        end = heads[index + 1].start() if index + 1 < len(heads) else len(line)
+        body = line[head.end():end]
+        if not _KV_RE.match(body):
+            continue
+        kvs = _extract_kvs(body)
+        groups.append({
+            "group": head.group("group"),
+            "raw": line[head.start("group"):end].strip(),
+            "kvs": kvs,
+        })
+    return groups
+
+
+def _parse_kv_value(raw: str) -> tuple[Any, bool]:
+    """Parse an aggregate value with the module's number convention.
+
+    Returns ``(value, numeric)``; unparseable values stay raw strings so no
+    spelling is lost (``raw`` on the field always keeps the captured token).
+    """
+    if not raw:
+        return raw, False
+    try:
+        return parse_number(raw, why="aggregate value"), True
+    except StateDecodeError:
+        return raw, False
+
+
+def _extract_kvs(text: str) -> list[dict[str, Any]]:
+    kvs = []
+    for match in _KV_RE.finditer(text):
+        raw_value = match.group("value")
+        value, numeric = _parse_kv_value(raw_value)
+        kvs.append({"key": match.group("key"), "raw": raw_value,
+                    "value": value, "numeric": numeric})
+    return kvs
+
+
+def _aggregate_chunk(group: dict[str, Any], base: int,
+                     source: str | None) -> dict:
+    return {
+        "label": AGGREGATE_PREFIX + group["group"],
+        "group": group["group"],
+        "base": base,
+        "words": {},
+        "format": AGGREGATE_FORMAT,
+        "source": source,
+        "raw": group["raw"],
+        "kvs": group["kvs"],
+    }
+
+
 def parse_text(text: str, *, default_label: str | None = None,
                default_base: int = 0, source: str | None = None) -> list[dict]:
-    """Parse the text/hexdump/row formats into chunks.
+    """Parse the text/hexdump/row/aggregate formats into chunks.
 
     Chunk names are case-insensitive: ``INSTANCE`` and ``inst`` land in one
-    chunk, displayed with the first spelling seen.  Every line that matches
-    none of the three patterns is ignored.
+    chunk, displayed with the first spelling seen.  Aggregate groups (format
+    ``aggregate``, label ``PHYS_<group>``) are kept as separate chunks in
+    line order after the word chunks.  Every line that matches none of the
+    four patterns is ignored.
     """
     chunks: dict[str, dict] = {}
+    aggregate: list[dict] = []
 
     def chunk_for(label: str) -> dict:
         key = label.lower()
@@ -423,7 +531,9 @@ def parse_text(text: str, *, default_label: str | None = None,
             for index, token in enumerate(match.group("words").split()):
                 put(label, offset + index * WORD_BYTES, int(token, 16), where)
             continue
-    return [chunks[key] for key in chunks]
+        for group in parse_aggregate_line(line):
+            aggregate.append(_aggregate_chunk(group, default_base, source))
+    return [chunks[key] for key in chunks] + aggregate
 
 
 def load_input(spec: dict) -> dict:
@@ -445,7 +555,10 @@ def load_input(spec: dict) -> dict:
         chunks = parse_text(text, default_label=label, default_base=base,
                             source=path)
         if chunks:
-            return _document(path, "text", label, base, chunks)
+            fmt = (AGGREGATE_FORMAT
+                   if all(chunk.get("format") == AGGREGATE_FORMAT
+                          for chunk in chunks) else "text")
+            return _document(path, fmt, label, base, chunks)
     if _looks_binary(data):
         chunk = _chunk_from_binary(data, label or "RAW", base, path)
         return _document(path, "binary", label, base, [chunk])
@@ -542,8 +655,49 @@ def _decode_field(entry: dict, value: int) -> dict:
     return field
 
 
+def _decode_aggregate_chunk(chunk: dict) -> dict:
+    """Decode an aggregate group chunk into the ``greeneye-state-decode/1`` shape.
+
+    There are no words: the named fields *are* the kv pairs, in line order,
+    with duplicate keys separated by a 0-based ``index``.
+    """
+    fields = []
+    occurrences: dict[str, int] = {}
+    for kv in chunk.get("kvs", []):
+        key = kv["key"]
+        index = occurrences.get(key, 0)
+        occurrences[key] = index + 1
+        fields.append({
+            "name": key,
+            "kind": KV_KIND,
+            "index": index,
+            "offset": None,
+            "width": 0,
+            "raw": kv["raw"],
+            "value": kv["value"],
+            "numeric": kv["numeric"],
+        })
+    return {
+        "label": chunk["label"],
+        "group": chunk.get("group"),
+        "base": f"{chunk['base']:#x}",
+        "source": chunk.get("source"),
+        "input_format": chunk.get("format"),
+        "word_width": WORD_BYTES,
+        "word_count": 0,
+        "words": {},
+        "fields": fields,
+        "unmapped": [],
+        "unmapped_count": 0,
+        "kv_count": len(fields),
+        "raw": chunk.get("raw"),
+    }
+
+
 def decode_chunk(chunk: dict, entries: dict[tuple[str, int], dict]) -> dict:
     """Decode one internal chunk into the ``greeneye-state-decode/1`` shape."""
+    if chunk.get("format") == AGGREGATE_FORMAT:
+        return _decode_aggregate_chunk(chunk)
     label = chunk["label"]
     words: dict[int, int] = chunk["words"]
     matched = sorted(
@@ -604,6 +758,9 @@ def decode_specs(specs: list[dict], entries: dict[tuple[str, int], dict],
             "is silence, not absence.",
             "Offsets are chunk-relative bytes; base is recorded provenance and "
             "does not shift field lookup.",
+            "An aggregate chunk (label PHYS_<group>, input_format aggregate) "
+            "carries its own kv fields; the map does not apply and kv values "
+            "carry no word/bit semantics.",
             "No timing (L2) or device-environment (L3) claim: this is a field "
             "decoder over captured words.",
         ],
@@ -653,9 +810,16 @@ def _changed_fields(left: dict | None, right: dict | None) -> list[dict]:
         if (left_field is not None and right_field is not None
                 and left_field["raw"] == right_field["raw"]):
             continue
+        if left_field is None:
+            status = "added"
+        elif right_field is None:
+            status = "removed"
+        else:
+            status = "changed"
         row: dict[str, Any] = {
             "offset": f"{offset:#x}",
             "name": name,
+            "status": status,
             "old": left_field["raw"] if left_field else None,
             "new": right_field["raw"] if right_field else None,
             "old_summary": _field_summary(left_field),
@@ -684,11 +848,93 @@ def _word_rows(words: dict[int, int]) -> list[dict]:
             for offset, raw in sorted(words.items())]
 
 
+def _is_aggregate_chunk(chunk: dict | None) -> bool:
+    return bool(chunk) and chunk.get("input_format") == AGGREGATE_FORMAT
+
+
+def _kv_index(chunk: dict | None) -> dict[tuple[str, int], dict]:
+    """Index kv fields by ``(name, occurrence)`` so duplicate keys survive."""
+    if chunk is None:
+        return {}
+    return {(field["name"], int(field.get("index", 0))): field
+            for field in chunk.get("fields", [])}
+
+
+def _kv_rows(left: dict | None, right: dict | None) -> list[dict]:
+    """Classified kv diff rows (added / removed / changed) in key order."""
+    left_kvs = _kv_index(left)
+    right_kvs = _kv_index(right)
+    rows = []
+    for name, index in sorted(set(left_kvs) | set(right_kvs),
+                              key=lambda key: (key[0].lower(), key[0], key[1])):
+        left_field = left_kvs.get((name, index))
+        right_field = right_kvs.get((name, index))
+        if left_field is not None and right_field is not None:
+            if (left_field.get("value") == right_field.get("value")
+                    and bool(left_field.get("numeric"))
+                    == bool(right_field.get("numeric"))):
+                continue
+            status = "changed"
+        elif right_field is not None:
+            status = "added"
+        else:
+            status = "removed"
+        rows.append({
+            "name": name,
+            "index": index,
+            "kind": KV_KIND,
+            "offset": None,
+            "status": status,
+            "old": left_field.get("value") if left_field else None,
+            "new": right_field.get("value") if right_field else None,
+            "old_raw": left_field.get("raw") if left_field else None,
+            "new_raw": right_field.get("raw") if right_field else None,
+            "old_numeric": left_field.get("numeric") if left_field else None,
+            "new_numeric": right_field.get("numeric") if right_field else None,
+        })
+    return rows
+
+
+def _classify_fields(rows: list[dict]) -> tuple[list[dict], list[dict], list[dict]]:
+    return ([row for row in rows if row.get("status") == "added"],
+            [row for row in rows if row.get("status") == "removed"],
+            [row for row in rows if row.get("status") == "changed"])
+
+
+def _diff_aggregate_chunk(left: dict | None, right: dict | None) -> dict:
+    """Diff two aggregate (kv) chunks with explicit added/removed/changed."""
+    present = right if left is None else left
+    rows = _kv_rows(left, right)
+    fields_added, fields_removed, fields_changed = _classify_fields(rows)
+    if left is None or right is None:
+        status = "right_only" if left is None else "left_only"
+    else:
+        status = "changed" if rows else "identical"
+    return {
+        "label": present["label"],
+        "base": present["base"],
+        "left_base": left["base"] if left else None,
+        "right_base": right["base"] if right else None,
+        "status": status,
+        "added": [],
+        "removed": [],
+        "changed": [],
+        "changed_fields": rows,
+        "fields_added": fields_added,
+        "fields_removed": fields_removed,
+        "fields_changed": fields_changed,
+    }
+
+
 def _diff_chunk(left: dict | None, right: dict | None) -> dict:
+    if _is_aggregate_chunk(left) or _is_aggregate_chunk(right):
+        return _diff_aggregate_chunk(left, right)
     if left is None or right is None:
         present = right if left is None else left
         words = _words_to_int(present)
         rows = _word_rows(words)
+        changed_fields = _changed_fields(left, right)
+        fields_added, fields_removed, fields_changed = _classify_fields(changed_fields)
         row = {
             "label": present["label"],
             "base": present["base"],
@@ -698,7 +944,10 @@ def _diff_chunk(left: dict | None, right: dict | None) -> dict:
             "added": rows if left is None else [],
             "removed": rows if right is None else [],
             "changed": [],
-            "changed_fields": _changed_fields(left, right),
+            "changed_fields": changed_fields,
+            "fields_added": fields_added,
+            "fields_removed": fields_removed,
+            "fields_changed": fields_changed,
         }
         return row
     left_words = _words_to_int(left)
@@ -712,6 +961,7 @@ def _diff_chunk(left: dict | None, right: dict | None) -> dict:
                for offset in sorted(set(left_words) & set(right_words))
                if left_words[offset] != right_words[offset]]
     changed_fields = _changed_fields(left, right)
+    fields_added, fields_removed, fields_changed = _classify_fields(changed_fields)
     status = "changed" if (added or removed or changed or changed_fields) \
         else "identical"
     return {
@@ -724,6 +974,9 @@ def _diff_chunk(left: dict | None, right: dict | None) -> dict:
         "removed": removed,
         "changed": changed,
         "changed_fields": changed_fields,
+        "fields_added": fields_added,
+        "fields_removed": fields_removed,
+        "fields_changed": fields_changed,
     }
 
 
@@ -754,6 +1007,12 @@ def diff_reports(left: dict, right: dict) -> dict:
         "removed": sum(len(chunk["removed"]) for chunk in chunks),
         "changed_words": sum(len(chunk["changed"]) for chunk in chunks),
         "changed_fields": sum(len(chunk["changed_fields"]) for chunk in chunks),
+        "fields_added": sum(len(chunk.get("fields_added", []))
+                            for chunk in chunks),
+        "fields_removed": sum(len(chunk.get("fields_removed", []))
+                              for chunk in chunks),
+        "fields_changed": sum(len(chunk.get("fields_changed", []))
+                              for chunk in chunks),
     }
     return {
         "format": DIFF_FORMAT,
@@ -769,6 +1028,9 @@ def diff_reports(left: dict, right: dict) -> dict:
             "behaviour.",
             "Chunks pair by label and occurrence order; a differing base is "
             "recorded but does not re-pair.",
+            "Aggregate kv fields compare by parsed value and numeric-ness: a "
+            "raw spelling change that keeps the same value (0x10 vs 16) is "
+            "not reported as changed.",
         ],
     }
 
@@ -781,6 +1043,18 @@ def render_decode(report: dict) -> str:
     """Readable field table for humans; the JSON report stays canonical."""
     lines = []
     for chunk in report["chunks"]:
+        if chunk.get("input_format") == AGGREGATE_FORMAT:
+            kv_count = chunk.get("kv_count", len(chunk["fields"]))
+            lines.append(
+                f"chunk {chunk['label']} base={chunk['base']} "
+                f"kv={kv_count} (aggregate)")
+            for field in chunk["fields"]:
+                index = field.get("index", 0)
+                name = (field["name"] if not index
+                        else f"{field['name']}[{index}]")
+                lines.append(f"  {'kv':>6}  {field['raw']:<12} {name:<20} "
+                             f"value={field['value']!r}")
+            continue
         lines.append(
             f"chunk {chunk['label']} base={chunk['base']} "
             f"words={chunk['word_count']} mapped={len(chunk['fields'])} "
@@ -817,6 +1091,17 @@ def render_diff(report: dict) -> str:
             lines.append(f"  changed  {word['offset']} "
                          f"{word['old']} -> {word['new']}")
         for field in chunk["changed_fields"]:
+            if field.get("kind") == KV_KIND or field.get("offset") is None:
+                index = field.get("index", 0)
+                name = (field["name"] if not index
+                        else f"{field['name']}[{index}]")
+                old_raw = field.get("old_raw")
+                new_raw = field.get("new_raw")
+                old_raw = "-" if old_raw is None else old_raw
+                new_raw = "-" if new_raw is None else new_raw
+                lines.append(f"    kv {name} {field.get('status', 'changed')}: "
+                             f"{old_raw} -> {new_raw}")
+                continue
             lines.append(f"    field {field['name']} @ {field['offset']}: "
                          f"{field['old']} -> {field['new']}  "
                          f"({field['old_summary']} -> {field['new_summary']})")
@@ -983,15 +1268,21 @@ def _write_json(path: str, doc: Any) -> str:
 def build_parser() -> argparse.ArgumentParser:
     parser = argparse.ArgumentParser(
         prog="state_decode.py",
-        description="Decode engine-visible channel state words into named "
-                    "fields, and diff two captures (offline).")
+        description="Decode engine-visible channel state words and aggregate "
+                    "GROUP: key=value state groups into named fields, and diff "
+                    "two captures (offline).")
     sub = parser.add_subparsers(dest="command")
 
     decode = sub.add_parser(
-        "decode", help="decode one or more raw inputs into a field table")
+        "decode", help="decode one or more raw inputs (state words, hexdumps, "
+                       "snapshot JSON, or aggregate key=value groups) into a "
+                       "field table")
     decode.add_argument("--input", action=_InputAction, dest="inputs",
                         default=None, metavar="FILE",
-                        help="input file; repeatable, auto-detected format")
+                        help="input file; repeatable, auto-detected format "
+                             "(LABEL[offset]=value, row/hexdump words, "
+                             "greeneye-server-snapshot/1 JSON, or aggregate "
+                             "'GROUP: key=value ...' state groups)")
     decode.add_argument("--bin", action=_InputAction, dest="inputs",
                         default=None, metavar="FILE",
                         help="raw little-endian u32 input; repeatable")
@@ -1007,7 +1298,9 @@ def build_parser() -> argparse.ArgumentParser:
                         help="write the decoded report JSON here")
 
     diff = sub.add_parser(
-        "diff", help="diff two decoded reports or two raw inputs")
+        "diff", help="diff two decoded reports or two raw inputs (state "
+                     "words or aggregate key=value groups); kv changes are "
+                     "reported as added/removed/changed")
     diff.add_argument("left", help=f"{DECODE_FORMAT} report or raw input")
     diff.add_argument("right", help=f"{DECODE_FORMAT} report or raw input")
     diff.add_argument("--left-label", dest="left_label", default=None,