summaryrefslogtreecommitdiffstats
path: root/debug/risc-v-gsp-rm-vm/sim
diff options
context:
space:
mode:
authorArthur Hanson Rasmusson <arthur@vgpu.io>2026-10-06 19:28:39 -0400
committerArthur Hanson Rasmusson <arthur@vgpu.io>2026-10-06 19:28:39 -0400
commit6c3fd5df61ff2f5cb8833ead6c05eaff9fdc92bf (patch)
tree3b3680902cd2223cbe60aaec0a0799cfc7b26be6 /debug/risc-v-gsp-rm-vm/sim
parenta5473c91b8869b017385582ea969783b33553ded (diff)
downloadopen-iov-6c3fd5df61ff2f5cb8833ead6c05eaff9fdc92bf.tar.gz
open-iov-6c3fd5df61ff2f5cb8833ead6c05eaff9fdc92bf.zip
risc-v-gsp-rm-debugger: land docs 45-53 + harness/sim tool set with tests
rm_capture (chunked decode), launch_audit, state_seed/state_decode, valprobe, rehost, contracts+run_contracts, host_traffic, gmmu_walk, msgq_attach, libos_port, log_decode, server_snapshot; doc index updates in the tool overview, operations pages and the repo README.
Diffstat (limited to 'debug/risc-v-gsp-rm-vm/sim')
-rw-r--r--debug/risc-v-gsp-rm-vm/sim/gmmu_walk.py488
-rw-r--r--debug/risc-v-gsp-rm-vm/sim/host_traffic.py1437
-rw-r--r--debug/risc-v-gsp-rm-vm/sim/libos_port.py386
-rw-r--r--debug/risc-v-gsp-rm-vm/sim/msgq_attach.py1122
4 files changed, 3433 insertions, 0 deletions
diff --git a/debug/risc-v-gsp-rm-vm/sim/gmmu_walk.py b/debug/risc-v-gsp-rm-vm/sim/gmmu_walk.py
new file mode 100644
index 0000000..50960ab
--- /dev/null
+++ b/debug/risc-v-gsp-rm-vm/sim/gmmu_walk.py
@@ -0,0 +1,488 @@
+#!/usr/bin/env python3
+"""Functional GMMU page-table walker (Turing VER2 geometry).
+
+The decoder in :mod:`sim.gmmu_pte` decodes one captured entry. This module
+follows those entries: given a page-directory base and a sparse physical-memory
+view of the table pages, it walks a virtual address to its leaf mapping and
+reports every level, the page size, the aperture and the access flags -- or the
+exact fault (missing table, invalid aperture, invalid leaf).
+
+Why it exists: the engine sees memory through the GMMU, not through the host's
+bookkeeping. When a forwarded submission goes silent (an engine-side write
+never appears), the first question is *where would the engine read/write this
+address?* A capture of the channel's page tables plus this walker answers it
+offline, deterministically, without hardware.
+
+Evidence and honesty:
+
+* Geometry, field positions and the entry encodings come from the same
+ evidence base as :mod:`sim.gmmu_pte` (see ``docs/43`` and ``docs/48``).
+* The 4K and 2M paths are **validated against a live captured channel PDB**
+ (see :func:`channel_pdb_fixture`); the 64K dual path is modelled from the
+ format definitions and reported with ``evidence: "modeled"`` until a capture
+ validates it.
+* Unsupported encodings fail closed (``E_UNSUPPORTED``); a video-memory leaf
+ address is reported with ``pa_complete: false`` because the decoder cannot
+ claim the high address bits there (never guessed).
+* No timing claim; this is the engine's *address* view (L1), not its behaviour.
+"""
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import struct
+import sys
+
+TOOL_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+if TOOL_ROOT not in sys.path:
+ sys.path.insert(0, TOOL_ROOT)
+
+from sim.gmmu_pte import ( # noqa: E402
+ GmmuDecodeError,
+ KIND_VER2,
+ KIND_VER2_DUAL,
+ LEVELS,
+ SUPPORTED_KINDS,
+ decode_dual_pde,
+ decode_entry,
+ decode_pde,
+ decode_pte,
+)
+
+#: Schema tag of a walk report.
+WALK_FORMAT = "greeneye-gmmu-walk/1"
+
+#: Page modes the walker accepts. ``auto`` prefers the 64K dual half for a
+#: 64K-aligned VA when that half is valid, else the 4K half.
+PAGE_MODES = ("4k", "64k", "auto")
+
+VA_MASK = (1 << 49) - 1
+
+#: Fault codes (never a guess; each names a concrete missing/invalid piece).
+FAULT_TABLE_MISSING = "table_missing"
+FAULT_INVALID_APERTURE = "invalid_aperture"
+FAULT_PTE_INVALID = "pte_invalid"
+FAULT_UNSUPPORTED = "unsupported"
+FAULT_OUT_OF_RANGE = "va_out_of_range"
+
+
+class WalkError(ValueError):
+ """A walk request the evidence cannot answer."""
+
+
+class MemoryView:
+ """Sparse physical memory: ``pa -> bytes`` blobs (table pages, targets)."""
+
+ def __init__(self, blobs: dict[int, bytes] | None = None):
+ self._blobs = dict(blobs or {})
+
+ def add(self, pa: int, data: bytes) -> None:
+ if pa < 0 or not isinstance(data, (bytes, bytearray)):
+ raise WalkError("add() needs a non-negative pa and bytes")
+ self._blobs[pa] = bytes(data)
+
+ def has(self, pa: int) -> bool:
+ return pa in self._blobs
+
+ def read(self, pa: int, size: int) -> bytes | None:
+ for base, blob in self._blobs.items():
+ if base <= pa and pa + size <= base + len(blob):
+ off = pa - base
+ return blob[off:off + size]
+ return None
+
+ def spans(self) -> list[dict]:
+ return [{"pa": f"{pa:#x}", "size": len(b)} for pa, b in sorted(self._blobs.items())]
+
+
+# ---------------------------------------------------------------------------
+# Encoders (fixtures/tests; mirror of the decoder's field positions)
+# ---------------------------------------------------------------------------
+
+def encode_pde_pointer(pa: int, *, volatile: bool = True) -> int:
+ """Encode a VER2 directory PDE: bit0=0, aperture 3 (sysmem non-coherent)."""
+ if pa & 0xFFF:
+ raise WalkError(f"table PA must be 4K aligned, got {pa:#x}")
+ return ((pa >> 12) << 8) | (3 << 1) | (8 if volatile else 0)
+
+
+def encode_pte(pa: int, *, page_size: str = "4k", read_only: bool = False,
+ volatile: bool = True) -> int:
+ """Encode a VER2 leaf PTE: bit0=1, aperture 3 (sysmem non-coherent)."""
+ if page_size == "4k":
+ align = 0xFFF
+ elif page_size == "64k":
+ align = 0xFFFF
+ elif page_size == "2m":
+ align = 0x1F_FFFF
+ else:
+ raise WalkError(f"unknown page_size {page_size!r}")
+ if pa & align:
+ raise WalkError(f"PA {pa:#x} is not {page_size}-aligned")
+ return ((pa >> 12) << 8) | (3 << 1) | 1 | (8 if volatile else 0) | (0x40 if read_only else 0)
+
+
+def encode_dual_slot(big_pt_pa: int | None, small_pt_pa: int) -> bytes:
+ """Encode a VER2 dual PDE0 slot; BIG = 64K table, SMALL = 4K table."""
+ big = 0
+ if big_pt_pa is not None:
+ if big_pt_pa & 0xFFF:
+ raise WalkError(f"big table PA must be 4K aligned, got {big_pt_pa:#x}")
+ big = (((big_pt_pa >> 8) & ~1) << 4) | (3 << 1) | 8
+ small = ((small_pt_pa >> 12) << 8) | (3 << 1) | 8
+ return struct.pack("<QQ", big, small)
+
+
+def _fault(code: str, message: str, steps: list, **extra) -> dict:
+ report = {
+ "format": WALK_FORMAT, "status": "fault", "fault": code,
+ "fault_message": message, "steps": steps,
+ }
+ report.update(extra)
+ return report
+
+
+def walk(
+ va: int,
+ *,
+ root_pa: int,
+ mem: MemoryView,
+ kind: str = KIND_VER2,
+ page_mode: str = "4k",
+) -> dict:
+ """Walk ``va`` through the page tables rooted at ``root_pa``.
+
+ Returns a ``greeneye-gmmu-walk/1`` report. ``status`` is ``ok`` with
+ ``pa``/``page_size``/``writable`` on success, otherwise ``fault`` with a
+ concrete ``fault`` code. Every level's raw entry and decode is kept in
+ ``steps`` so the result is auditable.
+ """
+ if kind not in SUPPORTED_KINDS:
+ return _fault(FAULT_UNSUPPORTED, f"kind {kind!r} not in {SUPPORTED_KINDS}",
+ [], va=va, root_pa=root_pa)
+ if page_mode not in PAGE_MODES:
+ raise WalkError(f"page_mode must be one of {PAGE_MODES}, got {page_mode!r}")
+ if not (0 <= va <= VA_MASK):
+ return _fault(FAULT_OUT_OF_RANGE, f"va {va:#x} outside 49-bit space", [],
+ va=va, root_pa=root_pa)
+
+ steps: list[dict] = []
+ pa_complete = True
+ page_used = page_mode
+ evidence = "validated"
+ cur = root_pa
+
+ for depth in range(5):
+ geo = LEVELS[depth]
+ _hi_bit, lo_bit = geo["va_bits"]
+ idx = (va >> lo_bit) & ((1 << geo["index_bits"]) - 1)
+ table_pa = cur
+ if depth == 3:
+ off = idx * 16
+ raw = mem.read(table_pa + off, 16)
+ if raw is None:
+ return _fault(FAULT_TABLE_MISSING,
+ f"depth-3 dual slot at pa={table_pa + off:#x} not in view",
+ steps, va=va, root_pa=root_pa)
+ value128 = int.from_bytes(raw, "little")
+ low = value128 & 0xFFFF_FFFF_FFFF_FFFF
+ step = {"depth": depth, "level": geo["hw"], "table_pa": f"{table_pa:#x}",
+ "index": idx, "entry_pa": f"{table_pa + off:#x}",
+ "raw": f"{value128:#x}"}
+ if low & 1: # 2 MB leaf in the low half
+ try:
+ pte = decode_entry(low, kind=KIND_VER2, role="pte", level=3)
+ except GmmuDecodeError as exc:
+ return _fault(FAULT_UNSUPPORTED, str(exc), steps + [step],
+ va=va, root_pa=root_pa)
+ step.update({"role": "pte", "page_size": "2m",
+ "aperture_name": pte["aperture_name"],
+ "valid": pte["valid"], "address": pte["address"]})
+ steps.append(step)
+ if not pte["valid"] or pte["address"] is None:
+ return _fault(FAULT_PTE_INVALID,
+ f"2M leaf at {table_pa + off:#x} invalid/unsupported",
+ steps, va=va, root_pa=root_pa)
+ if not pte["address_complete"]:
+ pa_complete = False
+ pa = pte["address"] + (va & 0x1F_FFFF)
+ return _ok(va, pa, "2m", pte, steps, root_pa, kind,
+ page_used="2m", evidence=evidence, pa_complete=pa_complete)
+ try:
+ dual = decode_dual_pde(value128, pde_kind=KIND_VER2_DUAL, level=3)
+ except GmmuDecodeError as exc:
+ return _fault(FAULT_UNSUPPORTED, str(exc), steps + [step],
+ va=va, root_pa=root_pa)
+ chosen = None
+ if page_mode == "64k":
+ chosen, page_used, evidence = dual["big"], "64k", "modeled"
+ elif page_mode == "4k":
+ chosen, page_used = dual["small"], "4k"
+ else: # auto: prefer BIG for a 64K-aligned VA when valid
+ if va % 0x10000 == 0 and dual["big"]["address"] is not None:
+ chosen, page_used, evidence = dual["big"], "64k", "modeled"
+ else:
+ chosen, page_used = dual["small"], "4k"
+ step.update({"role": "dual_pde", "page_mode_used": page_used,
+ "big": dual["big"], "small": dual["small"]})
+ steps.append(step)
+ if chosen["address"] is None:
+ return _fault(FAULT_INVALID_APERTURE,
+ f"dual PDE at {table_pa + off:#x} {page_used} half has "
+ f"aperture {chosen['aperture_name']}", steps,
+ va=va, root_pa=root_pa)
+ if not chosen["address_complete"]:
+ pa_complete = False
+ cur = chosen["address"]
+ continue
+
+ off = idx * geo["entry_size"]
+ if depth == 4:
+ # A 64K-granularity leaf table is indexed by VA bits 24:16.
+ idx = (va >> (16 if page_used == "64k" else 12)) & 0x1FF
+ off = idx * geo["entry_size"]
+ raw = mem.read(table_pa + off, geo["entry_size"])
+ if raw is None:
+ return _fault(FAULT_TABLE_MISSING,
+ f"level {depth} entry at pa={table_pa + off:#x} not in view",
+ steps, va=va, root_pa=root_pa)
+ value = int.from_bytes(raw, "little")
+ step = {"depth": depth, "level": geo["hw"], "table_pa": f"{table_pa:#x}",
+ "index": idx, "entry_pa": f"{table_pa + off:#x}", "raw": f"{value:#x}"}
+ if depth < 3:
+ try:
+ pde = decode_pde(value, pde_kind=KIND_VER2, level=depth)
+ except GmmuDecodeError as exc:
+ return _fault(FAULT_UNSUPPORTED, str(exc), steps + [step],
+ va=va, root_pa=root_pa)
+ step.update({"role": "pde", "aperture_name": pde["aperture_name"],
+ "address": pde["address"]})
+ steps.append(step)
+ if pde["address"] is None:
+ return _fault(FAULT_INVALID_APERTURE,
+ f"level {depth} entry at {table_pa + off:#x} has "
+ f"aperture {pde['aperture_name']}", steps,
+ va=va, root_pa=root_pa)
+ if not pde["address_complete"]:
+ pa_complete = False
+ cur = pde["address"]
+ continue
+
+ # depth 4: leaf page table
+ try:
+ pte = decode_pte(value, pte_kind=KIND_VER2, level=4)
+ except GmmuDecodeError as exc:
+ return _fault(FAULT_UNSUPPORTED, str(exc), steps + [step],
+ va=va, root_pa=root_pa)
+ leaf_mask = 0xFFFF if page_used == "64k" else 0xFFF
+ step.update({"role": "pte", "page_size": page_used,
+ "aperture_name": pte["aperture_name"], "valid": pte["valid"],
+ "address": pte["address"], "writable": pte["writable"]})
+ steps.append(step)
+ if not pte["valid"] or pte["address"] is None:
+ return _fault(FAULT_PTE_INVALID,
+ f"leaf PTE at {table_pa + off:#x} invalid/unsupported "
+ f"(raw {value:#x})", steps, va=va, root_pa=root_pa)
+ if not pte["address_complete"]:
+ pa_complete = False
+ pa = pte["address"] + (va & leaf_mask)
+ return _ok(va, pa, page_used, pte, steps, root_pa, kind,
+ page_used=page_used, evidence=evidence, pa_complete=pa_complete)
+
+ return _fault(FAULT_UNSUPPORTED, "walk exceeded 5 levels", steps,
+ va=va, root_pa=root_pa)
+
+
+def _ok(va, pa, page_size, leaf, steps, root_pa, kind, *, page_used, evidence,
+ pa_complete) -> dict:
+ report = {
+ "format": WALK_FORMAT,
+ "status": "ok",
+ "va": f"{va:#x}",
+ "root_pa": f"{root_pa:#x}",
+ "kind": kind,
+ "page_mode_used": page_used,
+ "page_size": page_size,
+ "evidence": evidence,
+ "pa": f"{pa:#x}",
+ "pa_complete": pa_complete,
+ "aperture_name": leaf.get("aperture_name"),
+ "writable": leaf.get("writable", True),
+ "read_only": leaf.get("read_only", False),
+ "kind_name": leaf.get("kind_name"),
+ "steps": steps,
+ "assumptions": [
+ "4K/2M paths are validated against a captured channel PDB; the 64K dual path is modelled and reported with evidence='modeled'.",
+ "A video-memory leaf reports pa_complete=false (high address bits are not claimed by the decoder).",
+ ],
+ }
+ if page_size == "64k":
+ report["assumptions"].append("64K page-mode selection for auto walks is an assumption (64K-aligned VA prefers the BIG half).")
+ return report
+
+
+def walk_many(vas, *, root_pa: int, mem: MemoryView, kind: str = KIND_VER2,
+ page_mode: str = "4k") -> list[dict]:
+ return [walk(va, root_pa=root_pa, mem=mem, kind=kind, page_mode=page_mode)
+ for va in vas]
+
+
+# ---------------------------------------------------------------------------
+# Captured golden vector: a live channel PDB (numeric only)
+# ---------------------------------------------------------------------------
+
+#: Captured channel-PDB entries from a live run (five VAs of one channel
+#: chunk). Used as the walker's regression vector: the walk must reproduce
+#: the DMA addresses the capture logged for these VAs.
+_CHANNEL_PDB = {
+ "root_pa": 0x12F16E000,
+ "root": {0: 0x0000_0000_12F16F0E},
+ "pd2": {0: 0x0000_0000_11065E0E},
+ "pd1": {0: 0x0000_0000_1106630E},
+ "pd0_small": {0: 0x0000_0000_15908D0E},
+ "pt": {
+ 256: 0x0000_0000_2045D90F,
+ 257: 0x0000_0000_121BD90F,
+ 258: 0x0000_0000_1521150F,
+ 259: 0x0000_0000_15A5F50F,
+ 260: 0x0000_0000_10CF270F,
+ },
+}
+
+_CHANNEL_VAS = {
+ 0x100000: ("gpfifo", 0x2045D9000),
+ 0x101000: ("push-buffer", 0x121BD9000),
+ 0x102000: ("copy-src", 0x152115000),
+ 0x103000: ("copy-dst", 0x15A5F5000),
+ 0x104000: ("semaphore", 0x10CF27000),
+}
+
+
+def channel_pdb_fixture() -> tuple[int, MemoryView]:
+ """Rebuild the captured channel PDB as a walkable memory view."""
+ mem = MemoryView()
+ root = bytearray(0x1000)
+ for idx, value in _CHANNEL_PDB["root"].items():
+ struct.pack_into("<Q", root, idx * 8, value)
+ mem.add(0x12F16E000, bytes(root))
+ for name, base in (("pd2", 0x12F16F000), ("pd1", 0x11065E000)):
+ page = bytearray(0x1000)
+ for idx, value in _CHANNEL_PDB[name].items():
+ struct.pack_into("<Q", page, idx * 8, value)
+ mem.add(base, bytes(page))
+ pd0 = bytearray(0x1000)
+ for idx, small in _CHANNEL_PDB["pd0_small"].items():
+ struct.pack_into("<16s", pd0, idx * 16, encode_dual_slot(None, (small >> 8) << 12))
+ mem.add(0x110663000, bytes(pd0))
+ pt = bytearray(0x1000)
+ for idx, value in _CHANNEL_PDB["pt"].items():
+ struct.pack_into("<Q", pt, idx * 8, value)
+ mem.add(0x15908D000, bytes(pt))
+ return _CHANNEL_PDB["root_pa"], mem
+
+
+def run_demo(out: str | None = None) -> dict:
+ """Walk the captured channel-PDB vector; PASS iff every VA matches."""
+ root_pa, mem = channel_pdb_fixture()
+ rows = []
+ ok = True
+ for va, (label, expected) in sorted(_CHANNEL_VAS.items()):
+ report = walk(va, root_pa=root_pa, mem=mem)
+ got = int(report["pa"], 16) if report["status"] == "ok" else None
+ match = got == expected
+ ok = ok and match
+ rows.append({"label": label, "va": f"{va:#x}", "expected_pa": f"{expected:#x}",
+ "walk": report, "match": match})
+ demo = {"format": "greeneye-gmmu-walk-demo/1", "ok": ok, "rows": rows,
+ "root_pa": f"{root_pa:#x}"}
+ if out:
+ os.makedirs(out, exist_ok=True)
+ path = os.path.join(out, "gmmu-walk-demo.json")
+ with open(path, "w", encoding="utf-8") as fh:
+ json.dump(demo, fh, indent=2, sort_keys=True)
+ fh.write("\n")
+ demo["artifacts"] = {"demo_json": os.path.abspath(path)}
+ return demo
+
+
+def _parse_int(text: str) -> int:
+ try:
+ return int(text, 0)
+ except ValueError as exc:
+ raise WalkError(f"bad integer {text!r}") from exc
+
+
+def _parse_mem(spec: str) -> tuple[int, bytes]:
+ if "=" not in spec:
+ raise WalkError(f"--mem needs <pa>=<hex>: {spec!r}")
+ pa_text, _, hex_text = spec.partition("=")
+ data = "".join(hex_text.split())
+ if len(data) % 2:
+ raise WalkError(f"odd hex length in {spec!r}")
+ return _parse_int(pa_text), bytes.fromhex(data)
+
+
+def build_parser() -> argparse.ArgumentParser:
+ parser = argparse.ArgumentParser(
+ prog="gmmu_walk.py",
+ description="Walk a GMMU VA through captured/constructed page tables (offline).",
+ )
+ sub = parser.add_subparsers(dest="command")
+ run = sub.add_parser("run", help="walk VAs through a page-table view")
+ run.add_argument("--pdb-pa", required=True, help="page-directory base PA")
+ run.add_argument("--va", action="append", required=True, help="virtual address (repeatable)")
+ run.add_argument("--mem", action="append", default=[], help="<pa>=<hex> table page (repeatable)")
+ run.add_argument("--kind", default=KIND_VER2, choices=list(SUPPORTED_KINDS))
+ run.add_argument("--page-mode", default="4k", choices=list(PAGE_MODES))
+ run.add_argument("--out", help="write the JSON report here")
+ demo = sub.add_parser("demo", help="walk the captured channel-PDB vector")
+ demo.add_argument("--out", help="output directory")
+ return parser
+
+
+def main(argv: list[str] | None = None) -> int:
+ parser = build_parser()
+ args = parser.parse_args(argv)
+ try:
+ if args.command == "run":
+ mem = MemoryView()
+ for spec in args.mem:
+ pa, data = _parse_mem(spec)
+ mem.add(pa, data)
+ reports = [walk(_parse_int(va), root_pa=_parse_int(args.pdb_pa), mem=mem,
+ kind=args.kind, page_mode=args.page_mode) for va in args.va]
+ doc = {"format": "greeneye-gmmu-walk-set/1", "reports": reports}
+ for report in reports:
+ if report["status"] == "ok":
+ print(f"OK {report['va']} -> {report['pa']} "
+ f"{report['page_size']} {report['aperture_name']} "
+ f"writable={report['writable']}")
+ else:
+ print(f"FAULT {report['va']} {report['fault']}: {report['fault_message']}")
+ if args.out:
+ os.makedirs(args.out, exist_ok=True)
+ path = os.path.join(args.out, "gmmu-walk.json")
+ with open(path, "w", encoding="utf-8") as fh:
+ json.dump(doc, fh, indent=2, sort_keys=True)
+ fh.write("\n")
+ print(json.dumps({"report_json": os.path.abspath(path)}, indent=2))
+ return 0 if all(r["status"] == "ok" for r in reports) else 3
+ if args.command == "demo":
+ demo = run_demo(args.out)
+ for row in demo["rows"]:
+ report = row["walk"]
+ got = report.get("pa", report.get("fault_message"))
+ print(f"{'PASS' if row['match'] else 'FAIL'} {row['label']:<12} "
+ f"{row['va']} -> {got} (expected {row['expected_pa']})")
+ print(json.dumps({"ok": demo["ok"]}, indent=2))
+ return 0 if demo["ok"] else 3
+ parser.print_help()
+ return 2
+ except WalkError as exc:
+ print(f"error: {exc}", file=sys.stderr)
+ return 2
+
+
+if __name__ == "__main__":
+ raise SystemExit(main())
diff --git a/debug/risc-v-gsp-rm-vm/sim/host_traffic.py b/debug/risc-v-gsp-rm-vm/sim/host_traffic.py
new file mode 100644
index 0000000..a9dd55d
--- /dev/null
+++ b/debug/risc-v-gsp-rm-vm/sim/host_traffic.py
@@ -0,0 +1,1437 @@
+"""Host-traffic message loop: doorbell + GSP msgq transport + replay (B4/B5).
+
+Status: **stub wave (in development)**. This module lands the transport half
+of the B4/B5 rung (docs/35-GREENEYE-SIMULATOR-DESIGN.md §3.3, docs/45,
+handoff-simulator.md §31 item B1): a modelled host doorbell, a command/status
+message-queue pair laid out with the real queue arithmetic, queue-element
+framing with the real checksum, a small stub dispatcher, and deterministic
+replay of an ``rm_capture`` stimulus. The real 570.144 firmware is **not**
+run by this module: no real message loop, no RM semantics, no firmware claim.
+
+Real structures mirrored here (provenance: public OpenRM headers and the
+header-derived layouts already used by the project's boot code; every constant
+is checked against ``harness/rm_capture.py`` where the two overlap):
+
+* ``NV_PGSP_QUEUE_HEAD(i) = 0x110c00 + i * 8`` (public Turing ``dev_gsp.h``),
+ 32-bit RW; the 570.144 host writes it after submitting a command
+ (``kgspSetCmdQueueHead``), and the command data itself lives in the sysmem
+ message queues. The simulator reaches the register at ``PRIV+0x110c00``.
+* ``msgqTxHeader`` (8 x u32: version/size/msgSize/msgCount/writePtr/flags/
+ rxHdrOff/entryOff) and ``msgqRxHeader`` (readPtr), with
+ ``rxHdrOff = align32(16) = 32`` and ``entryOff = align36(16) = 48`` for the
+ GSP header alignment of 4 (``msgqTxCreate``).
+* ``GSP_MSG_QUEUE_ELEMENT``: 16 B authTag + 16 B aad + checksum/seqNum/
+ elemCount/reserved (16 B) + the 32 B ``rpc_message_header_v`` (so the RPC
+ payload starts at 0x50). Checksum = XOR of 64-bit words, hi32 ^ lo32; the
+ stored field makes a receiver's recomputation zero (verified).
+* The real command queue is created with ``MSGQ_FLAGS_SWAP_RX``; the two rx
+ headers are cross-linked between the command and status buffers. That
+ cross-link is modelled, not simplified away.
+* The RPC call shape is the one ``harness/rm_capture.py`` decodes
+ (``rpc_message_header_v`` = 32 bytes, GSP_RM_CONTROL = 76, GSP_RM_ALLOC =
+ 103); the demo's 512-byte control is real-shaped but comes from no capture.
+
+What is stubbed: the dispatcher answers only ``RPC_FN_ECHO`` /
+``RPC_FN_STATUS_QUERY`` (stub-only function ids) and returns
+``NV_ERR_NOT_SUPPORTED`` for every real RM function. That reply is framed and
+transported exactly, but it is a stub verdict, never firmware behaviour.
+
+Determinism: the transport transcript carries no host timestamps and is
+byte-identical across runs; only the optional ``greeneye-session/1`` recorder
+events carry host ``t`` (documented as non-semantic by ``compare``). Every
+run is bounded: a message stages contiguously and splits across at most 16
+ring slots (the wire element maximum; larger messages are refused), ``poll``
+drains only the ring, ``run_cycles`` is capped. No hardware, module, mdev,
+guest, network or QEMU is touched.
+"""
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import struct
+import sys
+from dataclasses import dataclass
+from typing import Any, Callable
+
+from .devices.base import Device
+
+_TOOL_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+
+#: GSP doorbell register block (public Turing ``dev_gsp.h``, block-relative).
+QUEUE_HEAD_OFFSET = 0x110C00
+QUEUE_HEAD_STRIDE = 8
+QUEUE_HEAD_COUNT = 8
+#: PRIV aperture base used by the GSP profiles; doorbell window = PRIV + offset.
+PRIV_BASE = 0x2000000000000000
+DOORBELL_BASE = PRIV_BASE + QUEUE_HEAD_OFFSET
+DOORBELL_SIZE = QUEUE_HEAD_STRIDE * QUEUE_HEAD_COUNT
+#: Host writes the queue head to this register as the kick (observed 570.144
+#: submit path writes 0); the model records the value per queue index.
+DOORBELL_KICK_QUEUE = 0
+DOORBELL_KICK_VALUE = 0
+
+#: ``msgq_priv.h``.
+MSGQ_VERSION = 0
+MSGQ_FLAGS_SWAP_RX = 1
+MSGQ_TX_HEADER_BYTES = 32
+MSGQ_RX_HEADER_BYTES = 4
+MSGQ_HEADER_ALIGN = 4
+MSGQ_RX_HDR_OFF = 32
+MSGQ_ENTRY_OFF = 48
+MSGQ_TX_HEADER_FORMAT = "<8I"
+MSGQ_RX_HEADER_FORMAT = "<I"
+
+#: ``message_queue_priv.h``.
+GSP_MSG_QUEUE_ELEMENT_SIZE_MIN = 0x1000
+GSP_MSG_QUEUE_ELEMENT_SIZE_MAX = 0x10000
+GSP_MSG_QUEUE_HEADER_SIZE = 0x1000
+GSP_MSG_QUEUE_ELEMENT_HDR_SIZE = 0x30
+
+#: ``g_rpc-message-header.h`` / 570.144 wire constants.
+RPC_HEADER_BYTES = 32
+RPC_HEADER_FORMAT = "<8I"
+RPC_SIGNATURE = 0x43505256
+RPC_HEADER_VERSION = 0x03000000
+
+#: Default shared sysmem queue block and sizes (host ``message_queue_cpu.c``).
+DEFAULT_CMD_QUEUE_SIZE = 0x40000
+DEFAULT_STAT_QUEUE_SIZE = 0x40000
+SHARED_MEM_BASE = 0x81000000
+
+#: Real RM wire function ids used by the capture fixtures.
+GSP_RM_CONTROL = 76
+GSP_RM_ALLOC = 103
+GSP_GET_GSP_STATIC_INFO = 65
+
+#: Stub-only dispatcher ids (deliberately not RM function numbers).
+RPC_FN_ECHO = 0x40
+RPC_FN_STATUS_QUERY = 0x41
+
+#: ``nvstatuscodes.h``.
+NV_OK = 0
+NV_ERR_NOT_SUPPORTED = 0x56
+
+#: Bounded ``run_cycles`` cap (same ceiling as the boot runner's attempts).
+MAX_RUN_CYCLES = 2_000_000
+
+STATUS_MAGIC = b"GSTS"
+STATUS_VERSION = 1
+STATUS_FORMAT = "<4sIIII"
+
+DEMO_FORMAT = "greeneye-host-traffic-demo/1"
+REPLAY_FORMAT = "greeneye-host-traffic-replay/1"
+TRANSCRIPT_FORMAT = "greeneye-host-traffic-log/1"
+REPLY_FORMAT = "greeneye-host-traffic-reply/1"
+STIMULUS_FORMAT = "greeneye-rm-rpc-stimulus/1"
+CAPTURE_FORMAT = "greeneye-rm-rpc-capture/1"
+
+_TX_HEADER = struct.Struct(MSGQ_TX_HEADER_FORMAT)
+_RX_HEADER = struct.Struct(MSGQ_RX_HEADER_FORMAT)
+_RPC_HEADER = struct.Struct(RPC_HEADER_FORMAT)
+_ELEMENT_PREFIX = struct.Struct("<16s16sIIII")
+
+_LIMITS = [
+ "Stub wave: no real firmware message loop is run; the dispatcher handlers "
+ "are stubs and real RM functions get NV_ERR_NOT_SUPPORTED.",
+ "Multi-element messages are split across ring slots up to the wire element "
+ "maximum (0x10000 bytes); larger messages are refused, not truncated.",
+ "The shared-memory address and queue sizes are fixture defaults, not "
+ "values read from a real GSP_ARGUMENTS_CACHED transfer.",
+ "The MSI-X/event path by which the real GSP signals a status-queue write "
+ "is not modelled; replies are read by polling.",
+ "L0/L1 framing and ordering only; no timing (L2) and no device-environment "
+ "(L3) claim.",
+]
+
+
+class HostTrafficError(Exception):
+ """A framing, queue or replay input that cannot be represented."""
+
+
+def _align_up(value: int, align: int) -> int:
+ return (value + align - 1) // align * align
+
+
+# ---------------------------------------------------------------------------
+# Framing
+# ---------------------------------------------------------------------------
+
+
+@dataclass
+class RpcMessage:
+ """One decoded ``rpc_message_header_v`` plus its payload."""
+
+ header_version: int
+ signature: int
+ length: int
+ function: int
+ rpc_result: int
+ rpc_result_private: int
+ sequence: int
+ spare: int
+ params: bytes = b""
+
+ def to_bytes(self) -> bytes:
+ return _RPC_HEADER.pack(
+ self.header_version,
+ self.signature,
+ RPC_HEADER_BYTES + len(self.params),
+ self.function,
+ self.rpc_result,
+ self.rpc_result_private,
+ self.sequence,
+ self.spare,
+ ) + self.params
+
+
+@dataclass
+class RpcRequest:
+ """A command message handed to a dispatcher handler."""
+
+ function: int
+ sequence: int
+ params: bytes
+ element_index: int
+ raw: bytes
+
+
+@dataclass
+class RpcReply:
+ """A handler result before framing into a status element."""
+
+ function: int
+ sequence: int
+ params: bytes = b""
+ result: int = NV_OK
+
+
+def build_rpc_message(
+ function: int,
+ params: bytes = b"",
+ sequence: int = 0,
+ *,
+ result: int = NV_OK,
+ result_private: int = 0,
+ header_version: int = RPC_HEADER_VERSION,
+ signature: int = RPC_SIGNATURE,
+) -> bytes:
+ """Frame one ``rpc_message_header_v`` + payload (capture-tool shape)."""
+ payload = bytes(params)
+ return _RPC_HEADER.pack(
+ header_version,
+ signature,
+ RPC_HEADER_BYTES + len(payload),
+ int(function) & 0xFFFFFFFF,
+ int(result) & 0xFFFFFFFF,
+ int(result_private) & 0xFFFFFFFF,
+ int(sequence) & 0xFFFFFFFF,
+ 0,
+ ) + payload
+
+
+def parse_rpc_message(data: bytes) -> RpcMessage:
+ """Decode and validate one RPC message; raises on a bad frame."""
+ if len(data) < RPC_HEADER_BYTES:
+ raise HostTrafficError(
+ f"rpc message shorter than {RPC_HEADER_BYTES} bytes: {len(data)}"
+ )
+ (
+ header_version,
+ signature,
+ length,
+ function,
+ rpc_result,
+ rpc_result_private,
+ sequence,
+ spare,
+ ) = _RPC_HEADER.unpack_from(data, 0)
+ if signature != RPC_SIGNATURE:
+ raise HostTrafficError(
+ f"bad rpc signature {signature:#010x} (expected {RPC_SIGNATURE:#010x})"
+ )
+ if length < RPC_HEADER_BYTES or length > len(data):
+ raise HostTrafficError(
+ f"rpc length {length:#x} outside [{RPC_HEADER_BYTES:#x}, {len(data):#x}]"
+ )
+ return RpcMessage(
+ header_version=header_version,
+ signature=signature,
+ length=length,
+ function=function,
+ rpc_result=rpc_result,
+ rpc_result_private=rpc_result_private,
+ sequence=sequence,
+ spare=spare,
+ params=bytes(data[RPC_HEADER_BYTES:length]),
+ )
+
+
+def compute_message_checksum(data: bytes) -> int:
+ """The msgq checksum: XOR of 64-bit words, then hi32 ^ lo32.
+
+ Zero-pads to 8 bytes exactly like ``_checkSum32``; a correctly built queue
+ element recomputes to 0.
+ """
+ padded = bytes(data) + bytes((-len(data)) % 8)
+ total = 0
+ for offset in range(0, len(padded), 8):
+ total ^= int.from_bytes(padded[offset : offset + 8], "little")
+ return ((total >> 32) ^ (total & 0xFFFFFFFF)) & 0xFFFFFFFF
+
+
+def build_queue_element(
+ rpc_bytes: bytes, *, sequence: int = 0, elem_count: int | None = None
+) -> bytes:
+ """Wrap an RPC message in a ``GSP_MSG_QUEUE_ELEMENT`` with checksum."""
+ payload = bytes(rpc_bytes)
+ if elem_count is None:
+ elem_count = max(1, -(-(GSP_MSG_QUEUE_ELEMENT_HDR_SIZE + len(payload))
+ // GSP_MSG_QUEUE_ELEMENT_SIZE_MIN))
+ prefix = _ELEMENT_PREFIX.pack(b"\x00" * 16, b"\x00" * 16, 0,
+ sequence & 0xFFFFFFFF,
+ elem_count & 0xFFFFFFFF, 0)
+ body = prefix + payload
+ checksum = compute_message_checksum(body)
+ return _ELEMENT_PREFIX.pack(b"\x00" * 16, b"\x00" * 16, checksum,
+ sequence & 0xFFFFFFFF,
+ elem_count & 0xFFFFFFFF, 0) + payload
+
+
+def split_element_stream(stream: bytes, msg_size: int) -> list[bytes]:
+ """Split a contiguous element stream into ``msg_size`` ring slots.
+
+ The real producer stages the whole message in one buffer and copies
+ ``GSP_MSG_QUEUE_ELEMENT_SIZE_MIN`` bytes per ring slot
+ (``GspMsgQueueSendCommand``); the tail slot is zero-padded.
+ """
+ blob = bytes(stream)
+ if not blob:
+ raise HostTrafficError("empty element stream")
+ slots = []
+ for start in range(0, len(blob), msg_size):
+ piece = blob[start : start + msg_size]
+ slots.append(piece + bytes(msg_size - len(piece)))
+ return slots
+
+
+def parse_queue_element(data: bytes, *, verify: bool = True) -> tuple[dict, RpcMessage]:
+ """Decode one queue element stream; raises on a bad frame or checksum."""
+ if len(data) < GSP_MSG_QUEUE_ELEMENT_HDR_SIZE + RPC_HEADER_BYTES:
+ raise HostTrafficError(f"queue element too short: {len(data)} bytes")
+ auth_tag, aad, checksum, seq_num, elem_count, reserved = _ELEMENT_PREFIX.unpack_from(data, 0)
+ if not 1 <= elem_count <= GSP_MSG_QUEUE_ELEMENT_SIZE_MAX // GSP_MSG_QUEUE_ELEMENT_SIZE_MIN:
+ raise HostTrafficError(f"implausible elem_count {elem_count}")
+ rpc = parse_rpc_message(data[GSP_MSG_QUEUE_ELEMENT_HDR_SIZE:])
+ total = GSP_MSG_QUEUE_ELEMENT_HDR_SIZE + rpc.length
+ if total > len(data):
+ raise HostTrafficError(f"queue element truncated: need {total:#x}, have {len(data):#x}")
+ if verify and compute_message_checksum(bytes(data[:total])) != 0:
+ raise HostTrafficError("bad message checksum")
+ prefix = {
+ "auth_tag": auth_tag,
+ "aad": aad,
+ "checksum": checksum,
+ "seq_num": seq_num,
+ "elem_count": elem_count,
+ "reserved": reserved,
+ }
+ return prefix, rpc
+
+
+# ---------------------------------------------------------------------------
+# Memory adapters
+# ---------------------------------------------------------------------------
+
+
+class RegionMemory:
+ """A ``base``/``size`` bytearray window with Machine-like read/write."""
+
+ def __init__(self, base: int, size: int):
+ if size <= 0:
+ raise HostTrafficError(f"region size must be positive, got {size}")
+ self.base = int(base)
+ self.size = int(size)
+ self.data = bytearray(self.size)
+
+ def _check(self, addr: int, size: int) -> int:
+ offset = addr - self.base
+ if offset < 0 or offset + size > self.size:
+ raise HostTrafficError(
+ f"access outside region: addr={addr:#x} size={size}"
+ )
+ return offset
+
+ def read(self, addr: int, size: int) -> bytes:
+ offset = self._check(addr, size)
+ return bytes(self.data[offset : offset + size])
+
+ def write(self, addr: int, data: bytes) -> None:
+ blob = bytes(data)
+ offset = self._check(addr, len(blob))
+ self.data[offset : offset + len(blob)] = blob
+
+
+class _MachineMemory:
+ """Adapter from RegionMemory's surface to ``sim.machine.Machine``."""
+
+ def __init__(self, machine):
+ self.machine = machine
+
+ def read(self, addr: int, size: int) -> bytes:
+ return self.machine.read_mem(addr, size)
+
+ def write(self, addr: int, data: bytes) -> None:
+ self.machine.write_mem(addr, data)
+
+
+def _make_memory(machine, shared_base: int, size: int):
+ if machine is None:
+ return RegionMemory(shared_base, size)
+ return _MachineMemory(machine)
+
+
+# ---------------------------------------------------------------------------
+# Message queues
+# ---------------------------------------------------------------------------
+
+
+@dataclass
+class MsgqTxHeader:
+ """``msgqTxHeader``: 8 x u32 written by the queue producer."""
+
+ version: int
+ size: int
+ msg_size: int
+ msg_count: int
+ write_ptr: int
+ flags: int
+ rx_hdr_off: int
+ entry_off: int
+
+ def to_bytes(self) -> bytes:
+ return _TX_HEADER.pack(
+ self.version,
+ self.size,
+ self.msg_size,
+ self.msg_count,
+ self.write_ptr,
+ self.flags,
+ self.rx_hdr_off,
+ self.entry_off,
+ )
+
+ @classmethod
+ def from_bytes(cls, data: bytes) -> "MsgqTxHeader":
+ return cls(*_TX_HEADER.unpack_from(data, 0))
+
+
+class MsgqQueue:
+ """One msgq buffer: tx header at its base, entries at ``entryOff``.
+
+ ``rx_hdr_addr`` is the physical address of this queue's rx header. The
+ real command queue keeps it in the status buffer (``MSGQ_FLAGS_SWAP_RX``),
+ so the caller passes the cross-linked address.
+ """
+
+ def __init__(
+ self,
+ mem,
+ base: int,
+ size: int,
+ *,
+ msg_size: int = GSP_MSG_QUEUE_ELEMENT_SIZE_MIN,
+ header_align: int = MSGQ_HEADER_ALIGN,
+ flags: int = MSGQ_FLAGS_SWAP_RX,
+ rx_hdr_addr: int | None = None,
+ version: int = MSGQ_VERSION,
+ ):
+ if size < GSP_MSG_QUEUE_ELEMENT_SIZE_MIN:
+ raise HostTrafficError(f"queue size {size:#x} below one element")
+ self.mem = mem
+ self.tx_base = int(base)
+ self.size = int(size)
+ self.msg_size = int(msg_size)
+ self.header_align = int(header_align)
+ self.flags = int(flags)
+ self.version = int(version)
+ self.rx_hdr_addr = int(rx_hdr_addr) if rx_hdr_addr is not None else base + MSGQ_RX_HDR_OFF
+
+ def format(self) -> MsgqTxHeader:
+ """Write a fresh tx header and zero the rx header (msgqTxCreate)."""
+ rx_hdr_off = _align_up(MSGQ_TX_HEADER_BYTES, 1 << self.header_align)
+ entry_off = _align_up(rx_hdr_off + MSGQ_RX_HEADER_BYTES, 1 << self.header_align)
+ header = MsgqTxHeader(
+ version=self.version,
+ size=self.size,
+ msg_size=self.msg_size,
+ msg_count=(self.size - entry_off) // self.msg_size,
+ write_ptr=0,
+ flags=self.flags,
+ rx_hdr_off=rx_hdr_off,
+ entry_off=entry_off,
+ )
+ self.mem.write(self.tx_base, header.to_bytes())
+ self.mem.write(self.rx_hdr_addr, _RX_HEADER.pack(0))
+ return header
+
+ def header(self) -> MsgqTxHeader:
+ return MsgqTxHeader.from_bytes(self.mem.read(self.tx_base, MSGQ_TX_HEADER_BYTES))
+
+ def _write_header(self, header: MsgqTxHeader) -> None:
+ self.mem.write(self.tx_base, header.to_bytes())
+
+ @property
+ def write_ptr(self) -> int:
+ return self.header().write_ptr
+
+ @property
+ def read_ptr(self) -> int:
+ return _RX_HEADER.unpack_from(self.mem.read(self.rx_hdr_addr, MSGQ_RX_HEADER_BYTES), 0)[0]
+
+ def _entry_addr(self, index: int) -> int:
+ count = self.header().msg_count
+ if not 0 <= index < count:
+ raise HostTrafficError(f"entry index {index} outside 0..{count - 1}")
+ return self.tx_base + self.header().entry_off + index * self.msg_size
+
+ def write_entry(self, index: int, data: bytes) -> None:
+ blob = bytes(data)
+ if len(blob) > self.msg_size:
+ raise HostTrafficError(
+ f"entry of {len(blob)} bytes exceeds msgSize {self.msg_size:#x}"
+ )
+ self.mem.write(self._entry_addr(index), blob)
+
+ def read_entry(self, index: int, size: int | None = None) -> bytes:
+ return self.mem.read(self._entry_addr(index), self.msg_size if size is None else size)
+
+ def publish(self, write_ptr: int) -> int:
+ header = self.header()
+ header.write_ptr = int(write_ptr) % header.msg_count
+ self._write_header(header)
+ return header.write_ptr
+
+ def set_read(self, read_ptr: int) -> int:
+ count = self.header().msg_count
+ value = int(read_ptr) % count
+ self.mem.write(self.rx_hdr_addr, _RX_HEADER.pack(value))
+ return value
+
+ def state(self) -> dict:
+ header = self.header()
+ return {
+ "tx_base": self.tx_base,
+ "size": self.size,
+ "msg_size": header.msg_size,
+ "msg_count": header.msg_count,
+ "write_ptr": header.write_ptr,
+ "read_ptr": self.read_ptr,
+ "flags": header.flags,
+ "rx_hdr_off": header.rx_hdr_off,
+ "entry_off": header.entry_off,
+ "rx_hdr_addr": self.rx_hdr_addr,
+ }
+
+
+class SharedQueues:
+ """A command/status msgq pair with the real cross-linked rx headers.
+
+ Layout follows the host's ``GspMsgQueuesInit``: the status buffer starts
+ ``cmd_size`` bytes after the command buffer, both tx headers carry
+ ``MSGQ_FLAGS_SWAP_RX``, and each queue's rx header physically lives in the
+ other buffer (``msgqRxLink``).
+ """
+
+ def __init__(
+ self,
+ mem,
+ *,
+ cmd_base: int = SHARED_MEM_BASE,
+ stat_base: int | None = None,
+ cmd_size: int = DEFAULT_CMD_QUEUE_SIZE,
+ stat_size: int = DEFAULT_STAT_QUEUE_SIZE,
+ msg_size: int = GSP_MSG_QUEUE_ELEMENT_SIZE_MIN,
+ swap_rx: bool = True,
+ ):
+ self.mem = mem
+ self.cmd_base = int(cmd_base)
+ self.stat_base = int(stat_base) if stat_base is not None else self.cmd_base + cmd_size
+ flags = MSGQ_FLAGS_SWAP_RX if swap_rx else 0
+ self.command = MsgqQueue(
+ mem,
+ self.cmd_base,
+ cmd_size,
+ msg_size=msg_size,
+ flags=flags,
+ rx_hdr_addr=self.stat_base + MSGQ_RX_HDR_OFF,
+ )
+ self.status = MsgqQueue(
+ mem,
+ self.stat_base,
+ stat_size,
+ msg_size=msg_size,
+ flags=flags,
+ rx_hdr_addr=self.cmd_base + MSGQ_RX_HDR_OFF,
+ )
+
+ def format(self) -> None:
+ self.command.format()
+ self.status.format()
+
+ def _place(self, queue: MsgqQueue, element: bytes) -> dict:
+ slots = split_element_stream(element, queue.msg_size)
+ count = queue.header().msg_count
+ if len(slots) > count:
+ raise HostTrafficError(
+ f"message needs {len(slots)} elements, queue holds {count}"
+ )
+ index = queue.write_ptr
+ ptr = index
+ for slot in slots:
+ queue.write_entry(ptr, slot)
+ ptr = (ptr + 1) % count
+ write_ptr = queue.publish(ptr)
+ return {
+ "index": index,
+ "write_ptr": write_ptr,
+ "bytes": len(element),
+ "elements": len(slots),
+ }
+
+ def _take(self, queue: MsgqQueue) -> dict | None:
+ write_ptr = queue.write_ptr
+ read_ptr = queue.read_ptr
+ if read_ptr == write_ptr:
+ return None
+ count = queue.header().msg_count
+ first = queue.read_entry(read_ptr)
+ elements = int.from_bytes(first[0x28:0x2C], "little")
+ if not 1 <= elements <= count:
+ raise HostTrafficError(f"implausible elem_count {elements}")
+ slots = []
+ ptr = read_ptr
+ for _ in range(elements):
+ slots.append(queue.read_entry(ptr))
+ ptr = (ptr + 1) % count
+ stream = b"".join(slots)
+ total = GSP_MSG_QUEUE_ELEMENT_HDR_SIZE + int.from_bytes(
+ stream[0x38:0x3C], "little"
+ )
+ element = (
+ stream[:total]
+ if GSP_MSG_QUEUE_ELEMENT_HDR_SIZE + RPC_HEADER_BYTES <= total <= len(stream)
+ else stream
+ )
+ queue.set_read(ptr)
+ return {"index": read_ptr, "element": element, "elements": elements}
+
+ def submit_element(self, element: bytes) -> dict:
+ """Host -> GSP: append one element to the command ring."""
+ return self._place(self.command, element)
+
+ def take_command(self) -> dict | None:
+ """GSP side: consume the next command element, or ``None``."""
+ return self._take(self.command)
+
+ def publish_element(self, element: bytes) -> dict:
+ """GSP -> host: append one element to the status ring."""
+ return self._place(self.status, element)
+
+ def take_status(self) -> dict | None:
+ """Host side: consume the next status element, or ``None``."""
+ return self._take(self.status)
+
+ def state(self) -> dict:
+ return {"command": self.command.state(), "status": self.status.state()}
+
+
+# ---------------------------------------------------------------------------
+# Doorbell
+# ---------------------------------------------------------------------------
+
+
+class DoorbellModel:
+ """Recorded queue-head writes; the firmware-visible read-back surface."""
+
+ def __init__(self, *, count: int = QUEUE_HEAD_COUNT, max_writes: int = 1024):
+ self.count = int(count)
+ self.max_writes = int(max_writes)
+ self.heads = [0] * self.count
+ self.writes: list[dict] = []
+ self.dropped = 0
+ self.kicks = 0
+
+ def ring(self, queue: int, value: int = 0, *, pc: int | None = None) -> None:
+ queue = int(queue)
+ if not 0 <= queue < self.count:
+ raise ValueError(f"queue index {queue} outside 0..{self.count - 1}")
+ value = int(value) & 0xFFFFFFFF
+ self.heads[queue] = value
+ self.kicks += 1
+ record = {"queue": queue, "value": value}
+ if pc is not None:
+ record["pc"] = int(pc)
+ if len(self.writes) < self.max_writes:
+ self.writes.append(record)
+ else:
+ self.dropped += 1
+
+ def head(self, queue: int) -> int:
+ queue = int(queue)
+ if not 0 <= queue < self.count:
+ raise ValueError(f"queue index {queue} outside 0..{self.count - 1}")
+ return self.heads[queue]
+
+ def state(self) -> dict:
+ return {
+ "count": self.count,
+ "heads": list(self.heads),
+ "kicks": self.kicks,
+ "writes": [dict(entry) for entry in self.writes],
+ "dropped": self.dropped,
+ }
+
+ def reset(self) -> None:
+ self.heads = [0] * self.count
+ self.writes = []
+ self.dropped = 0
+ self.kicks = 0
+
+
+class DoorbellBlock(Device):
+ """``NV_PGSP_QUEUE_HEAD`` window over a :class:`DoorbellModel`.
+
+ Evidence: hardware register header (offset/width/access) for the register;
+ the model's read-back is storage of the last written value, and the
+ firmware's read use of this register is **not observed** (documented, not
+ claimed). Guest stores are recorded with their PC.
+ """
+
+ name = "gsp-doorbell"
+ wants_pc = True
+
+ def __init__(self, doorbell: DoorbellModel | None = None, *,
+ base: int = DOORBELL_BASE, size: int = DOORBELL_SIZE):
+ super().__init__(base=base, size=size)
+ self.doorbell = doorbell if doorbell is not None else DoorbellModel()
+
+ def reset(self) -> None:
+ if getattr(self, "doorbell", None) is None:
+ self.doorbell = DoorbellModel()
+ else:
+ self.doorbell.reset()
+
+ def _queue(self, offset: int, size: int) -> int | None:
+ if size != 4 or offset % QUEUE_HEAD_STRIDE:
+ return None
+ queue = offset // QUEUE_HEAD_STRIDE
+ if not 0 <= queue < QUEUE_HEAD_COUNT:
+ return None
+ return queue
+
+ def read(self, offset: int, size: int, pc: int = 0) -> int | None:
+ queue = self._queue(offset, size)
+ if queue is None:
+ return None
+ return self.doorbell.head(queue)
+
+ def write(self, offset: int, size: int, value: int, pc: int = 0) -> bool:
+ queue = self._queue(offset, size)
+ if queue is None:
+ return False
+ self.doorbell.ring(queue, value, pc=pc)
+ return True
+
+ def peek(self, offset: int, size: int) -> int | None:
+ return self.read(offset, size)
+
+ def state(self) -> dict:
+ return {"doorbell": self.doorbell.state()}
+
+ def load_state(self, data: dict) -> None:
+ state = data.get("doorbell", data)
+ self.doorbell.count = int(state.get("count", self.doorbell.count))
+ self.doorbell.heads = [int(v) for v in state.get("heads", [0] * self.doorbell.count)]
+ self.doorbell.kicks = int(state.get("kicks", 0))
+ self.doorbell.writes = [dict(entry) for entry in state.get("writes", [])]
+ self.doorbell.dropped = int(state.get("dropped", 0))
+
+
+# ---------------------------------------------------------------------------
+# Recorder compatibility
+# ---------------------------------------------------------------------------
+
+
+class _NullRecorder:
+ """Stand-in for ``greeneye.record.NullRecorder`` (keeps sim standalone)."""
+
+ dir: str | None = None
+
+ def event(self, event_type: str, **fields) -> None:
+ pass
+
+ def blob(self, name: str, data: bytes) -> str:
+ return name
+
+
+# ---------------------------------------------------------------------------
+# Host traffic loop
+# ---------------------------------------------------------------------------
+
+
+class HostTraffic:
+ """The modelled host-side transport: submit, dispatch (stub), read reply.
+
+ ``machine`` (optional) supplies simulated RAM for the queue block and lets
+ :meth:`run_cycles` step the core; without it the queues live in a private
+ :class:`RegionMemory`. ``recorder`` (optional) is any object with
+ ``event(type, **fields)`` / ``blob(name, data)``; the transport transcript
+ stays deterministic regardless.
+ """
+
+ def __init__(
+ self,
+ *,
+ machine=None,
+ recorder=None,
+ shared_base: int = SHARED_MEM_BASE,
+ cmd_size: int = DEFAULT_CMD_QUEUE_SIZE,
+ stat_size: int = DEFAULT_STAT_QUEUE_SIZE,
+ msg_size: int = GSP_MSG_QUEUE_ELEMENT_SIZE_MIN,
+ handlers: dict[int, Callable[..., Any]] | None = None,
+ ):
+ if machine is None and shared_base == SHARED_MEM_BASE:
+ mem = RegionMemory(shared_base, cmd_size + stat_size)
+ else:
+ mem = _make_memory(machine, shared_base, cmd_size + stat_size)
+ self.machine = machine
+ self.recorder = recorder if recorder is not None else _NullRecorder()
+ self.shared_base = int(shared_base)
+ self.queues = SharedQueues(
+ mem,
+ cmd_base=shared_base,
+ cmd_size=cmd_size,
+ stat_size=stat_size,
+ msg_size=msg_size,
+ )
+ self.queues.format()
+ self.doorbell = DoorbellModel()
+ self._tx_seq = 0
+ self._rx_seq = 0
+ self._next_sequence = 1
+ self._submitted = 0
+ self._dispatched = 0
+ self._replies = 0
+ self._events: list[dict] = []
+ self._closed = False
+ self._handlers: dict[int, tuple[str, Callable[..., Any]]] = {}
+ self.set_handler(RPC_FN_ECHO, self._handle_echo, name="stub:echo")
+ self.set_handler(RPC_FN_STATUS_QUERY, self._handle_status_query,
+ name="stub:status_query")
+ for function, handler in (handlers or {}).items():
+ self.set_handler(function, handler)
+ self._emit(
+ "host_traffic_config",
+ shared_base=self.shared_base,
+ cmd_base=self.queues.command.tx_base,
+ stat_base=self.queues.status.tx_base,
+ cmd_size=self.queues.command.size,
+ stat_size=self.queues.status.size,
+ msg_size=self.queues.command.msg_size,
+ handlers=sorted(self.handler_names().items()),
+ )
+
+ # -- handler table ------------------------------------------------------
+
+ def set_handler(self, function: int, handler: Callable[..., Any], *,
+ name: str | None = None) -> None:
+ label = name or getattr(handler, "__name__", "handler")
+ self._handlers[int(function)] = (label, handler)
+
+ def handler_names(self) -> dict[int, str]:
+ return {function: label for function, (label, _) in sorted(self._handlers.items())}
+
+ def _handle_echo(self, request: RpcRequest) -> RpcReply:
+ return RpcReply(function=request.function, sequence=request.sequence,
+ params=request.params)
+
+ def _handle_status_query(self, request: RpcRequest) -> RpcReply:
+ payload = struct.pack(
+ STATUS_FORMAT,
+ STATUS_MAGIC,
+ STATUS_VERSION,
+ self._submitted,
+ self._dispatched,
+ self._replies,
+ )
+ return RpcReply(function=request.function, sequence=request.sequence,
+ params=payload)
+
+ # -- recording ----------------------------------------------------------
+
+ def _emit(self, event_type: str, **fields) -> None:
+ self.recorder.event(event_type, **fields)
+
+ def _record(self, event: dict) -> None:
+ self._events.append(event)
+
+ def note_skip(self, index: int, reason: str) -> None:
+ """Record a replay message that was not submitted (documented)."""
+ self._record({"type": "skip", "index": int(index), "reason": reason})
+ self._emit("host_replay_skip", index=int(index), reason=reason)
+
+ # -- host side ----------------------------------------------------------
+
+ def submit(self, message: bytes) -> dict:
+ """Write one framed RPC message to the command ring, then kick."""
+ parsed = parse_rpc_message(message)
+ element = build_queue_element(message, sequence=self._tx_seq)
+ if len(element) > GSP_MSG_QUEUE_ELEMENT_SIZE_MAX:
+ raise HostTrafficError(
+ f"message of {len(element)} bytes exceeds the wire element "
+ f"maximum of {GSP_MSG_QUEUE_ELEMENT_SIZE_MAX:#x}"
+ )
+ placed = self.queues.submit_element(element)
+ self._tx_seq += 1
+ self.doorbell.ring(DOORBELL_KICK_QUEUE, DOORBELL_KICK_VALUE)
+ self._submitted += 1
+ self._emit("doorbell", queue=DOORBELL_KICK_QUEUE,
+ value=DOORBELL_KICK_VALUE, source="host")
+ result = {
+ "format": "greeneye-host-traffic-submit/1",
+ "index": placed["index"],
+ "write_ptr": placed["write_ptr"],
+ "function": parsed.function,
+ "sequence": parsed.sequence,
+ "msgq_sequence": self._tx_seq - 1,
+ "length": parsed.length,
+ "params_bytes": len(parsed.params),
+ "element_bytes": len(element),
+ "doorbell": {"queue": DOORBELL_KICK_QUEUE, "value": DOORBELL_KICK_VALUE},
+ }
+ self._record({
+ "type": "submit",
+ "index": placed["index"],
+ "function": parsed.function,
+ "sequence": parsed.sequence,
+ "params_bytes": len(parsed.params),
+ "element_bytes": len(element),
+ "write_ptr": placed["write_ptr"],
+ })
+ self._emit(
+ "host_submit",
+ function=parsed.function,
+ sequence=parsed.sequence,
+ length=parsed.length,
+ params_bytes=len(parsed.params),
+ element_index=placed["index"],
+ write_ptr=placed["write_ptr"],
+ )
+ return result
+
+ def submit_rpc(self, function: int, params: bytes = b"",
+ *, sequence: int | None = None) -> dict:
+ """Frame and submit one RPC; ``sequence`` defaults to a local counter."""
+ if sequence is None:
+ sequence = self._next_sequence
+ self._next_sequence = max(self._next_sequence, int(sequence) + 1)
+ return self.submit(build_rpc_message(function, params, sequence=sequence))
+
+ def _normalise_reply(self, produced: Any, request: RpcRequest) -> RpcReply:
+ if produced is None:
+ return RpcReply(function=request.function, sequence=request.sequence)
+ if isinstance(produced, RpcReply):
+ return produced
+ if isinstance(produced, (bytes, bytearray)):
+ return RpcReply(function=request.function, sequence=request.sequence,
+ params=bytes(produced))
+ raise HostTrafficError(
+ f"handler returned {type(produced).__name__}; expected RpcReply, "
+ "bytes or None"
+ )
+
+ def poll(self) -> int:
+ """GSP side: drain the command ring through the dispatcher (bounded)."""
+ drained = 0
+ while True:
+ taken = self.queues.take_command()
+ if taken is None:
+ break
+ prefix, request_message = parse_queue_element(taken["element"])
+ label, handler = self._handlers.get(
+ request_message.function, ("stub:unsupported", None)
+ )
+ self._dispatched += 1
+ request = RpcRequest(
+ function=request_message.function,
+ sequence=request_message.sequence,
+ params=request_message.params,
+ element_index=taken["index"],
+ raw=bytes(taken["element"]),
+ )
+ if handler is None:
+ reply = RpcReply(
+ function=request.function,
+ sequence=request.sequence,
+ params=b"",
+ result=NV_ERR_NOT_SUPPORTED,
+ )
+ else:
+ reply = self._normalise_reply(handler(request), request)
+ reply_bytes = build_rpc_message(
+ reply.function, reply.params, sequence=reply.sequence,
+ result=reply.result,
+ )
+ element = build_queue_element(reply_bytes, sequence=self._rx_seq)
+ self._rx_seq += 1
+ if len(element) > GSP_MSG_QUEUE_ELEMENT_SIZE_MAX:
+ raise HostTrafficError(
+ f"reply of {len(element)} bytes exceeds the wire element maximum"
+ )
+ self.queues.publish_element(element)
+ self._replies += 1
+ drained += 1
+ self._record({
+ "type": "dispatch",
+ "index": taken["index"],
+ "function": request.function,
+ "sequence": request.sequence,
+ "handler": label,
+ "result": reply.result,
+ "reply_bytes": len(reply.params),
+ })
+ self._emit(
+ "host_dispatch",
+ function=request.function,
+ sequence=request.sequence,
+ handler=label,
+ result=reply.result,
+ reply_bytes=len(reply.params),
+ element_index=taken["index"],
+ msgq_sequence=prefix["seq_num"],
+ )
+ return drained
+
+ def read_reply(self) -> dict | None:
+ """Host side: consume the next status element, or ``None``."""
+ taken = self.queues.take_status()
+ if taken is None:
+ return None
+ _prefix, message = parse_queue_element(taken["element"])
+ record = {
+ "format": REPLY_FORMAT,
+ "index": taken["index"],
+ "function": message.function,
+ "sequence": message.sequence,
+ "result": message.rpc_result,
+ "params": message.params,
+ "params_hex": message.params.hex(),
+ "params_bytes": len(message.params),
+ }
+ self._record({
+ "type": "reply",
+ "index": taken["index"],
+ "function": message.function,
+ "sequence": message.sequence,
+ "result": message.rpc_result,
+ "params_bytes": len(message.params),
+ })
+ self._emit(
+ "host_reply_read",
+ function=message.function,
+ sequence=message.sequence,
+ result=message.rpc_result,
+ params_bytes=len(message.params),
+ element_index=taken["index"],
+ )
+ return record
+
+ def run_cycles(self, budget: int) -> dict:
+ """Run the attached core for a bounded ``budget``, then ``poll``."""
+ if isinstance(budget, bool) or not isinstance(budget, int) or not 0 < budget <= MAX_RUN_CYCLES:
+ raise ValueError(f"budget must be an int in 1..{MAX_RUN_CYCLES}, got {budget!r}")
+ reason = pc = None
+ if self.machine is not None:
+ reason, pc, _a, _b = self.machine.run(budget)
+ dispatched = self.poll()
+ return {
+ "budget": budget,
+ "reason": reason,
+ "pc": pc,
+ "insns": self.machine.insns if self.machine is not None else 0,
+ "dispatched": dispatched,
+ "replies": self._replies,
+ }
+
+ # -- evidence -----------------------------------------------------------
+
+ def transcript(self) -> dict:
+ counts: dict[str, int] = {}
+ for event in self._events:
+ counts[event["type"]] = counts.get(event["type"], 0) + 1
+ return {
+ "format": TRANSCRIPT_FORMAT,
+ "events": [dict(event) for event in self._events],
+ "counts": counts,
+ }
+
+ def state(self) -> dict:
+ return {
+ "format": TRANSCRIPT_FORMAT,
+ "shared_base": self.shared_base,
+ "queues": self.queues.state(),
+ "doorbell": self.doorbell.state(),
+ "counts": {
+ "submitted": self._submitted,
+ "dispatched": self._dispatched,
+ "replies": self._replies,
+ },
+ "handlers": self.handler_names(),
+ }
+
+ def doorbell_block(self) -> DoorbellBlock:
+ return DoorbellBlock(self.doorbell)
+
+ def close(self) -> None:
+ if self._closed:
+ return
+ self._closed = True
+ payload = json.dumps(self.transcript(), indent=2, sort_keys=True).encode()
+ self.recorder.blob("host-traffic.json", payload)
+ self._emit("host_transcript", events=len(self._events))
+
+
+# ---------------------------------------------------------------------------
+# Stimulus replay
+# ---------------------------------------------------------------------------
+
+
+def _stimulus_from_capture(source: str | dict) -> dict:
+ if _TOOL_ROOT not in sys.path:
+ sys.path.insert(0, _TOOL_ROOT)
+ try:
+ from harness.rm_capture import load_capture, stimulus_from_capture
+ except ImportError as exc: # pragma: no cover - tool checkout broken
+ raise HostTrafficError(f"harness.rm_capture unavailable: {exc}") from exc
+ if isinstance(source, dict):
+ raise HostTrafficError(
+ "an inline capture document is not accepted; pass the capture JSON "
+ "path or a stimulus document"
+ )
+ capture = load_capture(source)
+ return stimulus_from_capture(capture)
+
+
+def load_stimulus(source: str | dict) -> dict:
+ """Load a replay stimulus from a doc, a stimulus JSON or a capture file."""
+ if isinstance(source, dict):
+ doc = source
+ else:
+ if not os.path.isfile(source):
+ raise HostTrafficError(f"stimulus not found: {source}")
+ with open(source, "r", encoding="utf-8", errors="replace") as fh:
+ head = fh.read(4096)
+ if not head.lstrip().startswith("{"):
+ return _stimulus_from_capture(source)
+ try:
+ with open(source, "r", encoding="utf-8", errors="replace") as fh:
+ doc = json.load(fh)
+ except (OSError, json.JSONDecodeError) as exc:
+ raise HostTrafficError(f"cannot read stimulus {source}: {exc}") from exc
+ if not isinstance(doc, dict):
+ raise HostTrafficError("stimulus document must be an object")
+ fmt = str(doc.get("format") or "")
+ if fmt == STIMULUS_FORMAT:
+ return doc
+ if fmt == CAPTURE_FORMAT:
+ return _stimulus_from_capture(source)
+ raise HostTrafficError(
+ f"unknown stimulus/capture format {fmt!r}; expected {STIMULUS_FORMAT!r} "
+ f"or {CAPTURE_FORMAT!r}"
+ )
+
+
+def _message_to_rpc(message: dict, index: int) -> tuple[int, bytes]:
+ """Map one stimulus message to ``(wire function, payload)``.
+
+ ``send`` messages carry the wire function and payload directly. ``control``
+ and ``alloc`` records carry a named id instead; the capture tool's mapping
+ (control -> 76, alloc -> 103) is applied, and the captured parameter block
+ is used as the payload. That identity is an assumption until a capture
+ with full send payloads exists (documented in docs/51).
+ """
+ kind = message.get("kind")
+ function = message.get("fn")
+ if function is None:
+ if kind == "control":
+ function = GSP_RM_CONTROL
+ elif kind == "alloc":
+ function = GSP_RM_ALLOC
+ if function is None:
+ raise HostTrafficError(f"message {index}: no wire function id")
+ hex_text = message.get("params_hex") or ""
+ try:
+ params = bytes.fromhex(hex_text)
+ except ValueError as exc:
+ raise HostTrafficError(f"message {index}: bad params_hex: {exc}") from exc
+ return int(function), params
+
+
+def replay_stimulus(host: HostTraffic, stimulus: str | dict, *,
+ strict: bool = True) -> dict:
+ """Submit a stimulus document's messages in order, then read the replies."""
+ doc = load_stimulus(stimulus)
+ messages = doc.get("messages")
+ if not isinstance(messages, list):
+ raise HostTrafficError("stimulus document has no 'messages' list")
+ pairs: list[dict] = []
+ skipped: list[dict] = []
+ for index, message in enumerate(messages):
+ if not isinstance(message, dict):
+ raise HostTrafficError(f"message {index} is not an object")
+ if message.get("truncated"):
+ reason = "truncated payload is not replayable as-is"
+ if strict:
+ raise HostTrafficError(f"message {index}: {reason}")
+ host.note_skip(index, reason)
+ skipped.append({"index": index, "reason": reason})
+ continue
+ function, params = _message_to_rpc(message, index)
+ submit = host.submit_rpc(function, params)
+ dispatched = host.poll()
+ reply = host.read_reply()
+ if reply is None or dispatched != 1:
+ raise HostTrafficError(
+ f"message {index}: transport did not produce exactly one reply"
+ )
+ pairs.append({
+ "seq": message.get("seq", index),
+ "kind": message.get("kind"),
+ "function": function,
+ "submit": submit,
+ "reply": reply,
+ })
+ return {
+ "format": REPLAY_FORMAT,
+ "source": doc.get("source"),
+ "submitted": len(pairs),
+ "skipped": skipped,
+ "pairs": pairs,
+ "transcript": host.transcript(),
+ "limits": list(_LIMITS),
+ }
+
+
+# ---------------------------------------------------------------------------
+# Demo
+# ---------------------------------------------------------------------------
+
+
+def _new_recorder(out_dir: str | None, name: str):
+ if out_dir is None:
+ return _NullRecorder(), False
+ from greeneye.record import Recorder # deferred: keep sim importable alone
+
+ meta = {"kind": name, "host_traffic": True}
+ return Recorder(out_dir, name=name, meta=meta), True
+
+
+def _json_reply(reply: dict | None) -> dict | None:
+ if reply is None:
+ return None
+ return {
+ "format": reply["format"],
+ "index": reply["index"],
+ "function": reply["function"],
+ "sequence": reply["sequence"],
+ "result": reply["result"],
+ "params_hex": reply["params_hex"],
+ "params_bytes": reply["params_bytes"],
+ "read": True,
+ }
+
+
+def run_demo(out_dir: str | None = None) -> dict:
+ """Deterministic end-to-end transport demo; no capture, no firmware."""
+ from .machine import Machine # deferred (module import stays light)
+
+ machine = Machine(memory_mb=32)
+ machine.load_segments(
+ [(machine.ram_base, bytes.fromhex("73001000"))], entry=machine.ram_base
+ )
+ recorder, owns = _new_recorder(out_dir, "host-traffic-demo")
+ host = HostTraffic(machine=machine, recorder=recorder)
+ replies: list[dict] = []
+ for sequence, (function, params) in enumerate(
+ (
+ (RPC_FN_ECHO, b"\x01\x02\x03\x04"),
+ (RPC_FN_STATUS_QUERY, b""),
+ ),
+ start=1,
+ ):
+ host.submit_rpc(function, params, sequence=sequence)
+ host.run_cycles(64)
+ reply = host.read_reply()
+ if reply is not None:
+ replies.append(_json_reply(reply))
+
+ control_cmd = 0x2080012B
+ control_params = (
+ (0xC2000006).to_bytes(4, "little")
+ + (0xABCD0080).to_bytes(4, "little")
+ + control_cmd.to_bytes(4, "little")
+ + bytes(500)
+ )
+ host.submit_rpc(GSP_RM_CONTROL, control_params, sequence=3)
+ host.run_cycles(64)
+ control_reply = host.read_reply()
+ if control_reply is not None:
+ replies.append(_json_reply(control_reply))
+
+ checks = {
+ "echo_reply_matches_payload": (
+ replies[0]["function"] == RPC_FN_ECHO
+ and replies[0]["params_hex"] == b"\x01\x02\x03\x04".hex()
+ and replies[0]["result"] == NV_OK
+ ),
+ "status_reply_has_magic": (
+ len(replies) > 1
+ and bytes.fromhex(replies[1]["params_hex"])[:4] == STATUS_MAGIC
+ ),
+ "control_replay_is_512_bytes": host.state()["counts"]["submitted"] == 3,
+ "control_reply_is_a_framed_stub_verdict": (
+ control_reply is not None
+ and control_reply["function"] == GSP_RM_CONTROL
+ and control_reply["result"] == NV_ERR_NOT_SUPPORTED
+ ),
+ "doorbell_ring_per_submit": host.doorbell.kicks == 3,
+ "all_replies_read": len(replies) == 3,
+ }
+ control = {
+ "function": GSP_RM_CONTROL,
+ "cmd": control_cmd,
+ "bytes": len(control_params),
+ "reply_function": control_reply["function"] if control_reply else None,
+ "reply_result": control_reply["result"] if control_reply else None,
+ "reply_params_bytes": control_reply["params_bytes"] if control_reply else None,
+ }
+ payload = {
+ "format": DEMO_FORMAT,
+ "ok": all(checks.values()),
+ "checks": checks,
+ "machine": {
+ "ram_base": machine.ram_base,
+ "insns": machine.insns,
+ "cycles": machine.cycles,
+ },
+ "shared_base": host.shared_base,
+ "queue": host.queues.command.state(),
+ "control": control,
+ "replies": replies,
+ "transcript": host.transcript(),
+ "limits": list(_LIMITS),
+ }
+ host.close()
+ if owns:
+ os.makedirs(out_dir, exist_ok=True)
+ payload["artifacts"] = {"session": os.path.abspath(out_dir)}
+ with open(os.path.join(out_dir, "host-traffic-demo.json"), "w",
+ encoding="utf-8") as fh:
+ json.dump(payload, fh, indent=1, sort_keys=True)
+ fh.write("\n")
+ recorder.close()
+ return payload
+
+
+# ---------------------------------------------------------------------------
+# CLI
+# ---------------------------------------------------------------------------
+
+
+def build_parser() -> argparse.ArgumentParser:
+ parser = argparse.ArgumentParser(
+ prog="sim.host_traffic",
+ description=(
+ "Modelled host traffic loop for the GSP message queues: doorbell, "
+ "command/status rings, stub dispatch and rm_capture stimulus "
+ "replay. Offline only; the simulator runs no real firmware here."
+ ),
+ )
+ sub = parser.add_subparsers(dest="command")
+ demo_cmd = sub.add_parser("demo", help="deterministic end-to-end demo")
+ demo_cmd.add_argument("--out", help="session/output directory (optional)")
+ demo_cmd.add_argument("--json", action="store_true", help="print full JSON")
+ replay_cmd = sub.add_parser("replay", help="replay an rm_capture stimulus")
+ replay_cmd.add_argument("--stimulus", required=True,
+ help="stimulus JSON, capture JSON or raw capture log")
+ replay_cmd.add_argument("--out", help="session/output directory (optional)")
+ replay_cmd.add_argument("--allow-truncated", action="store_true",
+ help="skip truncated messages instead of refusing")
+ replay_cmd.add_argument("--json", action="store_true", help="print full JSON")
+ return parser
+
+
+def _cli_summary(payload: dict) -> dict:
+ summary = {
+ "format": payload.get("format"),
+ "ok": payload.get("ok"),
+ }
+ if payload.get("checks") is not None:
+ summary["checks"] = payload["checks"]
+ if "submitted" in payload:
+ summary["submitted"] = payload["submitted"]
+ summary["skipped"] = len(payload.get("skipped", []))
+ if payload.get("artifacts"):
+ summary["artifacts"] = payload["artifacts"]
+ return summary
+
+
+def _jsonable(value):
+ if isinstance(value, bytes):
+ return value.hex()
+ if isinstance(value, (list, tuple)):
+ return [_jsonable(item) for item in value]
+ if isinstance(value, dict):
+ return {str(key): _jsonable(item) for key, item in value.items()}
+ return value
+
+
+def main(argv: list[str] | None = None) -> int:
+ parser = build_parser()
+ args = parser.parse_args(argv)
+ if args.command is None:
+ parser.print_help(sys.stderr)
+ return 2
+ try:
+ if args.command == "demo":
+ payload = run_demo(args.out)
+ else:
+ recorder, owns = _new_recorder(args.out, "host-traffic-replay")
+ host = HostTraffic(recorder=recorder)
+ payload = replay_stimulus(
+ host, args.stimulus, strict=not args.allow_truncated
+ )
+ host.close()
+ if owns:
+ os.makedirs(args.out, exist_ok=True)
+ payload["artifacts"] = {"session": os.path.abspath(args.out)}
+ with open(os.path.join(args.out, "host-traffic-replay.json"), "w",
+ encoding="utf-8") as fh:
+ json.dump(_jsonable(payload), fh, indent=1, sort_keys=True)
+ fh.write("\n")
+ recorder.close()
+ except HostTrafficError as exc:
+ print(f"error: {exc}", file=sys.stderr)
+ return 2
+ if args.json:
+ print(json.dumps(_jsonable(payload), indent=2, sort_keys=True))
+ else:
+ print(json.dumps(_jsonable(_cli_summary(payload)), indent=2, sort_keys=True))
+ if args.command == "demo" and not payload.get("ok"):
+ return 3
+ return 0
+
+
+if __name__ == "__main__":
+ raise SystemExit(main())
diff --git a/debug/risc-v-gsp-rm-vm/sim/libos_port.py b/debug/risc-v-gsp-rm-vm/sim/libos_port.py
new file mode 100644
index 0000000..eaa3f95
--- /dev/null
+++ b/debug/risc-v-gsp-rm-vm/sim/libos_port.py
@@ -0,0 +1,386 @@
+"""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,
+ NAMES,
+ TASK_PORT_SEND_RECV_WAIT,
+)
+
+#: Report schema tag for :meth:`LibosPortModel.state`.
+REPORT_FORMAT = "greeneye-msgq-port/1"
+
+#: ``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
+
+ 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)
+ return True, LIBOS_OK
+ if 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
+
+ # 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,
+ }
diff --git a/debug/risc-v-gsp-rm-vm/sim/msgq_attach.py b/debug/risc-v-gsp-rm-vm/sim/msgq_attach.py
new file mode 100644
index 0000000..f1530bd
--- /dev/null
+++ b/debug/risc-v-gsp-rm-vm/sim/msgq_attach.py
@@ -0,0 +1,1122 @@
+"""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())