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

#!/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 <bdf>                       # config space readback
  cat /sys/bus/pci/devices/<bdf>/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": "<date>"},
      {"name": "BOOT_42", "window": "pmc", "offset": "0xa00",
       "value": "0x164", "source": "sysfs-readback", "captured_at": "<date>"}
    ]
  }
  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())