"""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, LibosPortNetwork 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 #: Default extra slice cap for the scheduled second task (0 = run it until #: the total instruction budget is reached). The observed RM task completes #: its boot handshake with the init task inside a few thousand instructions; #: its own start-up then runs past the total cap, so the frontier is reported #: at the cap, never as success. DEFAULT_SECOND_TASK_BUDGET = 0 #: Wait reasons that mean "this task is blocked and another task may run". _BLOCK_REASONS = ("port-wait-blocks", "sync-recv-blocks", "sync-send-blocks") #: 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, tasks: dict | None = None, schedule: list | 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, "tasks": dict(tasks) if tasks is not None else {"format": "greeneye-image-tasks/1", "tasks": []}, "schedule": [dict(step) for step in (schedule or [])], "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 image_tasks(image: str) -> dict: """Discover the linked libos tasks from the image's section header table. Pure stdlib ELF64 parsing (no symbols are present in the image): the ``section_task__elf_text_instance`` sections give each task's text range and the ``section_task__stack_instance`` sections its stack; the task entry is the first pointer inside the text range found in the ``.manifest`` section (the firmware's own task record). Data-driven: an image without section names, manifest or match returns an empty table and the caller stays single-task. Returns ``{"format": "greeneye-image-tasks/1", "tasks": [...]}``. """ empty = {"format": "greeneye-image-tasks/1", "tasks": []} try: with open(image, "rb") as handle: blob = handle.read() except OSError: return empty if len(blob) < 64 or blob[:4] != b"\x7fELF" or blob[4] != 2: return empty shoff = int.from_bytes(blob[0x28:0x30], "little") shentsize = int.from_bytes(blob[0x3A:0x3C], "little") shnum = int.from_bytes(blob[0x3C:0x3E], "little") shstrndx = int.from_bytes(blob[0x3E:0x40], "little") if not (0 < shentsize and 0 < shnum and shstrndx < shnum and shoff + shentsize * shnum <= len(blob)): return empty shstr = None raw_sections = [] for index in range(shnum): base = shoff + index * shentsize if base + 64 > len(blob): return empty fields = blob[base:base + 64] name_off = int.from_bytes(fields[0:4], "little") addr = int.from_bytes(fields[16:24], "little") offset = int.from_bytes(fields[24:32], "little") size = int.from_bytes(fields[32:40], "little") raw_sections.append((name_off, addr, offset, size)) _, shstr_addr, shstr_off, shstr_size = raw_sections[shstrndx] shstr = blob[shstr_off:shstr_off + shstr_size] def name_at(name_off): end = shstr.find(b"\x00", name_off) return shstr[name_off:end].decode("latin1") if end >= 0 else "" sections = {} for name_off, addr, offset, size in raw_sections: name = name_at(name_off) if name: sections[name] = {"addr": addr, "offset": offset, "size": size, "name": name} tasks = {} for name, section in sections.items(): if name.startswith(".section_task_") and name.endswith("_elf_text_instance"): task = name[len(".section_task_"):-len("_elf_text_instance")] tasks.setdefault(task, {})["text"] = {"base": section["addr"], "size": section["size"]} elif name.startswith(".section_task_") and name.endswith("_stack_instance"): task = name[len(".section_task_"):-len("_stack_instance")] tasks.setdefault(task, {})["stack"] = { "base": section["addr"], "size": section["size"], "top": section["addr"] + section["size"]} manifest = sections.get(".manifest") if manifest: data = blob[manifest["offset"]:manifest["offset"] + manifest["size"]] for index in range(0, len(data) - 7, 8): value = int.from_bytes(data[index:index + 8], "little") for task, record in tasks.items(): text = record.get("text") if text and "entry" not in record \ and text["base"] <= value < text["base"] + text["size"]: record["entry"] = value ordered = [] for task in sorted(tasks): record = tasks[task] if "text" not in record or "entry" not in record: continue ordered.append({ "name": task, "entry": int(record["entry"]), "text_base": int(record["text"]["base"]), "text_size": int(record["text"]["size"]), "stack_base": int(record.get("stack", {}).get("base", 0)), "stack_size": int(record.get("stack", {}).get("size", 0)), "stack_top": int(record.get("stack", {}).get("top", 0)), }) return {"format": "greeneye-image-tasks/1", "tasks": ordered} def _task_for_pc(tasks_doc: dict, pc: int) -> dict | None: for task in tasks_doc.get("tasks", []): if task["text_base"] <= pc < task["text_base"] + task["text_size"]: return task return None def missing_semantic_after_second_task(*, tasks_doc: dict, schedule: list, frontier: dict, polls: list) -> dict: """The exact unmodelled semantics at the two-task frontier (pure). Recorded when the scheduled second task (the RM task discovered from the image) is blocked or out of its bounded slice without touching the rings. """ first = schedule[0] if schedule else {} second = schedule[-1] if schedule else {} return { "stage": "second-task-frontier", "tasks": [task["name"] for task in tasks_doc.get("tasks", [])], "observed": [ "the host doorbell was delivered as the RTOS external-interrupt " "report and the init task consumed it as a spurious/EINT report " "(modelled trap state, TASK_STATE_QUERY served)", "the init task re-posted its receive and blocked in its port wait", "the second task (the RM task, entry and stack discovered from " "the image section/manifest records) was scheduled and completed " "its boot port handshake with the init task", "no command/status ring or doorbell read occurred within the " "bounded instruction budget; the RM task's own start-up continues " "past the cap before its first further port syscall", ], "missing": [ "the queue consumer's start-up is not reachable inside the 2M " "instruction cap from its entry; a data-driven post-init resume " "point (or a qualified budget) is the smallest next semantic", "the GSP_ARGUMENTS_CACHED / message-queue init handoff: the " "simulator never delivers the queue base addresses the host " "actually gives the firmware, so the rings stay at the fixture " "block and no firmware queue poll can be observed", "the doorbell -> interrupt binding for the consumer task: the " "bound interrupt port recorded from INTERRUPT_ARM_EXTERNAL is " "not the port the RM task receives the queue kick on", ], "next_step": ( "extend the schedule with a data-driven post-init entry for the " "RM task (or raise the cap with an admission record), deliver the " "GSP_ARGUMENTS_CACHED/queue-init addresses from the host model, " "then re-run this attach and classify the first ring poll; do not " "infer a poll from this run"), "schedule": [dict(step) for step in schedule], "frontier": dict(frontier) if frontier else None, "polls": len(polls), "first_task": first.get("task"), "second_task": second.get("task"), } 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, second_task: bool = True, second_task_budget: int = DEFAULT_SECOND_TASK_BUDGET, 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. ``second_task`` (default on) additionally discovers the image's linked tasks (section/manifest records) and schedules the second one (the RM task) when the init task blocks: the boot port handshake is modelled with the firmware's own payloads, bounded by the total instruction budget (an optional tighter ``second_task_budget`` slice cap, 0 = until the total budget). The second task's own start-up runs past the cap, so the report names that frontier instead of manufacturing a poll; the single-task behaviour is kept with ``second_task=False``. """ 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}") if isinstance(second_task_budget, bool) or not isinstance(second_task_budget, int) \ or not 0 <= second_task_budget <= MAX_RUN_CYCLES: raise MsgqAttachError( f"second_task_budget must be an int in 0..{MAX_RUN_CYCLES} " f"(0 = until the total budget), got {second_task_budget!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)) tasks_doc = image_tasks(image) session = None notes: list[str] = [] divergence = None polls: list[dict] = [] schedule: 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 and second_task: # Two-task schedule: the image's second linked task (the RM task) # is data-discovered and seeded when the init task blocks. primary = _task_for_pc(tasks_doc, int(pc)) or (tasks_doc["tasks"][0] if tasks_doc["tasks"] else None) secondary = next((task for task in tasks_doc["tasks"] if task is not primary), None) if primary else None if primary is not None and secondary is not None: model = LibosPortNetwork( base=libos, tasks=(primary["name"], secondary["name"]), current=primary["name"]) 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, ) schedule: list[dict] = [] saved = {primary["name"]: None, secondary["name"]: None} started = {primary["name"]: True, secondary["name"]: False} entries = {primary["name"]: primary, secondary["name"]: secondary} reason2, pc2, a2, b2 = _run_watching(machine, max_cycles, geometry, milestones_hit, polls) for _ in range(8): stop = model.stopped if reason2 == R_ECALL else None if polls or stop is None \ or stop.get("reason") not in _BLOCK_REASONS \ or reason2 == R_POLL_LIMIT: break current = model.current saved[current] = {"registers": list(machine.x), "pc": int(machine.pc)} schedule.append({"task": current, "pc": int(pc2), "reason": stop.get("reason"), "insns": int(machine.insns)}) other = secondary["name"] if current == primary["name"] \ else primary["name"] if not started[other]: if machine.insns >= max_cycles: break started[other] = True record = entries[other] machine.x = [0] * 32 machine.x[2] = int(record.get("stack_top", 0)) machine.pc = int(record["entry"]) model.switch(other, pc=int(machine.pc)) model.stopped = None schedule.append({"task": other, "pc": int(machine.pc), "reason": "scheduled", "insns": int(machine.insns)}) slice_end = max_cycles if not second_task_budget \ else min(max_cycles, machine.insns + int(second_task_budget)) reason2, pc2, a2, b2 = _run_watching( machine, slice_end, geometry, milestones_hit, polls) continue if model.task_completions.get(other): machine.x = list(saved[other]["registers"]) machine.pc = int(saved[other]["pc"]) model.switch(other, pc=int(machine.pc)) model.stopped = None schedule.append({"task": other, "pc": int(machine.pc), "reason": "resumed", "insns": int(machine.insns)}) reason2, pc2, a2, b2 = _run_watching( machine, max_cycles, geometry, milestones_hit, polls) continue break 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, "second_task": 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), "tasks": list(model.tasks), "switches": len(model.switches), } 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 not polls: attack = (port_stopped or {}) missing_semantic = missing_semantic_after_second_task( tasks_doc=tasks_doc, schedule=schedule, frontier=attack, polls=polls) divergence = { "kind": "second-task-frontier", "reason": ( "after the recorded host wake the init task " "consumed the external-interrupt report and " "blocked; the discovered second task (the RM " "task) was scheduled, completed its boot port " "handshake and then blocked without touching " "the rings -- " f"{name2} (num={number2}) at pc={pc2:#x}"), "stop": dict(attack), } elif reason2 == R_BUDGET and not polls: frontier = { "task": model.current, "reason": "second-task-budget" if second_task_budget else "total-budget", "pc": int(pc2), } missing_semantic = missing_semantic_after_second_task( tasks_doc=tasks_doc, schedule=schedule, frontier=frontier, polls=polls) if second_task_budget: divergence = { "kind": "second-task-frontier", "reason": (f"the discovered second task ran its " f"bounded slice ({second_task_budget} " f"instructions) without a ring read " f"(total budget {max_cycles})"), } else: divergence = { "kind": "second-task-frontier", "reason": (f"the discovered second task ran to the " f"total budget of {max_cycles} " f"instructions without a ring read " f"(its start-up continues past the cap)"), } 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 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, tasks=tasks_doc, schedule=schedule, 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("--no-second-task", action="store_true", help="keep the single-task port model (do not schedule " "the image's second linked task)") run_cmd.add_argument("--second-task-budget", type=int, default=DEFAULT_SECOND_TASK_BUDGET, help="extra instruction slice cap for the scheduled " "second task (0 = until the total budget, " f"default {DEFAULT_SECOND_TASK_BUDGET})") 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) second_task = not getattr(args, "no_second_task", 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, second_task=second_task, second_task_budget=args.second_task_budget, 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())