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, 1437 insertions, 0 deletions

diff --git a/debug/risc-v-gsp-rm-vm/sim/host_traffic.py b/debug/risc-v-gsp-rm-vm/sim/host_traffic.py
new file mode 100644
index 0000000..a9dd55d
--- /dev/null
+++ b/debug/risc-v-gsp-rm-vm/sim/host_traffic.py
@@ -0,0 +1,1437 @@
+"""Host-traffic message loop: doorbell + GSP msgq transport + replay (B4/B5).
+
+Status: **stub wave (in development)**.  This module lands the transport half
+of the B4/B5 rung (docs/35-GREENEYE-SIMULATOR-DESIGN.md §3.3, docs/45,
+handoff-simulator.md §31 item B1): a modelled host doorbell, a command/status
+message-queue pair laid out with the real queue arithmetic, queue-element
+framing with the real checksum, a small stub dispatcher, and deterministic
+replay of an ``rm_capture`` stimulus.  The real 570.144 firmware is **not**
+run by this module: no real message loop, no RM semantics, no firmware claim.
+
+Real structures mirrored here (provenance: public OpenRM headers and the
+header-derived layouts already used by the project's boot code; every constant
+is checked against ``harness/rm_capture.py`` where the two overlap):
+
+* ``NV_PGSP_QUEUE_HEAD(i) = 0x110c00 + i * 8`` (public Turing ``dev_gsp.h``),
+  32-bit RW; the 570.144 host writes it after submitting a command
+  (``kgspSetCmdQueueHead``), and the command data itself lives in the sysmem
+  message queues.  The simulator reaches the register at ``PRIV+0x110c00``.
+* ``msgqTxHeader`` (8 x u32: version/size/msgSize/msgCount/writePtr/flags/
+  rxHdrOff/entryOff) and ``msgqRxHeader`` (readPtr), with
+  ``rxHdrOff = align32(16) = 32`` and ``entryOff = align36(16) = 48`` for the
+  GSP header alignment of 4 (``msgqTxCreate``).
+* ``GSP_MSG_QUEUE_ELEMENT``: 16 B authTag + 16 B aad + checksum/seqNum/
+  elemCount/reserved (16 B) + the 32 B ``rpc_message_header_v`` (so the RPC
+  payload starts at 0x50).  Checksum = XOR of 64-bit words, hi32 ^ lo32; the
+  stored field makes a receiver's recomputation zero (verified).
+* The real command queue is created with ``MSGQ_FLAGS_SWAP_RX``; the two rx
+  headers are cross-linked between the command and status buffers.  That
+  cross-link is modelled, not simplified away.
+* The RPC call shape is the one ``harness/rm_capture.py`` decodes
+  (``rpc_message_header_v`` = 32 bytes, GSP_RM_CONTROL = 76, GSP_RM_ALLOC =
+  103); the demo's 512-byte control is real-shaped but comes from no capture.
+
+What is stubbed: the dispatcher answers only ``RPC_FN_ECHO`` /
+``RPC_FN_STATUS_QUERY`` (stub-only function ids) and returns
+``NV_ERR_NOT_SUPPORTED`` for every real RM function.  That reply is framed and
+transported exactly, but it is a stub verdict, never firmware behaviour.
+
+Determinism: the transport transcript carries no host timestamps and is
+byte-identical across runs; only the optional ``greeneye-session/1`` recorder
+events carry host ``t`` (documented as non-semantic by ``compare``).  Every
+run is bounded: a message stages contiguously and splits across at most 16
+ring slots (the wire element maximum; larger messages are refused), ``poll``
+drains only the ring, ``run_cycles`` is capped.  No hardware, module, mdev,
+guest, network or QEMU is touched.
+"""
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import struct
+import sys
+from dataclasses import dataclass
+from typing import Any, Callable
+
+from .devices.base import Device
+
+_TOOL_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+
+#: GSP doorbell register block (public Turing ``dev_gsp.h``, block-relative).
+QUEUE_HEAD_OFFSET = 0x110C00
+QUEUE_HEAD_STRIDE = 8
+QUEUE_HEAD_COUNT = 8
+#: PRIV aperture base used by the GSP profiles; doorbell window = PRIV + offset.
+PRIV_BASE = 0x2000000000000000
+DOORBELL_BASE = PRIV_BASE + QUEUE_HEAD_OFFSET
+DOORBELL_SIZE = QUEUE_HEAD_STRIDE * QUEUE_HEAD_COUNT
+#: Host writes the queue head to this register as the kick (observed 570.144
+#: submit path writes 0); the model records the value per queue index.
+DOORBELL_KICK_QUEUE = 0
+DOORBELL_KICK_VALUE = 0
+
+#: ``msgq_priv.h``.
+MSGQ_VERSION = 0
+MSGQ_FLAGS_SWAP_RX = 1
+MSGQ_TX_HEADER_BYTES = 32
+MSGQ_RX_HEADER_BYTES = 4
+MSGQ_HEADER_ALIGN = 4
+MSGQ_RX_HDR_OFF = 32
+MSGQ_ENTRY_OFF = 48
+MSGQ_TX_HEADER_FORMAT = "<8I"
+MSGQ_RX_HEADER_FORMAT = "<I"
+
+#: ``message_queue_priv.h``.
+GSP_MSG_QUEUE_ELEMENT_SIZE_MIN = 0x1000
+GSP_MSG_QUEUE_ELEMENT_SIZE_MAX = 0x10000
+GSP_MSG_QUEUE_HEADER_SIZE = 0x1000
+GSP_MSG_QUEUE_ELEMENT_HDR_SIZE = 0x30
+
+#: ``g_rpc-message-header.h`` / 570.144 wire constants.
+RPC_HEADER_BYTES = 32
+RPC_HEADER_FORMAT = "<8I"
+RPC_SIGNATURE = 0x43505256
+RPC_HEADER_VERSION = 0x03000000
+
+#: Default shared sysmem queue block and sizes (host ``message_queue_cpu.c``).
+DEFAULT_CMD_QUEUE_SIZE = 0x40000
+DEFAULT_STAT_QUEUE_SIZE = 0x40000
+SHARED_MEM_BASE = 0x81000000
+
+#: Real RM wire function ids used by the capture fixtures.
+GSP_RM_CONTROL = 76
+GSP_RM_ALLOC = 103
+GSP_GET_GSP_STATIC_INFO = 65
+
+#: Stub-only dispatcher ids (deliberately not RM function numbers).
+RPC_FN_ECHO = 0x40
+RPC_FN_STATUS_QUERY = 0x41
+
+#: ``nvstatuscodes.h``.
+NV_OK = 0
+NV_ERR_NOT_SUPPORTED = 0x56
+
+#: Bounded ``run_cycles`` cap (same ceiling as the boot runner's attempts).
+MAX_RUN_CYCLES = 2_000_000
+
+STATUS_MAGIC = b"GSTS"
+STATUS_VERSION = 1
+STATUS_FORMAT = "<4sIIII"
+
+DEMO_FORMAT = "greeneye-host-traffic-demo/1"
+REPLAY_FORMAT = "greeneye-host-traffic-replay/1"
+TRANSCRIPT_FORMAT = "greeneye-host-traffic-log/1"
+REPLY_FORMAT = "greeneye-host-traffic-reply/1"
+STIMULUS_FORMAT = "greeneye-rm-rpc-stimulus/1"
+CAPTURE_FORMAT = "greeneye-rm-rpc-capture/1"
+
+_TX_HEADER = struct.Struct(MSGQ_TX_HEADER_FORMAT)
+_RX_HEADER = struct.Struct(MSGQ_RX_HEADER_FORMAT)
+_RPC_HEADER = struct.Struct(RPC_HEADER_FORMAT)
+_ELEMENT_PREFIX = struct.Struct("<16s16sIIII")
+
+_LIMITS = [
+    "Stub wave: no real firmware message loop is run; the dispatcher handlers "
+    "are stubs and real RM functions get NV_ERR_NOT_SUPPORTED.",
+    "Multi-element messages are split across ring slots up to the wire element "
+    "maximum (0x10000 bytes); larger messages are refused, not truncated.",
+    "The shared-memory address and queue sizes are fixture defaults, not "
+    "values read from a real GSP_ARGUMENTS_CACHED transfer.",
+    "The MSI-X/event path by which the real GSP signals a status-queue write "
+    "is not modelled; replies are read by polling.",
+    "L0/L1 framing and ordering only; no timing (L2) and no device-environment "
+    "(L3) claim.",
+]
+
+
+class HostTrafficError(Exception):
+    """A framing, queue or replay input that cannot be represented."""
+
+
+def _align_up(value: int, align: int) -> int:
+    return (value + align - 1) // align * align
+
+
+# ---------------------------------------------------------------------------
+# Framing
+# ---------------------------------------------------------------------------
+
+
+@dataclass
+class RpcMessage:
+    """One decoded ``rpc_message_header_v`` plus its payload."""
+
+    header_version: int
+    signature: int
+    length: int
+    function: int
+    rpc_result: int
+    rpc_result_private: int
+    sequence: int
+    spare: int
+    params: bytes = b""
+
+    def to_bytes(self) -> bytes:
+        return _RPC_HEADER.pack(
+            self.header_version,
+            self.signature,
+            RPC_HEADER_BYTES + len(self.params),
+            self.function,
+            self.rpc_result,
+            self.rpc_result_private,
+            self.sequence,
+            self.spare,
+        ) + self.params
+
+
+@dataclass
+class RpcRequest:
+    """A command message handed to a dispatcher handler."""
+
+    function: int
+    sequence: int
+    params: bytes
+    element_index: int
+    raw: bytes
+
+
+@dataclass
+class RpcReply:
+    """A handler result before framing into a status element."""
+
+    function: int
+    sequence: int
+    params: bytes = b""
+    result: int = NV_OK
+
+
+def build_rpc_message(
+    function: int,
+    params: bytes = b"",
+    sequence: int = 0,
+    *,
+    result: int = NV_OK,
+    result_private: int = 0,
+    header_version: int = RPC_HEADER_VERSION,
+    signature: int = RPC_SIGNATURE,
+) -> bytes:
+    """Frame one ``rpc_message_header_v`` + payload (capture-tool shape)."""
+    payload = bytes(params)
+    return _RPC_HEADER.pack(
+        header_version,
+        signature,
+        RPC_HEADER_BYTES + len(payload),
+        int(function) & 0xFFFFFFFF,
+        int(result) & 0xFFFFFFFF,
+        int(result_private) & 0xFFFFFFFF,
+        int(sequence) & 0xFFFFFFFF,
+        0,
+    ) + payload
+
+
+def parse_rpc_message(data: bytes) -> RpcMessage:
+    """Decode and validate one RPC message; raises on a bad frame."""
+    if len(data) < RPC_HEADER_BYTES:
+        raise HostTrafficError(
+            f"rpc message shorter than {RPC_HEADER_BYTES} bytes: {len(data)}"
+        )
+    (
+        header_version,
+        signature,
+        length,
+        function,
+        rpc_result,
+        rpc_result_private,
+        sequence,
+        spare,
+    ) = _RPC_HEADER.unpack_from(data, 0)
+    if signature != RPC_SIGNATURE:
+        raise HostTrafficError(
+            f"bad rpc signature {signature:#010x} (expected {RPC_SIGNATURE:#010x})"
+        )
+    if length < RPC_HEADER_BYTES or length > len(data):
+        raise HostTrafficError(
+            f"rpc length {length:#x} outside [{RPC_HEADER_BYTES:#x}, {len(data):#x}]"
+        )
+    return RpcMessage(
+        header_version=header_version,
+        signature=signature,
+        length=length,
+        function=function,
+        rpc_result=rpc_result,
+        rpc_result_private=rpc_result_private,
+        sequence=sequence,
+        spare=spare,
+        params=bytes(data[RPC_HEADER_BYTES:length]),
+    )
+
+
+def compute_message_checksum(data: bytes) -> int:
+    """The msgq checksum: XOR of 64-bit words, then hi32 ^ lo32.
+
+    Zero-pads to 8 bytes exactly like ``_checkSum32``; a correctly built queue
+    element recomputes to 0.
+    """
+    padded = bytes(data) + bytes((-len(data)) % 8)
+    total = 0
+    for offset in range(0, len(padded), 8):
+        total ^= int.from_bytes(padded[offset : offset + 8], "little")
+    return ((total >> 32) ^ (total & 0xFFFFFFFF)) & 0xFFFFFFFF
+
+
+def build_queue_element(
+    rpc_bytes: bytes, *, sequence: int = 0, elem_count: int | None = None
+) -> bytes:
+    """Wrap an RPC message in a ``GSP_MSG_QUEUE_ELEMENT`` with checksum."""
+    payload = bytes(rpc_bytes)
+    if elem_count is None:
+        elem_count = max(1, -(-(GSP_MSG_QUEUE_ELEMENT_HDR_SIZE + len(payload))
+                             // GSP_MSG_QUEUE_ELEMENT_SIZE_MIN))
+    prefix = _ELEMENT_PREFIX.pack(b"\x00" * 16, b"\x00" * 16, 0,
+                                  sequence & 0xFFFFFFFF,
+                                  elem_count & 0xFFFFFFFF, 0)
+    body = prefix + payload
+    checksum = compute_message_checksum(body)
+    return _ELEMENT_PREFIX.pack(b"\x00" * 16, b"\x00" * 16, checksum,
+                                sequence & 0xFFFFFFFF,
+                                elem_count & 0xFFFFFFFF, 0) + payload
+
+
+def split_element_stream(stream: bytes, msg_size: int) -> list[bytes]:
+    """Split a contiguous element stream into ``msg_size`` ring slots.
+
+    The real producer stages the whole message in one buffer and copies
+    ``GSP_MSG_QUEUE_ELEMENT_SIZE_MIN`` bytes per ring slot
+    (``GspMsgQueueSendCommand``); the tail slot is zero-padded.
+    """
+    blob = bytes(stream)
+    if not blob:
+        raise HostTrafficError("empty element stream")
+    slots = []
+    for start in range(0, len(blob), msg_size):
+        piece = blob[start : start + msg_size]
+        slots.append(piece + bytes(msg_size - len(piece)))
+    return slots
+
+
+def parse_queue_element(data: bytes, *, verify: bool = True) -> tuple[dict, RpcMessage]:
+    """Decode one queue element stream; raises on a bad frame or checksum."""
+    if len(data) < GSP_MSG_QUEUE_ELEMENT_HDR_SIZE + RPC_HEADER_BYTES:
+        raise HostTrafficError(f"queue element too short: {len(data)} bytes")
+    auth_tag, aad, checksum, seq_num, elem_count, reserved = _ELEMENT_PREFIX.unpack_from(data, 0)
+    if not 1 <= elem_count <= GSP_MSG_QUEUE_ELEMENT_SIZE_MAX // GSP_MSG_QUEUE_ELEMENT_SIZE_MIN:
+        raise HostTrafficError(f"implausible elem_count {elem_count}")
+    rpc = parse_rpc_message(data[GSP_MSG_QUEUE_ELEMENT_HDR_SIZE:])
+    total = GSP_MSG_QUEUE_ELEMENT_HDR_SIZE + rpc.length
+    if total > len(data):
+        raise HostTrafficError(f"queue element truncated: need {total:#x}, have {len(data):#x}")
+    if verify and compute_message_checksum(bytes(data[:total])) != 0:
+        raise HostTrafficError("bad message checksum")
+    prefix = {
+        "auth_tag": auth_tag,
+        "aad": aad,
+        "checksum": checksum,
+        "seq_num": seq_num,
+        "elem_count": elem_count,
+        "reserved": reserved,
+    }
+    return prefix, rpc
+
+
+# ---------------------------------------------------------------------------
+# Memory adapters
+# ---------------------------------------------------------------------------
+
+
+class RegionMemory:
+    """A ``base``/``size`` bytearray window with Machine-like read/write."""
+
+    def __init__(self, base: int, size: int):
+        if size <= 0:
+            raise HostTrafficError(f"region size must be positive, got {size}")
+        self.base = int(base)
+        self.size = int(size)
+        self.data = bytearray(self.size)
+
+    def _check(self, addr: int, size: int) -> int:
+        offset = addr - self.base
+        if offset < 0 or offset + size > self.size:
+            raise HostTrafficError(
+                f"access outside region: addr={addr:#x} size={size}"
+            )
+        return offset
+
+    def read(self, addr: int, size: int) -> bytes:
+        offset = self._check(addr, size)
+        return bytes(self.data[offset : offset + size])
+
+    def write(self, addr: int, data: bytes) -> None:
+        blob = bytes(data)
+        offset = self._check(addr, len(blob))
+        self.data[offset : offset + len(blob)] = blob
+
+
+class _MachineMemory:
+    """Adapter from RegionMemory's surface to ``sim.machine.Machine``."""
+
+    def __init__(self, machine):
+        self.machine = machine
+
+    def read(self, addr: int, size: int) -> bytes:
+        return self.machine.read_mem(addr, size)
+
+    def write(self, addr: int, data: bytes) -> None:
+        self.machine.write_mem(addr, data)
+
+
+def _make_memory(machine, shared_base: int, size: int):
+    if machine is None:
+        return RegionMemory(shared_base, size)
+    return _MachineMemory(machine)
+
+
+# ---------------------------------------------------------------------------
+# Message queues
+# ---------------------------------------------------------------------------
+
+
+@dataclass
+class MsgqTxHeader:
+    """``msgqTxHeader``: 8 x u32 written by the queue producer."""
+
+    version: int
+    size: int
+    msg_size: int
+    msg_count: int
+    write_ptr: int
+    flags: int
+    rx_hdr_off: int
+    entry_off: int
+
+    def to_bytes(self) -> bytes:
+        return _TX_HEADER.pack(
+            self.version,
+            self.size,
+            self.msg_size,
+            self.msg_count,
+            self.write_ptr,
+            self.flags,
+            self.rx_hdr_off,
+            self.entry_off,
+        )
+
+    @classmethod
+    def from_bytes(cls, data: bytes) -> "MsgqTxHeader":
+        return cls(*_TX_HEADER.unpack_from(data, 0))
+
+
+class MsgqQueue:
+    """One msgq buffer: tx header at its base, entries at ``entryOff``.
+
+    ``rx_hdr_addr`` is the physical address of this queue's rx header.  The
+    real command queue keeps it in the status buffer (``MSGQ_FLAGS_SWAP_RX``),
+    so the caller passes the cross-linked address.
+    """
+
+    def __init__(
+        self,
+        mem,
+        base: int,
+        size: int,
+        *,
+        msg_size: int = GSP_MSG_QUEUE_ELEMENT_SIZE_MIN,
+        header_align: int = MSGQ_HEADER_ALIGN,
+        flags: int = MSGQ_FLAGS_SWAP_RX,
+        rx_hdr_addr: int | None = None,
+        version: int = MSGQ_VERSION,
+    ):
+        if size < GSP_MSG_QUEUE_ELEMENT_SIZE_MIN:
+            raise HostTrafficError(f"queue size {size:#x} below one element")
+        self.mem = mem
+        self.tx_base = int(base)
+        self.size = int(size)
+        self.msg_size = int(msg_size)
+        self.header_align = int(header_align)
+        self.flags = int(flags)
+        self.version = int(version)
+        self.rx_hdr_addr = int(rx_hdr_addr) if rx_hdr_addr is not None else base + MSGQ_RX_HDR_OFF
+
+    def format(self) -> MsgqTxHeader:
+        """Write a fresh tx header and zero the rx header (msgqTxCreate)."""
+        rx_hdr_off = _align_up(MSGQ_TX_HEADER_BYTES, 1 << self.header_align)
+        entry_off = _align_up(rx_hdr_off + MSGQ_RX_HEADER_BYTES, 1 << self.header_align)
+        header = MsgqTxHeader(
+            version=self.version,
+            size=self.size,
+            msg_size=self.msg_size,
+            msg_count=(self.size - entry_off) // self.msg_size,
+            write_ptr=0,
+            flags=self.flags,
+            rx_hdr_off=rx_hdr_off,
+            entry_off=entry_off,
+        )
+        self.mem.write(self.tx_base, header.to_bytes())
+        self.mem.write(self.rx_hdr_addr, _RX_HEADER.pack(0))
+        return header
+
+    def header(self) -> MsgqTxHeader:
+        return MsgqTxHeader.from_bytes(self.mem.read(self.tx_base, MSGQ_TX_HEADER_BYTES))
+
+    def _write_header(self, header: MsgqTxHeader) -> None:
+        self.mem.write(self.tx_base, header.to_bytes())
+
+    @property
+    def write_ptr(self) -> int:
+        return self.header().write_ptr
+
+    @property
+    def read_ptr(self) -> int:
+        return _RX_HEADER.unpack_from(self.mem.read(self.rx_hdr_addr, MSGQ_RX_HEADER_BYTES), 0)[0]
+
+    def _entry_addr(self, index: int) -> int:
+        count = self.header().msg_count
+        if not 0 <= index < count:
+            raise HostTrafficError(f"entry index {index} outside 0..{count - 1}")
+        return self.tx_base + self.header().entry_off + index * self.msg_size
+
+    def write_entry(self, index: int, data: bytes) -> None:
+        blob = bytes(data)
+        if len(blob) > self.msg_size:
+            raise HostTrafficError(
+                f"entry of {len(blob)} bytes exceeds msgSize {self.msg_size:#x}"
+            )
+        self.mem.write(self._entry_addr(index), blob)
+
+    def read_entry(self, index: int, size: int | None = None) -> bytes:
+        return self.mem.read(self._entry_addr(index), self.msg_size if size is None else size)
+
+    def publish(self, write_ptr: int) -> int:
+        header = self.header()
+        header.write_ptr = int(write_ptr) % header.msg_count
+        self._write_header(header)
+        return header.write_ptr
+
+    def set_read(self, read_ptr: int) -> int:
+        count = self.header().msg_count
+        value = int(read_ptr) % count
+        self.mem.write(self.rx_hdr_addr, _RX_HEADER.pack(value))
+        return value
+
+    def state(self) -> dict:
+        header = self.header()
+        return {
+            "tx_base": self.tx_base,
+            "size": self.size,
+            "msg_size": header.msg_size,
+            "msg_count": header.msg_count,
+            "write_ptr": header.write_ptr,
+            "read_ptr": self.read_ptr,
+            "flags": header.flags,
+            "rx_hdr_off": header.rx_hdr_off,
+            "entry_off": header.entry_off,
+            "rx_hdr_addr": self.rx_hdr_addr,
+        }
+
+
+class SharedQueues:
+    """A command/status msgq pair with the real cross-linked rx headers.
+
+    Layout follows the host's ``GspMsgQueuesInit``: the status buffer starts
+    ``cmd_size`` bytes after the command buffer, both tx headers carry
+    ``MSGQ_FLAGS_SWAP_RX``, and each queue's rx header physically lives in the
+    other buffer (``msgqRxLink``).
+    """
+
+    def __init__(
+        self,
+        mem,
+        *,
+        cmd_base: int = SHARED_MEM_BASE,
+        stat_base: int | None = None,
+        cmd_size: int = DEFAULT_CMD_QUEUE_SIZE,
+        stat_size: int = DEFAULT_STAT_QUEUE_SIZE,
+        msg_size: int = GSP_MSG_QUEUE_ELEMENT_SIZE_MIN,
+        swap_rx: bool = True,
+    ):
+        self.mem = mem
+        self.cmd_base = int(cmd_base)
+        self.stat_base = int(stat_base) if stat_base is not None else self.cmd_base + cmd_size
+        flags = MSGQ_FLAGS_SWAP_RX if swap_rx else 0
+        self.command = MsgqQueue(
+            mem,
+            self.cmd_base,
+            cmd_size,
+            msg_size=msg_size,
+            flags=flags,
+            rx_hdr_addr=self.stat_base + MSGQ_RX_HDR_OFF,
+        )
+        self.status = MsgqQueue(
+            mem,
+            self.stat_base,
+            stat_size,
+            msg_size=msg_size,
+            flags=flags,
+            rx_hdr_addr=self.cmd_base + MSGQ_RX_HDR_OFF,
+        )
+
+    def format(self) -> None:
+        self.command.format()
+        self.status.format()
+
+    def _place(self, queue: MsgqQueue, element: bytes) -> dict:
+        slots = split_element_stream(element, queue.msg_size)
+        count = queue.header().msg_count
+        if len(slots) > count:
+            raise HostTrafficError(
+                f"message needs {len(slots)} elements, queue holds {count}"
+            )
+        index = queue.write_ptr
+        ptr = index
+        for slot in slots:
+            queue.write_entry(ptr, slot)
+            ptr = (ptr + 1) % count
+        write_ptr = queue.publish(ptr)
+        return {
+            "index": index,
+            "write_ptr": write_ptr,
+            "bytes": len(element),
+            "elements": len(slots),
+        }
+
+    def _take(self, queue: MsgqQueue) -> dict | None:
+        write_ptr = queue.write_ptr
+        read_ptr = queue.read_ptr
+        if read_ptr == write_ptr:
+            return None
+        count = queue.header().msg_count
+        first = queue.read_entry(read_ptr)
+        elements = int.from_bytes(first[0x28:0x2C], "little")
+        if not 1 <= elements <= count:
+            raise HostTrafficError(f"implausible elem_count {elements}")
+        slots = []
+        ptr = read_ptr
+        for _ in range(elements):
+            slots.append(queue.read_entry(ptr))
+            ptr = (ptr + 1) % count
+        stream = b"".join(slots)
+        total = GSP_MSG_QUEUE_ELEMENT_HDR_SIZE + int.from_bytes(
+            stream[0x38:0x3C], "little"
+        )
+        element = (
+            stream[:total]
+            if GSP_MSG_QUEUE_ELEMENT_HDR_SIZE + RPC_HEADER_BYTES <= total <= len(stream)
+            else stream
+        )
+        queue.set_read(ptr)
+        return {"index": read_ptr, "element": element, "elements": elements}
+
+    def submit_element(self, element: bytes) -> dict:
+        """Host -> GSP: append one element to the command ring."""
+        return self._place(self.command, element)
+
+    def take_command(self) -> dict | None:
+        """GSP side: consume the next command element, or ``None``."""
+        return self._take(self.command)
+
+    def publish_element(self, element: bytes) -> dict:
+        """GSP -> host: append one element to the status ring."""
+        return self._place(self.status, element)
+
+    def take_status(self) -> dict | None:
+        """Host side: consume the next status element, or ``None``."""
+        return self._take(self.status)
+
+    def state(self) -> dict:
+        return {"command": self.command.state(), "status": self.status.state()}
+
+
+# ---------------------------------------------------------------------------
+# Doorbell
+# ---------------------------------------------------------------------------
+
+
+class DoorbellModel:
+    """Recorded queue-head writes; the firmware-visible read-back surface."""
+
+    def __init__(self, *, count: int = QUEUE_HEAD_COUNT, max_writes: int = 1024):
+        self.count = int(count)
+        self.max_writes = int(max_writes)
+        self.heads = [0] * self.count
+        self.writes: list[dict] = []
+        self.dropped = 0
+        self.kicks = 0
+
+    def ring(self, queue: int, value: int = 0, *, pc: int | None = None) -> None:
+        queue = int(queue)
+        if not 0 <= queue < self.count:
+            raise ValueError(f"queue index {queue} outside 0..{self.count - 1}")
+        value = int(value) & 0xFFFFFFFF
+        self.heads[queue] = value
+        self.kicks += 1
+        record = {"queue": queue, "value": value}
+        if pc is not None:
+            record["pc"] = int(pc)
+        if len(self.writes) < self.max_writes:
+            self.writes.append(record)
+        else:
+            self.dropped += 1
+
+    def head(self, queue: int) -> int:
+        queue = int(queue)
+        if not 0 <= queue < self.count:
+            raise ValueError(f"queue index {queue} outside 0..{self.count - 1}")
+        return self.heads[queue]
+
+    def state(self) -> dict:
+        return {
+            "count": self.count,
+            "heads": list(self.heads),
+            "kicks": self.kicks,
+            "writes": [dict(entry) for entry in self.writes],
+            "dropped": self.dropped,
+        }
+
+    def reset(self) -> None:
+        self.heads = [0] * self.count
+        self.writes = []
+        self.dropped = 0
+        self.kicks = 0
+
+
+class DoorbellBlock(Device):
+    """``NV_PGSP_QUEUE_HEAD`` window over a :class:`DoorbellModel`.
+
+    Evidence: hardware register header (offset/width/access) for the register;
+    the model's read-back is storage of the last written value, and the
+    firmware's read use of this register is **not observed** (documented, not
+    claimed).  Guest stores are recorded with their PC.
+    """
+
+    name = "gsp-doorbell"
+    wants_pc = True
+
+    def __init__(self, doorbell: DoorbellModel | None = None, *,
+                 base: int = DOORBELL_BASE, size: int = DOORBELL_SIZE):
+        super().__init__(base=base, size=size)
+        self.doorbell = doorbell if doorbell is not None else DoorbellModel()
+
+    def reset(self) -> None:
+        if getattr(self, "doorbell", None) is None:
+            self.doorbell = DoorbellModel()
+        else:
+            self.doorbell.reset()
+
+    def _queue(self, offset: int, size: int) -> int | None:
+        if size != 4 or offset % QUEUE_HEAD_STRIDE:
+            return None
+        queue = offset // QUEUE_HEAD_STRIDE
+        if not 0 <= queue < QUEUE_HEAD_COUNT:
+            return None
+        return queue
+
+    def read(self, offset: int, size: int, pc: int = 0) -> int | None:
+        queue = self._queue(offset, size)
+        if queue is None:
+            return None
+        return self.doorbell.head(queue)
+
+    def write(self, offset: int, size: int, value: int, pc: int = 0) -> bool:
+        queue = self._queue(offset, size)
+        if queue is None:
+            return False
+        self.doorbell.ring(queue, value, pc=pc)
+        return True
+
+    def peek(self, offset: int, size: int) -> int | None:
+        return self.read(offset, size)
+
+    def state(self) -> dict:
+        return {"doorbell": self.doorbell.state()}
+
+    def load_state(self, data: dict) -> None:
+        state = data.get("doorbell", data)
+        self.doorbell.count = int(state.get("count", self.doorbell.count))
+        self.doorbell.heads = [int(v) for v in state.get("heads", [0] * self.doorbell.count)]
+        self.doorbell.kicks = int(state.get("kicks", 0))
+        self.doorbell.writes = [dict(entry) for entry in state.get("writes", [])]
+        self.doorbell.dropped = int(state.get("dropped", 0))
+
+
+# ---------------------------------------------------------------------------
+# Recorder compatibility
+# ---------------------------------------------------------------------------
+
+
+class _NullRecorder:
+    """Stand-in for ``greeneye.record.NullRecorder`` (keeps sim standalone)."""
+
+    dir: str | None = None
+
+    def event(self, event_type: str, **fields) -> None:
+        pass
+
+    def blob(self, name: str, data: bytes) -> str:
+        return name
+
+
+# ---------------------------------------------------------------------------
+# Host traffic loop
+# ---------------------------------------------------------------------------
+
+
+class HostTraffic:
+    """The modelled host-side transport: submit, dispatch (stub), read reply.
+
+    ``machine`` (optional) supplies simulated RAM for the queue block and lets
+    :meth:`run_cycles` step the core; without it the queues live in a private
+    :class:`RegionMemory`.  ``recorder`` (optional) is any object with
+    ``event(type, **fields)`` / ``blob(name, data)``; the transport transcript
+    stays deterministic regardless.
+    """
+
+    def __init__(
+        self,
+        *,
+        machine=None,
+        recorder=None,
+        shared_base: int = SHARED_MEM_BASE,
+        cmd_size: int = DEFAULT_CMD_QUEUE_SIZE,
+        stat_size: int = DEFAULT_STAT_QUEUE_SIZE,
+        msg_size: int = GSP_MSG_QUEUE_ELEMENT_SIZE_MIN,
+        handlers: dict[int, Callable[..., Any]] | None = None,
+    ):
+        if machine is None and shared_base == SHARED_MEM_BASE:
+            mem = RegionMemory(shared_base, cmd_size + stat_size)
+        else:
+            mem = _make_memory(machine, shared_base, cmd_size + stat_size)
+        self.machine = machine
+        self.recorder = recorder if recorder is not None else _NullRecorder()
+        self.shared_base = int(shared_base)
+        self.queues = SharedQueues(
+            mem,
+            cmd_base=shared_base,
+            cmd_size=cmd_size,
+            stat_size=stat_size,
+            msg_size=msg_size,
+        )
+        self.queues.format()
+        self.doorbell = DoorbellModel()
+        self._tx_seq = 0
+        self._rx_seq = 0
+        self._next_sequence = 1
+        self._submitted = 0
+        self._dispatched = 0
+        self._replies = 0
+        self._events: list[dict] = []
+        self._closed = False
+        self._handlers: dict[int, tuple[str, Callable[..., Any]]] = {}
+        self.set_handler(RPC_FN_ECHO, self._handle_echo, name="stub:echo")
+        self.set_handler(RPC_FN_STATUS_QUERY, self._handle_status_query,
+                        name="stub:status_query")
+        for function, handler in (handlers or {}).items():
+            self.set_handler(function, handler)
+        self._emit(
+            "host_traffic_config",
+            shared_base=self.shared_base,
+            cmd_base=self.queues.command.tx_base,
+            stat_base=self.queues.status.tx_base,
+            cmd_size=self.queues.command.size,
+            stat_size=self.queues.status.size,
+            msg_size=self.queues.command.msg_size,
+            handlers=sorted(self.handler_names().items()),
+        )
+
+    # -- handler table ------------------------------------------------------
+
+    def set_handler(self, function: int, handler: Callable[..., Any], *,
+                    name: str | None = None) -> None:
+        label = name or getattr(handler, "__name__", "handler")
+        self._handlers[int(function)] = (label, handler)
+
+    def handler_names(self) -> dict[int, str]:
+        return {function: label for function, (label, _) in sorted(self._handlers.items())}
+
+    def _handle_echo(self, request: RpcRequest) -> RpcReply:
+        return RpcReply(function=request.function, sequence=request.sequence,
+                        params=request.params)
+
+    def _handle_status_query(self, request: RpcRequest) -> RpcReply:
+        payload = struct.pack(
+            STATUS_FORMAT,
+            STATUS_MAGIC,
+            STATUS_VERSION,
+            self._submitted,
+            self._dispatched,
+            self._replies,
+        )
+        return RpcReply(function=request.function, sequence=request.sequence,
+                        params=payload)
+
+    # -- recording ----------------------------------------------------------
+
+    def _emit(self, event_type: str, **fields) -> None:
+        self.recorder.event(event_type, **fields)
+
+    def _record(self, event: dict) -> None:
+        self._events.append(event)
+
+    def note_skip(self, index: int, reason: str) -> None:
+        """Record a replay message that was not submitted (documented)."""
+        self._record({"type": "skip", "index": int(index), "reason": reason})
+        self._emit("host_replay_skip", index=int(index), reason=reason)
+
+    # -- host side ----------------------------------------------------------
+
+    def submit(self, message: bytes) -> dict:
+        """Write one framed RPC message to the command ring, then kick."""
+        parsed = parse_rpc_message(message)
+        element = build_queue_element(message, sequence=self._tx_seq)
+        if len(element) > GSP_MSG_QUEUE_ELEMENT_SIZE_MAX:
+            raise HostTrafficError(
+                f"message of {len(element)} bytes exceeds the wire element "
+                f"maximum of {GSP_MSG_QUEUE_ELEMENT_SIZE_MAX:#x}"
+            )
+        placed = self.queues.submit_element(element)
+        self._tx_seq += 1
+        self.doorbell.ring(DOORBELL_KICK_QUEUE, DOORBELL_KICK_VALUE)
+        self._submitted += 1
+        self._emit("doorbell", queue=DOORBELL_KICK_QUEUE,
+                   value=DOORBELL_KICK_VALUE, source="host")
+        result = {
+            "format": "greeneye-host-traffic-submit/1",
+            "index": placed["index"],
+            "write_ptr": placed["write_ptr"],
+            "function": parsed.function,
+            "sequence": parsed.sequence,
+            "msgq_sequence": self._tx_seq - 1,
+            "length": parsed.length,
+            "params_bytes": len(parsed.params),
+            "element_bytes": len(element),
+            "doorbell": {"queue": DOORBELL_KICK_QUEUE, "value": DOORBELL_KICK_VALUE},
+        }
+        self._record({
+            "type": "submit",
+            "index": placed["index"],
+            "function": parsed.function,
+            "sequence": parsed.sequence,
+            "params_bytes": len(parsed.params),
+            "element_bytes": len(element),
+            "write_ptr": placed["write_ptr"],
+        })
+        self._emit(
+            "host_submit",
+            function=parsed.function,
+            sequence=parsed.sequence,
+            length=parsed.length,
+            params_bytes=len(parsed.params),
+            element_index=placed["index"],
+            write_ptr=placed["write_ptr"],
+        )
+        return result
+
+    def submit_rpc(self, function: int, params: bytes = b"",
+                   *, sequence: int | None = None) -> dict:
+        """Frame and submit one RPC; ``sequence`` defaults to a local counter."""
+        if sequence is None:
+            sequence = self._next_sequence
+        self._next_sequence = max(self._next_sequence, int(sequence) + 1)
+        return self.submit(build_rpc_message(function, params, sequence=sequence))
+
+    def _normalise_reply(self, produced: Any, request: RpcRequest) -> RpcReply:
+        if produced is None:
+            return RpcReply(function=request.function, sequence=request.sequence)
+        if isinstance(produced, RpcReply):
+            return produced
+        if isinstance(produced, (bytes, bytearray)):
+            return RpcReply(function=request.function, sequence=request.sequence,
+                            params=bytes(produced))
+        raise HostTrafficError(
+            f"handler returned {type(produced).__name__}; expected RpcReply, "
+            "bytes or None"
+        )
+
+    def poll(self) -> int:
+        """GSP side: drain the command ring through the dispatcher (bounded)."""
+        drained = 0
+        while True:
+            taken = self.queues.take_command()
+            if taken is None:
+                break
+            prefix, request_message = parse_queue_element(taken["element"])
+            label, handler = self._handlers.get(
+                request_message.function, ("stub:unsupported", None)
+            )
+            self._dispatched += 1
+            request = RpcRequest(
+                function=request_message.function,
+                sequence=request_message.sequence,
+                params=request_message.params,
+                element_index=taken["index"],
+                raw=bytes(taken["element"]),
+            )
+            if handler is None:
+                reply = RpcReply(
+                    function=request.function,
+                    sequence=request.sequence,
+                    params=b"",
+                    result=NV_ERR_NOT_SUPPORTED,
+                )
+            else:
+                reply = self._normalise_reply(handler(request), request)
+            reply_bytes = build_rpc_message(
+                reply.function, reply.params, sequence=reply.sequence,
+                result=reply.result,
+            )
+            element = build_queue_element(reply_bytes, sequence=self._rx_seq)
+            self._rx_seq += 1
+            if len(element) > GSP_MSG_QUEUE_ELEMENT_SIZE_MAX:
+                raise HostTrafficError(
+                    f"reply of {len(element)} bytes exceeds the wire element maximum"
+                )
+            self.queues.publish_element(element)
+            self._replies += 1
+            drained += 1
+            self._record({
+                "type": "dispatch",
+                "index": taken["index"],
+                "function": request.function,
+                "sequence": request.sequence,
+                "handler": label,
+                "result": reply.result,
+                "reply_bytes": len(reply.params),
+            })
+            self._emit(
+                "host_dispatch",
+                function=request.function,
+                sequence=request.sequence,
+                handler=label,
+                result=reply.result,
+                reply_bytes=len(reply.params),
+                element_index=taken["index"],
+                msgq_sequence=prefix["seq_num"],
+            )
+        return drained
+
+    def read_reply(self) -> dict | None:
+        """Host side: consume the next status element, or ``None``."""
+        taken = self.queues.take_status()
+        if taken is None:
+            return None
+        _prefix, message = parse_queue_element(taken["element"])
+        record = {
+            "format": REPLY_FORMAT,
+            "index": taken["index"],
+            "function": message.function,
+            "sequence": message.sequence,
+            "result": message.rpc_result,
+            "params": message.params,
+            "params_hex": message.params.hex(),
+            "params_bytes": len(message.params),
+        }
+        self._record({
+            "type": "reply",
+            "index": taken["index"],
+            "function": message.function,
+            "sequence": message.sequence,
+            "result": message.rpc_result,
+            "params_bytes": len(message.params),
+        })
+        self._emit(
+            "host_reply_read",
+            function=message.function,
+            sequence=message.sequence,
+            result=message.rpc_result,
+            params_bytes=len(message.params),
+            element_index=taken["index"],
+        )
+        return record
+
+    def run_cycles(self, budget: int) -> dict:
+        """Run the attached core for a bounded ``budget``, then ``poll``."""
+        if isinstance(budget, bool) or not isinstance(budget, int) or not 0 < budget <= MAX_RUN_CYCLES:
+            raise ValueError(f"budget must be an int in 1..{MAX_RUN_CYCLES}, got {budget!r}")
+        reason = pc = None
+        if self.machine is not None:
+            reason, pc, _a, _b = self.machine.run(budget)
+        dispatched = self.poll()
+        return {
+            "budget": budget,
+            "reason": reason,
+            "pc": pc,
+            "insns": self.machine.insns if self.machine is not None else 0,
+            "dispatched": dispatched,
+            "replies": self._replies,
+        }
+
+    # -- evidence -----------------------------------------------------------
+
+    def transcript(self) -> dict:
+        counts: dict[str, int] = {}
+        for event in self._events:
+            counts[event["type"]] = counts.get(event["type"], 0) + 1
+        return {
+            "format": TRANSCRIPT_FORMAT,
+            "events": [dict(event) for event in self._events],
+            "counts": counts,
+        }
+
+    def state(self) -> dict:
+        return {
+            "format": TRANSCRIPT_FORMAT,
+            "shared_base": self.shared_base,
+            "queues": self.queues.state(),
+            "doorbell": self.doorbell.state(),
+            "counts": {
+                "submitted": self._submitted,
+                "dispatched": self._dispatched,
+                "replies": self._replies,
+            },
+            "handlers": self.handler_names(),
+        }
+
+    def doorbell_block(self) -> DoorbellBlock:
+        return DoorbellBlock(self.doorbell)
+
+    def close(self) -> None:
+        if self._closed:
+            return
+        self._closed = True
+        payload = json.dumps(self.transcript(), indent=2, sort_keys=True).encode()
+        self.recorder.blob("host-traffic.json", payload)
+        self._emit("host_transcript", events=len(self._events))
+
+
+# ---------------------------------------------------------------------------
+# Stimulus replay
+# ---------------------------------------------------------------------------
+
+
+def _stimulus_from_capture(source: str | dict) -> dict:
+    if _TOOL_ROOT not in sys.path:
+        sys.path.insert(0, _TOOL_ROOT)
+    try:
+        from harness.rm_capture import load_capture, stimulus_from_capture
+    except ImportError as exc:  # pragma: no cover - tool checkout broken
+        raise HostTrafficError(f"harness.rm_capture unavailable: {exc}") from exc
+    if isinstance(source, dict):
+        raise HostTrafficError(
+            "an inline capture document is not accepted; pass the capture JSON "
+            "path or a stimulus document"
+        )
+    capture = load_capture(source)
+    return stimulus_from_capture(capture)
+
+
+def load_stimulus(source: str | dict) -> dict:
+    """Load a replay stimulus from a doc, a stimulus JSON or a capture file."""
+    if isinstance(source, dict):
+        doc = source
+    else:
+        if not os.path.isfile(source):
+            raise HostTrafficError(f"stimulus not found: {source}")
+        with open(source, "r", encoding="utf-8", errors="replace") as fh:
+            head = fh.read(4096)
+        if not head.lstrip().startswith("{"):
+            return _stimulus_from_capture(source)
+        try:
+            with open(source, "r", encoding="utf-8", errors="replace") as fh:
+                doc = json.load(fh)
+        except (OSError, json.JSONDecodeError) as exc:
+            raise HostTrafficError(f"cannot read stimulus {source}: {exc}") from exc
+    if not isinstance(doc, dict):
+        raise HostTrafficError("stimulus document must be an object")
+    fmt = str(doc.get("format") or "")
+    if fmt == STIMULUS_FORMAT:
+        return doc
+    if fmt == CAPTURE_FORMAT:
+        return _stimulus_from_capture(source)
+    raise HostTrafficError(
+        f"unknown stimulus/capture format {fmt!r}; expected {STIMULUS_FORMAT!r} "
+        f"or {CAPTURE_FORMAT!r}"
+    )
+
+
+def _message_to_rpc(message: dict, index: int) -> tuple[int, bytes]:
+    """Map one stimulus message to ``(wire function, payload)``.
+
+    ``send`` messages carry the wire function and payload directly.  ``control``
+    and ``alloc`` records carry a named id instead; the capture tool's mapping
+    (control -> 76, alloc -> 103) is applied, and the captured parameter block
+    is used as the payload.  That identity is an assumption until a capture
+    with full send payloads exists (documented in docs/51).
+    """
+    kind = message.get("kind")
+    function = message.get("fn")
+    if function is None:
+        if kind == "control":
+            function = GSP_RM_CONTROL
+        elif kind == "alloc":
+            function = GSP_RM_ALLOC
+    if function is None:
+        raise HostTrafficError(f"message {index}: no wire function id")
+    hex_text = message.get("params_hex") or ""
+    try:
+        params = bytes.fromhex(hex_text)
+    except ValueError as exc:
+        raise HostTrafficError(f"message {index}: bad params_hex: {exc}") from exc
+    return int(function), params
+
+
+def replay_stimulus(host: HostTraffic, stimulus: str | dict, *,
+                    strict: bool = True) -> dict:
+    """Submit a stimulus document's messages in order, then read the replies."""
+    doc = load_stimulus(stimulus)
+    messages = doc.get("messages")
+    if not isinstance(messages, list):
+        raise HostTrafficError("stimulus document has no 'messages' list")
+    pairs: list[dict] = []
+    skipped: list[dict] = []
+    for index, message in enumerate(messages):
+        if not isinstance(message, dict):
+            raise HostTrafficError(f"message {index} is not an object")
+        if message.get("truncated"):
+            reason = "truncated payload is not replayable as-is"
+            if strict:
+                raise HostTrafficError(f"message {index}: {reason}")
+            host.note_skip(index, reason)
+            skipped.append({"index": index, "reason": reason})
+            continue
+        function, params = _message_to_rpc(message, index)
+        submit = host.submit_rpc(function, params)
+        dispatched = host.poll()
+        reply = host.read_reply()
+        if reply is None or dispatched != 1:
+            raise HostTrafficError(
+                f"message {index}: transport did not produce exactly one reply"
+            )
+        pairs.append({
+            "seq": message.get("seq", index),
+            "kind": message.get("kind"),
+            "function": function,
+            "submit": submit,
+            "reply": reply,
+        })
+    return {
+        "format": REPLAY_FORMAT,
+        "source": doc.get("source"),
+        "submitted": len(pairs),
+        "skipped": skipped,
+        "pairs": pairs,
+        "transcript": host.transcript(),
+        "limits": list(_LIMITS),
+    }
+
+
+# ---------------------------------------------------------------------------
+# Demo
+# ---------------------------------------------------------------------------
+
+
+def _new_recorder(out_dir: str | None, name: str):
+    if out_dir is None:
+        return _NullRecorder(), False
+    from greeneye.record import Recorder  # deferred: keep sim importable alone
+
+    meta = {"kind": name, "host_traffic": True}
+    return Recorder(out_dir, name=name, meta=meta), True
+
+
+def _json_reply(reply: dict | None) -> dict | None:
+    if reply is None:
+        return None
+    return {
+        "format": reply["format"],
+        "index": reply["index"],
+        "function": reply["function"],
+        "sequence": reply["sequence"],
+        "result": reply["result"],
+        "params_hex": reply["params_hex"],
+        "params_bytes": reply["params_bytes"],
+        "read": True,
+    }
+
+
+def run_demo(out_dir: str | None = None) -> dict:
+    """Deterministic end-to-end transport demo; no capture, no firmware."""
+    from .machine import Machine  # deferred (module import stays light)
+
+    machine = Machine(memory_mb=32)
+    machine.load_segments(
+        [(machine.ram_base, bytes.fromhex("73001000"))], entry=machine.ram_base
+    )
+    recorder, owns = _new_recorder(out_dir, "host-traffic-demo")
+    host = HostTraffic(machine=machine, recorder=recorder)
+    replies: list[dict] = []
+    for sequence, (function, params) in enumerate(
+        (
+            (RPC_FN_ECHO, b"\x01\x02\x03\x04"),
+            (RPC_FN_STATUS_QUERY, b""),
+        ),
+        start=1,
+    ):
+        host.submit_rpc(function, params, sequence=sequence)
+        host.run_cycles(64)
+        reply = host.read_reply()
+        if reply is not None:
+            replies.append(_json_reply(reply))
+
+    control_cmd = 0x2080012B
+    control_params = (
+        (0xC2000006).to_bytes(4, "little")
+        + (0xABCD0080).to_bytes(4, "little")
+        + control_cmd.to_bytes(4, "little")
+        + bytes(500)
+    )
+    host.submit_rpc(GSP_RM_CONTROL, control_params, sequence=3)
+    host.run_cycles(64)
+    control_reply = host.read_reply()
+    if control_reply is not None:
+        replies.append(_json_reply(control_reply))
+
+    checks = {
+        "echo_reply_matches_payload": (
+            replies[0]["function"] == RPC_FN_ECHO
+            and replies[0]["params_hex"] == b"\x01\x02\x03\x04".hex()
+            and replies[0]["result"] == NV_OK
+        ),
+        "status_reply_has_magic": (
+            len(replies) > 1
+            and bytes.fromhex(replies[1]["params_hex"])[:4] == STATUS_MAGIC
+        ),
+        "control_replay_is_512_bytes": host.state()["counts"]["submitted"] == 3,
+        "control_reply_is_a_framed_stub_verdict": (
+            control_reply is not None
+            and control_reply["function"] == GSP_RM_CONTROL
+            and control_reply["result"] == NV_ERR_NOT_SUPPORTED
+        ),
+        "doorbell_ring_per_submit": host.doorbell.kicks == 3,
+        "all_replies_read": len(replies) == 3,
+    }
+    control = {
+        "function": GSP_RM_CONTROL,
+        "cmd": control_cmd,
+        "bytes": len(control_params),
+        "reply_function": control_reply["function"] if control_reply else None,
+        "reply_result": control_reply["result"] if control_reply else None,
+        "reply_params_bytes": control_reply["params_bytes"] if control_reply else None,
+    }
+    payload = {
+        "format": DEMO_FORMAT,
+        "ok": all(checks.values()),
+        "checks": checks,
+        "machine": {
+            "ram_base": machine.ram_base,
+            "insns": machine.insns,
+            "cycles": machine.cycles,
+        },
+        "shared_base": host.shared_base,
+        "queue": host.queues.command.state(),
+        "control": control,
+        "replies": replies,
+        "transcript": host.transcript(),
+        "limits": list(_LIMITS),
+    }
+    host.close()
+    if owns:
+        os.makedirs(out_dir, exist_ok=True)
+        payload["artifacts"] = {"session": os.path.abspath(out_dir)}
+        with open(os.path.join(out_dir, "host-traffic-demo.json"), "w",
+                  encoding="utf-8") as fh:
+            json.dump(payload, fh, indent=1, sort_keys=True)
+            fh.write("\n")
+        recorder.close()
+    return payload
+
+
+# ---------------------------------------------------------------------------
+# CLI
+# ---------------------------------------------------------------------------
+
+
+def build_parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(
+        prog="sim.host_traffic",
+        description=(
+            "Modelled host traffic loop for the GSP message queues: doorbell, "
+            "command/status rings, stub dispatch and rm_capture stimulus "
+            "replay. Offline only; the simulator runs no real firmware here."
+        ),
+    )
+    sub = parser.add_subparsers(dest="command")
+    demo_cmd = sub.add_parser("demo", help="deterministic end-to-end demo")
+    demo_cmd.add_argument("--out", help="session/output directory (optional)")
+    demo_cmd.add_argument("--json", action="store_true", help="print full JSON")
+    replay_cmd = sub.add_parser("replay", help="replay an rm_capture stimulus")
+    replay_cmd.add_argument("--stimulus", required=True,
+                            help="stimulus JSON, capture JSON or raw capture log")
+    replay_cmd.add_argument("--out", help="session/output directory (optional)")
+    replay_cmd.add_argument("--allow-truncated", action="store_true",
+                            help="skip truncated messages instead of refusing")
+    replay_cmd.add_argument("--json", action="store_true", help="print full JSON")
+    return parser
+
+
+def _cli_summary(payload: dict) -> dict:
+    summary = {
+        "format": payload.get("format"),
+        "ok": payload.get("ok"),
+    }
+    if payload.get("checks") is not None:
+        summary["checks"] = payload["checks"]
+    if "submitted" in payload:
+        summary["submitted"] = payload["submitted"]
+        summary["skipped"] = len(payload.get("skipped", []))
+    if payload.get("artifacts"):
+        summary["artifacts"] = payload["artifacts"]
+    return summary
+
+
+def _jsonable(value):
+    if isinstance(value, bytes):
+        return value.hex()
+    if isinstance(value, (list, tuple)):
+        return [_jsonable(item) for item in value]
+    if isinstance(value, dict):
+        return {str(key): _jsonable(item) for key, item in value.items()}
+    return value
+
+
+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
+    try:
+        if args.command == "demo":
+            payload = run_demo(args.out)
+        else:
+            recorder, owns = _new_recorder(args.out, "host-traffic-replay")
+            host = HostTraffic(recorder=recorder)
+            payload = replay_stimulus(
+                host, args.stimulus, strict=not args.allow_truncated
+            )
+            host.close()
+            if owns:
+                os.makedirs(args.out, exist_ok=True)
+                payload["artifacts"] = {"session": os.path.abspath(args.out)}
+                with open(os.path.join(args.out, "host-traffic-replay.json"), "w",
+                          encoding="utf-8") as fh:
+                    json.dump(_jsonable(payload), fh, indent=1, sort_keys=True)
+                    fh.write("\n")
+                recorder.close()
+    except HostTrafficError as exc:
+        print(f"error: {exc}", file=sys.stderr)
+        return 2
+    if args.json:
+        print(json.dumps(_jsonable(payload), indent=2, sort_keys=True))
+    else:
+        print(json.dumps(_jsonable(_cli_summary(payload)), indent=2, sort_keys=True))
+    if args.command == "demo" and not payload.get("ok"):
+        return 3
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())