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

"""libos port receive-path model (the B5 message-loop blocker, first half).

Status: **offline development tool (new, additive)**.  This module models the
part of the GSP firmware's libos *port receive* machinery that the B3
message-loop task needs before it can make progress: the argument contract of
``TASK_PORT_SEND_RECV_WAIT`` (syscall 1), the asynchronous receive post, the
synchronous send/reply, the cross-payload handoff performed when a receive
completes, and the external-interrupt port retirement that a host doorbell wake
performs.

Evidence kind: **firmware use + RTOS port implementation** (the same evidence
class as :mod:`sim.libos`).  Every structure below was observed on the real
570.144 image, not inferred from a stub:

* the message-loop task posts four asynchronous receives through a wrapper at
  firmware vaddr ``0x4018000`` that moves its arguments ``a0-a3`` into
  ``a4-a7``, zeroes ``a0-a3`` and ``t1``/``t2``, and executes ``ecall`` with
  ``t0 = 1`` (an async receive post):
  ``(shuttle 10, port 6, payload 0, size 0)`` immediately followed by
  ``INTERRUPT_ARM_EXTERNAL(0xFF)`` -- the external-interrupt receive;
  ``(shuttle 11, port 7, payload 0x5ba8e10, size 4)``;
  ``(shuttle 9, port 8, payload 0x4015070, size 0x808)``;
  ``(shuttle 14, port 13, payload 0x5ba8e20, size 0x24)``;
* it then performs one synchronous send on the default shuttle id (1) to port
  12 with a zero-length payload, ``waitId = 1``, ``timeout = -1`` (infinite);
* it blocks in ``PortWait(LIBOS_SHUTTLE_ANY, timeout = -1)`` (ecall at
  ``0x4019df4``);
* on an external interrupt the port that was armed retires with
  ``error_code = LIBOS_OK`` and ``payload_size = 0`` (the report path), and the
  wait returns with ``a0 = status``, ``a4 = completedSize``,
  ``a5 = completedRemoteTaskId``, ``a6 = completedShuttle``.

The argument contract modelled (mirrors the RTOS ``libosPortSendRecvAndWait``):

===============  =========================================================
register         meaning at ``ecall``
===============  =========================================================
``t0``           syscall number (1)
``a0``           send shuttle id (0 = none)
``a1``           send port id
``a2``/``a3``    send payload address / size
``a4``           receive shuttle id (0 = none)
``a5``           receive port id
``a6``/``a7``    receive payload address / capacity
``t1``           wait id (0 = do not block, ``0xFFFF`` = any)
``t2``           timeout in ns (``0xFFFFFFFFFFFFFFFF`` = infinite)
===============  =========================================================

On return: ``a0`` = status, ``a4`` = completed size, ``a5`` = originating task
id, ``a6`` = completed shuttle (local handle).

Honesty rules (mandatory; docs/51 section 6, docs/06-ACCURACY.md):

* the model is **data-driven by observation**: port roles come from the posted
  receive sequence of the real image; nothing infers a port number that was
  not observed;
* every modelled event carries ``"modelled": true`` and a label -- the host
  doorbell wake is ``shim:host-doorbell-wake``, the boot handshake send is
  ``modelled:peer-consumes-on-send``.  These are not firmware behaviour;
* **receives are never completed with invented payloads**: a wake only retires
  the external-interrupt receive with the RTOS report contract (rare, zero
  length).  No command, status or event payload is fabricated;
* wait semantics are strictly "return the first completed shuttle or stop";
  the model never returns from a blocking wait that nothing completed;
* L0/L1 only: no timing (L2) and no device-environment (L3) claim; with a
  single modelled task there is no scheduler and no other task is run.

Development-only, offline: no GPU, kernel module, mdev, guest, network, QEMU,
``/dev`` access or reboot.  Nothing here is a product dependency and nothing
is injected into a guest.
"""
from __future__ import annotations

from sim.libos import (
    INTERRUPT_ARM_EXTERNAL,
    MEMORY_QUERY,
    NAMES,
    SHUTTLE_RESET,
    TASK_PORT_SEND_RECV_WAIT,
    TASK_STATE_QUERY,
)

#: Report schema tag for :meth:`LibosPortModel.state`.
REPORT_FORMAT = "greeneye-msgq-port/1"

#: Report schema tag for :meth:`LibosPortNetwork.state`.
NETWORK_REPORT_FORMAT = "greeneye-msgq-port-network/1"

