path: root/debug/risc-v-gsp-rm-vm/harness/launch_audit.py

#!/usr/bin/env python3
"""Launch memory-contract audit: where would the engine read/write?

A forwarded submission goes silent when one of the engine-visible memory legs
is wrong: the QMD destination the front-end copies to, the release semaphore
the engine writes, the ring the front-end fetches, the parameter/program
pages the QMD points at.  The host's bookkeeping (shadows, guest pages) is not
what the engine sees -- the *page tables* are.  This harness walks each
address of a submission's memory contract through a captured/constructed
page-table view (``sim.gmmu_walk``) and reports, per leg, the PA the engine
would use, its access flags, and any divergence from what the host believes.

It answers the class of question behind "the front-end never wrote the QMD"/
"the release never fired": *is the mapping the engine uses the mapping we are
checking?* -- offline, deterministically, without hardware.

``check`` closes the capture loop: run it against a partial or empty
``memory[]`` view and it reports, per contract leg, the FIRST table page the
walk needed but the view does not contain (with the level/index path), plus a
consolidated ``needed_pages[]`` list.  Dump exactly those pages, re-run, and
repeat -- one capture iteration per table level.

Schemas:

* ``greeneye-launch-audit-spec/1``  -- inputs (page tables + contract).
* ``greeneye-launch-audit/1``       -- the verdict report.
* ``greeneye-launch-audit-check/1`` -- the dump-completeness report.

Honesty: a verdict is only as good as the captured table view in the spec.
``capture-hints`` prints what to capture for a real audit and never executes
anything.  Fidelity is the address view (L1); no timing claim.
"""
from __future__ import annotations

import argparse
import json
import os
import struct
import sys
from typing import Any

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 KIND_VER2, LEVELS, SUPPORTED_KINDS  # noqa: E402
from sim.gmmu_walk import (  # noqa: E402
    FAULT_TABLE_MISSING,
    MemoryView,
    WalkError,
    channel_pdb_fixture,
    encode_dual_slot,
    encode_pde_pointer,
    encode_pte,
    walk,
)

SPEC_FORMAT = "greeneye-launch-audit-spec/1"
AUDIT_FORMAT = "greeneye-launch-audit/1"
CHECK_FORMAT = "greeneye-launch-audit-check/1"
DEMO_FORMAT = "greeneye-launch-audit-demo/1"

VERDICT_OK = "ok"
VERDICT_UNMAPPED = "unmapped"
VERDICT_PA_MISMATCH = "pa_mismatch"
VERDICT_ACCESS_DENIED = "access_denied"

#: Walk depth -> display label (root first, leaf page table last).  Mirrors
#: the ``LEVELS`` geometry in ``sim.gmmu_pte``; kept here as display names so
#: check reports read like the capture hints (root/pde2/pde1/pde0/pte).
LEVEL_LABELS = ("root", "pde2", "pde1", "pde0", "pte")


class AuditError(ValueError):
    """A spec or request the tool cannot honour."""


def _parse_int(value: Any) -> int:
    if isinstance(value, bool):
        raise AuditError(f"bad integer {value!r}")
    if isinstance(value, int):
        return value
    if isinstance(value, str):
        try:
            return int(value, 0)
        except ValueError as exc:
            raise AuditError(f"bad integer {value!r}") from exc
    raise AuditError(f"bad integer {value!r}")


def _parse_hex_bytes(text: str) -> bytes:
    data = "".join(str(text).split())
    if len(data) % 2:
        raise AuditError("odd hex length")
    try:
        return bytes.fromhex(data)
    except ValueError as exc:
        raise AuditError(f"bad hex payload: {exc}") from exc


def load_memory(items) -> MemoryView:
    mem = MemoryView()
    for item in items or ():
        if "pa" not in item or "hex" not in item:
            raise AuditError(f"memory item needs pa and hex: {item!r}")
        mem.add(_parse_int(item["pa"]), _parse_hex_bytes(item["hex"]))
    return mem


