risc-v-gsp-rm-debugger: land docs 45-53 + harness/sim tool set with tests

rm_capture (chunked decode), launch_audit, state_seed/state_decode,
valprobe, rehost, contracts+run_contracts, host_traffic, gmmu_walk,
msgq_attach, libos_port, log_decode, server_snapshot; doc index updates
in the tool overview, operations pages and the repo README.

1 files changed, 1122 insertions, 0 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
new file mode 100644
index 0000000..f1530bd
--- /dev/null
+++ b/debug/risc-v-gsp-rm-vm/sim/msgq_attach.py
@@ -0,0 +1,1122 @@
+"""Message-queue attach: real 570.144 GSP boot to B3 + host ring attach (B3/B4 seam).
+
+Status: **offline development tool (new, additive)**.  This module implements
+steps 1--3 of ``docs/51-HOST-TRAFFIC-LOOP.md`` section 7 ("the next step to
+B5"):
+
+1. boot the real 570.144 image to B3 -- the message-loop port wait, the
+   ``task_port_send_recv_wait`` ecall at ``0x4018038`` -- reusing
+   :mod:`sim.gsp_boot`'s profile, boot shim and libos service model;
+2. attach :class:`sim.host_traffic.DoorbellBlock` and the command/status rings
+   to the machine **before** the boot run, so the firmware's first queue-head
+   access is served by the model and recorded with its PC;
+3. with ``--doorbell``, submit a real-shaped ``GSP_RM_CONTROL`` from the host
+   model, wake the port wait exactly once, continue in bounded instructions,
+   and classify the first ring access the firmware performs (address, width,
+   value read, and whether it looks like a queue head / Tx header / Rx header /
+   entry).  By default the continuation runs under
+   :class:`sim.libos_port.LibosPortModel` (the ``--no-port-model`` flag keeps
+   the first wave's single-wake shim): the model reproduces the observed
+   async receive posts, the boot handshake send and the host wake's
+   external-interrupt handoff, and reports the exact unmodelled semantic that
+   stops the run (``report["missing_semantic"]``) instead of manufacturing a
+   poll.
+
+Schema: ``greeneye-msgq-attach/1`` (:func:`assemble_report`).
+
+Honesty rules (mandatory, docs/51 section 6, docs/06-ACCURACY.md):
+
+* the real image is executed; nothing here fakes a poll, a reply or a
+  completion.  If the firmware does not reach the wait, faults, or never
+  touches the rings, the report says so and records the first divergence --
+  a run that does not reach the loop is never presented as success;
+* the port-wait wake is a documented shim, labelled in
+  ``report["continuation"]["wake_log"]``; the wake result is not firmware
+  behaviour and no RM semantics are claimed.  Under the port model the wake
+  uses the RTOS external-interrupt report contract (error 0, zero payload);
+  without it the first-wave single-wake shim is used;
+* the queue layout and arithmetic are the :mod:`sim.host_traffic` models
+  (real msgq arithmetic), but the queue base addresses are the fixture block;
+  the real addresses live in ``GSP_ARGUMENTS_CACHED``/queue-init arguments,
+  which the simulator does not parse yet;
+* the dispatcher in :mod:`sim.host_traffic` is a stub and is not driven here:
+  the tool records what the *firmware* does, not a stub verdict;
+* L0/L1 only: no timing (L2) and no device-environment (L3) claim; the cycle
+  count is the modeled :mod:`sim.cycles` accounting, not a measurement.
+
+Development-only, offline: no GPU, kernel module, mdev, guest, network, QEMU,
+``/dev`` access or reboot.  Nothing here is a product dependency and nothing
+is injected into a guest.
+"""
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import os
+import sys
+from dataclasses import dataclass
+
+try:  # prefer the harness's established image/profile default (docs/51 seam)
+    from harness.rehost import DEFAULT_IMAGE, DEFAULT_PROFILE
+except Exception:  # pragma: no cover - harness import is not always on path
+    from sim.gsp_boot import DEFAULT_IMAGE, DEFAULT_PROFILE
+
+from greeneye.device_map import firmware_sha256
+from greeneye.elf import load_elf
+from greeneye.sim_backend import launch_native
+from sim.decode import K_LOAD, LOAD_SIZE, decode16, decode32
+from sim.devices import load_profile, registry_from_profile
+from sim.gsp_boot import (
+    MAX_ATTEMPT_INSTRUCTIONS,
+    MILESTONES,
+    STOP_REASONS,
+    _apply_boot_shim,
+    _rung_from_run,
+)
+from sim.host_traffic import (
+    DOORBELL_BASE,
+    DOORBELL_SIZE,
+    GSP_MSG_QUEUE_ELEMENT_HDR_SIZE,
+    GSP_RM_CONTROL,
+    MSGQ_RX_HDR_OFF,
+    MSGQ_TX_HEADER_BYTES,
+    HostTraffic,
+    parse_queue_element,
+)
+from sim.libos import LibosSyscalls, TASK_PORT_SEND_RECV_WAIT, TASK_STATE_QUERY
+from sim.libos_port import LibosPortModel
+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.
+REPORT_FORMAT = "greeneye-msgq-attach/1"
+
+#: ``run_cycles``-style cap: same ceiling as the boot runner's attempts.
+MAX_RUN_CYCLES = MAX_ATTEMPT_INSTRUCTIONS
+#: Default total instruction budget for one attach run.
+DEFAULT_MAX_CYCLES = MAX_ATTEMPT_INSTRUCTIONS
+#: Demo budget: boot-to-B3 needs ~474k instructions; leave a small tail.
+DEMO_MAX_CYCLES = 600_000
+#: Bounded evidence: the first ring access plus up to 16 subsequent ones.
+MAX_POLLS_RECORDED = 17
+#: Internal stop code: the bounded poll list was filled before the run ended.
+R_POLL_LIMIT = 199
+
+#: Doorbell label (block-relative register name; public Turing dev_gsp.h).
+DOORBELL_WINDOW = "PRIV+0x110c00"
+
+#: The real-shaped control used for the host kick (same shape as the docs/51
+#: demo control; real-shaped, from no capture).
+CONTROL_CMD = 0x2080012B
+
+#: ``msgqTxHeader`` field names by byte offset (public queue header).
+_TX_FIELDS = {
+    0: "version",
+    4: "size",
+    8: "msgSize",
+    12: "msgCount",
+    16: "writePtr",
+    20: "flags",
+    24: "rxHdrOff",
+    28: "entryOff",
+}
+
+_LIMITS = [
+    "The real 570.144 image is executed; no poll, reply or completion is "
+    "faked. A run that does not reach the message-loop wait reports its first "
+    "divergence instead of success.",
+    "The port-wait wake is a documented shim (the libos message-port delivery "
+    "path is not modelled); wake events carry an explicit label.",
+    "The port model (sim.libos_port) reproduces the observed receive posts, "
+    "the boot handshake send and the external-interrupt handoff only; it "
+    "fabricates no message payload and schedules no second task. The exact "
+    "unmodelled semantic that stops the run is reported in missing_semantic.",
+    "Queue base addresses are the sim.host_traffic fixture block; the real "
+    "addresses live in GSP_ARGUMENTS_CACHED/queue-init arguments, which are "
+    "not parsed yet.",
+    "No RM semantics: the sim.host_traffic stub dispatcher is not driven by "
+    "this tool, so no reply is claimed.",
+    "L0/L1 only: no timing (L2) and no device-environment (L3) claim; cycle "
+    "numbers are modeled accounting (sim.cycles), not measurements.",
+]
+
+
+class MsgqAttachError(Exception):
+    """An attach input (image, profile or budget) that cannot be run."""
+
+
+def _as_int(value, default=0):
+    if value is None:
+        return default
+    if isinstance(value, str):
+        return int(value, 0)
+    return int(value)
+
+
+def _sha256_text(text: str) -> str:
+    return hashlib.sha256(text.encode("utf-8")).hexdigest()
+
+
+# ---------------------------------------------------------------------------
+# Ring geometry and the pure poll classifier
+# ---------------------------------------------------------------------------
+
+
+@dataclass(frozen=True)
+class RingGeometry:
+    """Physical layout of the attached doorbell and command/status rings.
+
+    All fields are plain integers so the classifier is testable without a
+    machine: ``classify_ring_access`` takes one of these plus an access.
+    """
+
+    shared_base: int
+    cmd_base: int
+    stat_base: int
+    cmd_size: int
+    stat_size: int
+    msg_size: int
+    entry_off: int
+    rx_hdr_off: int
+    doorbell_base: int
+    doorbell_size: int
+
+    def to_dict(self) -> dict:
+        return {
+            "shared_base": self.shared_base,
+            "cmd_base": self.cmd_base,
+            "stat_base": self.stat_base,
+            "cmd_size": self.cmd_size,
+            "stat_size": self.stat_size,
+            "msg_size": self.msg_size,
+            "entry_off": self.entry_off,
+            "rx_hdr_off": self.rx_hdr_off,
+            "doorbell_base": self.doorbell_base,
+            "doorbell_size": self.doorbell_size,
+        }
+
+
+def ring_geometry_from_host(host: HostTraffic) -> RingGeometry:
+    """Read the attached host's queue/doorbell layout into a pure geometry."""
+    command = host.queues.command
+    header = command.header()
+    return RingGeometry(
+        shared_base=int(host.shared_base),
+        cmd_base=int(command.tx_base),
+        stat_base=int(host.queues.status.tx_base),
+        cmd_size=int(command.size),
+        stat_size=int(host.queues.status.size),
+        msg_size=int(header.msg_size),
+        entry_off=int(header.entry_off),
+        rx_hdr_off=int(MSGQ_RX_HDR_OFF),
+        doorbell_base=int(DOORBELL_BASE),
+        doorbell_size=int(DOORBELL_SIZE),
+    )
+
+
+def classify_ring_access(addr: int, width, value, geometry: RingGeometry) -> dict:
+    """Classify one firmware access against the ring geometry (pure).
+
+    Returns ``{addr, width, value, in_ring, kind, ...}`` where ``kind`` is one
+    of ``queue-head`` (the doorbell register), ``tx-header``, ``rx-header``,
+    ``entry``, ``ring-gap`` or ``other``.  The Rx-header note carries the real
+    ``MSGQ_FLAGS_SWAP_RX`` cross-link: the header physically in the command
+    buffer at ``+rx_hdr_off`` is the *status* queue's read pointer and vice
+    versa.
+    """
+    addr = int(addr)
+    entry = {
+        "addr": addr,
+        "width": width,
+        "value": value,
+        "in_ring": False,
+        "kind": "other",
+    }
+    if geometry.doorbell_base <= addr < geometry.doorbell_base + geometry.doorbell_size:
+        offset = addr - geometry.doorbell_base
+        queue = offset // 8 if offset % 8 == 0 else None
+        entry.update({
+            "in_ring": True,
+            "kind": "queue-head",
+            "window": DOORBELL_WINDOW,
+            "offset": offset,
+        })
+        if queue is not None and 0 <= queue < geometry.doorbell_size // 8:
+            entry["queue"] = queue
+            entry["register"] = f"NV_PGSP_QUEUE_HEAD({queue})"
+        return entry
+    for buffer_name, base, size in (
+        ("command", geometry.cmd_base, geometry.cmd_size),
+        ("status", geometry.stat_base, geometry.stat_size),
+    ):
+        if not base <= addr < base + size:
+            continue
+        offset = addr - base
+        entry.update({"in_ring": True, "buffer": buffer_name})
+        if offset < MSGQ_TX_HEADER_BYTES:
+            entry.update({
+                "kind": "tx-header",
+                "field_offset": offset,
+                "field": _TX_FIELDS.get(offset, "reserved"),
+            })
+        elif offset == geometry.rx_hdr_off:
+            owner = "status" if buffer_name == "command" else "command"
+            entry.update({
+                "kind": "rx-header",
+                "cross_linked_owner": owner,
+                "semantics": "readPtr",
+            })
+        elif offset >= geometry.entry_off:
+            relative = offset - geometry.entry_off
+            slot = relative % geometry.msg_size
+            entry.update({
+                "kind": "entry",
+                "entry_index": relative // geometry.msg_size,
+                "slot_offset": slot,
+                "region": "entry-header" if slot < GSP_MSG_QUEUE_ELEMENT_HDR_SIZE
+                          else "entry-payload",
+            })
+        else:
+            entry["kind"] = "ring-gap"
+        return entry
+    return entry
+
+
+# ---------------------------------------------------------------------------
+# Pure report assembly (hermetic-testable)
+# ---------------------------------------------------------------------------
+
+
+def _report_sha(report: dict) -> str:
+    clone = json.loads(json.dumps(report, sort_keys=True, default=str))
+    clone.get("determinism", {}).pop("report_sha256", None)
+    return _sha256_text(json.dumps(clone, sort_keys=True))
+
+
+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,
+                    limits=None, notes=None) -> dict:
+    """Assemble the ``greeneye-msgq-attach/1`` report from plain inputs.
+
+    Pure: no machine, no boot, no I/O.  ``polls`` is the ordered list of
+    classified ring accesses (first = index 0); the report keeps the first and
+    at most 16 subsequent ones.  ``boot["message_loop_wait_reached"]`` and
+    ``boot.get("skipped")`` select the outcome; a not-reached run keeps its
+    non-empty divergence reason, never a manufactured success.
+
+    ``port_model`` is the :class:`sim.libos_port.LibosPortModel` state when the
+    receive path was modelled (``{"enabled": False}`` otherwise), and
+    ``missing_semantic`` is the exact unmodelled semantic that stopped a
+    port-model continuation (``None`` when there is none).  Both are additive
+    fields, present on every report.
+    """
+    ordered = [dict(poll) for poll in polls]
+    first = ordered[0] if ordered else None
+    subsequent = ordered[1:1 + (MAX_POLLS_RECORDED - 1)]
+    if boot.get("skipped"):
+        outcome = "skipped"
+    elif not boot.get("message_loop_wait_reached"):
+        outcome = "not-reached"
+    elif first is not None:
+        outcome = "wait-reached-poll-classified"
+    else:
+        outcome = "wait-reached-no-poll"
+    ok = outcome in ("wait-reached-poll-classified", "wait-reached-no-poll")
+    report = {
+        "format": REPORT_FORMAT,
+        "ok": ok,
+        "outcome": outcome,
+        "image": dict(image),
+        "profile": dict(profile),
+        "boot": dict(boot),
+        "attach": dict(attach),
+        "doorbell": dict(doorbell),
+        "continuation": dict(continuation or {}),
+        "first_poll": first,
+        "subsequent_polls": subsequent,
+        "poll_count": len(ordered),
+        "divergence": divergence,
+        "missing_mmio": dict(missing_mmio),
+        "port_model": dict(port_model) if port_model is not None
+                      else {"enabled": False},
+        "missing_semantic": dict(missing_semantic) if missing_semantic is not None
+                            else None,
+        "determinism": dict(determinism),
+        "limits": list(limits if limits is not None else _LIMITS),
+        "notes": list(notes or []),
+    }
+    report["determinism"]["report_sha256"] = _report_sha(report)
+    return report
+
+
+def _skipped_report(image: str, profile_path: str, reason: str,
+                    *, doorbell: bool) -> dict:
+    return assemble_report(
+        image={"path": os.path.abspath(image), "sha256": None, "size": None},
+        profile={"name": os.path.basename(profile_path),
+                 "path": os.path.abspath(profile_path)},
+        boot={"ran": False, "skipped": True, "reason": reason,
+              "message_loop_wait_reached": False, "rung": None,
+              "instructions_executed": 0},
+        attach={},
+        doorbell={"requested": bool(doorbell)},
+        polls=[],
+        divergence=None,
+        missing_mmio={"count": 0, "first": None},
+        determinism={},
+    )
+
+
+# ---------------------------------------------------------------------------
+# Execution helpers (machine stepping stays in sim.machine)
+# ---------------------------------------------------------------------------
+
+
+class _PortWaitWake:
+    """Host-traffic-gated wake of ``task_port_send_recv_wait`` (documented shim).
+
+    The real libos kernel unblocks the task when a message is delivered through
+    the port machinery; the simulator does not model that path, so the only
+    modelled wake is explicit: after the host has rung the doorbell, the first
+    wait returns LIBOS_OK (0) exactly once and the event is recorded.  A
+    re-entry is recorded and stops the run -- the wait is never silently
+    repeated as if progress had been made.
+    """
+
+    def __init__(self, base_handler, *, max_wakes: int = 1):
+        self.base_handler = base_handler
+        self.max_wakes = int(max_wakes)
+        self.wakes = 0
+        self.wake_log: list[dict] = []
+        self.reentries: list[dict] = []
+
+    def handle(self, machine, pc: int = 0):
+        number = int(machine.x[5])
+        if number == TASK_PORT_SEND_RECV_WAIT:
+            record = {
+                "num": number,
+                "name": "task_port_send_recv_wait",
+                "pc": int(pc),
+                "a4": int(machine.x[14]),
+                "a5": int(machine.x[15]),
+                "a6": int(machine.x[16]),
+                "a7": int(machine.x[17]),
+            }
+            if self.wakes < self.max_wakes:
+                self.wakes += 1
+                record["result"] = "shim:port_wait_wake"
+                record["returned"] = 0
+                self.wake_log.append(record)
+                return True, 0
+            record["result"] = "stop:reentry"
+            self.reentries.append(record)
+            return False, None
+        return self.base_handler.handle(machine, pc)
+
+    def state(self) -> dict:
+        return {
+            "wakes": self.wakes,
+            "wake_log": [dict(entry) for entry in self.wake_log],
+            "reentries": [dict(entry) for entry in self.reentries],
+        }
+
+
+def _watchers(geometry: RingGeometry):
+    """Read-watchpoints over the doorbell and both ring buffers (mask = loads)."""
+    return (
+        (geometry.doorbell_base, geometry.doorbell_base + geometry.doorbell_size,
+         "doorbell", 2),
+        (geometry.cmd_base, geometry.cmd_base + geometry.cmd_size, "rings", 2),
+        (geometry.stat_base, geometry.stat_base + geometry.stat_size, "rings", 2),
+    )
+
+
+def _decoded_load(machine, pc: int):
+    """``(size, rd)`` of the cached load at ``pc``, or ``(None, None)``."""
+    decoded = getattr(machine, "_cache", {}).get(pc)
+    if decoded is None:
+        try:
+            fetch = pc
+            if machine.mpu.active:
+                translated = machine.mpu.translate(pc)
+                if translated is not None:
+                    fetch = translated
+            word = int.from_bytes(machine.read_mem(fetch, 4), "little")
+            if word & 3 == 3:
+                decoded = decode32(pc, word)
+            else:
+                decoded = decode16(pc, word & 0xFFFF)
+        except Exception:
+            return None, None
+    if decoded[0] != K_LOAD:
+        return None, None
+    return int(LOAD_SIZE[decoded[1]]), int(decoded[2])
+
+
+def _capture_poll(machine, geometry: RingGeometry, pc: int, addr: int) -> dict:
+    """Record and step over the watched load (a watch fires before execution).
+
+    The load's width/rd come from the decoded instruction; the value is read
+    from the destination register after the single step, so the recorded value
+    is the value the firmware actually received.
+    """
+    size, rd = _decoded_load(machine, pc)
+    step_reason, _spc, fault_cause, fault_tval = machine.step()
+    value = None
+    if rd is not None and step_reason != R_FAULT and rd != 0:
+        value = machine.x[rd]
+    poll = classify_ring_access(addr, size, value, geometry)
+    poll["pc"] = int(pc)
+    poll["executed"] = step_reason == R_STEP
+    if step_reason == R_FAULT:
+        poll["step_fault"] = {"cause": int(fault_cause), "tval": int(fault_tval)}
+    return poll
+
+
+def _run_watching(machine, budget: int, geometry: RingGeometry,
+                  milestones_hit: list, polls: list,
+                  *, max_polls: int = MAX_POLLS_RECORDED):
+    """Run to ``budget`` (absolute instructions), recording milestone hits and
+    ring reads; same loop shape as ``sim.gsp_boot._run_with_milestones`` plus
+    watchpoint step-over.  Returns ``(reason, pc, a, b)``.
+    """
+    pending = {addr for name, addr in MILESTONES if name not in milestones_hit}
+    while True:
+        remaining = budget - machine.insns
+        if remaining <= 0:
+            return R_BUDGET, machine.pc, 0, 0
+        watchers = _watchers(geometry) if len(polls) < max_polls else ()
+        reason, pc, a, b = machine.run(remaining, breakpoints=pending,
+                                       watchers=watchers)
+        if reason == R_BP:
+            for name, addr in MILESTONES:
+                if addr == pc and name not in milestones_hit:
+                    milestones_hit.append(name)
+                    break
+            pending = {addr for name, addr in MILESTONES
+                       if name not in milestones_hit}
+            continue
+        if reason == R_WATCH:
+            polls.append(_capture_poll(machine, geometry, pc, a))
+            if len(polls) >= max_polls:
+                return R_POLL_LIMIT, machine.pc, a, b
+            continue
+        return reason, pc, a, b
+
+
+def _control_params() -> bytes:
+    """A real-shaped 512-byte ``GSP_RM_CONTROL`` parameter block (docs/51)."""
+    return (
+        (0xC2000006).to_bytes(4, "little")
+        + (0xABCD0080).to_bytes(4, "little")
+        + CONTROL_CMD.to_bytes(4, "little")
+        + bytes(500)
+    )
+
+
+def _verify_command_element(host: HostTraffic, submit: dict) -> dict:
+    """Parse the just-submitted command element back and verify its checksum."""
+    try:
+        raw = host.queues.command.read_entry(submit["index"],
+                                             submit["element_bytes"])
+        prefix, message = parse_queue_element(raw)
+        return {
+            "checksum_ok": True,
+            "elem_count": int(prefix["elem_count"]),
+            "function": int(message.function),
+            "sequence": int(message.sequence),
+            "length": int(message.length),
+        }
+    except Exception as exc:  # pragma: no cover - defensive evidence
+        return {"checksum_ok": False, "error": str(exc)}
+
+
+def missing_semantic_after_handoff(*, stopped: dict, polls: list) -> dict:
+    """The exact unmodelled semantics that stopped the port-model continuation.
+
+    Recorded when, after the recorded host wake, the message-loop task
+    completed the external-interrupt receive, dispatched it, and issued
+    ``TASK_STATE_QUERY`` (the libos trap/report path) without touching the
+    command/status rings.  Pure: takes the stop record and the classified poll
+    list; no machine, no I/O.
+    """
+    return {
+        "stage": "after-extintr-handoff",
+        "stop": dict(stopped),
+        "observed": [
+            "host doorbell write NV_PGSP_QUEUE_HEAD(0) recorded",
+            "external-interrupt receive retired with the RTOS report contract "
+            "(error 0, payload size 0) on the recorded wake",
+            "completed shuttle dispatched; next action is a task-state query "
+            "for the reported remote task",
+            "no command/status ring or doorbell read occurred before the stop",
+        ],
+        "missing": [
+            "doorbell-write -> falcon interrupt bit mapping: the host submit "
+            "path writes only NV_PGSP_QUEUE_HEAD; no IRQ register write is "
+            "observed and the firmware armed mask 0xFF",
+            "libos interrupt-trap delivery and the interrupted task's saved "
+            "xcause/xepc consumed by TASK_STATE_QUERY",
+            "the command-queue consumer (msgq receive) task and the "
+            "GSP_ARGUMENTS_CACHED/queue-init address handoff; this single-task "
+            "model schedules no other task",
+        ],
+        "next_step": (
+            "model TASK_STATE_QUERY's trap state or establish the queue-service "
+            "task's wake path, then re-run this attach and classify the first "
+            "ring poll; do not infer a poll from this run"),
+        "polls": len(polls),
+    }
+
+
+# ---------------------------------------------------------------------------
+# The attach run
+# ---------------------------------------------------------------------------
+
+
+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,
+               record_dir: str | None = None) -> dict:
+    """Boot to the port wait, attach the rings, optionally kick and continue.
+
+    Bounded and deterministic for a fixed image/profile/budget.  Returns the
+    assembled report; on an absent image/profile it returns a clean
+    ``outcome: "skipped"`` report instead of raising.  Never fakes a poll: the
+    outcome is the observed one, with the first divergence recorded when the
+    wait is not reached.
+
+    ``port_model`` (default on) installs
+    :class:`sim.libos_port.LibosPortModel` for the continuation: the async
+    receive posts, the boot handshake send and the host wake's
+    external-interrupt handoff are modelled, so the run advances to the
+    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.
+    """
+    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}")
+    image = os.path.abspath(image)
+    profile_path = os.path.abspath(profile_path)
+    if not os.path.isfile(image):
+        return _skipped_report(image, profile_path, f"image absent: {image}",
+                               doorbell=doorbell)
+    if not os.path.isfile(profile_path):
+        return _skipped_report(image, profile_path,
+                               f"profile absent: {profile_path}",
+                               doorbell=doorbell)
+
+    sha = firmware_sha256(image)
+    size = os.path.getsize(image)
+    profile = load_profile(profile_path)
+    load = profile.get("load", {})
+    ram_base = _as_int(load.get("ram_base", 0))
+    memory_mb = _as_int(load.get("memory_mb", 12288))
+
+    session = None
+    notes: list[str] = []
+    divergence = None
+    polls: list[dict] = []
+    continuation: dict = {"ran": False}
+    port_model_state: dict | None = None
+    missing_semantic: dict | None = None
+    try:
+        session = launch_native(
+            kernel=image,
+            memory_mb=memory_mb,
+            ram_base=ram_base,
+            record=record_dir if record_dir else False,
+            name="msgq-attach",
+            device_map=None,
+            unknown_mmio="lenient",
+            bias=0,
+        )
+        machine = session.target.machine
+        machine.devices = registry_from_profile(profile, machine=machine)
+        machine.translator = getattr(machine.devices, "dma", None)
+        libos = LibosSyscalls()
+        machine.syscall_handler = libos.handle
+        shim_meta = {"enabled": True, "assumptions": []}
+        shim = dict(profile.get("boot_shim", {}))
+        shim["_image_path"] = image
+        _apply_boot_shim(machine, shim, load_elf(image), shim_meta)
+
+        host = HostTraffic(machine=machine, recorder=session.recorder)
+        machine.devices.add(host.doorbell_block())
+        geometry = ring_geometry_from_host(host)
+
+        submit = None
+        element_check = None
+        if doorbell:
+            submit = host.submit_rpc(GSP_RM_CONTROL, _control_params(),
+                                     sequence=1)
+            element_check = _verify_command_element(host, submit)
+
+        milestones_hit: list[str] = []
+        reason, pc, a, b = _run_watching(machine, max_cycles, geometry,
+                                         milestones_hit, polls)
+        stopped = dict(libos.stopped) if libos.stopped else None
+        wait_reached = bool(stopped) and stopped.get("num") == TASK_PORT_SEND_RECV_WAIT
+        stop_reason = STOP_REASONS.get(reason, str(reason))
+        if reason == R_POLL_LIMIT:
+            stop_reason = "poll-limit"
+        fault = {"cause": int(a), "tval": int(b), "pc": int(pc)} \
+            if reason == R_FAULT else None
+        rung = _rung_from_run(machine, shim_meta, milestones_hit, libos)
+
+        if not wait_reached:
+            if reason == R_FAULT:
+                divergence = {
+                    "kind": "fault",
+                    "reason": (f"fault before the message-loop port wait: "
+                               f"cause={a:#x} tval={b:#x} pc={pc:#x}"),
+                }
+            elif reason == R_POLL_LIMIT:
+                divergence = {
+                    "kind": "poll-limit",
+                    "reason": (f"{MAX_POLLS_RECORDED} ring accesses observed "
+                               f"before the port wait; run stopped by the "
+                               f"bounded poll limit"),
+                }
+            elif reason == R_ECALL:
+                name = (stopped or {}).get("name", "unhandled ecall")
+                number = (stopped or {}).get("num")
+                divergence = {
+                    "kind": "ecall",
+                    "reason": (f"unhandled libos syscall before the "
+                               f"message-loop port wait: {name} "
+                               f"(num={number}) at pc={pc:#x}"),
+                }
+            else:
+                divergence = {
+                    "kind": "budget",
+                    "reason": (f"message-loop port wait not reached within "
+                               f"{max_cycles} instructions (rung {rung}, "
+                               f"stop={stop_reason} at pc={pc:#x})"),
+                }
+
+        if 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
+            model.host_kick(
+                queue=int(last_kick["queue"]) if last_kick else None,
+                value=int(last_kick["value"]) if last_kick else None,
+            )
+            reason2, pc2, a2, b2 = _run_watching(machine, max_cycles, geometry,
+                                                 milestones_hit, polls)
+            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,
+                "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),
+            }
+            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 number2 == TASK_STATE_QUERY:
+                    missing_semantic = missing_semantic_after_handoff(
+                        stopped=port_stopped or {"num": number2, "pc": int(pc2)},
+                        polls=polls)
+                    divergence = {
+                        "kind": "task-state-query",
+                        "reason": (
+                            "after the recorded host wake the external "
+                            "interrupt receive completed (error 0, size 0) and "
+                            "the message-loop task dispatched it; its next "
+                            "syscall is task_state_query "
+                            f"(num={number2}) at pc={pc2:#x}, the libos "
+                            "trap/report path -- no command/status ring or "
+                            "doorbell read occurred, so no poll is reported"),
+                        "stop": dict(port_stopped or {}),
+                    }
+                elif number2 == TASK_PORT_SEND_RECV_WAIT:
+                    divergence = {
+                        "kind": "port-wait-reentry",
+                        "reason": (
+                            "the message-loop task re-entered "
+                            "task_port_send_recv_wait after the recorded host "
+                            "wake; nothing completed the wait, so the firmware "
+                            "produced no queue poll"),
+                        "wait_pc": int(pc2),
+                    }
+                else:
+                    divergence = {
+                        "kind": "ecall",
+                        "reason": (f"unhandled libos syscall after the host "
+                                   f"wake: {name2} (num={number2}) at "
+                                   f"pc={pc2:#x}"),
+                    }
+            elif reason2 == R_BUDGET and not polls:
+                divergence = {
+                    "kind": "budget",
+                    "reason": (f"no ring access observed within the total "
+                               f"budget of {max_cycles} instructions"),
+                }
+            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:
+            wake = _PortWaitWake(libos, max_wakes=1)
+            machine.syscall_handler = wake.handle
+            reason2, pc2, a2, b2 = _run_watching(machine, max_cycles, geometry,
+                                                 milestones_hit, polls)
+            stop_reason2 = STOP_REASONS.get(reason2, str(reason2))
+            if reason2 == R_POLL_LIMIT:
+                stop_reason2 = "poll-limit"
+            continuation = {
+                "ran": True,
+                "port_model": False,
+                "stop_reason": stop_reason2,
+                "stop_pc": int(pc2),
+                "instructions_executed": int(machine.insns),
+                "wakes": int(wake.wakes),
+                "wake_log": [dict(entry) for entry in wake.wake_log],
+                "reentries": [dict(entry) for entry in wake.reentries],
+                "polls": len(polls),
+            }
+            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:
+                number2 = int(machine.x[5])
+                if number2 == TASK_PORT_SEND_RECV_WAIT:
+                    divergence = {
+                        "kind": "port-wait-reentry",
+                        "reason": (
+                            "the message-loop task re-entered "
+                            "task_port_send_recv_wait after the recorded host "
+                            "wake; libos message-port delivery is not modelled, "
+                            "so the firmware produced no queue poll"),
+                        "wait_pc": int(pc2),
+                    }
+                else:
+                    divergence = {
+                        "kind": "ecall",
+                        "reason": (f"unhandled libos syscall after the host "
+                                   f"wake: num={number2} at pc={pc2:#x}"),
+                    }
+            elif reason2 == R_BUDGET and not polls:
+                divergence = {
+                    "kind": "budget",
+                    "reason": (f"no ring access observed within the total "
+                               f"budget of {max_cycles} instructions"),
+                }
+            elif reason2 == R_POLL_LIMIT:
+                notes.append(
+                    f"continuation stopped after {MAX_POLLS_RECORDED} recorded "
+                    "ring accesses (bounded by design)")
+        elif wait_reached and not doorbell:
+            notes.append(
+                "continuation not requested (--doorbell off); the port wait was "
+                "reached but no host kick was sent and no poll was classified")
+
+        ring_reads = [
+            access for access in machine.devices.accesses
+            if access["rw"] == "r"
+            and geometry.doorbell_base <= access["addr"]
+            < geometry.doorbell_base + geometry.doorbell_size
+        ]
+        ring_writes = [
+            access for access in machine.devices.accesses
+            if access["rw"] == "w"
+            and geometry.doorbell_base <= access["addr"]
+            < geometry.doorbell_base + geometry.doorbell_size
+        ]
+        doorbell_write = None
+        if submit is not None and host.doorbell.writes:
+            last = host.doorbell.writes[-1]
+            doorbell_write = {
+                "window": DOORBELL_WINDOW,
+                "base": int(DOORBELL_BASE),
+                "offset": int(last["queue"]) * 8,
+                "queue": int(last["queue"]),
+                "value": int(last["value"]),
+            }
+        doorbell_report = {
+            "requested": bool(doorbell),
+            "write": doorbell_write,
+            "host_kicks": int(host.doorbell.kicks),
+            "head_read_back": int(host.doorbell.head(0)),
+            "guest_reads": len(ring_reads),
+            "guest_writes": len(ring_writes),
+        }
+
+        attach = dict(geometry.to_dict())
+        attach["command_queue"] = host.queues.command.state()
+        attach["status_queue"] = host.queues.status.state()
+        attach["command_element"] = element_check
+
+        device_state = json.dumps(machine.devices.state(), sort_keys=True,
+                                  default=str)
+        queue_state = json.dumps(host.queues.state(), sort_keys=True)
+        doorbell_state = json.dumps(host.doorbell.state(), sort_keys=True)
+        determinism = {
+            "instructions": int(machine.insns),
+            "cycles_modeled": int(machine.cycles),
+            "cycle_model": machine.cycle_model.name
+                           if getattr(machine, "cycle_model", None) else None,
+            "max_cycles": int(max_cycles),
+            "ram_base": int(machine.ram_base),
+            "ram_size": int(machine.ram_size),
+            "device_state_sha256": _sha256_text(device_state),
+            "access_log_len": len(machine.devices.accesses),
+            "queue_state_sha256": _sha256_text(queue_state),
+            "doorbell_state_sha256": _sha256_text(doorbell_state),
+        }
+
+        boot = {
+            "ran": True,
+            "skipped": False,
+            "rung": rung,
+            "message_loop_wait_reached": wait_reached,
+            "instructions_executed": int(machine.insns),
+            "cycles_modeled": int(machine.cycles),
+            "cycle_model": determinism["cycle_model"],
+            "instructions_budget": int(max_cycles),
+            "stop": {"reason": stop_reason, "pc": int(pc), "fault": fault},
+            "wait": stopped if wait_reached else None,
+            "libos_stopped": stopped,
+            "milestones": list(milestones_hit),
+            "mmio_unknown": int(machine.mmio_unknown),
+            "device_unhandled": int(machine.devices.unhandled),
+        }
+        missing_mmio = {
+            "count": int(machine.mmio_unknown),
+            "first": dict(machine.mmio_log[0]) if machine.mmio_log else None,
+        }
+        report = assemble_report(
+            image={"path": image, "sha256": sha, "size": size},
+            profile={"name": profile.get("name"), "path": profile_path},
+            boot=boot,
+            attach=attach,
+            doorbell=doorbell_report,
+            polls=polls,
+            divergence=divergence,
+            missing_mmio=missing_mmio,
+            determinism=determinism,
+            continuation=continuation,
+            port_model=port_model_state,
+            missing_semantic=missing_semantic,
+            notes=notes,
+        )
+        if getattr(session.recorder, "dir", None) is not None:
+            session.recorder.meta.update({
+                "kind": "msgq-attach",
+                "outcome": report["outcome"],
+                "rung": rung,
+                "instructions_executed": int(machine.insns),
+                "message_loop_wait_reached": wait_reached,
+                "port_model": bool(report["port_model"].get("enabled")),
+                "missing_semantic": bool(report["missing_semantic"]),
+                "first_poll": None if report["first_poll"] is None
+                              else report["first_poll"]["kind"],
+            })
+            session.recorder.blob(
+                "msgq-attach.json",
+                json.dumps(report, indent=1, sort_keys=True).encode(),
+            )
+            session.note(
+                "msgq_attach",
+                outcome=report["outcome"],
+                rung=rung,
+                instructions=machine.insns,
+                wait_reached=wait_reached,
+                polls=report["poll_count"],
+            )
+        if json_out:
+            parent = os.path.dirname(os.path.abspath(json_out))
+            os.makedirs(parent, exist_ok=True)
+            with open(json_out, "w", encoding="utf-8") as handle:
+                json.dump(report, handle, indent=1, sort_keys=True)
+                handle.write("\n")
+    finally:
+        if session is not None:
+            session.close()
+    return report
+
+
+# ---------------------------------------------------------------------------
+# Demo and CLI
+# ---------------------------------------------------------------------------
+
+
+def run_demo(*, image: str = DEFAULT_IMAGE, profile_path: str = DEFAULT_PROFILE,
+             print_json: bool = False, port_model: bool = True) -> dict:
+    """Bounded attach demo; cleanly skips (exit 0) when the image is absent."""
+    report = run_attach(image=image, profile_path=profile_path,
+                        max_cycles=DEMO_MAX_CYCLES, doorbell=True,
+                        port_model=port_model)
+    if report["outcome"] == "skipped":
+        print(f"[msgq-attach] skipped: {report['boot'].get('reason')}")
+        return report
+    if print_json:
+        print(json.dumps(report, indent=1, sort_keys=True))
+    else:
+        _print_report(report)
+    return report
+
+
+def _print_report(report: dict) -> None:
+    image = report["image"]
+    boot = report["boot"]
+    print(f"format: {report['format']}")
+    print(f"image: {image['path']}")
+    print(f"sha256: {image['sha256']}  size: {image['size']}")
+    print(f"outcome: {report['outcome']} (ok={report['ok']})")
+    print(f"rung: {boot.get('rung')}  instructions: {boot.get('instructions_executed')}"
+          f"  cycles (modeled): {boot.get('cycles_modeled')}")
+    wait = boot.get("wait")
+    if wait:
+        print(f"message-loop wait: reached ({wait.get('name')} @ "
+              f"{wait.get('pc'):#x})")
+    else:
+        print("message-loop wait: not reached")
+    write = report["doorbell"].get("write")
+    if write:
+        print(f"doorbell: host write {write['window']} offset={write['offset']:#x} "
+              f"value={write['value']:#x} (guest reads={report['doorbell']['guest_reads']})")
+    model = report.get("port_model") or {}
+    if model.get("enabled"):
+        posts = model.get("posts", [])
+        roles = ", ".join(f"port {post['port']}:{post['role']}"
+                          for post in posts)
+        print(f"port model: enabled  posts=[{roles}]  "
+              f"host kicks={model.get('host_kicks', 0)}")
+        for wait_record in model.get("waits", []):
+            if wait_record.get("completed") is not None:
+                print(f"  wait @ {wait_record['pc']:#x}: completed shuttle "
+                      f"{wait_record['completed']} ({wait_record['role']}) "
+                      f"size={wait_record['payload_size']} "
+                      f"contract={wait_record['contract']}")
+    else:
+        print("port model: disabled")
+    first = report["first_poll"]
+    if first:
+        print(f"first poll: addr={first['addr']:#x} width={first['width']} "
+              f"value={first['value']} kind={first['kind']}")
+    else:
+        print("first poll: none observed")
+    print(f"polls recorded: {report['poll_count']}"
+          f" (subsequent listed: {len(report['subsequent_polls'])})")
+    if report["divergence"]:
+        print(f"divergence: {report['divergence']['reason']}")
+    missing = report.get("missing_semantic")
+    if missing:
+        print(f"missing semantic ({missing.get('stage')}):")
+        for item in missing.get("missing", []):
+            print(f"  - {item}")
+    print("note: development-only, offline; L0/L1 evidence, no timing claim")
+
+
+def build_parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(
+        prog="sim.msgq_attach",
+        description=(
+            "Boot the real 570.144 GSP image to the message-loop port wait "
+            "(B3), attach the host doorbell/msgq rings before the run, "
+            "optionally kick, continue bounded and classify the first ring "
+            "poll. Offline only; no hardware, module, guest or network."),
+    )
+    sub = parser.add_subparsers(dest="command")
+    run_cmd = sub.add_parser(
+        "run", help="bounded boot + attach + optional kick + poll classify")
+    run_cmd.add_argument("--image", default=DEFAULT_IMAGE,
+                         help="GSP firmware image (default: the 570.144 image)")
+    run_cmd.add_argument("--profile", default=DEFAULT_PROFILE,
+                         help="greeneye-sim-profile/1 document")
+    run_cmd.add_argument("--max-cycles", type=int, default=DEFAULT_MAX_CYCLES,
+                         help=f"total instruction budget (1..{MAX_RUN_CYCLES})")
+    run_cmd.add_argument("--doorbell", action="store_true",
+                         help="submit a host command and wake the port wait once")
+    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("--json", metavar="OUT", default=None,
+                         help="write the full report JSON to OUT")
+    run_cmd.add_argument("--record", metavar="DIR", default=None,
+                         help="write a greeneye-session/1 directory to DIR")
+    demo_cmd = sub.add_parser(
+        "demo", help="bounded canned attach run (skips cleanly if image absent)")
+    demo_cmd.add_argument("--image", default=DEFAULT_IMAGE,
+                          help="GSP firmware image (default: the 570.144 image)")
+    demo_cmd.add_argument("--profile", default=DEFAULT_PROFILE,
+                          help="greeneye-sim-profile/1 document")
+    demo_cmd.add_argument("--no-port-model", action="store_true",
+                          help="disable the sim.libos_port receive-path model")
+    demo_cmd.add_argument("--json", action="store_true",
+                          help="print the full report JSON")
+    return parser
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = build_parser()
+    args = parser.parse_args(argv)
+    if args.command is None:
+        parser.print_help(sys.stderr)
+        return 2
+    port_model = not getattr(args, "no_port_model", False)
+    if args.command == "demo":
+        run_demo(image=args.image, profile_path=args.profile,
+                 print_json=args.json, port_model=port_model)
+        return 0
+    try:
+        report = run_attach(
+            image=args.image,
+            profile_path=args.profile,
+            max_cycles=args.max_cycles,
+            doorbell=args.doorbell,
+            port_model=port_model,
+            json_out=args.json,
+            record_dir=args.record,
+        )
+    except MsgqAttachError as exc:
+        print(f"[msgq-attach] error: {exc}", file=sys.stderr)
+        return 1
+    if report["outcome"] == "skipped":
+        print(f"[msgq-attach] skipped: {report['boot'].get('reason')}",
+              file=sys.stderr)
+        return 2
+    _print_report(report)
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())