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

8 files changed, 1811 insertions, 39 deletions

diff --git a/debug/risc-v-gsp-rm-vm/docs/51-HOST-TRAFFIC-LOOP.md b/debug/risc-v-gsp-rm-vm/docs/51-HOST-TRAFFIC-LOOP.md
index e56094c..30c2804 100644
--- a/debug/risc-v-gsp-rm-vm/docs/51-HOST-TRAFFIC-LOOP.md
+++ b/debug/risc-v-gsp-rm-vm/docs/51-HOST-TRAFFIC-LOOP.md
@@ -229,6 +229,21 @@ Next step for B5, concretely (each point is bounded and session-recorded):
 5. Then replay one captured stimulus message with the real image and record
    the reply/writes as the B5 session. Until a supervised capture exists,
    this stops at step 3–4 with the stub dispatcher.
+6. **Two-task wave (landed 2026-10-06).** `sim/msgq_attach.py` now discovers
+   the image's linked tasks from its own section/manifest records and
+   schedules the second one (the RM task) when the init task blocks:
+   `sim.libos_port.LibosPortNetwork` routes sends/receives/replies between
+   the modelled tasks with the firmware's own payload bytes, serves
+   `TASK_STATE_QUERY` with a modelled spurious/EINT trap state, and fixes the
+   synchronous receive semantics (a receive with a non-zero wait id blocks
+   instead of completing). Measured frontier: the RM task completes the boot
+   port handshake and then runs its own start-up past the 2M instruction cap
+   before its first further port syscall, so **no ring poll is reachable from
+   its entry inside the cap**; the report names the exact next semantics
+   (`missing_semantic.stage = second-task-frontier`): a data-driven post-init
+   resume point, the `GSP_ARGUMENTS_CACHED`/queue-init address handoff, and
+   the doorbell→consumer interrupt binding. Single-task behaviour is kept
+   with `--no-second-task` (and the old `task-state-query` classification).
 
 The seam for step 3 already exists: `HostTraffic(machine=..., shared_base=...)`
 uses the machine's RAM for the rings and `run_cycles` steps the core; the
diff --git a/debug/risc-v-gsp-rm-vm/docs/53-FIRMWARE-OBSERVABILITY-WORKFLOW.md b/debug/risc-v-gsp-rm-vm/docs/53-FIRMWARE-OBSERVABILITY-WORKFLOW.md
index 4106244..336f1b6 100644
--- a/debug/risc-v-gsp-rm-vm/docs/53-FIRMWARE-OBSERVABILITY-WORKFLOW.md
+++ b/debug/risc-v-gsp-rm-vm/docs/53-FIRMWARE-OBSERVABILITY-WORKFLOW.md
@@ -119,14 +119,21 @@ python3 -m sim.msgq_attach demo          # bounded run + honest skip if no image
 ```
 
 Measured today: **B3 reached** (~474k instructions), doorbell write performed
-and checksum-verified; the **libos port model** then completes the
+and checksum-verified; the **libos port model** completes the
 external-interrupt receive handoff (`wait @ 0x4019df4: completed shuttle 10
-(extintr) size=0`), and the loop task's next syscall is `TASK_STATE_QUERY` at
-`0x401a46c` — the libos trap/report path. **First ring poll: still none**, and
-the report says exactly why (trap state for the query; the command-queue
-consumer task is never scheduled in this single-hart model; the
-`GSP_ARGUMENTS_CACHED` address handoff is not delivered). No fabricated poll;
-two runs are byte-identical.
+(extintr) size=0`) and serves the following `TASK_STATE_QUERY` with a modelled
+spurious/EINT trap state, so the init task consumes the report and re-blocks.
+The image's **two linked tasks are discovered from its own section/manifest
+records** (an init task and the RM task); the **scheduled second task** then
+completes its boot port handshake with the init task inside the bounded
+budget (`modelled:cross-task-send` / `modelled:cross-task-reply`, payloads
+copied from firmware memory), after which its own start-up continues past the
+2M instruction cap before its first further port syscall. **First ring poll:
+still none**, and the report says exactly why
+(`missing_semantic.stage = second-task-frontier`: the consumer start-up is not
+reachable from its entry inside the cap; the `GSP_ARGUMENTS_CACHED`
+queue-address handoff is not delivered; the doorbell→consumer binding is not
+modelled). No fabricated poll; two runs are byte-identical.
 
 ---
 
@@ -243,3 +250,4 @@ Ordered by what each unblocks; each is bounded and session-recorded.
 | Date | Change |
 |---|---|
 | 2026-10-06 | Created with the four new tools (`state_decode`, `log_decode`, `server_snapshot`, `msgq_attach`), the layer map, the run-to-the-wall procedure and the continuation loop. |
+| 2026-10-06 | Continuation: `sim/libos_port.py` + two-task B5 attach (extintr handoff completes; first poll still blocked on post-init resume + `GSP_ARGUMENTS_CACHED`); `state_decode` gains aggregate `key: value` state parsing (m26→m29 diff). |
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,
diff --git a/debug/risc-v-gsp-rm-vm/sim/libos_port.py b/debug/risc-v-gsp-rm-vm/sim/libos_port.py
index eaa3f95..b1f596d 100644
--- a/debug/risc-v-gsp-rm-vm/sim/libos_port.py
+++ b/debug/risc-v-gsp-rm-vm/sim/libos_port.py
@@ -73,13 +73,33 @@ from __future__ import annotations
 
 from sim.libos import (
     INTERRUPT_ARM_EXTERNAL,
+    MEMORY_QUERY,
     NAMES,
+    SHUTTLE_RESET,
     TASK_PORT_SEND_RECV_WAIT,
+    TASK_STATE_QUERY,
 )
 
 #: Report schema tag for :meth:`LibosPortModel.state`.
 REPORT_FORMAT = "greeneye-msgq-port/1"
 
+#: Report schema tag for :meth:`LibosPortNetwork.state`.
+NETWORK_REPORT_FORMAT = "greeneye-msgq-port-network/1"
+
+#: The 8-byte "SPURIOUS" sentinel the firmware compares a queried task's
+#: ``xepc`` against (the little-endian bytes of the literal, observed in the
+#: image rodata and compared as a u64).  A modelled external-interrupt report
+#: carries it when no task trapped.
+SPURIOUS_XEPC = 0x53554F4952555053
+
+#: Modelled external-interrupt cause (interrupt bit | machine external
+#: interrupt code) for the doorbell wake's report state.
+EXTERNAL_INTERRUPT_XCAUSE = (1 << 63) | 11
+
+#: ``LibosTaskState`` size (32 registers + xepc + xcause + xbadaddr +
+#: nvriscv), the payload of ``TASK_STATE_QUERY``.
+TASK_STATE_BYTES = 288
+
 #: ``LIBOS_SHUTTLE_ANY`` (RTOS port API).
 SHUTTLE_ANY = 0xFFFF
 #: The task's default shuttle id (RTOS port API).
@@ -263,6 +283,10 @@ class LibosPortModel:
 
         send = a0 != 0 or a1 != 0
         recv = a4 != 0 or a5 != 0
+        # A receive with a non-zero wait id is a *synchronous* receive (or the
+        # receive half of a sendrecv): the kernel blocks until a message is
+        # delivered.  Only a wait id of 0 is an asynchronous post.
+        sync_recv = recv and t1 != 0
 
         if recv:
             self._post(shuttle=a4, port=a5, payload=a6, capacity=a7, pc=pc,
@@ -270,11 +294,54 @@ class LibosPortModel:
         if send:
             self._send(shuttle=a0, port=a1, payload=a2, size=a3, wait_id=t1,
                        timeout=t2, pc=pc, machine=machine)
-            return True, LIBOS_OK
-        if recv:
+            if not recv:
+                return True, LIBOS_OK
+        if recv and not sync_recv:
             # Async receive post: the kernel returns LIBOS_OK without touching
             # the wait outputs (a4-a6 still hold the receive arguments).
             return True, LIBOS_OK
+        if sync_recv:
+            if not self.completed:
+                self._try_kick()
+            if self.completed:
+                entry = self.completed.pop(0)
+                machine.x[14] = entry["payload_size"] & 0xFFFFFFFFFFFFFFFF
+                machine.x[15] = entry["origin_task"] & 0xFFFFFFFFFFFFFFFF
+                machine.x[16] = entry["shuttle"] & 0xFFFFFFFFFFFFFFFF
+                machine.x[17] = 0
+                record = {
+                    "pc": int(pc),
+                    "wait_id": t1,
+                    "timeout": t2,
+                    "completed": entry["shuttle"],
+                    "payload_size": entry["payload_size"],
+                    "origin_task": entry["origin_task"],
+                    "role": entry["role"],
+                    "contract": entry["contract"],
+                }
+                self.waits.append(record)
+                self._event("sync_recv_complete", **record)
+                return True, LIBOS_OK
+            record = {
+                "pc": int(pc),
+                "wait_id": t1,
+                "timeout": t2,
+                "reason": "sync-recv-blocks",
+                "port": int(a5),
+                "shuttle": int(a4),
+                "pending": sorted(self.pending),
+            }
+            self.waits.append(record)
+            self._event("sync_recv_block", **record)
+            if self.stopped is None:
+                self.stopped = {
+                    "num": TASK_PORT_SEND_RECV_WAIT,
+                    "name": "task_port_send_recv_wait",
+                    "pc": int(pc),
+                    "reason": "sync-recv-blocks",
+                    "pending": sorted(self.pending),
+                }
+            return False, None
 
         # Pure blocking wait: return the first completed shuttle, or stop.
         if not self.completed:
@@ -384,3 +451,443 @@ class LibosPortModel:
             "events": [dict(event) for event in self.events],
             "stopped": dict(self.stopped) if self.stopped else None,
         }
