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

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