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

#!/usr/bin/env python3
"""Contract registry: every solved wall becomes a re-runnable regression.

Why this exists
---------------
The GR campaign's dominant cost is wall-hell: one blocking firmware problem
after another, each triaged by supervised hardware cycles. Several wall
classes now have offline, deterministic tools (docs/45-48), but their value is
only durable if the *answer* survives as a re-runnable check. This registry
pins the proven entry points of those tools to one explicit expectation each
and re-runs them: fast contracts run in the main suite, slow ones on demand.

It wraps existing entry points as-is (no measurement logic is reimplemented)
and every contract names its wall class from ``docs/50-WALL-WORKFLOW.md``, so
per-domain coverage is visible - including the classes with no offline
contract by design (engine execution awaits supervised captures).

Offline only: never touches a GPU, kernel module, mdev, guest or network.
"""
from __future__ import annotations

import os
import time
from dataclasses import dataclass
from typing import Any, Callable

TOOL_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))

from harness.rehost import DEFAULT_IMAGE, DEMO_ELF  # noqa: E402

#: Schema tag of a registry run summary.
RUN_FORMAT = "greeneye-contract-run/1"

FAST = "fast"
SLOW = "slow"

#: Wall classes from docs/50 section 2.  A contract names exactly one.
WALL_CLASSES = {
    "a": "firmware boot/status",
    "b": "RM server behaviour",
    "c": "engine memory contract",
    "d": "engine execution/dispatch",
    "e": "guest-driver/protocol interface",
    "f": "version skew/portability",
}


class ContractError(Exception):
    """A contract record or registry request is invalid."""


class ContractSkip(Exception):
    """A contract precondition is missing at run time (not a pass/fail)."""


@dataclass(frozen=True)
class Contract:
    """One re-runnable regression over an existing tool entry point."""

    name: str
    wall_class: str          # key into WALL_CLASSES (docs/50 section 2)
    doc: str                 # where the wall and the tool are documented
    entry: str               # exact wrapped entry point, e.g. "sim.gmmu_walk:run_demo()"
    speed: str               # FAST runs in the suite; SLOW is --all only
    expect: str              # the pass condition, in words
    run: Callable[[], dict]  # no-arg callable returning the tool's report
    check: Callable[[dict], tuple[bool, str]]  # (passed, detail)
    requires: tuple[str, ...] = ()  # paths that must exist; otherwise skip

    @property
    def domain(self) -> str:
        return WALL_CLASSES.get(self.wall_class, self.wall_class)


@dataclass(frozen=True)
class ContractResult:
    """Outcome of one contract run; ``status`` is pass | fail | skip."""

    name: str
    wall_class: str
    speed: str
    doc: str
    entry: str
    expect: str
    status: str
    detail: str
    seconds: float

    @property
    def domain(self) -> str:
        return WALL_CLASSES.get(self.wall_class, self.wall_class)

    @property
    def passed(self) -> bool:
        return self.status == "pass"

    def to_dict(self) -> dict:
        return {
            "name": self.name,
            "wall_class": self.wall_class,
            "domain": self.domain,
            "speed": self.speed,
            "doc": self.doc,
            "entry": self.entry,
            "expect": self.expect,
            "status": self.status,
            "detail": self.detail,
            "seconds": round(self.seconds, 4),
        }


# ---------------------------------------------------------------------------
# Runners: thin wrappers over the proven tool entry points.
#
# Every wrapper only imports, calls and (where the demo writes artifacts)
# suppresses output paths.  The expectation lives in the checker next to it.
# ---------------------------------------------------------------------------

def _run_fecs_matrix() -> dict:
    from harness.valprobe import fecs_spec, run_spec

    return run_spec(spec=fecs_spec(), record=False, name="contract-fecs-matrix")


def _check_fecs_matrix(report: dict) -> tuple[bool, str]:
    variants = report.get("variants", [])
    evaluated = report.get("evaluated")
    failed = report.get("failed")
    ok = (
        report.get("format") == "greeneye-valprobe-matrix/1"
        and len(variants) == 15
        and evaluated == 15
        and failed == 0
        and all(row.get("pass") is True for row in variants)
    )
    return ok, f"{evaluated}/{len(variants)} PROMOTE_CTX variants labelled, {failed} failed"


def _run_fecs_determinism() -> dict:
    from harness.valprobe import run_demo

    return run_demo()