+
+
+def task_state_bytes(*, xcause: int, xepc: int, xbadaddr: int = 0,
+                     registers=None, priv_err_info: int = 0) -> bytes:
+    """Serialise a ``LibosTaskState`` payload (pure; no machine).
+
+    Layout: 32 u64 registers, then ``xepc``, ``xcause``, ``xbadaddr`` and the
+    NVRISCV error word (``priv_err_info`` u32 + reserved u32).
+    """
+    buf = bytearray(TASK_STATE_BYTES)
+    for index, value in enumerate(registers or ()):
+        if 0 <= index < 32:
+            buf[index * 8:index * 8 + 8] = (int(value) & 0xFFFFFFFFFFFFFFFF).to_bytes(8, "little")
+    buf[256:264] = (int(xepc) & 0xFFFFFFFFFFFFFFFF).to_bytes(8, "little")
+    buf[264:272] = (int(xcause) & 0xFFFFFFFFFFFFFFFF).to_bytes(8, "little")
+    buf[272:280] = (int(xbadaddr) & 0xFFFFFFFFFFFFFFFF).to_bytes(8, "little")
+    buf[280:284] = (int(priv_err_info) & 0xFFFFFFFF).to_bytes(4, "little")
+    return bytes(buf)
+
+
+class LibosPortNetwork(LibosPortModel):
+    """Two-task libos port network (additive, opt-in).
+
+    The real 570.144 GSP image links **two** tasks (an init daemon and the RM
+    task); the single-hart linear model only ever runs one of them.  This
+    subclass models the libos port machinery between the modelled tasks:
+
+    * receives/posts/completions are kept per task (shuttle ids are
+      task-local handles and collide across tasks);
+    * a send is routed to a receive waiting on the same port id in another
+      task, copying the firmware's own payload bytes (never an invented
+      payload); the receiver's completion records the sender's task index and
+      pairs a reply back to the sender's receive half (sendrecv);
+    * ``TASK_STATE_QUERY`` serves a modelled trap state for the external
+      interrupt report (the queried state's ``xcause`` is the interrupt cause
+      and ``xepc`` the "SPURIOUS" sentinel the firmware checks; both are
+      labelled model values, not firmware observations);
+    * ``MEMORY_QUERY`` and ``SHUTTLE_RESET`` are acknowledged from the flat
+      RAM model (labelled);
+    * ``INTERRUPT_ARM_EXTERNAL`` records the second argument (the port the
+      report is bound to) in addition to the mask.
+
+    The single-task :class:`LibosPortModel` behaviour and report schema are
+    unchanged; this class is used only when the runner schedules a second
+    task.
+    """
+
+    name = "libos-port-network"
+
+    def __init__(self, base=None, *, tasks=("init", "rm"), current=None):
+        task = current or tasks[0]
+        super().__init__(base=base, task=task)
+        self.tasks = tuple(tasks)
+        self.current = task
+        self.task_completions = {name: [] for name in self.tasks}
+        self.task_recvs = {name: {} for name in self.tasks}
+        self.send_queue = {port: [] for port in ()}
+        self.reply_map = {}
+        self.cross_task_payloads = []
+        self.services = []
+        self.queries = []
+        self.arm_ports = []
+        self.report_trap = None
+        self.switches = []
+        #: Blocked syscall per task (so a resumed ecall is not re-executed as
+        #: a fresh send/post: the kernel would switch back *after* the wait).
+        self.blocked = {}
+
+    # -- task identity -----------------------------------------------------
+
+    def switch(self, task: str, pc: int = 0) -> dict:
+        """Set the executing task (runner-side cooperative switch)."""
+        self.current = task
+        self.task = task
+        event = self._event("task_switch", pc=int(pc), to=task)
+        self.switches.append({"pc": int(pc), "to": task})
+        return event
+
+    def _task_index(self, name: str) -> int:
+        return self.tasks.index(name) if name in self.tasks else 0
+
+    # -- routing helpers ---------------------------------------------------
+
+    def _find_receiver(self, port: int):
+        for name in self.tasks:
+            for shuttle in sorted(self.task_recvs[name]):
+                post = self.task_recvs[name][shuttle]
+                if post["port"] == int(port):
+                    return post
+        return None
+
+    def _post(self, *, shuttle: int, port: int, payload: int, capacity: int,
+              pc: int, wait_id: int, timeout: int) -> dict:
+        post = {
+            "shuttle": int(shuttle),
+            "port": int(port),
+            "role": role_for(port),
+            "payload": int(payload),
+            "capacity": int(capacity),
+            "pc": int(pc),
+            "wait_id": int(wait_id),
+            "timeout": int(timeout),
+            "task": self.current,
+        }
+        self.posts.append(post)
+        self.task_recvs[self.current][int(shuttle)] = post
+        self._event("receive_post", **post)
+        self._try_kick()
+        return post
+
+    def _complete(self, shuttle: int, *, payload_size: int, origin_task: int,
+                  contract: str, reason: str) -> dict:
+        owner = self.current
+        post = None
+        for name in self.tasks:
+            if int(shuttle) in self.task_recvs[name]:
+                owner = name
+                post = self.task_recvs[name].pop(int(shuttle))
+                break
+        if post is None:  # pragma: no cover - caller only completes posts
+            raise KeyError(f"shuttle {shuttle} is not posted")
+        entry = {
+            "shuttle": int(shuttle),
+            "port": int(post["port"]),
+            "role": post["role"],
+            "payload": int(post["payload"]),
+            "capacity": int(post["capacity"]),
+            "payload_size": int(payload_size),
+            "origin_task": int(origin_task),
+            "contract": contract,
+            "reason": reason,
+            "post_pc": int(post["pc"]),
+            "task": owner,
+        }
+        self.completed.append(entry)
+        self.task_completions[owner].append(entry)
+        self._event("receive_complete", **entry)
+        return entry
+
+    def _copy(self, machine, dst: int, src: int, size: int) -> bool:
+        if size <= 0:
+            return True
+        try:
+            data = machine.read_mem(int(src), int(size))
+            machine.write_mem(int(dst), data)
+            return True
+        except Exception:
+            return False
+
+    def _take_completion(self, machine, pc: int):
+        queue = self.task_completions[self.current]
+        if not queue:
+            return None
+        entry = queue.pop(0)
+        machine.x[14] = entry["payload_size"] & 0xFFFFFFFFFFFFFFFF
+        machine.x[15] = entry["origin_task"] & 0xFFFFFFFFFFFFFFFF
+        machine.x[16] = entry["shuttle"] & 0xFFFFFFFFFFFFFFFF
+        machine.x[17] = 0
+        record = {
+            "pc": int(pc),
+            "completed": entry["shuttle"],
+            "payload_size": entry["payload_size"],
+            "origin_task": entry["origin_task"],
+            "role": entry["role"],
+            "contract": entry["contract"],
+            "task": self.current,
+        }
+        self.waits.append(record)
+        self._event("wait_complete", **record)
+        return record
+
+    def _block(self, machine, pc: int, *, reason: str, port: int,
+               shuttle: int) -> None:
+        kind = {"port-wait-blocks": "wait", "sync-recv-blocks": "recv",
+                "sync-send-blocks": "send"}.get(reason, "wait")
+        self.blocked[self.current] = {"kind": kind, "reason": reason,
+                                      "pc": int(pc), "port": int(port),
+                                      "shuttle": int(shuttle)}
+        record = {
+            "pc": int(pc),
+            "reason": reason,
+            "port": int(port),
+            "shuttle": int(shuttle),
+            "pending": sorted(s for name in self.tasks
+                              for s in self.task_recvs[name]),
+            "task": self.current,
+        }
+        self.waits.append(record)
+        self._event("wait_block", **record)
+        if self.stopped is None:
+            self.stopped = {
+                "num": TASK_PORT_SEND_RECV_WAIT,
+                "name": "task_port_send_recv_wait",
+                "pc": int(pc),
+                "reason": reason,
+                "pending": record["pending"],
+                "task": self.current,
+            }
+
+    # -- syscall surface ---------------------------------------------------
+
+    def handle(self, machine, pc: int = 0):
+        num = int(machine.x[5])
+        if num == TASK_PORT_SEND_RECV_WAIT:
+            return self._handle_network(machine, pc)
+        if num == INTERRUPT_ARM_EXTERNAL:
+            mask, port = int(machine.x[10]), int(machine.x[11])
+            self.armed_mask |= mask
+            self.arm_ports.append({"mask": mask, "port": port, "pc": int(pc),
+                                   "task": self.current})
+            self._event("interrupt_arm_external", mask=mask, port=port,
+                        pc=pc, task=self.current)
+            self._try_kick()
+            return True, LIBOS_OK
+        if num == TASK_STATE_QUERY:
+            self.queries.append({"pc": int(pc), "task_index": int(machine.x[10]),
+                                 "buffer": int(machine.x[11]),
+                                 "task": self.current})
+            trap = self.report_trap
+            if trap is None:
+                # No modelled trap state: the kernel would return an access
+                # error for an invalid target; return LIBOS_ERROR_INVALID_ACCESS
+                # (labelled model value) and let the firmware error path run.
+                return True, 0x80000000
+            state = task_state_bytes(
+                xcause=trap["xcause"], xepc=trap["xepc"],
+                xbadaddr=trap["xbadaddr"], registers=trap.get("registers"),
+            )
+            try:
+                machine.write_mem(int(machine.x[11]), state)
+            except Exception:
+                return True, 0x80000000
+            self._event("task_state_query", **self.queries[-1],
+                        xcause=hex(trap["xcause"]), xepc=hex(trap["xepc"]))
+            return True, LIBOS_OK
+        if num == MEMORY_QUERY:
+            remote, pointer = int(machine.x[10]), int(machine.x[11])
+            attrs = 0x1 | 0x2 | 0x8  # read | write | cached (flat model)
+            resolved = pointer
+            translator = getattr(machine, "translator", None)
+            if translator is not None:
+                sim = translator.translate(pointer)
+                if sim is not None:
+                    resolved = sim
+            machine.x[11] = attrs
+            machine.x[12] = pointer & ~0xFFF
+            machine.x[13] = 0x1000
+            machine.x[14] = 1  # LIBOS_MEMORY_APERTURE_SYSCOH
+            machine.x[15] = resolved
+            record = {"pc": int(pc), "remote_task": remote, "pointer": pointer,
+                      "resolved": resolved, "task": self.current}
+            self.services.append(dict(record, service="memory_query"))
+            self._event("memory_query", **record)
+            return True, LIBOS_OK
+        if num == SHUTTLE_RESET:
+            record = {"pc": int(pc), "shuttle": int(machine.x[10]),
+                      "task": self.current}
+            self.services.append(dict(record, service="shuttle_reset"))
+            self._event("shuttle_reset", **record)
+            return True, LIBOS_OK
+        return super().handle(machine, pc)
+
+    def _handle_network(self, machine, pc: int):
+        resuming = self.blocked.pop(self.current, None)
+        if resuming is not None and resuming["kind"] in ("recv", "wait"):
+            record = self._take_completion(machine, pc)
+            if record is not None:
+                return True, LIBOS_OK
+            self._block(machine, pc, reason=resuming["reason"],
+                        port=resuming["port"], shuttle=resuming["shuttle"])
+            return False, None
+        if resuming is not None and resuming["kind"] == "send":
+            self._block(machine, pc, reason=resuming["reason"],
+                        port=resuming["port"], shuttle=resuming["shuttle"])
+            return False, None
+        a0, a1, a2, a3 = (int(machine.x[10]), int(machine.x[11]),
+                          int(machine.x[12]), int(machine.x[13]))
+        a4, a5, a6, a7 = (int(machine.x[14]), int(machine.x[15]),
+                          int(machine.x[16]), int(machine.x[17]))
+        t1, t2 = int(machine.x[6]), int(machine.x[7])
+        send = a0 != 0 or a1 != 0
+        recv = a4 != 0 or a5 != 0
+        sync_recv = recv and t1 != 0
+
+        if send and a1 == 0:
+            # Reply: route through the received shuttle's partner.
+            target = self.reply_map.pop((self.current, a0), None)
+            if target is not None:
+                name, shuttle = target
+                post = self.task_recvs[name].get(shuttle)
+                if post is not None:
+                    size = min(a3, int(post["capacity"]))
+                    ok = self._copy(machine, post["payload"], a2, size)
+                    entry = self._complete(
+                        shuttle, payload_size=size if ok else 0,
+                        origin_task=self._task_index(self.current),
+                        contract="modelled:cross-task-reply",
+                        reason=f"reply from {self.current} ({size} bytes)")
+                    self.sends.append({"shuttle": a0, "port": 0, "size": a3,
+                                       "pc": int(pc), "task": self.current,
+                                       "queued": False, "delivered_to": entry["shuttle"]})
+                    self.cross_task_payloads.append({
+                        "port": 0, "size": size, "from": self.current,
+                        "to": entry.get("task", name), "kind": "reply"})
+                    machine.x[14] = size
+                    machine.x[15] = 0
+                    machine.x[16] = a0
+                    machine.x[17] = 0
+                    if t1 == a0 or not recv:
+                        return True, LIBOS_OK
+
+        if send and a1 != 0:
+            partner = self._find_receiver(a1)
+            if partner is None:
+                size = a3
+                self.sends.append({"shuttle": a0, "port": a1, "size": size,
+                                   "pc": int(pc), "task": self.current,
+                                   "queued": True, "payload": a2})
+                return self._unrouted_send(machine, pc, a0, a1, a3, t1,
+                                           recv, sync_recv)
+            size = min(a3, int(partner["capacity"]))
+            ok = self._copy(machine, partner["payload"], a2, size)
+            entry = self._complete(
+                int(partner["shuttle"]), payload_size=size if ok else 0,
+                origin_task=self._task_index(self.current),
+                contract="modelled:cross-task-send",
+                reason=f"send from {self.current} port {a1} copied {size}")
+            self.sends.append({"shuttle": a0, "port": a1, "size": a3,
+                               "pc": int(pc), "task": self.current,
+                               "queued": False, "delivered_to": entry["shuttle"]})
+            self.cross_task_payloads.append({
+                "port": int(a1), "size": size, "from": self.current,
+                "to": entry.get("task"), "kind": "send"})
+            if a4 or a5:
+                self.reply_map[(entry.get("task"), entry["shuttle"])] = (self.current, a4)
+            if t1 == a0:
+                machine.x[14] = size
+                machine.x[15] = 0
+                machine.x[16] = a0
+                machine.x[17] = 0
+                return True, LIBOS_OK
+            if not recv:
+                return True, LIBOS_OK
+
+        if recv:
+            self._post(shuttle=a4, port=a5, payload=a6, capacity=a7, pc=pc,
+                       wait_id=t1, timeout=t2)
+            if sync_recv:
+                record = self._take_completion(machine, pc)
+                if record is not None:
+                    return True, LIBOS_OK
+                self._block(machine, pc, reason="sync-recv-blocks", port=a5,
+                            shuttle=a4)
+                return False, None
+            return True, LIBOS_OK
+
+        record = self._take_completion(machine, pc)
+        if record is not None:
+            return True, LIBOS_OK
+        self._block(machine, pc, reason="port-wait-blocks", port=0, shuttle=0)
+        return False, None
+
+    def _unrouted_send(self, machine, pc, shuttle, port, size, wait_id,
+                       recv, sync_recv):
+        """A send with no modelled receiver: keep the boot-shape shim.
+
+        The zero-length boot handshake send had no modelled peer and is an
+        explicit, labelled assumption (``modelled:peer-consumes-on-send``);
+        every other unrouted send stays queued and blocks its sync sender.
+        """
+        if size != 0:
+            if wait_id == shuttle:
+                self._block(machine, pc, reason="sync-send-blocks", port=port,
+                            shuttle=shuttle)
+                return False, None
+            return True, LIBOS_OK
+        machine.x[14] = 0
+        machine.x[15] = 0
+        machine.x[16] = shuttle
+        machine.x[17] = 0
+        self._event("send_complete", shuttle=shuttle, port=port, size=0,
+                    pc=int(pc), task=self.current,
+                    contract="modelled:peer-consumes-on-send")
+        if wait_id == shuttle or not recv:
+            return True, LIBOS_OK
+        if sync_recv:
+            record = self._take_completion(machine, pc)
+            if record is not None:
+                return True, LIBOS_OK
+            self._block(machine, pc, reason="sync-recv-blocks", port=0,
+                        shuttle=0)
+            return False, None
+        return True, LIBOS_OK
+
+    def _try_kick(self):
+        """Deliver a pending host wake to the extintr receive (documented shim).
+
+        The bound interrupt port is recorded from the arm call (the real kernel
+        delivers to the armed port); the modelled report retires the waiting
+        external-interrupt receive and records the modelled trap state its
+        ``TASK_STATE_QUERY`` consumer observes.
+        """
+        if not self.kicks_pending:
+            return None
+        for name in self.tasks:
+            for shuttle in sorted(self.task_recvs[name]):
+                post = self.task_recvs[name][shuttle]
+                if post["role"] != "extintr":
+                    continue
+                self.kicks_pending -= 1
+                self.report_trap = {
+                    "xcause": EXTERNAL_INTERRUPT_XCAUSE,
+                    "xepc": SPURIOUS_XEPC,
+                    "xbadaddr": 0,
+                    "origin_task": 0,
+                }
+                self._complete(shuttle, payload_size=0, origin_task=0,
+                               contract=EXTINTR_WAKE_CONTRACT,
+                               reason="external-interrupt report")
+                return shuttle
+        return None
+
+    def state(self) -> dict:
+        base = super().state()
+        base.update({
+            "format": NETWORK_REPORT_FORMAT,
+            "tasks": list(self.tasks),
+            "current": self.current,
+            "task_completions": {name: [dict(e) for e in entries]
+                                 for name, entries in sorted(self.task_completions.items())},
+            "recvs": {name: sorted(self.task_recvs[name])
+                      for name in self.tasks},
+            "cross_task_payloads": [dict(p) for p in self.cross_task_payloads],
+            "services": [dict(s) for s in self.services],
+            "queries": [dict(q) for q in self.queries],
+            "arm_ports": [dict(a) for a in self.arm_ports],
+            "report_trap": dict(self.report_trap) if self.report_trap else None,
+            "switches": [dict(s) for s in self.switches],
+        })
+        return base
diff --git a/debug/risc-v-gsp-rm-vm/sim/msgq_attach.py b/debug/risc-v-gsp-rm-vm/sim/msgq_attach.py
index f1530bd..01fc84a 100644
--- a/debug/risc-v-gsp-rm-vm/sim/msgq_attach.py
+++ b/debug/risc-v-gsp-rm-vm/sim/msgq_attach.py
@@ -85,7 +85,7 @@ from sim.host_traffic import (
     parse_queue_element,
 )
 from sim.libos import LibosSyscalls, TASK_PORT_SEND_RECV_WAIT, TASK_STATE_QUERY