#: The 8-byte "SPURIOUS" sentinel the firmware compares a queried task's
#: ``xepc`` against (the little-endian bytes of the literal, observed in the
#: image rodata and compared as a u64).  A modelled external-interrupt report
#: carries it when no task trapped.
SPURIOUS_XEPC = 0x53554F4952555053

#: Modelled external-interrupt cause (interrupt bit | machine external
#: interrupt code) for the doorbell wake's report state.
EXTERNAL_INTERRUPT_XCAUSE = (1 << 63) | 11

#: ``LibosTaskState`` size (32 registers + xepc + xcause + xbadaddr +
#: nvriscv), the payload of ``TASK_STATE_QUERY``.
TASK_STATE_BYTES = 288

#: ``LIBOS_SHUTTLE_ANY`` (RTOS port API).
SHUTTLE_ANY = 0xFFFF
#: The task's default shuttle id (RTOS port API).
SHUTTLE_DEFAULT = 1
#: ``LIBOS_TIMEOUT_INFINITE``.
TIMEOUT_INFINITE = 0xFFFFFFFFFFFFFFFF
#: LIBOS_OK.
LIBOS_OK = 0

#: Observed port roles for the real 570.144 message-loop task.  The values are
#: the port ids carried by the task's posted receives (resource handles); the
#: labels describe behaviour only.
PORT_ROLES = {
    6: "extintr",      # external-interrupt receive (armed with 0xFF)
    7: "notification", # 4-byte notification buffer
    8: "request",      # 0x808-byte request/reply buffer
    12: "handshake",   # zero-length boot handshake send target
    13: "event",       # 36-byte event buffer
}

#: Port label used when a port was not part of the observed post sequence.
ROLE_DATA = "data"

#: The wake contract observed for the external-interrupt receive: the RTOS
#: report retires the waiting shuttle with error LIBOS_OK and a zero payload.
EXTINTR_WAKE_CONTRACT = "shim:host-doorbell-wake"
HANDSHAKE_CONTRACT = "modelled:peer-consumes-on-send"


def role_for(port: int) -> str:
    """Role label for a port id (observed table; ``"data"`` otherwise)."""
    return PORT_ROLES.get(int(port), ROLE_DATA)