def audit(spec: dict) -> dict:
    """Run one audit spec; return a ``greeneye-launch-audit/1`` report."""
    fmt = spec.get("format", SPEC_FORMAT)
    if fmt != SPEC_FORMAT:
        raise AuditError(f"unsupported spec format {fmt!r} (want {SPEC_FORMAT})")
    if "pdb_pa" not in spec or "contract" not in spec:
        raise AuditError("spec needs pdb_pa and contract")
    kind = spec.get("gmmu", KIND_VER2)
    if kind not in SUPPORTED_KINDS:
        raise AuditError(f"unsupported gmmu kind {kind!r} (support: {SUPPORTED_KINDS})")
    page_mode = spec.get("page_mode", "auto")
    root_pa = _parse_int(spec["pdb_pa"])
    mem = load_memory(spec.get("memory"))

    rows = []
    for item in spec["contract"]:
        label = item.get("label") or "unnamed"
        if "va" not in item:
            raise AuditError(f"contract item needs va: {item!r}")
        va = _parse_int(item["va"])
        access = item.get("access", "read")
        report = walk(va, root_pa=root_pa, mem=mem, kind=kind, page_mode=page_mode)
        row: dict[str, Any] = {
            "label": label,
            "va": f"{va:#x}",
            "access": access,
            "note": item.get("note"),
            "verdict": None,
            "walk": report,
        }
        if report["status"] != "ok":
            row["verdict"] = VERDICT_UNMAPPED
            row["detail"] = report.get("fault_message")
        else:
            expect_pa = item.get("expect_pa")
            if access == "write" and not report.get("writable", False):
                row["verdict"] = VERDICT_ACCESS_DENIED
                row["detail"] = "PTE is read-only but the engine must write here"
            elif expect_pa is not None and _parse_int(expect_pa) != _parse_int(report["pa"]):
                row["verdict"] = VERDICT_PA_MISMATCH
                row["detail"] = (f"engine walk -> {report['pa']}; host believes "
                                 f"{_parse_int(expect_pa):#x}")
            else:
                row["verdict"] = VERDICT_OK
        rows.append(row)

    summary = {v: sum(1 for r in rows if r["verdict"] == v)
               for v in (VERDICT_OK, VERDICT_UNMAPPED, VERDICT_PA_MISMATCH,
                         VERDICT_ACCESS_DENIED)}
    report = {
        "format": AUDIT_FORMAT,
        "name": spec.get("name"),
        "firmware": spec.get("firmware"),
        "gmmu": kind,
        "page_mode": page_mode,
        "pdb_pa": f"{root_pa:#x}",
        "contract": rows,
        "summary": summary,
        "ok": summary[VERDICT_UNMAPPED] == 0 and summary[VERDICT_PA_MISMATCH] == 0
              and summary[VERDICT_ACCESS_DENIED] == 0,
        "limits": [
            "Address view only (L1): a clean audit proves the engine CAN reach these PAs, not that it runs.",
            "Verdicts depend on the captured table view in the spec; an incomplete view reports unmapped, not unreachability.",
        ],
    }
    return report


# ---------------------------------------------------------------------------
# Check: which table pages does the walk still need?
# ---------------------------------------------------------------------------

def _path_from_steps(steps: list[dict]) -> list[dict]:
    """The completed walk hops, in walk order, as check ``path`` elements."""
    return [
        {
            "level": LEVEL_LABELS[step["depth"]],
            "depth": step["depth"],
            "table_pa": step["table_pa"],
            "index": step["index"],
            "entry_pa": step["entry_pa"],
            "present": True,
        }
        for step in steps
    ]


def _missing_table(report: dict, root_pa: int, va: int) -> dict:
    """Describe the first table page a ``table_missing`` walk needed.

    ``report["steps"]`` holds the levels that completed before the failing
    read; the missing table is the page that read was addressing.  The
    returned dict carries the page base (``pa`` and ``table_pa``), the level
    label, the walk depth, the index inside the table and the exact entry
    address.  ``report`` must be a ``table_missing`` fault.
    """
    depth = len(report["steps"])
    if depth == 0:
        table_pa = root_pa
    else:
        prev = report["steps"][-1]
        if prev.get("role") == "dual_pde":
            half = prev.get("page_mode_used")
            table_pa = (prev["big"]["address"] if half == "64k"
                        else prev["small"]["address"])
        else:
            table_pa = prev["address"]
    if depth not in range(len(LEVEL_LABELS)):
        # Cannot happen for a walker fault, but never guess a geometry.
        return {"pa": f"{table_pa:#x}", "table_pa": f"{table_pa:#x}",
                "depth": depth, "level": f"depth{depth}", "index": None,
                "entry_pa": None, "present": False}
    if depth == 4:
        prev = report["steps"][-1]
        shift = 16 if prev.get("page_mode_used") == "64k" else 12
        index = (va >> shift) & 0x1FF
        entry_size = 8
    else:
        geo = LEVELS[depth]
        index = (va >> geo["va_bits"][1]) & ((1 << geo["index_bits"]) - 1)
        entry_size = geo["entry_size"]
    return {
        "pa": f"{table_pa:#x}",
        "table_pa": f"{table_pa:#x}",
        "depth": depth,
        "level": LEVEL_LABELS[depth],
        "index": index,
        "entry_pa": f"{table_pa + index * entry_size:#x}",
        "present": False,
    }


