risc-v-gsp-rm-debugger: land docs 45-53 + harness/sim tool set with tests

rm_capture (chunked decode), launch_audit, state_seed/state_decode,
valprobe, rehost, contracts+run_contracts, host_traffic, gmmu_walk,
msgq_attach, libos_port, log_decode, server_snapshot; doc index updates
in the tool overview, operations pages and the repo README.

1 files changed, 651 insertions, 0 deletions

diff --git a/debug/risc-v-gsp-rm-vm/harness/valprobe.py b/debug/risc-v-gsp-rm-vm/harness/valprobe.py
new file mode 100644
index 0000000..55952b9
--- /dev/null
+++ b/debug/risc-v-gsp-rm-vm/harness/valprobe.py
@@ -0,0 +1,651 @@
+#!/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())