-from sim.libos_port import LibosPortModel
+from sim.libos_port import LibosPortModel, LibosPortNetwork
 from sim.machine import R_BP, R_BUDGET, R_ECALL, R_FAULT, R_STEP, R_WATCH
 
 #: Report schema tag.  Every session/JSON artifact carries it.
@@ -102,6 +102,16 @@ MAX_POLLS_RECORDED = 17
 #: Internal stop code: the bounded poll list was filled before the run ended.
 R_POLL_LIMIT = 199
 
+#: Default extra slice cap for the scheduled second task (0 = run it until
+#: the total instruction budget is reached).  The observed RM task completes
+#: its boot handshake with the init task inside a few thousand instructions;
+#: its own start-up then runs past the total cap, so the frontier is reported
+#: at the cap, never as success.
+DEFAULT_SECOND_TASK_BUDGET = 0
+
+#: Wait reasons that mean "this task is blocked and another task may run".
+_BLOCK_REASONS = ("port-wait-blocks", "sync-recv-blocks", "sync-send-blocks")
+
 #: Doorbell label (block-relative register name; public Turing dev_gsp.h).
 DOORBELL_WINDOW = "PRIV+0x110c00"
 
@@ -297,6 +307,7 @@ def assemble_report(*, image: dict, profile: dict, boot: dict, attach: dict,
                     doorbell: dict, polls: list, divergence, missing_mmio: dict,
                     determinism: dict, continuation: dict | None = None,
                     port_model: dict | None = None, missing_semantic: dict | None = None,
+                    tasks: dict | None = None, schedule: list | None = None,
                     limits=None, notes=None) -> dict:
     """Assemble the ``greeneye-msgq-attach/1`` report from plain inputs.
 
@@ -343,6 +354,9 @@ def assemble_report(*, image: dict, profile: dict, boot: dict, attach: dict,
                       else {"enabled": False},
         "missing_semantic": dict(missing_semantic) if missing_semantic is not None
                             else None,
+        "tasks": dict(tasks) if tasks is not None
+                 else {"format": "greeneye-image-tasks/1", "tasks": []},
+        "schedule": [dict(step) for step in (schedule or [])],
         "determinism": dict(determinism),
         "limits": list(limits if limits is not None else _LIMITS),
         "notes": list(notes or []),
@@ -533,6 +547,153 @@ def _verify_command_element(host: HostTraffic, submit: dict) -> dict:
         return {"checksum_ok": False, "error": str(exc)}
 
 
+def image_tasks(image: str) -> dict:
+    """Discover the linked libos tasks from the image's section header table.
+
+    Pure stdlib ELF64 parsing (no symbols are present in the image): the
+    ``section_task_<name>_elf_text_instance`` sections give each task's text
+    range and the ``section_task_<name>_stack_instance`` sections its stack;
+    the task entry is the first pointer inside the text range found in the
+    ``.manifest`` section (the firmware's own task record).  Data-driven: an
+    image without section names, manifest or match returns an empty table and
+    the caller stays single-task.
+
+    Returns ``{"format": "greeneye-image-tasks/1", "tasks": [...]}``.
+    """
+    empty = {"format": "greeneye-image-tasks/1", "tasks": []}
+    try:
+        with open(image, "rb") as handle:
+            blob = handle.read()
+    except OSError:
+        return empty
+    if len(blob) < 64 or blob[:4] != b"\x7fELF" or blob[4] != 2:
+        return empty
+    shoff = int.from_bytes(blob[0x28:0x30], "little")
+    shentsize = int.from_bytes(blob[0x3A:0x3C], "little")
+    shnum = int.from_bytes(blob[0x3C:0x3E], "little")
+    shstrndx = int.from_bytes(blob[0x3E:0x40], "little")
+    if not (0 < shentsize and 0 < shnum and shstrndx < shnum
+            and shoff + shentsize * shnum <= len(blob)):
+        return empty
+    shstr = None
+    raw_sections = []
+    for index in range(shnum):
+        base = shoff + index * shentsize
+        if base + 64 > len(blob):
+            return empty
+        fields = blob[base:base + 64]
+        name_off = int.from_bytes(fields[0:4], "little")
+        addr = int.from_bytes(fields[16:24], "little")
+        offset = int.from_bytes(fields[24:32], "little")
+        size = int.from_bytes(fields[32:40], "little")
+        raw_sections.append((name_off, addr, offset, size))
+    _, shstr_addr, shstr_off, shstr_size = raw_sections[shstrndx]
+    shstr = blob[shstr_off:shstr_off + shstr_size]
+
+    def name_at(name_off):
+        end = shstr.find(b"\x00", name_off)
+        return shstr[name_off:end].decode("latin1") if end >= 0 else ""
+
+    sections = {}
+    for name_off, addr, offset, size in raw_sections:
+        name = name_at(name_off)
+        if name:
+            sections[name] = {"addr": addr, "offset": offset, "size": size, "name": name}
+    tasks = {}
+    for name, section in sections.items():
+        if name.startswith(".section_task_") and name.endswith("_elf_text_instance"):
+            task = name[len(".section_task_"):-len("_elf_text_instance")]
+            tasks.setdefault(task, {})["text"] = {"base": section["addr"],
+                                                  "size": section["size"]}
+        elif name.startswith(".section_task_") and name.endswith("_stack_instance"):
+            task = name[len(".section_task_"):-len("_stack_instance")]
+            tasks.setdefault(task, {})["stack"] = {
+                "base": section["addr"], "size": section["size"],
+                "top": section["addr"] + section["size"]}
+    manifest = sections.get(".manifest")
+    if manifest:
+        data = blob[manifest["offset"]:manifest["offset"] + manifest["size"]]
+        for index in range(0, len(data) - 7, 8):
+            value = int.from_bytes(data[index:index + 8], "little")
+            for task, record in tasks.items():
+                text = record.get("text")
+                if text and "entry" not in record \
+                        and text["base"] <= value < text["base"] + text["size"]:
+                    record["entry"] = value
+    ordered = []
+    for task in sorted(tasks):
+        record = tasks[task]
+        if "text" not in record or "entry" not in record:
+            continue
+        ordered.append({
+            "name": task,
+            "entry": int(record["entry"]),
+            "text_base": int(record["text"]["base"]),
+            "text_size": int(record["text"]["size"]),
+            "stack_base": int(record.get("stack", {}).get("base", 0)),
+            "stack_size": int(record.get("stack", {}).get("size", 0)),
+            "stack_top": int(record.get("stack", {}).get("top", 0)),
+        })
+    return {"format": "greeneye-image-tasks/1", "tasks": ordered}
+
+
+def _task_for_pc(tasks_doc: dict, pc: int) -> dict | None:
+    for task in tasks_doc.get("tasks", []):
+        if task["text_base"] <= pc < task["text_base"] + task["text_size"]:
+            return task
+    return None
+
+
+def missing_semantic_after_second_task(*, tasks_doc: dict, schedule: list,
+                                       frontier: dict, polls: list) -> dict:
+    """The exact unmodelled semantics at the two-task frontier (pure).
+
+    Recorded when the scheduled second task (the RM task discovered from the
+    image) is blocked or out of its bounded slice without touching the rings.
+    """
+    first = schedule[0] if schedule else {}
+    second = schedule[-1] if schedule else {}
+    return {
+        "stage": "second-task-frontier",
+        "tasks": [task["name"] for task in tasks_doc.get("tasks", [])],
+        "observed": [
+            "the host doorbell was delivered as the RTOS external-interrupt "
+            "report and the init task consumed it as a spurious/EINT report "
+            "(modelled trap state, TASK_STATE_QUERY served)",
+            "the init task re-posted its receive and blocked in its port wait",
+            "the second task (the RM task, entry and stack discovered from "
+            "the image section/manifest records) was scheduled and completed "
+            "its boot port handshake with the init task",
+            "no command/status ring or doorbell read occurred within the "
+            "bounded instruction budget; the RM task's own start-up continues "
+            "past the cap before its first further port syscall",
+        ],
+        "missing": [
+            "the queue consumer's start-up is not reachable inside the 2M "
+            "instruction cap from its entry; a data-driven post-init resume "
+            "point (or a qualified budget) is the smallest next semantic",
+            "the GSP_ARGUMENTS_CACHED / message-queue init handoff: the "
+            "simulator never delivers the queue base addresses the host "
+            "actually gives the firmware, so the rings stay at the fixture "
+            "block and no firmware queue poll can be observed",
+            "the doorbell -> interrupt binding for the consumer task: the "
+            "bound interrupt port recorded from INTERRUPT_ARM_EXTERNAL is "
+            "not the port the RM task receives the queue kick on",
+        ],
+        "next_step": (
+            "extend the schedule with a data-driven post-init entry for the "
+            "RM task (or raise the cap with an admission record), deliver the "
+            "GSP_ARGUMENTS_CACHED/queue-init addresses from the host model, "
+            "then re-run this attach and classify the first ring poll; do not "
+            "infer a poll from this run"),
+        "schedule": [dict(step) for step in schedule],
+        "frontier": dict(frontier) if frontier else None,
+        "polls": len(polls),
+        "first_task": first.get("task"),
+        "second_task": second.get("task"),
+    }
+
+
 def missing_semantic_after_handoff(*, stopped: dict, polls: list) -> dict:
     """The exact unmodelled semantics that stopped the port-model continuation.
 
@@ -578,7 +739,9 @@ def missing_semantic_after_handoff(*, stopped: dict, polls: list) -> dict:
 
 def run_attach(*, image: str = DEFAULT_IMAGE, profile_path: str = DEFAULT_PROFILE,
                max_cycles: int = DEFAULT_MAX_CYCLES, doorbell: bool = False,
-               port_model: bool = True, json_out: str | None = None,
+               port_model: bool = True, second_task: bool = True,
+               second_task_budget: int = DEFAULT_SECOND_TASK_BUDGET,
+               json_out: str | None = None,
                record_dir: str | None = None) -> dict:
     """Boot to the port wait, attach the rings, optionally kick and continue.
 
@@ -595,11 +758,25 @@ def run_attach(*, image: str = DEFAULT_IMAGE, profile_path: str = DEFAULT_PROFIL
     task-state-query the interrupt dispatch performs (or further).  With the
     model off, the documented single-wake shim of the first wave is used and
     the run reports ``port-wait-reentry`` instead.
+
+    ``second_task`` (default on) additionally discovers the image's linked
+    tasks (section/manifest records) and schedules the second one (the RM
+    task) when the init task blocks: the boot port handshake is modelled with
+    the firmware's own payloads, bounded by the total instruction budget (an
+    optional tighter ``second_task_budget`` slice cap, 0 = until the total
+    budget).  The second task's own start-up runs past the cap, so the report
+    names that frontier instead of manufacturing a poll; the single-task
+    behaviour is kept with ``second_task=False``.
     """
     if isinstance(max_cycles, bool) or not isinstance(max_cycles, int) \
             or not 0 < max_cycles <= MAX_RUN_CYCLES:
         raise MsgqAttachError(
             f"max_cycles must be an int in 1..{MAX_RUN_CYCLES}, got {max_cycles!r}")
+    if isinstance(second_task_budget, bool) or not isinstance(second_task_budget, int) \
+            or not 0 <= second_task_budget <= MAX_RUN_CYCLES:
+        raise MsgqAttachError(
+            f"second_task_budget must be an int in 0..{MAX_RUN_CYCLES} "
+            f"(0 = until the total budget), got {second_task_budget!r}")
     image = os.path.abspath(image)
     profile_path = os.path.abspath(profile_path)
     if not os.path.isfile(image):
@@ -616,11 +793,13 @@ def run_attach(*, image: str = DEFAULT_IMAGE, profile_path: str = DEFAULT_PROFIL
     load = profile.get("load", {})
     ram_base = _as_int(load.get("ram_base", 0))
     memory_mb = _as_int(load.get("memory_mb", 12288))
+    tasks_doc = image_tasks(image)
 
     session = None
     notes: list[str] = []
     divergence = None
     polls: list[dict] = []
+    schedule: list[dict] = []
     continuation: dict = {"ran": False}
     port_model_state: dict | None = None
     missing_semantic: dict | None = None
@@ -699,7 +878,163 @@ def run_attach(*, image: str = DEFAULT_IMAGE, profile_path: str = DEFAULT_PROFIL
                                f"stop={stop_reason} at pc={pc:#x})"),
                 }
 
-        if wait_reached and doorbell and port_model:
+        if wait_reached and doorbell and port_model and second_task:
+            # Two-task schedule: the image's second linked task (the RM task)
+            # is data-discovered and seeded when the init task blocks.
+            primary = _task_for_pc(tasks_doc, int(pc)) or (tasks_doc["tasks"][0]
+                                                           if tasks_doc["tasks"] else None)
+            secondary = next((task for task in tasks_doc["tasks"] if task is not primary),
+                             None) if primary else None
+            if primary is not None and secondary is not None:
+                model = LibosPortNetwork(
+                    base=libos, tasks=(primary["name"], secondary["name"]),
+                    current=primary["name"])
+                machine.syscall_handler = model.handle
+                last_kick = host.doorbell.writes[-1] if host.doorbell.writes else None
+                model.host_kick(
+                    queue=int(last_kick["queue"]) if last_kick else None,
+                    value=int(last_kick["value"]) if last_kick else None,
+                )
+                schedule: list[dict] = []
+                saved = {primary["name"]: None, secondary["name"]: None}
+                started = {primary["name"]: True, secondary["name"]: False}
+                entries = {primary["name"]: primary, secondary["name"]: secondary}
+                reason2, pc2, a2, b2 = _run_watching(machine, max_cycles, geometry,
+                                                     milestones_hit, polls)
+                for _ in range(8):
+                    stop = model.stopped if reason2 == R_ECALL else None
+                    if polls or stop is None \
+                            or stop.get("reason") not in _BLOCK_REASONS \
+                            or reason2 == R_POLL_LIMIT:
+                        break
+                    current = model.current
+                    saved[current] = {"registers": list(machine.x),
+                                      "pc": int(machine.pc)}
+                    schedule.append({"task": current, "pc": int(pc2),
+                                     "reason": stop.get("reason"),
+                                     "insns": int(machine.insns)})
+                    other = secondary["name"] if current == primary["name"] \
+                        else primary["name"]
+                    if not started[other]:
+                        if machine.insns >= max_cycles:
+                            break
+                        started[other] = True
+                        record = entries[other]
+                        machine.x = [0] * 32
+                        machine.x[2] = int(record.get("stack_top", 0))
+                        machine.pc = int(record["entry"])
+                        model.switch(other, pc=int(machine.pc))
+                        model.stopped = None
+                        schedule.append({"task": other, "pc": int(machine.pc),
+                                         "reason": "scheduled",
+                                         "insns": int(machine.insns)})
+                        slice_end = max_cycles if not second_task_budget \
+                            else min(max_cycles,
+                                     machine.insns + int(second_task_budget))
+                        reason2, pc2, a2, b2 = _run_watching(
+                            machine, slice_end, geometry, milestones_hit, polls)
+                        continue
+                    if model.task_completions.get(other):
+                        machine.x = list(saved[other]["registers"])
+                        machine.pc = int(saved[other]["pc"])
+                        model.switch(other, pc=int(machine.pc))
+                        model.stopped = None
+                        schedule.append({"task": other, "pc": int(machine.pc),
+                                         "reason": "resumed",
+                                         "insns": int(machine.insns)})
+                        reason2, pc2, a2, b2 = _run_watching(
+                            machine, max_cycles, geometry, milestones_hit, polls)
+                        continue
+                    break
+                stop_reason2 = STOP_REASONS.get(reason2, str(reason2))
+                if reason2 == R_POLL_LIMIT:
+                    stop_reason2 = "poll-limit"
+                port_model_state = {"enabled": True, **model.state()}
+                continuation = {
+                    "ran": True,
+                    "port_model": True,
+                    "second_task": True,
+                    "stop_reason": stop_reason2,
+                    "stop_pc": int(pc2),
+                    "instructions_executed": int(machine.insns),
+                    "wakes": int(model.host_kicks),
+                    "wake_log": [
+                        {
+                            "result": entry["contract"],
+                            "returned": 0,
+                            "delivered": entry.get("delivered"),
+                            "pending": entry.get("pending"),
+                            "queue": entry.get("queue"),
+                            "value": entry.get("value"),
+                        }
+                        for entry in model.events
+                        if entry["event"] == "host_kick"
+                    ],
+                    "reentries": [],
+                    "polls": len(polls),
+                    "tasks": list(model.tasks),
+                    "switches": len(model.switches),
+                }
+                if reason2 == R_FAULT:
+                    divergence = {
+                        "kind": "fault",
+                        "reason": (f"fault after the host wake: cause={a2:#x} "
+                                   f"tval={b2:#x} pc={pc2:#x}"),
+                    }
+                elif reason2 == R_ECALL:
+                    port_stopped = model.stopped or (
+                        dict(libos.stopped) if libos.stopped else None)
+                    number2 = int(port_stopped["num"]) if port_stopped \
+                        else int(machine.x[5])
+                    name2 = (port_stopped or {}).get("name", "unhandled ecall")
+                    if not polls:
+                        attack = (port_stopped or {})
+                        missing_semantic = missing_semantic_after_second_task(
+                            tasks_doc=tasks_doc, schedule=schedule,
+                            frontier=attack, polls=polls)
+                        divergence = {
+                            "kind": "second-task-frontier",
+                            "reason": (
+                                "after the recorded host wake the init task "
+                                "consumed the external-interrupt report and "
+                                "blocked; the discovered second task (the RM "
+                                "task) was scheduled, completed its boot port "
+                                "handshake and then blocked without touching "
+                                "the rings -- "
+                                f"{name2} (num={number2}) at pc={pc2:#x}"),
+                            "stop": dict(attack),
+                        }
+                elif reason2 == R_BUDGET and not polls:
+                    frontier = {
+                        "task": model.current,
+                        "reason": "second-task-budget" if second_task_budget
+                                  else "total-budget",
+                        "pc": int(pc2),
+                    }
+                    missing_semantic = missing_semantic_after_second_task(
+                        tasks_doc=tasks_doc, schedule=schedule,
+                        frontier=frontier, polls=polls)
+                    if second_task_budget:
+                        divergence = {
+                            "kind": "second-task-frontier",
+                            "reason": (f"the discovered second task ran its "
+                                       f"bounded slice ({second_task_budget} "
+                                       f"instructions) without a ring read "
+                                       f"(total budget {max_cycles})"),
+                        }
+                    else:
+                        divergence = {
+                            "kind": "second-task-frontier",
+                            "reason": (f"the discovered second task ran to the "
+                                       f"total budget of {max_cycles} "
+                                       f"instructions without a ring read "
+                                       f"(its start-up continues past the cap)"),
+                        }
+                elif reason2 == R_POLL_LIMIT:
+                    notes.append(
+                        f"continuation stopped after {MAX_POLLS_RECORDED} "
+                        "recorded ring accesses (bounded by design)")
+        elif wait_reached and doorbell and port_model:
             model = LibosPortModel(base=libos)
             machine.syscall_handler = model.handle
             last_kick = host.doorbell.writes[-1] if host.doorbell.writes else None
@@ -935,6 +1270,8 @@ def run_attach(*, image: str = DEFAULT_IMAGE, profile_path: str = DEFAULT_PROFIL
             continuation=continuation,
             port_model=port_model_state,
             missing_semantic=missing_semantic,
+            tasks=tasks_doc,
+            schedule=schedule,
             notes=notes,
         )
         if getattr(session.recorder, "dir", None) is not None:
@@ -1069,6 +1406,14 @@ def build_parser() -> argparse.ArgumentParser:
     run_cmd.add_argument("--no-port-model", action="store_true",
                          help="disable the sim.libos_port receive-path model "
                               "(use the first-wave single-wake shim)")
+    run_cmd.add_argument("--no-second-task", action="store_true",
+                         help="keep the single-task port model (do not schedule "
+                              "the image's second linked task)")
+    run_cmd.add_argument("--second-task-budget", type=int,
+                         default=DEFAULT_SECOND_TASK_BUDGET,
+                         help="extra instruction slice cap for the scheduled "
+                              "second task (0 = until the total budget, "
+                              f"default {DEFAULT_SECOND_TASK_BUDGET})")
     run_cmd.add_argument("--json", metavar="OUT", default=None,
                          help="write the full report JSON to OUT")
     run_cmd.add_argument("--record", metavar="DIR", default=None,
@@ -1093,6 +1438,7 @@ def main(argv: list[str] | None = None) -> int:
         parser.print_help(sys.stderr)
         return 2
     port_model = not getattr(args, "no_port_model", False)
+    second_task = not getattr(args, "no_second_task", False)
     if args.command == "demo":
         run_demo(image=args.image, profile_path=args.profile,
                  print_json=args.json, port_model=port_model)
@@ -1104,6 +1450,8 @@ def main(argv: list[str] | None = None) -> int:
             max_cycles=args.max_cycles,
             doorbell=args.doorbell,
             port_model=port_model,
+            second_task=second_task,
+            second_task_budget=args.second_task_budget,
             json_out=args.json,
             record_dir=args.record,
         )
diff --git a/debug/risc-v-gsp-rm-vm/tests/test_libos_port.py b/debug/risc-v-gsp-rm-vm/tests/test_libos_port.py
index 0f01c01..befa14f 100644
--- a/debug/risc-v-gsp-rm-vm/tests/test_libos_port.py
+++ b/debug/risc-v-gsp-rm-vm/tests/test_libos_port.py
@@ -29,11 +29,15 @@ if TOOL_ROOT not in sys.path:
 from sim.libos import INTERRUPT_ARM_EXTERNAL, TASK_PORT_SEND_RECV_WAIT  # noqa: E402
 from sim.libos_port import (  # noqa: E402
     EXTINTR_WAKE_CONTRACT,
+    EXTERNAL_INTERRUPT_XCAUSE,
     HANDSHAKE_CONTRACT,
+    NETWORK_REPORT_FORMAT,
     REPORT_FORMAT,
     SHUTTLE_ANY,
+    SPURIOUS_XEPC,
     TIMEOUT_INFINITE,
     LibosPortModel,
+    LibosPortNetwork,
     role_for,
 )
 
@@ -50,6 +54,26 @@ class FakeMachine:
         self.x[index] = value & 0xFFFFFFFFFFFFFFFF
 
 
+class FakeMemoryMachine(FakeMachine):
+    """Register file plus flat memory for the routing model tests."""
+
+    def __init__(self, size: int = 0x10000):
+        super().__init__()
+        self.ram = bytearray(size)
+        self.ram_base = 0
+        self.ram_size = size
+
+    def read_mem(self, addr: int, size: int) -> bytes:
+        if addr < 0 or addr + size > self.ram_size:
+            raise ValueError("out of range")
+        return bytes(self.ram[addr:addr + size])
+
+    def write_mem(self, addr: int, data: bytes) -> None:
+        if addr < 0 or addr + len(data) > self.ram_size:
+            raise ValueError("out of range")
+        self.ram[addr:addr + len(data)] = data
+
+
 class FakeBase:
     """Minimal base syscall model: handles arm, records and stops the rest."""
 
@@ -287,5 +311,167 @@ class StateTests(unittest.TestCase):
             self.assertNotIn(str(key), ("t", "ts", "timestamp", "time"))
 
 
+class SyncPortTests(unittest.TestCase):
+    """Synchronous receive semantics (waitId != 0 must block, not fake OK)."""
+
+    def test_sync_recv_without_completion_blocks(self):
+        model = LibosPortModel(base=FakeBase())
+        machine = FakeMachine()
+        _port_call(machine, recv=(9, 8, 0x100, 0x10), wait=9)
+        handled, result = model.handle(machine, pc=0x4019B00)
+        self.assertFalse(handled)
+        self.assertIsNone(result)
+        self.assertEqual(model.stopped["reason"], "sync-recv-blocks")
+        self.assertEqual(model.waits[-1]["reason"], "sync-recv-blocks")
+        self.assertEqual(model.waits[-1]["shuttle"], 9)
+
+    def test_sync_recv_returns_a_queued_completion(self):
+        model = LibosPortModel(base=FakeBase())
+        machine = FakeMachine()
+        model.host_kick()
+        _port_call(machine, recv=(10, 6, 0, 0))
+        model.handle(machine, pc=0x4018038)
+        machine.set(5, INTERRUPT_ARM_EXTERNAL)
+        machine.set(10, 0xFF)
+        model.handle(machine, pc=0x4018054)
+        _port_call(machine, recv=(9, 8, 0x200, 0x10), wait=9)
+        handled, result = model.handle(machine, pc=0x4019B00)
+        self.assertTrue(handled)
+        self.assertEqual(result, 0)
+        self.assertEqual(machine.x[16], 10)
+        self.assertEqual(model.waits[-1]["completed"], 10)
+
+
+class NetworkModelTests(unittest.TestCase):
+    """The two-task routing model (opt-in, additive)."""
+
+    def _network(self):
+        return LibosPortNetwork(base=FakeBase(), tasks=("init", "rm"),
+                                current="init")
+
+    def test_state_format_is_the_network_schema(self):
+        model = self._network()
+        self.assertEqual(model.state()["format"], NETWORK_REPORT_FORMAT)
+        self.assertEqual(model.state()["tasks"], ["init", "rm"])
+
+    def test_cross_task_send_copies_the_firmware_payload(self):
+        model = self._network()
+        machine = FakeMemoryMachine()
+        model.switch("init", pc=0)
+        _port_call(machine, recv=(9, 8, 0x1000, 0x808))
+        model.handle(machine, pc=0x4019B00)
+        machine.write_mem(0x2000, b"\x11\x22\x33\x44")
+        model.switch("rm", pc=0)
+        _port_call(machine, send=(1, 8, 0x2000, 4), recv=(2, 11, 0x3000, 0x100),
+                   wait=2)
+        handled, _ = model.handle(machine, pc=0x5AD507C)
+        self.assertFalse(handled)  # the sync receive blocks
+        self.assertEqual(model.stopped["reason"], "sync-recv-blocks")
+        completions = model.task_completions["init"]
+        self.assertEqual(len(completions), 1)
+        self.assertEqual(completions[0]["payload_size"], 4)
+        self.assertEqual(completions[0]["origin_task"], 1)
+        self.assertEqual(machine.read_mem(0x1000, 4), b"\x11\x22\x33\x44")
+        self.assertEqual(model.cross_task_payloads,
+                         [{"port": 8, "size": 4, "from": "rm", "to": "init",
+                           "kind": "send"}])
+
+    def test_reply_completes_the_sender_receive_and_resume_is_idempotent(self):
+        model = self._network()
+        machine = FakeMemoryMachine()
+        model.switch("init", pc=0)
+        _port_call(machine, recv=(9, 8, 0x1000, 0x808))
+        model.handle(machine, pc=0x4019B00)
+        machine.write_mem(0x2000, b"\xab\xcd\xef\x01")
+        model.switch("rm", pc=0)
+        _port_call(machine, send=(1, 8, 0x2000, 4), recv=(2, 11, 0x3000, 0x100),
+                   wait=2)
+        model.handle(machine, pc=0x5AD507C)
+        model.stopped = None
+        model.switch("init", pc=0)
+        _port_call(machine, wait=SHUTTLE_ANY)
+        handled, _ = model.handle(machine, pc=0x4019DF4)
+        self.assertTrue(handled)
+        self.assertEqual(machine.x[16], 9)
+        machine.write_mem(0x1000, b"\x7f")
+        _port_call(machine, send=(9, 0, 0x1000, 1), wait=9)
+        handled, _ = model.handle(machine, pc=0x401829C)
+        self.assertTrue(handled)
+        model.stopped = None
+        model.switch("rm", pc=0)
+        _port_call(machine, send=(1, 8, 0x2000, 4), recv=(2, 11, 0x3000, 0x100),
+                   wait=2)
+        handled, _ = model.handle(machine, pc=0x5AD507C)
+        self.assertTrue(handled)
+        self.assertEqual(machine.x[16], 2)
+        self.assertEqual(machine.read_mem(0x3000, 1), b"\x7f")
+        # the resumed receive did not re-post a stale entry
+        self.assertNotIn(2, model.task_recvs["rm"])
+        self.assertEqual(model.cross_task_payloads[-1],
+                         {"port": 0, "size": 1, "from": "init", "to": "rm",
+                          "kind": "reply"})
+
+    def test_task_state_query_serves_the_modelled_trap_state(self):
+        model = self._network()
+        machine = FakeMemoryMachine()
+        model.switch("init", pc=0)
+        model.host_kick()
+        _port_call(machine, recv=(10, 6, 0, 0))
+        model.handle(machine, pc=0x4018038)
+        machine.set(5, INTERRUPT_ARM_EXTERNAL)
+        machine.set(10, 0xFF)
+        model.handle(machine, pc=0x4018054)
+        _port_call(machine, wait=SHUTTLE_ANY)
+        model.handle(machine, pc=0x4019DF4)
+        machine.set(5, 6)
+        machine.set(10, 1)
+        machine.set(11, 0x4000)
+        handled, result = model.handle(machine, pc=0x401A46C)
+        self.assertTrue(handled)
+        self.assertEqual(result, 0)
+        state = machine.read_mem(0x4000, 288)
+        self.assertEqual(int.from_bytes(state[256:264], "little"), SPURIOUS_XEPC)
+        self.assertEqual(int.from_bytes(state[264:272], "little"),
+                         EXTERNAL_INTERRUPT_XCAUSE)
+        self.assertEqual(model.queries[-1]["task_index"], 1)
+
+    def test_memory_query_and_shuttle_reset_are_services(self):
+        model = self._network()
+        machine = FakeMemoryMachine()
+        machine.set(5, 5)
+        machine.set(10, 0xFF)
+        machine.set(11, 0x1000)
+        handled, result = model.handle(machine, pc=0x5AC44FC)
+        self.assertTrue(handled)
+        self.assertEqual(result, 0)
+        self.assertEqual(machine.x[12], 0x1000)
+        self.assertEqual(machine.x[14], 1)
+        machine.set(5, 7)
+        machine.set(10, 1)
+        handled, result = model.handle(machine, pc=0x5AB7D8C)
+        self.assertTrue(handled)
+        self.assertEqual(result, 0)
+        self.assertEqual([s["service"] for s in model.services],
+                         ["memory_query", "shuttle_reset"])
+
+    def test_network_state_is_byte_identical_across_runs(self):
+        def scenario():
+            model = self._network()
+            machine = FakeMemoryMachine()
+            model.switch("init", pc=0)
+            _port_call(machine, recv=(9, 8, 0x1000, 0x808))
+            model.handle(machine, pc=0x4019B00)
+            model.switch("rm", pc=0)
+            _port_call(machine, send=(1, 8, 0x2000, 4),
+                       recv=(2, 11, 0x3000, 0x100), wait=2)
+            model.handle(machine, pc=0x5AD507C)
+            return model.state()
+
+        first = scenario()
+        second = scenario()
+        self.assertEqual(json.dumps(first, sort_keys=True),
+                         json.dumps(second, sort_keys=True))
+
+
 if __name__ == "__main__":
     unittest.main()
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
index 749157e..90643e9 100644
--- a/debug/risc-v-gsp-rm-vm/tests/test_msgq_attach.py
+++ b/debug/risc-v-gsp-rm-vm/tests/test_msgq_attach.py
@@ -43,7 +43,9 @@ from sim.msgq_attach import (  # noqa: E402
     RingGeometry,
     assemble_report,
     classify_ring_access,
+    image_tasks,
     missing_semantic_after_handoff,
+    missing_semantic_after_second_task,
     ring_geometry_from_host,
     run_attach,
 )
@@ -236,6 +238,69 @@ class ReportTests(unittest.TestCase):
                        max_cycles=MAX_RUN_CYCLES + 1)
         with self.assertRaises(MsgqAttachError):
             run_attach(image="/nonexistent/gsp-image.bin", max_cycles=True)
