#!/usr/bin/env python3 """Host-accuracy oracle comparison for GMMU/FBIF register observations. Ingests two files and compares them semantically: * an **oracle capture** (schema ``greeneye-gmmu-oracle/1``): a list of ``{name, window, offset, value, source, captured_at}`` items produced by a human from read-only RM queries or supervised captures; * a **simulator observation** (G1's ``greeneye-gmmu-fbif-trace/1``, or a session directory containing one), ingested by :func:`harness.gmmu_debug.load_sim_trace`. The comparison reports per-register value agreement, coverage (oracle items with no simulator observation), and simulator accesses with no oracle counterpart, with an explicit fidelity label per row (observed vs modelled vs unknown). It emits ``gmmu-oracle-report.json`` and ``.md``. ``--print-capture-hints`` prints (and only prints) the supervised, read-only commands a human would run to produce an oracle. Physical work is supervised-only per RUNBOOK P1/P7; an agent must never run those commands. No cycle/timing claim is made anywhere. This tool only reads files and writes reports; it never touches a GPU, module, mdev, guest or network. """ from __future__ import annotations import argparse import json import os import sys from dataclasses import dataclass, field 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 harness.gmmu_debug import ( # noqa: E402 GmmuInputError, SimTrace, TraceObservation, load_sim_trace, parse_int, sha256_file, ) #: Schema tag of the human-produced oracle capture. ORACLE_FORMAT = "greeneye-gmmu-oracle/1" #: Schema tag of the report this module emits. REPORT_FORMAT = "greeneye-gmmu-oracle-report/1" CAPTURE_HINTS = """\ ================================================================================ GMMU/FBIF ORACLE CAPTURE HINTS — SUPERVISED, READ-ONLY, HUMAN-RUN ONLY ================================================================================ This command prints instructions; it executes NOTHING. HARD RULES (RUNBOOK P1/P7, SIMULATOR-WORK-BOUNDARY.txt): * Physical work is supervised-only: a named human present, one cycle at a time. An agent MUST NOT run any command below. * Read-only means exactly that: no module load/unload, no mdev create or claim, no bind/unbind, no guest access, no /sys/fs/pstore, no reboot. * The RM tools used here allocate only an RM client/device/subdevice handle pair and issue allowlisted read-only controls (docs/17-RM-HARNESS.md §6). * A simulator observation is not silicon truth; only a supervised capture is. Product changes still need physical reproduction. STEP 1 — read-only RM inventory (record the session, human-run): cd open-iov/debug/risc-v-gsp-rm-vm ./rmharness/harness.sh --session dumps/real-rm-before --name gmmu-oracle \\ -- rmtool gpu list --format=json ./rmharness/harness.sh --session dumps/real-rm-before --name gmmu-oracle \\ -- rmtool engine list --format=json ./rmharness/harness.sh --session dumps/real-rm-before --name gmmu-oracle \\ -- rmtool mem stats --format=json ./rmharness/harness.sh --session dumps/real-rm-before --name gmmu-oracle \\ -- rmtop --once # allowlisted read-only control values (never a write control): ./rmharness/rmtool.sh nv2080 method call --help STEP 2 — sysfs/proc/debugfs readbacks (read-only `cat`): cat /proc/driver/nvidia/gpus/*/information lspci -vv -s # config space readback cat /sys/bus/pci/devices//resource # BAR ranges/sizes # vgsp debugfs (host module; read-only reads): ls /sys/kernel/debug/vgsp/ cat /sys/kernel/debug/vgsp/gsp-phys-* # name the exact file per readback STEP 3 — optional supervised first-touch capture: A simulator observation is produced without hardware (safe, agent-allowed): python3 -m sim.gsp_boot --instructions 2000000 --record dumps/gmmu-sim A physical first-touch capture is a supervised human step and must not be scripted by an agent. STEP 4 — write the oracle file (human, from the readbacks above): { "format": "greeneye-gmmu-oracle/1", "captures": [ {"name": "FBIF_CTL", "window": "falcon", "offset": "0x624", "value": "0x1", "source": "rm-readback", "captured_at": ""}, {"name": "BOOT_42", "window": "pmc", "offset": "0xa00", "value": "0x164", "source": "sysfs-readback", "captured_at": ""} ] } Then compare (safe, offline): python3 harness/gmmu_oracle.py compare \\ --oracle oracle.json --sim dumps/gmmu-sim --out dumps/gmmu-oracle-report ================================================================================ """ @dataclass class OracleItem: index: int name: str window: str offset: int | None addr: int | None value: int | None source: str captured_at: str raw: dict = field(default_factory=dict) @dataclass class OracleCapture: path: str format: str sha256: str items: list[OracleItem] = field(default_factory=list) def load_oracle(path: str) -> OracleCapture: """Load a ``greeneye-gmmu-oracle/1`` capture (object wrapper or bare list). A missing ``format`` is tolerated and reported as ``unlabeled-list`` / ``unlabeled-object``; a different explicit format is a clean error. """ if not os.path.isfile(path): raise GmmuInputError(f"oracle not found: {path}") digest = sha256_file(path) try: with open(path, "r", encoding="utf-8", errors="replace") as fh: payload = json.load(fh) except (OSError, json.JSONDecodeError) as exc: raise GmmuInputError(f"cannot read oracle {path}: {exc}") from exc if isinstance(payload, list): fmt = "unlabeled-list" items_payload: list = payload elif isinstance(payload, dict): fmt = str(payload.get("format") or payload.get("schema") or "unlabeled-object") if fmt not in (ORACLE_FORMAT, "unlabeled-object"): raise GmmuInputError( f"{path}: format {fmt!r} is not {ORACLE_FORMAT!r}" ) found = None for key in ("captures", "items", "registers", "observations", "oracle"): candidate = payload.get(key) if isinstance(candidate, list): found = candidate break if found is None: raise GmmuInputError(f"{path}: no capture list (expected 'captures')") items_payload = found else: raise GmmuInputError(f"{path}: oracle must be an object or list") items: list[OracleItem] = [] for index, raw in enumerate(items_payload): if not isinstance(raw, dict): continue name = str(raw.get("name") or raw.get("register") or "") window = str(raw.get("window") or raw.get("block") or "") offset = parse_int(raw.get("offset", raw.get("reg_offset"))) addr = parse_int(raw.get("addr", raw.get("address"))) value = parse_int(raw.get("value")) if not name: if offset is not None: name = f"0x{offset:x}" elif addr is not None: name = f"0x{addr:x}" else: name = f"item_{index}" items.append( OracleItem( index=index, name=name, window=window, offset=offset, addr=addr, value=value, source=str(raw.get("source") or "unlabelled"), captured_at=str(raw.get("captured_at") or ""), raw=raw, ) ) return OracleCapture( path=os.path.abspath(path), format=fmt, sha256=digest, items=items ) def _obs_keys(obs: TraceObservation) -> list[tuple]: keys: list[tuple] = [] if obs.offset is not None: if obs.window: keys.append(("window-offset", obs.window, obs.offset)) keys.append(("offset", obs.offset)) if obs.name and obs.window: keys.append(("window-name", obs.window, obs.name)) if obs.name: keys.append(("name", obs.name)) if obs.addr is not None: keys.append(("addr", obs.addr)) return keys def _item_keys(item: OracleItem) -> list[tuple]: keys: list[tuple] = [] if item.offset is not None: if item.window: keys.append(("window-offset", item.window, item.offset)) keys.append(("offset", item.offset)) if item.name and item.window: keys.append(("window-name", item.window, item.name)) if item.name: keys.append(("name", item.name)) if item.addr is not None: keys.append(("addr", item.addr)) return keys def _group_observations(observations: list[TraceObservation]) -> tuple[list[dict], dict]: """Cluster trace observations into registers; return (groups, key index).""" groups: list[dict] = [] index: dict[tuple, int] = {} for obs in observations: keys = _obs_keys(obs) gidx = None for key in keys: if key in index: gidx = index[key] break if gidx is None: gidx = len(groups) groups.append({"observations": []}) groups[gidx]["observations"].append(obs) for key in keys: index.setdefault(key, gidx) return groups, index def _group_descriptor(group: dict) -> dict: observations: list[TraceObservation] = group["observations"] first = observations[0] last = observations[-1] writes = [o for o in observations if o.kind == "write"] reads = [o for o in observations if o.kind == "read"] value_obs = writes[-1] if writes else (reads[-1] if reads else last) out: dict[str, Any] = { "register": first.name or ( f"0x{first.offset:x}" if first.offset is not None else (f"0x{first.addr:x}" if first.addr is not None else "?") ), "observations": len(observations), "writes": len(writes), "reads": len(reads), } if first.window: out["window"] = first.window if first.offset is not None: out["offset"] = first.offset if first.addr is not None: out["addr"] = first.addr if value_obs.value is not None: out["last_value"] = value_obs.value out["fidelity"] = "modelled (no oracle counterpart)" else: out["fidelity"] = "modelled (value unknown; no oracle counterpart)" pcs = [o.pc for o in observations if o.pc is not None] if pcs: out["first_pc"] = pcs[0] out["last_pc"] = pcs[-1] return out def compare(oracle: OracleCapture, trace: SimTrace) -> dict: """Semantic oracle-vs-simulator comparison; no timing claims.""" groups, index = _group_observations(trace.observations) matched_groups: set[int] = set() rows: list[dict] = [] unmatched_oracle: list[dict] = [] for item in oracle.items: gidx = None matched_by = "" for key in _item_keys(item): if key in index: gidx = index[key] matched_by = key[0] break if gidx is None: unmatched_oracle.append( { "register": item.name, "window": item.window, "offset": item.offset, "oracle_value": item.value, "source": item.source, "captured_at": item.captured_at, "fidelity": "observed (no simulator counterpart)", } ) rows.append( { "register": item.name, "window": item.window, "offset": item.offset, "oracle_value": item.value, "sim_value": None, "status": "oracle-only", "matched_by": "", "fidelity": "observed (no simulator counterpart)", "source": item.source, "captured_at": item.captured_at, } ) continue matched_groups.add(gidx) group = groups[gidx] observations: list[TraceObservation] = group["observations"] writes = [o for o in observations if o.kind == "write"] reads = [o for o in observations if o.kind == "read"] value_obs = writes[-1] if writes else (reads[-1] if reads else observations[-1]) sim_value = value_obs.value if item.value is None or sim_value is None: status = "no-value" fidelity = "observed vs modelled (value unknown)" elif sim_value == item.value: status = "agree" fidelity = "observed vs modelled" else: status = "mismatch" fidelity = "observed vs modelled" value_ever = any(o.value == item.value for o in observations if o.value is not None) pcs = [o.pc for o in observations if o.pc is not None] row = { "register": item.name, "window": item.window, "offset": item.offset, "oracle_value": item.value, "sim_value": sim_value, "status": status, "matched_by": matched_by, "fidelity": fidelity, "source": item.source, "captured_at": item.captured_at, "sim_observations": len(observations), "sim_writes": len(writes), "sim_reads": len(reads), "sim_value_ever_agreed": value_ever, "sim_first_pc": pcs[0] if pcs else None, "sim_last_pc": pcs[-1] if pcs else None, } rows.append(row) unmatched_simulator = [ _group_descriptor(group) for gidx, group in enumerate(groups) if gidx not in matched_groups ] agreed = sum(1 for row in rows if row["status"] == "agree") mismatched = sum(1 for row in rows if row["status"] == "mismatch") no_value = sum(1 for row in rows if row["status"] == "no-value") compared = agreed + mismatched + no_value oracle_items = len(oracle.items) denominator = agreed + mismatched agreement_pct = round(100.0 * agreed / denominator, 2) if denominator else None coverage_pct = round(100.0 * compared / oracle_items, 2) if oracle_items else None sim_registers = len(groups) simulator_match_pct = ( round(100.0 * len(matched_groups) / sim_registers, 2) if sim_registers else None ) return { "format": REPORT_FORMAT, "inputs": { "oracle": { "path": oracle.path, "format": oracle.format, "sha256": oracle.sha256, "items": oracle_items, }, "simulator": { "path": trace.path, "format": trace.format, "sha256": trace.sha256, "observations": len(trace.observations), "registers": sim_registers, "source": trace.source, }, }, "summary": { "oracle_items": oracle_items, "simulator_registers": sim_registers, "simulator_observations": len(trace.observations), "compared": compared, "agreed": agreed, "mismatched": mismatched, "no_value": no_value, "oracle_only": len(unmatched_oracle), "simulator_only": len(unmatched_simulator), "agreement_pct": agreement_pct, "coverage_pct": coverage_pct, "simulator_match_pct": simulator_match_pct, "claim": ( "per-register value agreement on the oracle/simulator intersection; " "not a timing, silicon or full-coverage claim" ), }, "rows": rows, "unmatched_oracle": unmatched_oracle, "unmatched_simulator": unmatched_simulator, "fidelity": { "oracle": "observed (human read-only capture; supervised for physical work)", "simulator": "modelled (device model / recorded trace)", "levels": "L0/L1 only; L2 timing and L3 device environment not claimed (docs/06)", }, "limits": [ "No live oracle capture was taken by this tool; it only ingests a file a human produced.", "Agreement is per-register on the matched intersection, never a global accuracy percentage.", "An oracle-only register means the simulator did not touch it under this stimulus; it is not proof of a device-model gap.", "A simulator-only register means no oracle counterpart exists yet; it is not evidence of a wrong value.", "Values are point observations, not timing or ordering claims.", ], } def render_oracle_markdown(report: dict) -> str: summary = report["summary"] lines = [ "# GMMU/FBIF oracle vs simulator report", "", f"> Format `{report['format']}`. Fidelity: {report['fidelity']['levels']}.", "> No cycle/timing claim. Underclaim every number (docs/06, docs/44).", "", "## Inputs", "", f"* oracle: `{report['inputs']['oracle']['path']}` " f"(format `{report['inputs']['oracle']['format']}`, " f"{report['inputs']['oracle']['items']} items, " f"sha256 `{report['inputs']['oracle']['sha256'][:16]}…`)", f"* simulator: `{report['inputs']['simulator']['path']}` " f"(format `{report['inputs']['simulator']['format']}`, " f"{report['inputs']['simulator']['registers']} registers / " f"{report['inputs']['simulator']['observations']} observations, " f"sha256 `{report['inputs']['simulator']['sha256'][:16]}…`)", "", "## Summary", "", "| oracle items | compared | agreed | mismatched | no-value | oracle-only | sim-only | agreement % | coverage % |", "|---|---|---|---|---|---|---|---|---|", f"| {summary['oracle_items']} | {summary['compared']} | {summary['agreed']} | " f"{summary['mismatched']} | {summary['no_value']} | {summary['oracle_only']} | " f"{summary['simulator_only']} | " f"{summary['agreement_pct'] if summary['agreement_pct'] is not None else 'n/a'} | " f"{summary['coverage_pct'] if summary['coverage_pct'] is not None else 'n/a'} |", "", f"* {summary['claim']}", "", "## Per-register rows", "", "| register | window | offset | oracle | simulator | status | fidelity | matched by |", "|---|---|---|---|---|---|---|---|", ] for row in report["rows"]: offset = f"{row['offset']:#x}" if row.get("offset") is not None else "" oracle_value = f"{row['oracle_value']:#x}" if row.get("oracle_value") is not None else "n/a" sim_value = f"{row['sim_value']:#x}" if row.get("sim_value") is not None else "n/a" lines.append( f"| `{row['register']}` | {row.get('window', '')} | {offset} | {oracle_value} | " f"{sim_value} | {row['status']} | {row['fidelity']} | {row.get('matched_by', '')} |" ) lines += ["", "## Oracle items with no simulator observation", ""] if report["unmatched_oracle"]: for item in report["unmatched_oracle"]: offset = f"{item['offset']:#x}" if item.get("offset") is not None else "?" value = f"{item['oracle_value']:#x}" if item.get("oracle_value") is not None else "n/a" lines.append( f"* `{item['register']}` ({item.get('window', '')} @ {offset}) = {value} " f"[{item['source']}]" ) else: lines.append("* none") lines += ["", "## Simulator registers with no oracle counterpart", ""] if report["unmatched_simulator"]: for item in report["unmatched_simulator"]: offset = f"{item['offset']:#x}" if item.get("offset") is not None else "?" value = f"{item['last_value']:#x}" if item.get("last_value") is not None else "n/a" lines.append( f"* `{item['register']}` ({item.get('window', '')} @ {offset}) last={value} " f"[{item['writes']}w/{item['reads']}r]" ) else: lines.append("* none") lines += ["", "## Limits", ""] for limit in report["limits"]: lines.append(f"* {limit}") lines.append("") return "\n".join(lines) def run_compare(oracle_path: str, sim_path: str, out_dir: str) -> dict: """Compare an oracle capture with a simulator trace; write JSON + MD.""" oracle = load_oracle(oracle_path) trace = load_sim_trace(sim_path) report = compare(oracle, trace) os.makedirs(out_dir, exist_ok=True) report_json = os.path.join(out_dir, "gmmu-oracle-report.json") report_md = os.path.join(out_dir, "gmmu-oracle-report.md") with open(report_json, "w", encoding="utf-8") as fh: json.dump(report, fh, indent=2, sort_keys=True) fh.write("\n") with open(report_md, "w", encoding="utf-8") as fh: fh.write(render_oracle_markdown(report)) report["artifacts"] = { "report_json": os.path.abspath(report_json), "report_md": os.path.abspath(report_md), } return report def print_capture_hints(out=None) -> None: """Print the supervised, read-only capture instructions; execute nothing.""" print(CAPTURE_HINTS, file=out or sys.stdout) def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser( prog="gmmu_oracle.py", description=( "Compare a human oracle capture with a simulator GMMU/FBIF trace. " "Offline only; never executes physical commands." ), ) parser.add_argument( "--print-capture-hints", action="store_true", help="print the supervised read-only commands a human would run (executes nothing)", ) sub = parser.add_subparsers(dest="command") compare_cmd = sub.add_parser("compare", help="compare oracle vs simulator observation") compare_cmd.add_argument("--oracle", required=True, help="greeneye-gmmu-oracle/1 file") compare_cmd.add_argument("--sim", required=True, help="trace file or session directory") compare_cmd.add_argument("--out", required=True, help="output directory") return parser def main(argv: list[str] | None = None) -> int: parser = build_parser() args = parser.parse_args(argv) if args.print_capture_hints: print_capture_hints() return 0 if args.command == "compare": try: report = run_compare(args.oracle, args.sim, args.out) except GmmuInputError as exc: print(f"error: {exc}", file=sys.stderr) return 2 print( json.dumps( { "format": report["format"], "summary": report["summary"], "artifacts": report["artifacts"], }, indent=2, sort_keys=True, ) ) return 0 parser.print_help() return 2 if __name__ == "__main__": raise SystemExit(main())