class LibosPortModel:
    """Deterministic model of the libos receive path for one GSP task.

    The model is a ``Machine.syscall_handler``: it owns syscall 1
    (``TASK_PORT_SEND_RECV_WAIT``) and the interrupt-arm call, and delegates
    every other syscall to ``base`` (:class:`sim.libos.LibosSyscalls`, which
    keeps recording).  ``machine.x`` is the only machine state touched.
    """

    name = "libos-port-model"

    def __init__(self, base=None, *, task: str = "gsp-rm-task"):
        self.base = base
        self.task = task
        #: Async receive posts in observation order.
        self.posts: list[dict] = []
        #: Pending receives by shuttle id.
        self.pending: dict[int, dict] = {}
        #: Completed handoffs awaiting a wait (FIFO).
        self.completed: list[dict] = []
        #: Synchronous sends observed (send payloads are not copied; the peer
        #: side is not modelled).
        self.sends: list[dict] = []
        #: Blocking waits observed.
        self.waits: list[dict] = []
        #: Labelled model events (every one is explicitly ``modelled``).
        self.events: list[dict] = []
        #: External interrupt mask last armed (RTOS ``IRQMSET`` semantics).
        #: An installed base model may already have observed an arm before this
        #: model was attached (the recorded boot arm), so seed from it.
        self.armed_mask = int(getattr(base, "armed_mask", 0))
        #: Recorded host doorbell wakes.
        self.host_kicks = 0
        #: Host wakes not yet handed to a receive (the firmware may post and
        #: arm after the host has rung the doorbell; the pending interrupt is
        #: delivered at the first armed receive).
        self.kicks_pending = 0
        #: The request that stopped the run (unmodelled or blocking), if any.
        self.stopped: dict | None = None

    # -- observation -------------------------------------------------------

    def _event(self, name: str, **fields) -> dict:
        event = {"event": name, "modelled": True, "task": self.task}
        event.update(fields)
        self.events.append(event)
        return event

    def host_kick(self, *, label=EXTINTR_WAKE_CONTRACT, queue=None, value=None,
                  bit=None) -> dict:
        """Record the host doorbell write; deliver it when a receive is armed.

        The host submit path writes ``NV_PGSP_QUEUE_HEAD(0)``; the mapping from
        that write to a falcon interrupt bit is **not modelled** (unobserved).
        The wake is delivered to the external-interrupt receive once it has
        been posted and interrupts are armed (the firmware may do both after
        the host has already rung).  Returns the event; ``"delivered"`` is the
        retired shuttle id, or ``None`` when the wake remains pending.
        """
        self.host_kicks += 1
        self.kicks_pending += 1
        delivered = self._try_kick()
        return self._event(
            "host_kick",
            contract=label,
            queue=queue,
            value=value,
            falcon_irq_bit=bit,
            delivered=delivered,
            pending=bool(self.kicks_pending),
            note=("queue-head write -> falcon interrupt bit is unobserved; "
                  "the wake is the documented shim"),
        )

    def _try_kick(self):
        """Deliver one pending host wake to an armed extintr receive, if any."""
        if not self.kicks_pending or not self.armed_mask:
            return None
        for shuttle in sorted(self.pending):
            post = self.pending[shuttle]
            if post["role"] != "extintr":
                continue
            self.kicks_pending -= 1
            self._complete(
                shuttle,
                payload_size=0,
                origin_task=0,
                contract=EXTINTR_WAKE_CONTRACT,
                reason="external-interrupt report (armed mask "
                       f"{self.armed_mask:#x}, pending host doorbell wake)",
            )
            return shuttle
        return None

    def _complete(self, shuttle: int, *, payload_size: int, origin_task: int,
                  contract: str, reason: str) -> dict:
        post = self.pending.pop(int(shuttle))
        entry = {
            "shuttle": int(shuttle),
            "port": int(post["port"]),
            "role": post["role"],
            "payload": int(post["payload"]),
            "capacity": int(post["capacity"]),
            "payload_size": int(payload_size),
            "origin_task": int(origin_task),
            "contract": contract,
            "reason": reason,
            "post_pc": int(post["pc"]),
        }
        self.completed.append(entry)
        self._event("receive_complete", **entry)
        return entry

    # -- syscall surface ---------------------------------------------------

    def handle(self, machine, pc: int = 0):
        """``Machine`` syscall hook: ``(handled, result)``."""
        num = int(machine.x[5])
        if num == TASK_PORT_SEND_RECV_WAIT:
            return self._handle_port(machine, pc)
        if num == INTERRUPT_ARM_EXTERNAL:
            self.armed_mask |= int(machine.x[10])
            self._event("interrupt_arm_external", mask=int(machine.x[10]), pc=pc)
            self._try_kick()
        if self.base is not None:
            handled, result = self.base.handle(machine, pc)
        else:
            handled, result = (True, LIBOS_OK) if num == INTERRUPT_ARM_EXTERNAL \
                else (False, None)
        if not handled and self.stopped is None:
            entry = getattr(self.base, "stopped", None)
            number = int(entry["num"]) if entry else num
            self.stopped = {
                "num": number,
                "name": (entry or {}).get("name", NAMES.get(number, f"unknown_{number}")),
                "pc": int(pc),
                "reason": "unmodelled-syscall",
            }
        return handled, result

    def _handle_port(self, machine, pc: int):
        a0, a1, a2, a3 = (int(machine.x[10]), int(machine.x[11]),
                          int(machine.x[12]), int(machine.x[13]))
        a4, a5, a6, a7 = (int(machine.x[14]), int(machine.x[15]),
                          int(machine.x[16]), int(machine.x[17]))
        t1, t2 = int(machine.x[6]), int(machine.x[7])

        send = a0 != 0 or a1 != 0
        recv = a4 != 0 or a5 != 0
        # A receive with a non-zero wait id is a *synchronous* receive (or the
        # receive half of a sendrecv): the kernel blocks until a message is
        # delivered.  Only a wait id of 0 is an asynchronous post.
        sync_recv = recv and t1 != 0

        if recv:
            self._post(shuttle=a4, port=a5, payload=a6, capacity=a7, pc=pc,
                       wait_id=t1, timeout=t2)
        if send:
            self._send(shuttle=a0, port=a1, payload=a2, size=a3, wait_id=t1,
                       timeout=t2, pc=pc, machine=machine)
            if not recv:
                return True, LIBOS_OK
        if recv and not sync_recv:
            # Async receive post: the kernel returns LIBOS_OK without touching
            # the wait outputs (a4-a6 still hold the receive arguments).
            return True, LIBOS_OK
        if sync_recv:
            if not self.completed:
                self._try_kick()
            if self.completed:
                entry = self.completed.pop(0)
                machine.x[14] = entry["payload_size"] & 0xFFFFFFFFFFFFFFFF
                machine.x[15] = entry["origin_task"] & 0xFFFFFFFFFFFFFFFF
                machine.x[16] = entry["shuttle"] & 0xFFFFFFFFFFFFFFFF
                machine.x[17] = 0
                record = {
                    "pc": int(pc),
                    "wait_id": t1,
                    "timeout": t2,
                    "completed": entry["shuttle"],
                    "payload_size": entry["payload_size"],
                    "origin_task": entry["origin_task"],
                    "role": entry["role"],
                    "contract": entry["contract"],
                }
                self.waits.append(record)
                self._event("sync_recv_complete", **record)
                return True, LIBOS_OK
            record = {
                "pc": int(pc),
                "wait_id": t1,
                "timeout": t2,
                "reason": "sync-recv-blocks",
                "port": int(a5),
                "shuttle": int(a4),
                "pending": sorted(self.pending),
            }
            self.waits.append(record)
            self._event("sync_recv_block", **record)
            if self.stopped is None:
                self.stopped = {
                    "num": TASK_PORT_SEND_RECV_WAIT,
                    "name": "task_port_send_recv_wait",
                    "pc": int(pc),
                    "reason": "sync-recv-blocks",
                    "pending": sorted(self.pending),
                }
            return False, None

        # Pure blocking wait: return the first completed shuttle, or stop.
        if not self.completed:
            self._try_kick()
        if self.completed:
            entry = self.completed.pop(0)
            machine.x[14] = entry["payload_size"] & 0xFFFFFFFFFFFFFFFF
            machine.x[15] = entry["origin_task"] & 0xFFFFFFFFFFFFFFFF
            machine.x[16] = entry["shuttle"] & 0xFFFFFFFFFFFFFFFF
            machine.x[17] = 0
            record = {
                "pc": int(pc),
                "wait_id": t1,
                "timeout": t2,
                "completed": entry["shuttle"],
                "payload_size": entry["payload_size"],
                "origin_task": entry["origin_task"],
                "role": entry["role"],
                "contract": entry["contract"],
            }
            self.waits.append(record)
            self._event("wait_complete", **record)
            return True, LIBOS_OK

        record = {
            "pc": int(pc),
            "wait_id": t1,
            "timeout": t2,
            "reason": "port-wait-blocks",
            "pending": sorted(self.pending),
        }
        self.waits.append(record)
        self._event("wait_block", **record)
        if self.stopped is None:
            self.stopped = {
                "num": TASK_PORT_SEND_RECV_WAIT,
                "name": "task_port_send_recv_wait",
                "pc": int(pc),
                "reason": "port-wait-blocks",
                "pending": sorted(self.pending),
            }
        return False, None

    def _post(self, *, shuttle: int, port: int, payload: int, capacity: int,
              pc: int, wait_id: int, timeout: int) -> dict:
        post = {
            "shuttle": int(shuttle),
            "port": int(port),
            "role": role_for(port),
            "payload": int(payload),
            "capacity": int(capacity),
            "pc": int(pc),
            "wait_id": int(wait_id),
            "timeout": int(timeout),
        }
        self.posts.append(post)
        self.pending[int(shuttle)] = post
        self._event("receive_post", **post)
        self._try_kick()
        return post

    def _send(self, *, shuttle: int, port: int, payload: int, size: int,
              wait_id: int, timeout: int, pc: int, machine) -> dict:
        """Model a synchronous send/reply as completed by its peer.

        Only the boot handshake shape was observed (zero-length payload to port
        12 on the default shuttle).  The peer is not modelled: consumption is
        an explicit, labelled assumption.  The wait outputs receive the RTOS
        transcribe contract for a completed send (``a4`` = copied size,
        ``a5`` = 0, ``a6`` = send shuttle).
        """
        record = {
            "shuttle": int(shuttle),
            "port": int(port),
            "role": role_for(port) if port else "reply",
            "payload": int(payload),
            "size": int(size),
            "pc": int(pc),
            "wait_id": int(wait_id),
            "timeout": int(timeout),
            "contract": HANDSHAKE_CONTRACT,
        }
        self.sends.append(record)
        machine.x[14] = int(size) & 0xFFFFFFFFFFFFFFFF
        machine.x[15] = 0
        machine.x[16] = int(shuttle) & 0xFFFFFFFFFFFFFFFF
        machine.x[17] = 0
        self._event("send_complete", **record)
        return record

    # -- report surface ----------------------------------------------------

    def state(self) -> dict:
        """Deterministic serialisable state for the ``greeneye-msgq-attach/1``
        report (no host timestamps)."""
        return {
            "format": REPORT_FORMAT,
            "task": self.task,
            "armed_mask": int(self.armed_mask),
            "host_kicks": int(self.host_kicks),
            "kicks_pending": int(self.kicks_pending),
            "posts": [dict(post) for post in self.posts],
            "pending": sorted(self.pending),
            "completed": [dict(entry) for entry in self.completed],
            "sends": [dict(record) for record in self.sends],
            "waits": [dict(record) for record in self.waits],
            "events": [dict(event) for event in self.events],
            "stopped": dict(self.stopped) if self.stopped else None,
        }