def check(spec: dict) -> dict:
    """Report the table pages a partial ``memory[]`` view still needs.

    Runs the same walks as :func:`audit` but, instead of verdicts, reports per
    contract leg the FIRST table page the walk could not read (the page the
    dump is missing) with the level/index path that reached it, and a
    consolidated ``needed_pages`` list.  Iterate: run ``check`` with a partial
    or empty dump, dump exactly the named pages, repeat until ``complete`` is
    true, then run :func:`audit` for verdicts.  A leg whose walk faults for a
    non-dump reason (invalid aperture/PTE, unsupported encoding) is reported
    as ``fault`` and needs no more pages.

    An empty spec memory is valid: every leg then reports the root page as
    its first needed page.
    """
    fmt = spec.get("format", SPEC_FORMAT)
    if fmt != SPEC_FORMAT:
        raise AuditError(f"unsupported spec format {fmt!r} (want {SPEC_FORMAT})")
    if "pdb_pa" not in spec or "contract" not in spec:
        raise AuditError("spec needs pdb_pa and contract")
    kind = spec.get("gmmu", KIND_VER2)
    if kind not in SUPPORTED_KINDS:
        raise AuditError(f"unsupported gmmu kind {kind!r} (support: {SUPPORTED_KINDS})")
    page_mode = spec.get("page_mode", "auto")
    root_pa = _parse_int(spec["pdb_pa"])
    mem = load_memory(spec.get("memory"))

    rows = []
    needed: dict[int, dict[str, Any]] = {}
    for item in spec["contract"]:
        label = item.get("label") or "unnamed"
        if "va" not in item:
            raise AuditError(f"contract item needs va: {item!r}")
        va = _parse_int(item["va"])
        report = walk(va, root_pa=root_pa, mem=mem, kind=kind, page_mode=page_mode)
        row: dict[str, Any] = {
            "label": label,
            "va": f"{va:#x}",
            "access": item.get("access", "read"),
            "expect_pa": None,
            "status": None,
            "resolved_pa": None,
            "writable": None,
            "first_missing": None,
            "fault": report.get("fault"),
            "detail": report.get("fault_message"),
            "path": _path_from_steps(report["steps"]),
        }
        if item.get("expect_pa") is not None:
            row["expect_pa"] = f"{_parse_int(item['expect_pa']):#x}"
        if report["status"] == "ok":
            row["status"] = "complete"
            row["resolved_pa"] = report["pa"]
            row["writable"] = report.get("writable")
            row["fault"] = None
            row["detail"] = None
        elif report.get("fault") == FAULT_TABLE_MISSING:
            missing = _missing_table(report, root_pa, va)
            row["status"] = "missing"
            row["first_missing"] = missing
            row["path"].append(missing)
            entry = needed.setdefault(int(missing["pa"], 16),
                                      {"levels": set(), "legs": []})
            entry["levels"].add(missing["level"])
            entry["legs"].append(label)
        else:
            row["status"] = "fault"
        rows.append(row)

    needed_pages = [
        {"pa": f"{pa:#x}", "levels": sorted(value["levels"]),
         "legs": sorted(value["legs"])}
        for pa, value in sorted(needed.items())
    ]
    return {
        "format": CHECK_FORMAT,
        "name": spec.get("name"),
        "firmware": spec.get("firmware"),
        "gmmu": kind,
        "page_mode": page_mode,
        "pdb_pa": f"{root_pa:#x}",
        "memory_pages": len(mem.spans()),
        "contract": rows,
        "needed_pages": needed_pages,
        "complete": all(row["status"] != "missing" for row in rows),
        "limits": [
            "Dump-completeness view only: `complete` means every table page the walk needed is in the spec, not that any mapping is correct.",
            "A `fault` leg needs no more pages: the walk read its entries and refused them (invalid aperture/PTE or unsupported encoding).",
            "`needed_pages` is the FIRST missing page per leg; dumping it can reveal the next one, so iterate.",
        ],
    }


