#!/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(" 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())