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

#!/usr/bin/env python3
"""Shared plumbing for the risc-v-gsp-rm-debugger value-validation demos.

Every demo maps one *real* Open-IOV scenario (cited from the handoffs) onto a
runnable demonstration on the bundled demo firmware / native RV64 simulator,
prints the question, the risc-v-gsp-rm-debugger commands, the answer, the evidence artifact
and the measured wall-clock, and emits one machine-readable ``RESULT_JSON``
line so ``run_all.py`` can aggregate.

Stdlib only; no hardware; runs from the tool root
(``open-iov/debug/risc-v-gsp-rm-vm``).
"""
from __future__ import annotations

import hashlib
import json
import os
import shutil
import sys
import tempfile
import time

HERE = os.path.dirname(os.path.abspath(__file__))
TOOL_ROOT = os.path.dirname(os.path.dirname(HERE))
if TOOL_ROOT not in sys.path:
    sys.path.insert(0, TOOL_ROOT)

ARTIFACT_ROOT = os.path.join(HERE, "artifacts")
DEMO_ELF = os.path.join(TOOL_ROOT, "firmware", "uproc_demo", "build", "uproc_demo.elf")
SCHEMA_JSON = os.path.join(TOOL_ROOT, "firmware", "uproc_demo", "gsp_ctrl.schema.json")
SDK_ELF = os.path.join(TOOL_ROOT, "sdk", "build", "injected_demo.elf")

CTRL_MAGIC = 0x47535031
CTRL_RPC_COUNT = 0x10
CTRL_LAST_RPC = 0x18
CTRL_MAILBOX = 0x20
CTRL_ARG0 = 0x28
CTRL_ARG1 = 0x30
CTRL_ARG2 = 0x38
CTRL_RESULT = 0x40
CTRL_BUF = 0x48
CTRL_SIZE = 0x190
PING_XOR = 0xBEEF
INJECT_CMD_BASE = 16
SERVICE_SLOT = 6

#: Per-demo wall-clock ceiling the runner enforces (subprocess timeout).
DEMO_TIMEOUT_S = 45.0
#: Runner ceiling so the whole suite stays well under ~3 min.
SUITE_BUDGET_S = 180.0


class DemoError(RuntimeError):
    """A demo precondition failed in a way the caller should report, not mask."""


def qemu_available() -> bool:
    return bool(shutil.which("qemu-system-riscv64"))


def demo_available() -> bool:
    return os.path.isfile(DEMO_ELF)


def choose_backend(preferred: str = "native") -> str:
    """Environment override wins; QEMU is used only when preferred and present."""
    env = os.environ.get("VALUE_DEMO_BACKEND")
    if env:
        return env
    if preferred == "qemu":
        return "qemu" if qemu_available() else "native"
    return "native"


def sha256_file(path: str) -> str:
    digest = hashlib.sha256()
    with open(path, "rb") as fh:
        for chunk in iter(lambda: fh.read(1 << 20), b""):
            digest.update(chunk)
    return digest.hexdigest()


class Stopwatch:
    def __init__(self) -> None:
        self.t0 = time.monotonic()
        self.elapsed = 0.0

    def stop(self) -> float:
        self.elapsed = time.monotonic() - self.t0
        return self.elapsed


def artifact_dir(name: str) -> str:
    path = os.path.join(ARTIFACT_ROOT, name)
    os.makedirs(path, exist_ok=True)
    return path


def session_dir(root: str, label: str = "session") -> str:
    """A unique recording dir under ``root``.

    Demos share a fixed artifact root so their reports are discoverable, but
    two concurrent demo runs (tests, agents) must never share one
    ``events.jsonl`` or a reader can see a half-written recording.
    """
    os.makedirs(root, exist_ok=True)
    return tempfile.mkdtemp(prefix=f"{label}-", dir=root)


def write_json(path: str, payload) -> str:
    os.makedirs(os.path.dirname(path), exist_ok=True)
    tmp = f"{path}.tmp-{os.getpid()}"
    with open(tmp, "w", encoding="utf-8") as fh:
        json.dump(payload, fh, indent=2, sort_keys=True)
        fh.write("\n")
    os.replace(tmp, path)
    return path


def write_text(path: str, text: str) -> str:
    os.makedirs(os.path.dirname(path), exist_ok=True)
    with open(path, "w", encoding="utf-8") as fh:
        fh.write(text)
    return path


def load_schema(ge):
    with open(SCHEMA_JSON, "r", encoding="utf-8") as fh:
        return ge.schema_from_json(json.load(fh))


def decode_ctrl(ge, data: bytes, schema=None) -> dict:
    schema = schema or load_schema(ge)
    out = schema.decode(data, base=0)
    if isinstance(out.get("buf"), (bytes, bytearray)):
        out["buf"] = bytes(out["buf"]).split(b"\x00", 1)[0].decode("utf-8", "replace")
    return out


def short_hex(value) -> str:
    if value is None:
        return "None"
    if isinstance(value, str):
        return value
    return f"{value:#x}" if isinstance(value, int) else str(value)


def render_banner(demo: str, question: str, scenario: str, baseline: str, commands) -> None:
    print(f"=== {demo} ===")
    print(f"real scenario : {scenario}")
    print(f"baseline      : {baseline}")
    print(f"question      : {question}")
    print("commands      :")
    for command in commands:
        print(f"  - {command}")


def finish(result: dict, artifact: str, elapsed: float) -> dict:
    """Persist the per-demo result and print the human + machine summary."""
    result["elapsed_s"] = round(elapsed, 4)
    result["artifact"] = os.path.relpath(artifact, TOOL_ROOT) if artifact else None
    result.setdefault("success", True)
    if artifact:
        write_json(os.path.join(artifact, "result.json"), result)
    print(f"answer        : {result.get('answer')}")
    print(f"evidence      : {result.get('artifact')}")
    print(f"analog time   : {elapsed:.3f} s ({result.get('backend')} backend)")
    if result.get("real_baseline"):
        print(f"real baseline : {result['real_baseline']}")
    if result.get("projected"):
        print(f"projected     : {result['projected']}")
    if result.get("blocked_on_device_model"):
        print("blocked       : real case blocked on H2 device model; workflow shown on analog")
    if not result.get("success", True):
        print(f"status        : FAILED - {result.get('failure', 'unknown')}")
    print("RESULT_JSON: " + json.dumps(result, sort_keys=True))
    sys.stdout.flush()
    return result


def timed_demo(fn, artifact: str) -> dict:
    watch = Stopwatch()
    try:
        result = fn()
    except Exception as exc:  # report, never hide
        import traceback

        result = {
            "success": False,
            "failure": f"{type(exc).__name__}: {exc}",
            "traceback": traceback.format_exc()[-2000:],
        }
    return finish(result, artifact, watch.stop())


def scenario(
    *,
    ids,
    baseline,
    projection,
    blocked=False,
    device=None,
) -> dict:
    return {
        "citation_ids": ids,
        "static_baseline": baseline,
        "projection": projection,
        "blocked_on_device_model": blocked,
        "device_model_needed": device,
    }