+        with self.assertRaises(MsgqAttachError):
+            run_attach(image="/nonexistent/gsp-image.bin",
+                       second_task_budget=-1)
+        with self.assertRaises(MsgqAttachError):
+            run_attach(image="/nonexistent/gsp-image.bin",
+                       second_task_budget=MAX_RUN_CYCLES + 1)
+
+
+class TaskTableTests(unittest.TestCase):
+    """Data-driven task discovery from the real image section table."""
+
+    @unittest.skipUnless(os.path.isfile(DEFAULT_IMAGE), "real GSP firmware absent")
+    def test_two_tasks_are_discovered_with_entries_and_stacks(self):
+        doc = image_tasks(DEFAULT_IMAGE)
+        self.assertEqual(doc["format"], "greeneye-image-tasks/1")
+        names = [task["name"] for task in doc["tasks"]]
+        self.assertEqual(names, ["init", "rm"])
+        for task in doc["tasks"]:
+            self.assertGreaterEqual(task["entry"], task["text_base"])
+            self.assertLess(task["entry"],
+                            task["text_base"] + task["text_size"])
+            self.assertGreater(task["stack_top"], task["stack_base"])
+            self.assertGreater(task["text_size"], 0)
+        rm = doc["tasks"][1]
+        self.assertGreater(rm["text_size"], doc["tasks"][0]["text_size"])
+
+    def test_absent_image_yields_an_empty_table(self):
+        doc = image_tasks("/nonexistent/image.bin")
+        self.assertEqual(doc, {"format": "greeneye-image-tasks/1", "tasks": []})
+
+    def test_second_task_missing_semantic_is_precise(self):
+        doc = image_tasks(DEFAULT_IMAGE) if os.path.isfile(DEFAULT_IMAGE) \
+            else {"format": "greeneye-image-tasks/1",
+                  "tasks": [{"name": "init"}, {"name": "rm"}]}
+        missing = missing_semantic_after_second_task(
+            tasks_doc=doc,
+            schedule=[{"task": "init", "reason": "port-wait-blocks"},
+                      {"task": "rm", "reason": "scheduled"}],
+            frontier={"task": "rm", "reason": "second-task-slice-budget"},
+            polls=[])
+        self.assertEqual(missing["stage"], "second-task-frontier")
+        self.assertEqual(missing["polls"], 0)
+        self.assertEqual(missing["second_task"], "rm")
+        self.assertGreaterEqual(len(missing["missing"]), 3)
+        self.assertTrue(missing["next_step"])
+
+    def test_report_carries_tasks_and_schedule_additively(self):
+        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},
+            tasks={"format": "greeneye-image-tasks/1",
+                   "tasks": [{"name": "init"}, {"name": "rm"}]},
+            schedule=[{"task": "rm", "reason": "scheduled"}],
+        )
+        self.assertEqual([task["name"] for task in report["tasks"]["tasks"]],
+                         ["init", "rm"])
+        self.assertEqual(report["schedule"][0]["task"], "rm")
+        # the schema tag and the original keys are untouched
+        self.assertEqual(report["format"], REPORT_FORMAT)
+        self.assertIn("first_poll", report)
+        self.assertIn("missing_semantic", report)
 
 
 class PortModelReportTests(unittest.TestCase):