# ---------------------------------------------------------------------------
# Demo: the captured channel-PDB contract + the QMD-destination hypotheses
# ---------------------------------------------------------------------------

def _tree_for_va(root_pa: int, target_va: int, target_pa: int | None,
                 *, tables_base: int = 0x50000000) -> MemoryView:
    """Build a minimal VER2 tree mapping one VA to one PA (None = unmapped)."""
    va = target_va
    i0 = (va >> 47) & 0x3
    i1 = (va >> 38) & 0x1FF
    i2 = (va >> 29) & 0x1FF
    i3 = (va >> 21) & 0xFF
    i4 = (va >> 12) & 0x1FF
    pd2 = tables_base + 0x1000
    pd1 = tables_base + 0x2000
    pd0 = tables_base + 0x3000
    pt = tables_base + 0x4000
    mem = MemoryView()
    root = bytearray(0x1000)
    struct.pack_into("<Q", root, i0 * 8, encode_pde_pointer(pd2))
    mem.add(root_pa, bytes(root))
    page = bytearray(0x1000)
    struct.pack_into("<Q", page, i1 * 8, encode_pde_pointer(pd1))
    mem.add(pd2, bytes(page))
    page = bytearray(0x1000)
    struct.pack_into("<Q", page, i2 * 8, encode_pde_pointer(pd0))
    mem.add(pd1, bytes(page))
    page = bytearray(0x1000)
    struct.pack_into("<16s", page, i3 * 16, encode_dual_slot(None, pt))
    mem.add(pd0, bytes(page))
    page = bytearray(0x1000)
    if target_pa is not None:
        struct.pack_into("<Q", page, i4 * 8, encode_pte(target_pa))
    mem.add(pt, bytes(page))
    return mem


