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