def _check_fecs_determinism(report: dict) -> tuple[bool, str]:
    ok = (
        report.get("format") == "greeneye-valprobe-demo/1"
        and report.get("ok") is True
        and report.get("deterministic") is True
        and report.get("failed") == 0
    )
    return ok, (
        f"evaluated={report.get('evaluated')} failed={report.get('failed')} "
        f"deterministic={report.get('deterministic')}"
    )


def _run_gmmu_demo() -> dict:
    from sim.gmmu_walk import run_demo

    return run_demo()


def _check_gmmu_demo(report: dict) -> tuple[bool, str]:
    rows = report.get("rows", [])
    matched = sum(1 for row in rows if row.get("match") is True)
    ok = report.get("format") == "greeneye-gmmu-walk-demo/1" and report.get("ok") is True \
        and matched == len(rows) == 5
    return ok, f"{matched}/{len(rows)} VAs walk to the captured DMA addresses"


def _run_launch_audit_demo() -> dict:
    from harness.launch_audit import run_demo

    return run_demo()


def _check_launch_audit_demo(report: dict) -> tuple[bool, str]:
    from harness.launch_audit import VERDICT_OK, VERDICT_PA_MISMATCH, VERDICT_UNMAPPED

    legs = report.get("captured_channel", {}).get("contract", [])
    ok_legs = sum(1 for leg in legs if leg.get("verdict") == VERDICT_OK)
    classes = {
        (scenario.get("report", {}).get("contract") or [{}])[0].get("verdict")
        for scenario in report.get("qmd_hypotheses", [])
    }
    ok = (
        report.get("format") == "greeneye-launch-audit-demo/1"
        and report.get("ok") is True
        and ok_legs == len(legs) == 5
        and {VERDICT_PA_MISMATCH, VERDICT_UNMAPPED} <= classes
    )
    return ok, (
        f"captured channel {ok_legs}/{len(legs)} legs ok; "
        f"QMD verdict classes {sorted(c for c in classes if c)}"
    )


def _run_rehost_demo() -> dict:
    from harness.rehost import run_demo

    return run_demo()


def _check_rehost_demo(report: dict) -> tuple[bool, str]:
    first = report.get("first", {})
    second = report.get("second", {})
    agree = (
        first.get("status") == second.get("status")
        and first.get("retval") == second.get("retval")
    )
    ok = (
        report.get("format") == "greeneye-rehost-demo/1"
        and report.get("ok") is True
        and first.get("status") == "return"
        and first.get("retval") == report.get("expected")
    )
    return ok, (
        f"status={first.get('status')} retval={first.get('retval')} "
        f"(expected {report.get('expected')}); two runs agree={agree}"
    )


def _run_rm_capture_demo() -> dict:
    from harness.rm_capture import run_demo

    return run_demo()


def _check_rm_capture_demo(report: dict) -> tuple[bool, str]:
    summary = report.get("summary", {})
    ok = report.get("format") == "greeneye-rm-rpc-capture-demo/1" and report.get("ok") is True
    return ok, (
        f"records {report.get('records_a')}/{report.get('records_b')}; "
        f"stimulus messages {report.get('messages')}; "
        f"agree={summary.get('agree')} shape_agree={summary.get('shape_agree')} "
        f"a_only={summary.get('a_only')} b_only={summary.get('b_only')}"
    )


# ---------------------------------------------------------------------------
# Registry
# ---------------------------------------------------------------------------

