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

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

1 files changed, 351 insertions, 3 deletions

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