def task_state_bytes(*, xcause: int, xepc: int, xbadaddr: int = 0,
                     registers=None, priv_err_info: int = 0) -> bytes:
    """Serialise a ``LibosTaskState`` payload (pure; no machine).

    Layout: 32 u64 registers, then ``xepc``, ``xcause``, ``xbadaddr`` and the
    NVRISCV error word (``priv_err_info`` u32 + reserved u32).
    """
    buf = bytearray(TASK_STATE_BYTES)
    for index, value in enumerate(registers or ()):
        if 0 <= index < 32:
            buf[index * 8:index * 8 + 8] = (int(value) & 0xFFFFFFFFFFFFFFFF).to_bytes(8, "little")
    buf[256:264] = (int(xepc) & 0xFFFFFFFFFFFFFFFF).to_bytes(8, "little")
    buf[264:272] = (int(xcause) & 0xFFFFFFFFFFFFFFFF).to_bytes(8, "little")
    buf[272:280] = (int(xbadaddr) & 0xFFFFFFFFFFFFFFFF).to_bytes(8, "little")
    buf[280:284] = (int(priv_err_info) & 0xFFFFFFFF).to_bytes(4, "little")
    return bytes(buf)


class LibosPortNetwork(LibosPortModel):
    """Two-task libos port network (additive, opt-in).

    The real 570.144 GSP image links **two** tasks (an init daemon and the RM
    task); the single-hart linear model only ever runs one of them.  This
    subclass models the libos port machinery between the modelled tasks:

    * receives/posts/completions are kept per task (shuttle ids are
      task-local handles and collide across tasks);
    * a send is routed to a receive waiting on the same port id in another
      task, copying the firmware's own payload bytes (never an invented
      payload); the receiver's completion records the sender's task index and
      pairs a reply back to the sender's receive half (sendrecv);
    * ``TASK_STATE_QUERY`` serves a modelled trap state for the external
      interrupt report (the queried state's ``xcause`` is the interrupt cause
      and ``xepc`` the "SPURIOUS" sentinel the firmware checks; both are
      labelled model values, not firmware observations);
    * ``MEMORY_QUERY`` and ``SHUTTLE_RESET`` are acknowledged from the flat
      RAM model (labelled);
    * ``INTERRUPT_ARM_EXTERNAL`` records the second argument (the port the
      report is bound to) in addition to the mask.

    The single-task :class:`LibosPortModel` behaviour and report schema are
    unchanged; this class is used only when the runner schedules a second
    task.
    """

    name = "libos-port-network"

    def __init__(self, base=None, *, tasks=("init", "rm"), current=None):
        task = current or tasks[0]
        super().__init__(base=base, task=task)
        self.tasks = tuple(tasks)
        self.current = task
        self.task_completions = {name: [] for name in self.tasks}
        self.task_recvs = {name: {} for name in self.tasks}
        self.send_queue = {port: [] for port in ()}
        self.reply_map = {}
        self.cross_task_payloads = []
        self.services = []
        self.queries = []
        self.arm_ports = []
        self.report_trap = None
        self.switches = []
        #: Blocked syscall per task (so a resumed ecall is not re-executed as
        #: a fresh send/post: the kernel would switch back *after* the wait).
        self.blocked = {}

    # -- task identity -----------------------------------------------------

    def switch(self, task: str, pc: int = 0) -> dict:
        """Set the executing task (runner-side cooperative switch)."""
        self.current = task
        self.task = task
        event = self._event("task_switch", pc=int(pc), to=task)
        self.switches.append({"pc": int(pc), "to": task})
        return event

    def _task_index(self, name: str) -> int:
        return self.tasks.index(name) if name in self.tasks else 0

    # -- routing helpers ---------------------------------------------------

    def _find_receiver(self, port: int):
        for name in self.tasks:
            for shuttle in sorted(self.task_recvs[name]):
                post = self.task_recvs[name][shuttle]
                if post["port"] == int(port):
                    return post
        return None

    def _post(self, *, shuttle: int, port: int, payload: int, capacity: int,
              pc: int, wait_id: int, timeout: int) -> dict:
        post = {
            "shuttle": int(shuttle),
            "port": int(port),
            "role": role_for(port),
            "payload": int(payload),
            "capacity": int(capacity),
            "pc": int(pc),
            "wait_id": int(wait_id),
            "timeout": int(timeout),
            "task": self.current,
        }
        self.posts.append(post)
        self.task_recvs[self.current][int(shuttle)] = post
        self._event("receive_post", **post)
        self._try_kick()
        return post

    def _complete(self, shuttle: int, *, payload_size: int, origin_task: int,
                  contract: str, reason: str) -> dict:
        owner = self.current
        post = None
        for name in self.tasks:
            if int(shuttle) in self.task_recvs[name]:
                owner = name
                post = self.task_recvs[name].pop(int(shuttle))
                break
        if post is None:  # pragma: no cover - caller only completes posts
            raise KeyError(f"shuttle {shuttle} is not posted")
        entry = {
            "shuttle": int(shuttle),
            "port": int(post["port"]),
            "role": post["role"],
            "payload": int(post["payload"]),
            "capacity": int(post["capacity"]),
            "payload_size": int(payload_size),
            "origin_task": int(origin_task),
            "contract": contract,
            "reason": reason,
            "post_pc": int(post["pc"]),
            "task": owner,
        }
        self.completed.append(entry)
        self.task_completions[owner].append(entry)
        self._event("receive_complete", **entry)
        return entry

    def _copy(self, machine, dst: int, src: int, size: int) -> bool:
        if size <= 0:
            return True
        try:
            data = machine.read_mem(int(src), int(size))
            machine.write_mem(int(dst), data)
            return True
        except Exception:
            return False

    def _take_completion(self, machine, pc: int):
        queue = self.task_completions[self.current]
        if not queue:
            return None
        entry = queue.pop(0)
        machine.x[14] = entry["payload_size"] & 0xFFFFFFFFFFFFFFFF
        machine.x[15] = entry["origin_task"] & 0xFFFFFFFFFFFFFFFF
        machine.x[16] = entry["shuttle"] & 0xFFFFFFFFFFFFFFFF
        machine.x[17] = 0
        record = {
            "pc": int(pc),
            "completed": entry["shuttle"],
            "payload_size": entry["payload_size"],
            "origin_task": entry["origin_task"],
            "role": entry["role"],
            "contract": entry["contract"],
            "task": self.current,
        }
        self.waits.append(record)
        self._event("wait_complete", **record)
        return record

    def _block(self, machine, pc: int, *, reason: str, port: int,
               shuttle: int) -> None:
        kind = {"port-wait-blocks": "wait", "sync-recv-blocks": "recv",
                "sync-send-blocks": "send"}.get(reason, "wait")
        self.blocked[self.current] = {"kind": kind, "reason": reason,
                                      "pc": int(pc), "port": int(port),
                                      "shuttle": int(shuttle)}
        record = {
            "pc": int(pc),
            "reason": reason,
            "port": int(port),
            "shuttle": int(shuttle),
            "pending": sorted(s for name in self.tasks
                              for s in self.task_recvs[name]),
            "task": self.current,
        }
        self.waits.append(record)
        self._event("wait_block", **record)
        if self.stopped is None:
            self.stopped = {
                "num": TASK_PORT_SEND_RECV_WAIT,
                "name": "task_port_send_recv_wait",
                "pc": int(pc),
                "reason": reason,
                "pending": record["pending"],
                "task": self.current,
            }

    # -- syscall surface ---------------------------------------------------

    def handle(self, machine, pc: int = 0):
        num = int(machine.x[5])
        if num == TASK_PORT_SEND_RECV_WAIT:
            return self._handle_network(machine, pc)
        if num == INTERRUPT_ARM_EXTERNAL:
            mask, port = int(machine.x[10]), int(machine.x[11])
            self.armed_mask |= mask
            self.arm_ports.append({"mask": mask, "port": port, "pc": int(pc),
                                   "task": self.current})
            self._event("interrupt_arm_external", mask=mask, port=port,
                        pc=pc, task=self.current)
            self._try_kick()
            return True, LIBOS_OK
        if num == TASK_STATE_QUERY:
            self.queries.append({"pc": int(pc), "task_index": int(machine.x[10]),
                                 "buffer": int(machine.x[11]),
                                 "task": self.current})
            trap = self.report_trap
            if trap is None:
                # No modelled trap state: the kernel would return an access
                # error for an invalid target; return LIBOS_ERROR_INVALID_ACCESS
                # (labelled model value) and let the firmware error path run.
                return True, 0x80000000
            state = task_state_bytes(
                xcause=trap["xcause"], xepc=trap["xepc"],
                xbadaddr=trap["xbadaddr"], registers=trap.get("registers"),
            )
            try:
                machine.write_mem(int(machine.x[11]), state)
            except Exception:
                return True, 0x80000000
            self._event("task_state_query", **self.queries[-1],
                        xcause=hex(trap["xcause"]), xepc=hex(trap["xepc"]))
            return True, LIBOS_OK
        if num == MEMORY_QUERY:
            remote, pointer = int(machine.x[10]), int(machine.x[11])
            attrs = 0x1 | 0x2 | 0x8  # read | write | cached (flat model)
            resolved = pointer
            translator = getattr(machine, "translator", None)
            if translator is not None:
                sim = translator.translate(pointer)
                if sim is not None:
                    resolved = sim
            machine.x[11] = attrs
            machine.x[12] = pointer & ~0xFFF
            machine.x[13] = 0x1000
            machine.x[14] = 1  # LIBOS_MEMORY_APERTURE_SYSCOH
            machine.x[15] = resolved
            record = {"pc": int(pc), "remote_task": remote, "pointer": pointer,
                      "resolved": resolved, "task": self.current}
            self.services.append(dict(record, service="memory_query"))
            self._event("memory_query", **record)
            return True, LIBOS_OK
        if num == SHUTTLE_RESET:
            record = {"pc": int(pc), "shuttle": int(machine.x[10]),
                      "task": self.current}
            self.services.append(dict(record, service="shuttle_reset"))
            self._event("shuttle_reset", **record)
            return True, LIBOS_OK
        return super().handle(machine, pc)

    def _handle_network(self, machine, pc: int):
        resuming = self.blocked.pop(self.current, None)
        if resuming is not None and resuming["kind"] in ("recv", "wait"):
            record = self._take_completion(machine, pc)
            if record is not None:
                return True, LIBOS_OK
            self._block(machine, pc, reason=resuming["reason"],
                        port=resuming["port"], shuttle=resuming["shuttle"])
            return False, None
        if resuming is not None and resuming["kind"] == "send":
            self._block(machine, pc, reason=resuming["reason"],
                        port=resuming["port"], shuttle=resuming["shuttle"])
            return False, None
        a0, a1, a2, a3 = (int(machine.x[10]), int(machine.x[11]),
                          int(machine.x[12]), int(machine.x[13]))
        a4, a5, a6, a7 = (int(machine.x[14]), int(machine.x[15]),
                          int(machine.x[16]), int(machine.x[17]))
        t1, t2 = int(machine.x[6]), int(machine.x[7])
        send = a0 != 0 or a1 != 0
        recv = a4 != 0 or a5 != 0
        sync_recv = recv and t1 != 0

        if send and a1 == 0:
            # Reply: route through the received shuttle's partner.
            target = self.reply_map.pop((self.current, a0), None)
            if target is not None:
                name, shuttle = target
                post = self.task_recvs[name].get(shuttle)
                if post is not None:
                    size = min(a3, int(post["capacity"]))
                    ok = self._copy(machine, post["payload"], a2, size)
                    entry = self._complete(
                        shuttle, payload_size=size if ok else 0,
                        origin_task=self._task_index(self.current),
                        contract="modelled:cross-task-reply",
                        reason=f"reply from {self.current} ({size} bytes)")
                    self.sends.append({"shuttle": a0, "port": 0, "size": a3,
                                       "pc": int(pc), "task": self.current,
                                       "queued": False, "delivered_to": entry["shuttle"]})
                    self.cross_task_payloads.append({
                        "port": 0, "size": size, "from": self.current,
                        "to": entry.get("task", name), "kind": "reply"})
                    machine.x[14] = size
                    machine.x[15] = 0
                    machine.x[16] = a0
                    machine.x[17] = 0
                    if t1 == a0 or not recv:
                        return True, LIBOS_OK

        if send and a1 != 0:
            partner = self._find_receiver(a1)
            if partner is None:
                size = a3
                self.sends.append({"shuttle": a0, "port": a1, "size": size,
                                   "pc": int(pc), "task": self.current,
                                   "queued": True, "payload": a2})
                return self._unrouted_send(machine, pc, a0, a1, a3, t1,
                                           recv, sync_recv)
            size = min(a3, int(partner["capacity"]))
            ok = self._copy(machine, partner["payload"], a2, size)
            entry = self._complete(
                int(partner["shuttle"]), payload_size=size if ok else 0,
                origin_task=self._task_index(self.current),
                contract="modelled:cross-task-send",
                reason=f"send from {self.current} port {a1} copied {size}")
            self.sends.append({"shuttle": a0, "port": a1, "size": a3,
                               "pc": int(pc), "task": self.current,
                               "queued": False, "delivered_to": entry["shuttle"]})
            self.cross_task_payloads.append({
                "port": int(a1), "size": size, "from": self.current,
                "to": entry.get("task"), "kind": "send"})
            if a4 or a5:
                self.reply_map[(entry.get("task"), entry["shuttle"])] = (self.current, a4)
            if t1 == a0:
                machine.x[14] = size
                machine.x[15] = 0
                machine.x[16] = a0
                machine.x[17] = 0
                return True, LIBOS_OK
            if not recv:
                return True, LIBOS_OK

        if recv:
            self._post(shuttle=a4, port=a5, payload=a6, capacity=a7, pc=pc,
                       wait_id=t1, timeout=t2)
            if sync_recv:
                record = self._take_completion(machine, pc)
                if record is not None:
                    return True, LIBOS_OK
                self._block(machine, pc, reason="sync-recv-blocks", port=a5,
                            shuttle=a4)
                return False, None
            return True, LIBOS_OK

        record = self._take_completion(machine, pc)
        if record is not None:
            return True, LIBOS_OK
        self._block(machine, pc, reason="port-wait-blocks", port=0, shuttle=0)
        return False, None

    def _unrouted_send(self, machine, pc, shuttle, port, size, wait_id,
                       recv, sync_recv):
        """A send with no modelled receiver: keep the boot-shape shim.

        The zero-length boot handshake send had no modelled peer and is an
        explicit, labelled assumption (``modelled:peer-consumes-on-send``);
        every other unrouted send stays queued and blocks its sync sender.
        """
        if size != 0:
            if wait_id == shuttle:
                self._block(machine, pc, reason="sync-send-blocks", port=port,
                            shuttle=shuttle)
                return False, None
            return True, LIBOS_OK
        machine.x[14] = 0
        machine.x[15] = 0
        machine.x[16] = shuttle
        machine.x[17] = 0
        self._event("send_complete", shuttle=shuttle, port=port, size=0,
                    pc=int(pc), task=self.current,
                    contract="modelled:peer-consumes-on-send")
        if wait_id == shuttle or not recv:
            return True, LIBOS_OK
        if sync_recv:
            record = self._take_completion(machine, pc)
            if record is not None:
                return True, LIBOS_OK
            self._block(machine, pc, reason="sync-recv-blocks", port=0,
                        shuttle=0)
            return False, None
        return True, LIBOS_OK

    def _try_kick(self):
        """Deliver a pending host wake to the extintr receive (documented shim).

        The bound interrupt port is recorded from the arm call (the real kernel
        delivers to the armed port); the modelled report retires the waiting
        external-interrupt receive and records the modelled trap state its
        ``TASK_STATE_QUERY`` consumer observes.
        """
        if not self.kicks_pending:
            return None
        for name in self.tasks:
            for shuttle in sorted(self.task_recvs[name]):
                post = self.task_recvs[name][shuttle]
                if post["role"] != "extintr":
                    continue
                self.kicks_pending -= 1
                self.report_trap = {
                    "xcause": EXTERNAL_INTERRUPT_XCAUSE,
                    "xepc": SPURIOUS_XEPC,
                    "xbadaddr": 0,
                    "origin_task": 0,
                }
                self._complete(shuttle, payload_size=0, origin_task=0,
                               contract=EXTINTR_WAKE_CONTRACT,
                               reason="external-interrupt report")
                return shuttle
        return None

    def state(self) -> dict:
        base = super().state()
        base.update({
            "format": NETWORK_REPORT_FORMAT,
            "tasks": list(self.tasks),
            "current": self.current,
            "task_completions": {name: [dict(e) for e in entries]
                                 for name, entries in sorted(self.task_completions.items())},
            "recvs": {name: sorted(self.task_recvs[name])
                      for name in self.tasks},
            "cross_task_payloads": [dict(p) for p in self.cross_task_payloads],
            "services": [dict(s) for s in self.services],
            "queries": [dict(q) for q in self.queries],
            "arm_ports": [dict(a) for a in self.arm_ports],
            "report_trap": dict(self.report_trap) if self.report_trap else None,
            "switches": [dict(s) for s in self.switches],
        })
        return base