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

#!/usr/bin/env python3
"""Validation-predicate probe: run a firmware predicate branch with crafted state.

Why this exists
---------------
The live GR campaign keeps burning supervised hardware cycles to ask the same
question class: *which control/entry shapes does the 570.144 GSP-RM firmware
accept or refuse, and why?*  The answer lives in firmware branches that are
buried behind host traffic (B4/B5) and RM object state, so a plain rehost of
the full handler early-outs before the predicate.  This harness instead runs
one **predicate branch** of the real image with the minimum crafted state it
needs: registers, memory, and (where a branch calls out to a lock/table walk
that is not the predicate) one recorded inline stub.  Each run stops on a
labelled outcome address, so the verdict is the branch the firmware took, not
an inference.

It is deliberately not a boot and not a model: the report carries the exact
entry, seeded state, stubs and outcome map, and every limitation in `limits`.
Fidelity is L0/L1 (docs/06); no timing claim.

Bundled target: ``sweep --target fecs-promote`` reproduces the live V6 FECS
matrix (`PM/PATCH/PRIV/UNR accepted, FECS init refused 0x1f, FECS VA-only
accepted`) and evaluates the untested shapes offline, including the
sysmem-aperture FECS-init variant the handoff proposed paying a hardware cycle
for.

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

import argparse
import json
import os
import struct
import sys
import time
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 greeneye.elf import load_elf  # noqa: E402
from greeneye.errors import GreenEyeError  # noqa: E402
from greeneye.sim_backend import launch_native  # noqa: E402
from sim.devices import load_profile, registry_from_profile  # noqa: E402
from sim.gsp_boot import _apply_boot_shim, _run_with_milestones, _rung_from_run  # noqa: E402
from sim.libos import LibosSyscalls  # noqa: E402
from sim.machine import R_BP, R_BUDGET, R_ECALL, R_FAULT  # noqa: E402

from harness.rehost import (  # noqa: E402
    DEFAULT_IMAGE,
    DEFAULT_PROFILE,
    RUN_CHUNK,
    _parse_u64,
    _sha256_file,
    parse_hex_bytes,
)

#: Schema tag of a single probe run.
PROBE_FORMAT = "greeneye-valprobe/1"
#: Schema tag of a probe spec.
SPEC_FORMAT = "greeneye-valprobe-spec/1"
#: Schema tag of a matrix run (a spec with variants).
MATRIX_FORMAT = "greeneye-valprobe-matrix/1"
#: Schema tag of the bundled demo.
DEMO_FORMAT = "greeneye-valprobe-demo/1"

DEFAULT_BUDGET = 50_000
DEFAULT_TIMEOUT = 20.0
BOOT_INSTRUCTIONS = 2_000_000
EBREAK = 0x00100073

REGISTERS = {
    "zero": 0, "ra": 1, "sp": 2, "gp": 3, "tp": 4,
    "t0": 5, "t1": 6, "t2": 7, "s0": 8, "fp": 8, "s1": 9,
    "a0": 10, "a1": 11, "a2": 12, "a3": 13, "a4": 14, "a5": 15,
    "a6": 16, "a7": 17, "s2": 18, "s3": 19, "s4": 20, "s5": 21,
    "s6": 22, "s7": 23, "s8": 24, "s9": 25, "s10": 26, "s11": 27,
    "t3": 28, "t4": 29, "t5": 30, "t6": 31,
}
for _i in range(32):
    REGISTERS.setdefault(f"x{_i}", _i)

RUNG_RANK = {"B0": 0, "B1": 1, "B2": 2, "B3": 3}


class ValprobeError(GreenEyeError):
    """A probe spec, image or scenario is invalid."""


def reg_index(name: str) -> int:
    key = str(name).strip().lower()
    if key not in REGISTERS:
        raise ValprobeError(f"unknown register {name!r}")
    return REGISTERS[key]


def encode_return_stub(value: int) -> bytes:
    """Encode ``lui a0,hi; addi a0,a0,lo; jalr zero,0(ra)`` (RV64IMAC)."""
    value &= 0xFFFF_FFFF_FFFF_FFFF
    hi = (value + 0x800) >> 12
    lo = value - (hi << 12)
    lui = ((hi & 0xF_FFFF) << 12) | (10 << 7) | 0x37
    addi = ((lo & 0xFFF) << 20) | (10 << 15) | (10 << 7) | 0x13
    ret = 0x0000_8067
    return struct.pack("<III", lui, addi, ret)


def decode_memory_spec(item: dict) -> tuple[int, bytes]:
    """Parse one memory item: ``{addr, hex|u64|u32|u8}``."""
    if "addr" not in item:
        raise ValprobeError(f"memory item needs addr: {item!r}")
    addr = _parse_u64(item["addr"])
    for key, fmt, size in (("hex", None, None), ("u64", "<Q", 8), ("u32", "<I", 4),
                           ("u16", "<H", 2), ("u8", "<B", 1)):
        if key in item:
            if key == "hex":
                return addr, parse_hex_bytes(str(item["hex"]))
            return addr, struct.pack(fmt, _parse_u64(item[key]) & ((1 << (8 * size)) - 1))
    raise ValprobeError(f"memory item needs one of hex/u64/u32/u16/u8: {item!r}")


def _load_spec(path: str) -> dict:
    if not os.path.isfile(path):
        raise ValprobeError(f"spec not found: {path}")
    with open(path, "r", encoding="utf-8") as fh:
        spec = json.load(fh)
    fmt = spec.get("format", SPEC_FORMAT)
    if fmt != SPEC_FORMAT:
        raise ValprobeError(f"unsupported spec format {fmt!r} (want {SPEC_FORMAT})")
    if not spec.get("variants"):
        raise ValprobeError("spec has no variants")
    return spec


def _apply_state(machine, *, registers: dict, memory, stubs, default_sp: int) -> list[dict]:
    """Write stubs, memory and registers; return the applied stub notes."""
    stub_notes = []
    for stub in stubs or ():
        addr = _parse_u64(stub["addr"])
        if "return" not in stub:
            raise ValprobeError(f"stub needs return: {stub!r}")
        code = encode_return_stub(_parse_u64(stub["return"]))
        machine.write_mem(addr, code)
        stub_notes.append({"addr": f"{addr:#x}", "return": f"{_parse_u64(stub['return']):#x}",
                           "code_hex": code.hex()})
    for item in memory or ():
        addr, data = decode_memory_spec(item)
        machine.write_mem(addr, data)
    for name, value in (registers or {}).items():
        machine.x[reg_index(name)] = _parse_u64(value)
    if machine.x[2] == 0:
        machine.x[2] = default_sp
    return stub_notes


def _read_outcomes(machine, reads) -> list[dict]:
    out = []
    for item in reads or ():
        addr = _parse_u64(item["addr"])
        size = int(item.get("size", 8))
        data = bytes(machine.read_mem(addr, size))
        record = {"addr": f"{addr:#x}", "size": size, "hex": data.hex()}
        if item.get("label"):
            record["label"] = item["label"]
        if size == 8:
            record["u64"] = int.from_bytes(data, "little")
        elif size == 4:
            record["u32"] = int.from_bytes(data, "little")
        out.append(record)
    return out


def run_probe(
    *,
    image: str = DEFAULT_IMAGE,
    profile_path: str | None = DEFAULT_PROFILE,
    spec: dict,
    variant: dict,
    out: str | None = None,
    record: bool | None = None,
    name: str = "valprobe",
) -> dict:
    """Run one variant of a probe spec in a fresh machine; return its report."""
    if not os.path.isfile(image):
        raise ValprobeError(f"image not found: {image}")
    image_obj = load_elf(image)
    if not image_obj.is_riscv:
        raise ValprobeError(f"{image}: not a RISC-V ELF image")

    profile = load_profile(profile_path) if profile_path else None
    load = (profile or {}).get("load", {})
    ram_base = _parse_u64(load.get("ram_base", 0))
    memory_mb = int(load.get("memory_mb", 256))
    ram_top = ram_base + memory_mb * 1024 * 1024

    entry = _parse_u64(variant.get("entry", spec.get("target")))
    outcomes = {_parse_u64(a): label for a, label in
                (variant.get("outcomes") or spec.get("outcomes") or {}).items()}
    if not outcomes:
        raise ValprobeError("probe needs at least one outcome address")

    budget = int(variant.get("budget", spec.get("budget", DEFAULT_BUDGET)))
    timeout = float(variant.get("timeout", spec.get("timeout", DEFAULT_TIMEOUT)))
    label = variant.get("label") or f"variant-{entry:#x}"
    boot_to = spec.get("boot_to") or variant.get("boot_to")

    if record is None:
        record = bool(out)
    session = launch_native(
        kernel=image,
        memory_mb=memory_mb,
        ram_base=ram_base,
        record=record,
        name=f"{name}-{label}"[:80],
        device_map=None,
        unknown_mmio="lenient",
        bias=0,
    )
    machine = session.target.machine
    shim_meta: dict[str, Any] = {"enabled": False, "assumptions": []}
    libos = LibosSyscalls()
    boot: dict | None = None
    try:
        if profile is not None:
            machine.devices = registry_from_profile(profile, machine=machine)
            machine.translator = getattr(machine.devices, "dma", None)
            machine.syscall_handler = libos.handle
            shim = dict(profile.get("boot_shim", {}))
            shim["_image_path"] = image
            _apply_boot_shim(machine, shim, image_obj, shim_meta)
            shim_meta["enabled"] = True

        if boot_to:
            milestones: list[str] = []
            reason, pc, _a, _b = _run_with_milestones(machine, BOOT_INSTRUCTIONS, milestones)
            rung = _rung_from_run(machine, shim_meta, milestones, libos)
            boot = {
                "requested": boot_to,
                "rung": rung,
                "instructions": machine.insns,
                "stop_pc": f"{pc:#x}",
                "milestones": milestones,
                "reached": RUNG_RANK.get(rung, -1) >= RUNG_RANK.get(str(boot_to), 99),
            }

        sentinel = (ram_top - 0x100) & ~0xF
        default_sp = (ram_top - 0x8000) & ~0xF
        machine.write_mem(sentinel, EBREAK.to_bytes(4, "little"))
        breakpoints = set(outcomes)
        breakpoints.add(sentinel)

        registers = dict(spec.get("registers") or {})
        registers.update(variant.get("registers") or {})
        registers.setdefault("ra", sentinel)
        memory = list(spec.get("memory") or ()) + list(variant.get("memory") or ())
        stubs = list(spec.get("stubs") or ()) + list(variant.get("stubs") or ())
        stubs_applied = _apply_state(
            machine, registers=registers, memory=memory, stubs=stubs, default_sp=default_sp
        )
        machine.pc = entry

        entered = {f"x{i}": machine.x[i] for i in range(32) if i and machine.x[i]}
        entered["pc"] = entry
        started = time.monotonic()
        status = "budget"
        stop_pc = entry
        fault: dict | None = None
        outcome: str | None = None
        remaining = budget
        while remaining > 0:
            if time.monotonic() - started > timeout:
                status = "timeout"
                break
            chunk = min(RUN_CHUNK, remaining)
            reason, stop_pc, a, b = machine.run(chunk, breakpoints=breakpoints)
            remaining -= chunk
            if reason == R_BP:
                if stop_pc == sentinel:
                    status = "return"
                    break
                outcome = outcomes.get(stop_pc, "unlabelled")
                status = "outcome"
                break
            if reason == R_FAULT:
                status = "fault"
                fault = {"cause": a, "tval": f"{b:#x}"}
                break
            if reason == R_ECALL:
                handled, result = (machine.syscall_handler(machine, stop_pc)
                                   if machine.syscall_handler else (False, None))
                if not handled:
                    status = "ecall"
                    break
                if result is not None:
                    machine.x[10] = result & 0xFFFF_FFFF_FFFF_FFFF
                continue
            if reason == R_BUDGET:
                status = "budget"

        report: dict[str, Any] = {
            "format": PROBE_FORMAT,
            "probe": spec.get("name"),
            "label": label,
            "image": os.path.abspath(image),
            "image_sha256": _sha256_file(image),
            "profile": os.path.basename(profile_path) if profile_path else None,
            "entry": f"{entry:#x}",
            "status": status,
            "outcome": outcome,
            "expect": variant.get("expect"),
            "pass": (None if variant.get("expect") is None else outcome == variant.get("expect")),
            "stop_pc": f"{stop_pc:#x}",
            "fault": fault,
            "retval": machine.x[10],
            "instructions": machine.insns,
            "wall_seconds": round(time.monotonic() - started, 4),
            "boot": boot,
            "stubs": stubs_applied,
            "registers_entered": {k: f"{v:#x}" if isinstance(v, int) else v
                                  for k, v in entered.items()},
            "reads": _read_outcomes(machine, variant.get("reads") or spec.get("reads")),
            "limits": [
                "Predicate-branch probe, not a boot: only the seeded state exists.",
                "Stubs (if any) replace an out-of-predicate dependency; they are listed.",
                "No timing/cycle claim (L0/L1 fidelity, docs/06).",
                "A fault is the first missing state, not a product defect.",
            ],
        }
        if getattr(session.recorder, "dir", None) is not None:
            session.note("valprobe", label=label, outcome=outcome, status=status)
            report["session"] = session.recorder.dir
        return report
    finally:
        session.close()


def run_spec(
    *,
    spec: dict,
    image: str | None = None,
    profile_path: str | None = None,
    out: str | None = None,
    record: bool | None = None,
    name: str | None = None,
) -> dict:
    """Run every variant of a spec; return a matrix report."""
    image = image or spec.get("image") or DEFAULT_IMAGE
    profile_path = profile_path or spec.get("profile") or DEFAULT_PROFILE
    started = time.monotonic()
    rows = []
    for variant in spec["variants"]:
        rows.append(run_probe(
            image=image, profile_path=profile_path, spec=spec, variant=variant,
            out=out, record=record, name=name or (spec.get("name") or "valprobe"),
        ))
    matrix: dict[str, Any] = {
        "format": MATRIX_FORMAT,
        "spec": spec.get("name"),
        "spec_format": spec.get("format"),
        "image": os.path.abspath(image),
        "image_sha256": _sha256_file(image),
        "target": spec.get("target"),
        "boot_to": spec.get("boot_to"),
        "variants": rows,
        "passed": sum(1 for r in rows if r["pass"] is True),
        "failed": sum(1 for r in rows if r["pass"] is False),
        "evaluated": sum(1 for r in rows if r["pass"] is not None),
        "wall_seconds": round(time.monotonic() - started, 4),
    }
    if out:
        os.makedirs(out, exist_ok=True)
        path = os.path.join(out, "valprobe-matrix.json")
        with open(path, "w", encoding="utf-8") as fh:
            json.dump(matrix, fh, indent=2, sort_keys=True)
            fh.write("\n")
        matrix["report_json"] = os.path.abspath(path)
    return matrix


# ---------------------------------------------------------------------------
# Bundled target: the GR PROMOTE_CTX FECS predicate (570.144 firmware).
#
# Firmware facts used (all verified by disassembly of the shipped image):
#   * per-buffer validation branch point: 0x5290dd4 (in FUN_ram_05290a48)
#   * bInitialize=1 and bufferId==9 -> FECS branch at 0x5290e3c
#   * FECS branch: instance = pGpu[(idx+0x37c)], idx = *(u32*)(getter+0xD28);
#     table = *(u64*)(instance+0x190); slot = table[mapped(bufferId)*8];
#     refuse (0x1f) iff slot != 0 -- regardless of physAttr/size/VA
#   * bufferId -> slot index via the firmware's own 0x531b944 jump table
#     (FECS => 6)
#   * bInitialize=0: bNonmapped!=0 => 0x1f, else accept
#
# The instance-index getter (0x524a5b8, a lock/table walk outside the
# predicate) is stubbed with a recorded constant return so the branch can run
# with crafted state; the predicate itself, the mapping table call and the
# slot read are the firmware's own code.
# ---------------------------------------------------------------------------

PROMOTE_VALIDATION = 0x5290DD4
FECS_INIT_BRANCH = 0x5290E3C
INSTANCE_GETTER = 0x524A5B8
OUTCOMES = {
    "0x5290cd0": "accept-commit-path",
    "0x5290ce4": "accept-skip-path",
    "0x5290df8": "refuse-invalid-argument-0x1f",
    "0x5290ebc": "refuse-fecs-slot-exists-0x1f",
}

#: Buffer ids (NV2080_CTRL_GPU_PROMOTE_CTX_BUFFER_ID_*), 570.144 interface order.
BUFFER_ID = {
    "MAIN": 0, "PM": 1, "PATCH": 2, "BUNDLE_CB": 3, "PAGEPOOL": 4,
    "ATTRIBUTE_CB": 5, "RTV_CB_GLOBAL": 6, "GFXP_POOL": 7, "GFXP_CTRL_BLK": 8,
    "FECS_EVENT": 9, "PRIV_ACCESS_MAP": 10, "UNRESTRICTED_PRIV_ACCESS_MAP": 11,
    "GLOBAL_PRIV_ACCESS_MAP": 12,
}


#: Scratch base for crafted predicate state.  Must be encodable by the
#: ``lui+addi`` return stub (below 2^31), inside RAM, outside the image
#: (max vaddr ~0x5bb9000), outside the 0x0..0x8000000 DMA window and below
#: the 0x100000000 DMEM aperture the firmware uses after relocation.
FECS_SCRATCH_BASE = 0x10000000


def _fecs_layout() -> dict:
    """Deterministic scratch layout (see module doc)."""
    scratch = FECS_SCRATCH_BASE
    return {
        "scratch": scratch,
        "sp": scratch + 0x4000,
        "s0": scratch + 0x5000,
        "pgpu": scratch + 0x1000,
        "getter_out": scratch + 0x1800,
        "instance": scratch + 0x3800,
        "entry": scratch + 0x6000,
    }


def _fecs_variant(layout: dict, *, label: str, buffer_id: int, b_initialize: int,
                  b_nonmapped: int, phys_attr: int = 4, size: int = 0x10000,
                  phys: int = 0x1F4150000, virt: int = 0x204608000,
                  slot: int | None = None, expect: str | None = None) -> dict:
    lay = layout
    entry = lay["entry"]
    # table = *(u64*)(instance+0x190) = instance+0x300; FECS maps to slot 6.
    slot_addr = lay["instance"] + 0x300 + 6 * 8
    memory = [
        {"addr": f"{entry + 0x30:#x}", "u64": b_initialize and phys or 0},
        {"addr": f"{entry + 0x38:#x}", "u64": virt},
        {"addr": f"{entry + 0x40:#x}", "u64": size},
        {"addr": f"{entry + 0x48:#x}", "u32": phys_attr},
        {"addr": f"{entry + 0x4c:#x}", "u16": buffer_id},
        {"addr": f"{entry + 0x4e:#x}", "u8": b_initialize},
        {"addr": f"{entry + 0x4f:#x}", "u8": b_nonmapped},
        # pGpu -> instance pointer; getter result -> idx 0; instance -> table.
        {"addr": f"{lay['pgpu'] + 0x1be0:#x}", "u64": lay["instance"]},
        {"addr": f"{lay['instance'] + 0x190:#x}", "u64": lay["instance"] + 0x300},
    ]
    if slot is not None:
        memory.append({"addr": f"{slot_addr:#x}", "u64": slot})
    return {
        "label": label,
        "registers": {
            "s0": f"{lay['s0']:#x}",
            "s2": f"{lay['pgpu']:#x}",
            "s5": f"{entry:#x}",
            "sp": f"{lay['sp']:#x}",
            "a3": buffer_id,
            "t1": 0xC,  # loop bound set by the function before our entry point
        },
        "stubs": [{"addr": f"{INSTANCE_GETTER:#x}", "return": f"{lay['getter_out']:#x}"}],
        "memory": memory,
        "reads": [{"addr": f"{lay['s0'] - 0x1c8:#x}", "size": 4,
                   "label": "firmware-mapped-slot-index"},
                  {"addr": f"{slot_addr:#x}", "size": 8, "label": "fecs-slot-value"}],
        "expect": expect,
    }


def fecs_spec(*, image: str = DEFAULT_IMAGE, profile_path: str = DEFAULT_PROFILE) -> dict:
    """Build the FECS promote-validation matrix spec (bundled target)."""
    layout = _fecs_layout()
    variants = [
        # Live m3j: FECS init (PA) refused -- slot already populated.
        _fecs_variant(layout, label="FECS init vidmem, slot non-NULL",
                      buffer_id=9, b_initialize=1, b_nonmapped=0, slot=1,
                      expect="refuse-fecs-slot-exists-0x1f"),
        # Predicate inverse: the same shape would be accepted on an empty slot.
        _fecs_variant(layout, label="FECS init vidmem, slot NULL",
                      buffer_id=9, b_initialize=1, b_nonmapped=0, slot=0,
                      expect="accept-commit-path"),
        # The handoff's untested variant (a): does a sysmem aperture change it?
        _fecs_variant(layout, label="FECS init sysmem attr 0x6, slot non-NULL",
                      buffer_id=9, b_initialize=1, b_nonmapped=0, phys_attr=6, slot=1,
                      expect="refuse-fecs-slot-exists-0x1f"),
        _fecs_variant(layout, label="FECS init sysmem attr 0x6, slot NULL",
                      buffer_id=9, b_initialize=1, b_nonmapped=0, phys_attr=6, slot=0,
                      expect="accept-commit-path"),
        # Size/PA content must not matter to the predicate.
        _fecs_variant(layout, label="FECS init size 0x1000, slot non-NULL",
                      buffer_id=9, b_initialize=1, b_nonmapped=0, size=0x1000, slot=1,
                      expect="refuse-fecs-slot-exists-0x1f"),
        # Live m7: FECS VA-only accepted (bInitialize=0 path).
        _fecs_variant(layout, label="FECS no-init mapped, slot non-NULL",
                      buffer_id=9, b_initialize=0, b_nonmapped=0, slot=1,
                      expect="accept-skip-path"),
        _fecs_variant(layout, label="FECS no-init nonmapped (bookkeeping refusal)",
                      buffer_id=9, b_initialize=0, b_nonmapped=1, slot=1,
                      expect="refuse-invalid-argument-0x1f"),
        # Live m3j: PM and PATCH init accepted.
        _fecs_variant(layout, label="PM init",
                      buffer_id=1, b_initialize=1, b_nonmapped=0,
                      expect="accept-commit-path"),
        _fecs_variant(layout, label="PATCH init",
                      buffer_id=2, b_initialize=1, b_nonmapped=0,
                      expect="accept-commit-path"),
        _fecs_variant(layout, label="GFXP_CTRL_BLK init (id 8)",
                      buffer_id=8, b_initialize=1, b_nonmapped=0,
                      expect="accept-skip-path"),
        _fecs_variant(layout, label="BUNDLE_CB init (id 3)",
                      buffer_id=3, b_initialize=1, b_nonmapped=0,
                      expect="refuse-invalid-argument-0x1f"),
        _fecs_variant(layout, label="PRIV_ACCESS_MAP init (id 10)",
                      buffer_id=10, b_initialize=1, b_nonmapped=0,
                      expect="accept-skip-path"),
        _fecs_variant(layout, label="UNRESTRICTED_PRIV init (id 11)",
                      buffer_id=11, b_initialize=1, b_nonmapped=0,
                      expect="accept-skip-path"),
        _fecs_variant(layout, label="GLOBAL_PRIV_ACCESS_MAP init (id 12)",
                      buffer_id=12, b_initialize=1, b_nonmapped=0,
                      expect="refuse-invalid-argument-0x1f"),
    ]
    # id > 0xc is refused one block earlier (0x5290cc4, still in the per-entry
    # loop); enter there with the loop's a4/t1 state.
    unknown_id = _fecs_variant(layout, label="unknown buffer id 13",
                               buffer_id=13, b_initialize=1, b_nonmapped=0,
                               expect="refuse-invalid-argument-0x1f")
    unknown_id["entry"] = "0x5290cc0"
    unknown_id["registers"]["a4"] = 13
    variants.append(unknown_id)
    return {
        "format": SPEC_FORMAT,
        "name": "fecs-promote-validation",
        "target": f"{PROMOTE_VALIDATION:#x}",
        "outcomes": OUTCOMES,
        "budget": 20_000,
        "timeout": 20.0,
        "image": image,
        "profile": profile_path,
        "note": (
            "570.144 per-buffer PROMOTE_CTX validation; instance-index getter "
            f"({INSTANCE_GETTER:#x}, lock/table walk outside the predicate) stubbed "
            "to a recorded constant; the predicate, the id->slot mapping call and "
            "the memDesc slot read are the firmware's own code."
        ),
        "variants": variants,
    }


def run_demo(out: str | None = None) -> dict:
    """Run the bundled FECS matrix on the real image, twice for determinism."""
    spec = fecs_spec()
    first = run_spec(spec=spec, out=out, record=False, name="valprobe-fecs")
    second = run_spec(spec=spec, record=False, name="valprobe-fecs")
    verdicts_a = [(r["label"], r["outcome"]) for r in first["variants"]]
    verdicts_b = [(r["label"], r["outcome"]) for r in second["variants"]]
    result = {
        "format": DEMO_FORMAT,
        "ok": first["failed"] == 0 and verdicts_a == verdicts_b,
        "evaluated": first["evaluated"],
        "failed": first["failed"],
        "deterministic": verdicts_a == verdicts_b,
        "verdicts": [{"label": label, "outcome": outcome} for label, outcome in verdicts_a],
    }
    if out:
        os.makedirs(out, exist_ok=True)
        path = os.path.join(out, "valprobe-demo.json")
        with open(path, "w", encoding="utf-8") as fh:
            json.dump(result, fh, indent=2, sort_keys=True)
            fh.write("\n")
        result["artifacts"] = {"demo_json": os.path.abspath(path)}
    return result


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        prog="valprobe.py",
        description=(
            "Run one firmware validation-predicate branch with crafted state "
            "(offline; never touches hardware)."
        ),
    )
    sub = parser.add_subparsers(dest="command")
    sweep = sub.add_parser("sweep", help="run a bundled target matrix")
    sweep.add_argument("--target", default="fecs-promote", choices=["fecs-promote"])
    sweep.add_argument("--image", default=None)
    sweep.add_argument("--profile", default=None)
    sweep.add_argument("--boot-to", default=None, help="boot the image to a rung first (e.g. B3)")
    sweep.add_argument("--out", help="output directory (matrix JSON + sessions)")
    sweep.add_argument("--no-record", action="store_true", help="skip session recording")
    run = sub.add_parser("run", help="run a probe spec file")
    run.add_argument("--spec", required=True)
    run.add_argument("--image", default=None)
    run.add_argument("--profile", default=None)
    run.add_argument("--out", help="output directory (matrix JSON + sessions)")
    run.add_argument("--no-record", action="store_true", help="skip session recording")
    demo = sub.add_parser("demo", help="bundled FECS matrix, run twice (determinism)")
    demo.add_argument("--out", help="output directory")
    return parser


def main(argv: list[str] | None = None) -> int:
    parser = build_parser()
    args = parser.parse_args(argv)
    try:
        if args.command == "sweep":
            spec = fecs_spec(image=args.image or DEFAULT_IMAGE,
                             profile_path=args.profile or DEFAULT_PROFILE)
            if args.boot_to:
                spec["boot_to"] = args.boot_to
            matrix = run_spec(spec=spec, image=args.image, profile_path=args.profile,
                              out=args.out, record=False if args.no_record else None)
        elif args.command == "run":
            spec = _load_spec(args.spec)
            matrix = run_spec(spec=spec, image=args.image, profile_path=args.profile,
                              out=args.out, record=False if args.no_record else None)
        elif args.command == "demo":
            matrix = run_demo(args.out)
        else:
            parser.print_help()
            return 2
    except ValprobeError as exc:
        print(f"error: {exc}", file=sys.stderr)
        return 2
    if matrix.get("format") == MATRIX_FORMAT:
        for row in matrix["variants"]:
            mark = {True: "PASS", False: "FAIL", None: "----"}[row["pass"]]
            print(f"{mark}  {row['label']:<48} {row['outcome'] or row['status']}")
        print(json.dumps({k: matrix[k] for k in
                          ("format", "spec", "evaluated", "passed", "failed",
                           "wall_seconds", "report_json") if k in matrix},
                         indent=2, sort_keys=True))
        return 3 if matrix["failed"] else 0
    print(json.dumps(matrix, indent=2, sort_keys=True))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())