1 files changed, 396 insertions, 0 deletions
diff --git a/debug/risc-v-gsp-rm-vm/harness/contracts.py b/debug/risc-v-gsp-rm-vm/harness/contracts.py
new file mode 100644
index 0000000..ec387fe
--- /dev/null
+++ b/debug/risc-v-gsp-rm-vm/harness/contracts.py
@@ -0,0 +1,396 @@
+#!/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],
+    }