def _registry() -> list[Contract]:
    return [
        Contract(
            name="fecs-predicate-matrix",
            wall_class="b",
            doc="docs/47, docs/50",
            entry="harness.valprobe:run_spec(fecs_spec())",
            speed=FAST,
            expect="15/15 PROMOTE_CTX predicate variants reach their labelled outcome (failed == 0)",
            requires=(DEFAULT_IMAGE,),
            run=_run_fecs_matrix,
            check=_check_fecs_matrix,
        ),
        Contract(
            name="fecs-predicate-determinism",
            wall_class="b",
            doc="docs/47, docs/25, docs/50",
            entry="harness.valprobe:run_demo()",
            speed=SLOW,
            expect="the bundled matrix runs twice with identical verdicts (ok and deterministic)",
            requires=(DEFAULT_IMAGE,),
            run=_run_fecs_determinism,
            check=_check_fecs_determinism,
        ),
        Contract(
            name="gmmu-golden-vector",
            wall_class="c",
            doc="docs/48, docs/50",
            entry="sim.gmmu_walk:run_demo()",
            speed=FAST,
            expect="the captured channel-PDB vector walks 5/5 VAs to the logged DMA addresses",
            run=_run_gmmu_demo,
            check=_check_gmmu_demo,
        ),
        Contract(
            name="launch-audit-captured-contract",
            wall_class="c",
            doc="docs/48, docs/50",
            entry="harness.launch_audit:run_demo()",
            speed=FAST,
            expect="captured contract all ok; QMD hypotheses classify pa_mismatch and unmapped",
            run=_run_launch_audit_demo,
            check=_check_launch_audit_demo,
        ),
        Contract(
            name="rehost-demo",
            wall_class="b",
            doc="docs/46, docs/25, docs/50",
            entry="harness.rehost:run_demo()",
            speed=FAST,
            expect="the demo function returns the expected sum twice (ok == true)",
            requires=(DEMO_ELF,),
            run=_run_rehost_demo,
            check=_check_rehost_demo,
        ),
        Contract(
            name="rm-capture-decode-demo",
            wall_class="e",
            doc="docs/45, docs/25, docs/50",
            entry="harness.rm_capture:run_demo()",
            speed=FAST,
            expect="synthetic capture decodes, compares and yields a stimulus (ok == true)",
            run=_run_rm_capture_demo,
            check=_check_rm_capture_demo,
        ),
    ]


def all_contracts() -> list[Contract]:
    """Every contract in the registry, fast and slow."""
    return _registry()


def fast_contracts() -> list[Contract]:
    """Contracts the main suite runs."""
    return [c for c in all_contracts() if c.speed == FAST]


def slow_contracts() -> list[Contract]:
    """Contracts only ``run_contracts.py --all`` runs."""
    return [c for c in all_contracts() if c.speed == SLOW]


def _result(contract: Contract, status: str, detail: str, seconds: float) -> ContractResult:
    return ContractResult(
        name=contract.name,
        wall_class=contract.wall_class,
        speed=contract.speed,
        doc=contract.doc,
        entry=contract.entry,
        expect=contract.expect,
        status=status,
        detail=detail,
        seconds=seconds,
    )


def run_contract(contract: Contract) -> ContractResult:
    """Run one contract, classifying every outcome without raising."""
    missing = [path for path in contract.requires if not os.path.exists(path)]
    if missing:
        return _result(contract, "skip", "prerequisite missing: " + ", ".join(missing), 0.0)
    started = time.monotonic()
    try:
        report = contract.run()
    except ContractSkip as exc:
        return _result(contract, "skip", str(exc), time.monotonic() - started)
    except Exception as exc:  # noqa: BLE001 - a raising entry point is a failed contract
        return _result(
            contract, "fail", f"{type(exc).__name__}: {exc}", time.monotonic() - started
        )
    elapsed = time.monotonic() - started
    try:
        passed, detail = contract.check(report)
    except Exception as exc:  # noqa: BLE001 - a broken checker is a failed contract
        return _result(contract, "fail", f"checker error: {type(exc).__name__}: {exc}", elapsed)
    return _result(contract, "pass" if passed else "fail", str(detail), elapsed)


def run_contracts(contracts: list[Contract]) -> list[ContractResult]:
    """Run the given contracts in order; never raises for a contract outcome."""
    return [run_contract(contract) for contract in contracts]


def summarize(results: list[ContractResult]) -> dict:
    """Machine-readable run summary with per-wall-class coverage."""
    counts = {"pass": 0, "fail": 0, "skip": 0}
    domains: dict[str, dict[str, Any]] = {
        key: {
            "wall_class": key,
            "domain": label,
            "total": 0,
            "pass": 0,
            "fail": 0,
            "skip": 0,
        }
        for key, label in WALL_CLASSES.items()
    }
    for result in results:
        if result.status not in counts:
            raise ContractError(f"unknown contract status {result.status!r}")
        counts[result.status] += 1
        bucket = domains[result.wall_class]
        bucket["total"] += 1
        bucket[result.status] += 1
    return {
        "format": RUN_FORMAT,
        "ok": counts["fail"] == 0,
        "counts": counts,
        "domains": [domains[key] for key in sorted(domains)],
        "contracts": [result.to_dict() for result in results],
    }