def run_demo(out: str | None = None) -> dict:
    """Audit the captured channel contract and the fork classes.

    Expected verdicts (all pinned by tests):

    * captured channel contract -- all 5 low-window legs ``ok``;
    * QMD destination (VA ``0x204608000``): engine resolves to the host shadow
      while the host checks the guest page -> ``pa_mismatch``; bookkeeping
      trusting the shadow -> ``ok``; unmapped leaf -> ``unmapped``;
    * engine-write through shadow (the next wall class: CUDA output buffers at
      high VAs, where Tier-1 refuses 0x1f and the leg resolves through the
      guest scratch PT pool): the write leg resolves to a shadow PA while the
      host bookkeeping still expects the guest PA -> ``pa_mismatch``; when the
      bookkeeping is aligned to the engine-visible shadow PA -> ``ok``.
      The shadow/guest PAs in this scenario are synthetic (documented in
      docs/48), not m7 captures.
    """
    # 1) Captured channel PDB contract (validated environment probe).
    root_pa, mem = channel_pdb_fixture()
    captured = {
        "format": SPEC_FORMAT,
        "name": "captured-channel-pdb",
        "firmware": "570.144",
        "gmmu": KIND_VER2,
        "page_mode": "4k",
        "pdb_pa": f"{root_pa:#x}",
        "memory": [{"pa": f"{pa:#x}", "hex": b.hex()} for pa, b in _memory_items(mem)],
        "contract": [
            {"label": "gpfifo-ring", "va": "0x100000", "access": "read",
             "expect_pa": "0x2045d9000", "note": "front-end GPFIFO fetch"},
            {"label": "push-buffer", "va": "0x101000", "access": "write",
             "expect_pa": "0x121bd9000"},
            {"label": "copy-src", "va": "0x102000", "access": "read",
             "expect_pa": "0x152115000"},
            {"label": "copy-dst", "va": "0x103000", "access": "write",
             "expect_pa": "0x15a5f5000"},
            {"label": "semaphore", "va": "0x104000", "access": "write",
             "expect_pa": "0x10cf27000"},
        ],
    }
    captured_report = audit(captured)

    # 2) QMD-destination hypotheses, same VA (0x204608000):
    #    A: engine resolves to the host shadow (0x15e8fa000) -- host checks
    #       the guest page (0x204608000) -> pa_mismatch (the "wrong page" bug).
    #    B: no mapping / invalid leaf -> unmapped (the "copy dropped" bug).
    qmd_va = 0x204608000
    shadow_pa = 0x15E8FA000
    scenarios = []
    for label, target, expect in (
        ("qmd-dest-shadow-trusted", shadow_pa, shadow_pa),
        ("qmd-dest-host-checks-guest", shadow_pa, 0x204608000),
        ("qmd-dest-unmapped", None, None),
    ):
        root = 0x60000000
        tree = _tree_for_va(root, qmd_va, target)
        spec = {
            "format": SPEC_FORMAT, "name": label, "firmware": "570.144",
            "gmmu": KIND_VER2, "page_mode": "4k", "pdb_pa": f"{root:#x}",
            "memory": [{"pa": f"{pa:#x}", "hex": b.hex()} for pa, b in _memory_items(tree)],
            "contract": [{"label": "inline-qmd-destination", "va": f"{qmd_va:#x}",
                          "access": "write", "expect_pa": f"{expect:#x}" if expect else None,
                          "note": "front-end copies the inline QMD here"}],
        }
        report = audit(spec)
        scenarios.append({"name": label, "report": report})

    # 3) Engine-write through shadow (the next wall class).  A high guest VA
    #    is refused by Tier-1 and resolved through the guest scratch PT pool
    #    to a shadow page; the host bookkeeping for the output may still track
    #    the guest page.  Same walk, two bookkeeping states:
    #    A: host expects the guest PA -> pa_mismatch (the write is invisible).
    #    B: bookkeeping aligned to the engine-visible shadow PA -> ok.
    shadow_va = 0x210000
    shadow_target = 0x13D2A0000
    guest_pa = 0x2045C0000
    shadow_scenarios = []
    for label, expect, want in (
        ("engine-write-shadow-host-checks-guest", guest_pa, VERDICT_PA_MISMATCH),
        ("engine-write-shadow-bookkeeping-aligned", shadow_target, VERDICT_OK),
    ):
        root = 0x70000000
        tree = _tree_for_va(root, shadow_va, shadow_target)
        spec = {
            "format": SPEC_FORMAT, "name": label, "firmware": "570.144",
            "gmmu": KIND_VER2, "page_mode": "4k", "pdb_pa": f"{root:#x}",
            "memory": [{"pa": f"{pa:#x}", "hex": b.hex()} for pa, b in _memory_items(tree)],
            "contract": [{"label": "high-va-output", "va": f"{shadow_va:#x}",
                          "access": "write", "expect_pa": f"{expect:#x}",
                          "note": "Tier-2 shadow alias for a high-VA write leg"}],
        }
        report = audit(spec)
        shadow_scenarios.append({"name": label, "expected_verdict": want,
                                 "report": report})

    demo = {
        "format": DEMO_FORMAT,
        "ok": (captured_report["ok"]
               and scenarios[0]["report"]["ok"]
               and scenarios[1]["report"]["contract"][0]["verdict"] == VERDICT_PA_MISMATCH
               and scenarios[2]["report"]["contract"][0]["verdict"] == VERDICT_UNMAPPED
               and all(s["report"]["contract"][0]["verdict"] == s["expected_verdict"]
                       for s in shadow_scenarios)),
        "captured_channel": captured_report,
        "qmd_hypotheses": scenarios,
        "engine_write_shadow": shadow_scenarios,
    }
    if out:
        os.makedirs(out, exist_ok=True)
        path = os.path.join(out, "launch-audit-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 _memory_items(mem: MemoryView):
    return [(int(item["pa"], 16), mem.read(int(item["pa"], 16), item["size"]))
            for item in mem.spans()]


CAPTURE_HINTS = """\
CAPTURE HINTS (print-only; nothing is executed by this tool)
============================================================
For a real launch audit you need the *engine's* table view, not the host's
bookkeeping.  Capture, for one failing submission:

1. The channel's page-directory base PA (PDB) -- from the channel's INSTANCE
   dump field, or the SET_PAGE_DIRECTORY / GPU_PDB_ADDRESS your stack programs.
2. The table pages reachable from that root for the contract VAs: root, PDE2,
   PDE1, PDE0 (dual slot) and the leaf page table(s).  A raw PA->bytes dump of
   those 4 KB pages is enough; keep them byte-exact (no pretty-printing).
3. The submitted stream's own pointers: GPFIFO ring, inline-QMD destination
   (SET_INLINE_QMD_ADDRESS_A/B), the QMD's parameter/program/release-slot
   addresses, and the channel report semaphore.
4. Your shadow/guest bookkeeping map for those same VAs (what you currently
   check after a timeout).
5. Put 1-4 into a `greeneye-launch-audit-spec/1` JSON (see docs/48).

ITERATE WITH `check` (one capture round per table level):
  # start with only pdb_pa + contract and `"memory": []` (or a partial dump)
  python3 harness/launch_audit.py check --spec m7-launch.json
  # -> prints the FIRST missing table page per leg + `needed_pages[]`
  # dump exactly those pages, append them to memory[], re-run check
  # when check says complete:
  python3 harness/launch_audit.py run --spec m7-launch.json

CURRENT LIVE WALL (m7, 2026-10-06) -- exact dump list:
  Host-side chain (the validated 5/5 fixture; dump all five raw pages):
    root  0x12f16e000
    pd2   0x12f16f000
    pd1   0x11065e000
    pd0   0x110663000   (dual slot pd0[0].small -> pt)
    pt    0x15908d000   (pt entries 256..260 = the low-window legs)
  Forward-window PT pages (Tier-1 DMA_UPDATE_PDE_2 windows 4128 / 4131):
    0x1b8706000  (pdeIndex 4128)
    0x1197a6000  (pdeIndex 4131)
  Geometry rule: pdeIndex = (pd1_idx << 8) | pd0_idx, so the forward windows
  land at pd1 index 16.  Their pd1[16] and pd0-level chain entries are
  written by the SERVER on DMA_UPDATE_PDE_2 acceptance and are in no host
  log -- include them (and any page they point at) if they fall outside the
  host pages above; `check` names the exact PA.
  Tier-2 shadows (high-VA legs):
    * the 16 guest-scratch-PT page DMAs -- module stage-2 chunk list,
      built as `guest scratch PT built: pages=16 vaBase=0x200000
      capacity=512`; shadow VA = 0x200000 + (page << 21) + (line << 16);
    * the rmcall_gr_shadow chunk DMAs -- engine-write shadow pages for the
      high-VA legs (the next wall class after dispatch clears).
  Key contract VAs:
    qmd-dest       0x204608000  write  (QMD destination prewritten by host;
                                        post-mortem reads the same hpa)
    release-slot   0x204027f7c  write  (engine release; relw never fired)
    ring           0x100000..0x104000  read/write (GPFIFO ring, push-buffer,
                                        copy src/dst, channel semaphore)
    param-alias    0x210000     read   (Tier-2 shadow alias; NOT directly
                                        mapped; PTEs come from guest scratch)
    program-alias  0x220000     read   (same)
    indirect-alias 0x260000     read   (shadow slot 6; same)
    The high-VA param/program guest VAs (0x7e06f0.../0x735aa0...) are NOT
    directly mapped: Tier-1 refuses status 0x1f and they resolve through the
    shadow aliases above.

COPY-PASTE SPEC SKELETON (start with empty memory and let `check` drive; fill
each `hex` with the 4096-byte raw dump of the page -- no pretty-printing):
{
  "format": "greeneye-launch-audit-spec/1",
  "name": "m7-launch",
  "firmware": "570.144",
  "gmmu": "ver2",
  "page_mode": "4k",
  "pdb_pa": "0x12f16e000",
  "memory": [],
  "contract": [
    {"label": "qmd-dest",      "va": "0x204608000", "access": "write", "expect_pa": "0x204608000"},
    {"label": "release-slot",  "va": "0x204027f7c", "access": "write"},
    {"label": "gpfifo-ring",   "va": "0x100000",    "access": "read"},
    {"label": "push-buffer",   "va": "0x101000",    "access": "write"},
    {"label": "copy-src",      "va": "0x102000",    "access": "read"},
    {"label": "copy-dst",      "va": "0x103000",    "access": "write"},
    {"label": "semaphore",     "va": "0x104000",    "access": "write"},
    {"label": "param-alias",   "va": "0x210000",    "access": "read"},
    {"label": "program-alias", "va": "0x220000",    "access": "read"},
    {"label": "indirect-alias","va": "0x260000",    "access": "read"}
  ]
}
Then: python3 harness/launch_audit.py check --spec m7-launch.json

Reading the result: `pa_mismatch` means the engine reads/writes a different PA
than the one you are checking (look at the shadow/copy bookkeeping);
`unmapped` means the walk cannot reach any PA (the engine's access is dropped
or faults); `access_denied` means the leaf is read-only for a write leg.
`ok` means the engine CAN reach every leg -- the silence is downstream of the
address path (front-end/engine/context), not in the mapping.
"""


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        prog="launch_audit.py",
        description="Audit a submission's engine-visible memory contract (offline).",
    )
    sub = parser.add_subparsers(dest="command")
    run = sub.add_parser("run", help="run an audit spec")
    run.add_argument("--spec", required=True)
    run.add_argument("--out", help="write the report JSON to this directory")
    check_cmd = sub.add_parser(
        "check", help="report the first table pages a partial/empty dump is missing")
    check_cmd.add_argument("--spec", required=True)
    check_cmd.add_argument("--out", help="write the report JSON to this directory")
    demo = sub.add_parser("demo", help="captured contract + QMD/shadow hypotheses")
    demo.add_argument("--out", help="output directory")
    sub.add_parser("capture-hints", help="print what to capture for a real audit")
    return parser


