path: root/debug/risc-v-gsp-rm-vm/sim/msgq_attach.py

"""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())