@@ -285,12 +350,12 @@ 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.
+    needs ~474k).  The scheduled second task (the image-discovered RM task)
+    completes its boot port handshake inside the bounded slice and then runs
+    its own start-up past the cap, so the report is expected to take the
+    honest ``wait-reached-no-poll`` frontier path -- this test accepts any of
+    the allowed outcomes and asserts determinism, never success.  A fabricated
+    poll is never accepted.
     """
 
     def test_bounded_run_is_deterministic_and_honest(self):
@@ -306,6 +371,12 @@ class ImageTests(unittest.TestCase):
             self.assertGreater(report["image"]["size"], 0)
             self.assertEqual(len(report["image"]["sha256"]), 64)
             self.assertTrue(report["port_model"]["enabled"])
+            self.assertEqual(report["port_model"]["format"],
+                             "greeneye-msgq-port-network/1")
+            self.assertEqual([task["name"] for task in report["tasks"]["tasks"]],
+                             ["init", "rm"])
+            self.assertTrue(report["schedule"])
+            self.assertIn("rm", [step.get("task") for step in report["schedule"]])
             if not report["boot"]["message_loop_wait_reached"]:
                 self.assertIsNotNone(report["divergence"])
                 self.assertTrue(report["divergence"]["reason"])
@@ -313,17 +384,22 @@ class ImageTests(unittest.TestCase):
                 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.
+            if divergence and divergence.get("kind") == "second-task-frontier":
+                # The doorbell wake was consumed by the init task and the
+                # discovered RM task completed the boot handshake before its
+                # start-up ran past the bounded slice; the exact frontier 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")
+                                 "second-task-frontier")
                 completed = [wait.get("completed")
                              for wait in report["port_model"].get("waits", [])]
                 self.assertIn(10, completed)
+                payloads = report["port_model"].get("cross_task_payloads", [])
+                self.assertTrue(any(item.get("from") == "rm"
+                                    and item.get("to") == "init"
+                                    for item in payloads))
         self.assertEqual(first["boot"]["message_loop_wait_reached"],
                          second["boot"]["message_loop_wait_reached"])
         self.assertEqual(first["boot"]["instructions_executed"],
@@ -334,9 +410,25 @@ class ImageTests(unittest.TestCase):
         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["tasks"], second["tasks"])
+        self.assertEqual(first["schedule"], second["schedule"])
         self.assertEqual(first["determinism"]["report_sha256"],
                          second["determinism"]["report_sha256"])
 
+    def test_single_task_mode_keeps_the_handoff_classification(self):
+        report = run_attach(image=DEFAULT_IMAGE, max_cycles=SMALL_BUDGET,
+                            doorbell=True, second_task=False)
+        self.assertIn(report["outcome"],
+                      ("not-reached", "wait-reached-no-poll",
+                       "wait-reached-poll-classified"))
+        self.assertEqual(report["port_model"]["format"], "greeneye-msgq-port/1")
+        self.assertFalse(report["schedule"])
+        divergence = report["divergence"]
+        if divergence and divergence.get("kind") == "task-state-query":
+            self.assertIsNone(report["first_poll"])
+            self.assertEqual(report["missing_semantic"]["stage"],
+                             "after-extintr-handoff")
+
 
 if __name__ == "__main__":
     unittest.main()
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()