def _write_out(report: dict, out: str, name: str) -> str:
    os.makedirs(out, exist_ok=True)
    path = os.path.join(out, name)
    with open(path, "w", encoding="utf-8") as fh:
        json.dump(report, fh, indent=2, sort_keys=True)
        fh.write("\n")
    return os.path.abspath(path)


def main(argv: list[str] | None = None) -> int:
    parser = build_parser()
    args = parser.parse_args(argv)
    try:
        if args.command == "run":
            if not os.path.isfile(args.spec):
                raise AuditError(f"spec not found: {args.spec}")
            with open(args.spec, "r", encoding="utf-8") as fh:
                spec = json.load(fh)
            report = audit(spec)
            for row in report["contract"]:
                print(f"{row['verdict']:<14} {row['label']:<28} {row['va']}"
                      + (f"  {row['detail']}" if row.get("detail") else ""))
            summary = {"format": report["format"], "ok": report["ok"],
                       "summary": report["summary"]}
            if args.out:
                summary["report_json"] = _write_out(report, args.out, "launch-audit.json")
            print(json.dumps(summary, indent=2, sort_keys=True))
            return 0 if report["ok"] else 3
        if args.command == "check":
            if not os.path.isfile(args.spec):
                raise AuditError(f"spec not found: {args.spec}")
            with open(args.spec, "r", encoding="utf-8") as fh:
                spec = json.load(fh)
            report = check(spec)
            for row in report["contract"]:
                if row["status"] == "missing":
                    miss = row["first_missing"]
                    print(f"missing        {row['label']:<28} {row['va']}"
                          f"  need {miss['pa']} ({miss['level']}[{miss['index']}])")
                elif row["status"] == "complete":
                    print(f"complete       {row['label']:<28} {row['va']}"
                          f"  -> {row['resolved_pa']}")
                else:
                    print(f"fault          {row['label']:<28} {row['va']}"
                          f"  {row['fault']}: {row['detail']}")
            if report["needed_pages"]:
                print("dump next:")
                for page in report["needed_pages"]:
                    print(f"  {page['pa']}  {'/'.join(page['levels'])}"
                          f"  legs: {', '.join(page['legs'])}")
            summary = {
                "format": report["format"],
                "complete": report["complete"],
                "needed_pages": [page["pa"] for page in report["needed_pages"]],
            }
            if args.out:
                summary["report_json"] = _write_out(
                    report, args.out, "launch-audit-check.json")
            print(json.dumps(summary, indent=2, sort_keys=True))
            return 0 if report["complete"] else 3
        if args.command == "demo":
            demo = run_demo(args.out)
            cap = demo["captured_channel"]
            print(f"captured channel contract: ok={cap['ok']} {cap['summary']}")
            for scenario in demo["qmd_hypotheses"]:
                row = scenario["report"]["contract"][0]
                print(f"{row['verdict']:<14} {scenario['name']}  {row.get('detail') or ''}")
            for scenario in demo["engine_write_shadow"]:
                row = scenario["report"]["contract"][0]
                print(f"{row['verdict']:<14} {scenario['name']}  {row.get('detail') or ''}")
            print(json.dumps({"ok": demo["ok"]}, indent=2))
            return 0 if demo["ok"] else 3
        if args.command == "capture-hints":
            print(CAPTURE_HINTS)
            return 0
        parser.print_help()
        return 2
    except (AuditError, WalkError) as exc:
        print(f"error: {exc}", file=sys.stderr)
        return 2


if __name__ == "__main__":
    raise SystemExit(main())