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

#!/usr/bin/env python3
"""One-command new-version GSP firmware triage (docs/40, goal U46).

For a NEW GSP firmware image (optionally against a ``--baseline``) this tool
answers, in one bounded run and with one report:

* **what it is** - sha256, size, ELF shape and version label (explicit
  ``--version``, the ``gsp_version.txt`` sidecar convention, parent directory,
  file stem, in that order);
* **where it boots to** - the existing QEMU probe path (first exception /
  first touch) and the S1 native boot ladder (``python3 -m sim.gsp_boot``,
  rungs B0-B4);
* **how the native backend compares** - the existing ``harness/sim_qemu_line``
  A/B harness (QEMU vs native, deterministic repeat, frozen thresholds);
* **what changed vs the baseline** - the existing ``harness/version_diff``
  (image/symbol/corpus/probe/behavior sections), when ``--baseline`` is given;
* **what capability surface is present** - hooks/watch/memory/session API,
  native simulator, TCG trace plugin, MCP, demo, boot ladder;
* **what work ``vgsp`` support needs** - a ``support_readiness`` verdict with
  blockers and ordered next actions.

Every step is wrapped: a failing step is recorded (``status: error``) and the
others still run. Every session a step uses is listed with sha256 hashes.
Nothing is downloaded; no hardware, GPU, mdev, kernel module or guest is
touched; QEMU is the only process launched besides the bounded native boot.

This is the flagship workflow of the simulator program (handoff
``U46``) and remains **in development** (docs/40). The frozen value
thresholds below were recorded on 2026-10-03 *before* the first measurement
(docs/40 section 5); never move them silently.

Run from the tool root::

    python3 harness/triage.py --firmware ../../firmware/57X-GSP/570.144/gsp-570.144.fwimage.bin --version 570.144 --out dumps/triage-570.144
    python3 harness/triage.py --firmware <new> --baseline <old> --out dumps/triage-570.211 --dry-run
"""
from __future__ import annotations

import argparse
import importlib.util
import json
import os
import re
import shutil
import subprocess
import sys
import time
from dataclasses import dataclass

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

from harness.common import (  # noqa: E402
    DEMO_ELF,
    ensure_dir,
    qemu_available,
    sha256_file,
    write_json,
    write_text,
)
from harness.sim_qemu_line import (  # noqa: E402
    MAX_INSTRUCTIONS,
    run_harness,
    run_qemu_baseline,
    rung_ordinal,
    session_hashes,
)
from harness.version_diff import (  # noqa: E402
    RM610_TOKEN,
    Rm610GateError,
    VersionDiffOptions,
    check_rm610_gate,
    run_version_diff,
)

FORMAT = "greeneye-new-version-triage/1"
PLAN_FORMAT = "greeneye-new-version-triage-plan/1"

#: Instruction budget bounds (task rule: <= 2M per attempt).
DEFAULT_BUDGET = 200_000
#: QEMU probe wall-clock bound (matches the S4 harness default).
DEFAULT_QEMU_DURATION_S = 0.5
#: Per-step wall-clock net (subprocess wedge protection; no unbounded runs).
DEFAULT_STEP_TIMEOUT_S = 300.0
#: Native rung at which runtime observation (not just first touch) is possible.
OBSERVABLE_RUNG = "B2"

#: Frozen 2026-10-03, before the first measured triage run (docs/40 section 5).
#: ``manual_baseline_s`` is the *conservative understatement* of the manual
#: steps this one command replaces (run+interpret probe, native boot ladder,
#: A/B, session hash collection and write-up for one new image).
FROZEN_THRESHOLDS = {
    "frozen_on": "2026-10-03",
    "manual_baseline_s": 1800.0,
    "triage_wall_s_max": 600.0,
    "required_steps": ("identity", "probe", "native_boot", "ab_eval", "features"),
}

VERSION_RE = re.compile(r"v?(\d{3}\.\d+)([A-Za-z0-9._-]*)?")
_UNAVAILABLE_MARKERS = (
    "no module named",
    "modulenotfounderror",
    "importerror",
    "cannot import",
    "profile absent",
    "image absent",
    "unavailable",
)


# ---------------------------------------------------------------------------
# options and helpers
# ---------------------------------------------------------------------------


@dataclass
class TriageOptions:
    firmware: str
    baseline: str | None = None
    version: str | None = None
    baseline_label: str | None = None
    out: str | None = None
    budget: int = DEFAULT_BUDGET
    qemu_duration: float = DEFAULT_QEMU_DURATION_S
    step_timeout: float = DEFAULT_STEP_TIMEOUT_S
    release: str | None = None
    approval_token: str | None = None
    dry_run: bool = False

    def labels(self) -> list[str]:
        return [
            value
            for value in (
                self.release,
                self.version,
                self.baseline_label,
                os.path.basename(self.firmware) if self.firmware else None,
                os.path.basename(self.baseline) if self.baseline else None,
            )
            if value
        ]


def _rm_tree(path: str) -> None:
    if path and os.path.isdir(path):
        shutil.rmtree(path, ignore_errors=True)


def _rel(path: str, root: str) -> str:
    try:
        return os.path.relpath(path, root)
    except ValueError:
        return path


def _abs(path: str | None) -> str | None:
    """Absolute, normalised path; session dirs from the harness may be tool-root relative."""
    if not path:
        return None
    return os.path.normpath(path if os.path.isabs(path) else os.path.join(TOOL_ROOT, path))


def _first_line(text: str | None, limit: int = 300) -> str | None:
    if not text:
        return None
    line = text.strip().splitlines()[-1].strip() if text.strip() else ""
    return line[:limit] or None


def _looks_unavailable(text: str) -> bool:
    lowered = (text or "").lower()
    return any(marker in lowered for marker in _UNAVAILABLE_MARKERS)


def extract_version(firmware: str, explicit: str | None = None) -> tuple[str | None, str]:
    """Version label: explicit, ``gsp_version.txt``, parent dir, file stem.

    The sidecar convention mirrors the existing probe/device-map helper
    (``greeneye/device_map.py``); no firmware content is guessed.
    """
    if explicit:
        return explicit, "explicit --version"
    sidecar = os.path.join(os.path.dirname(os.path.abspath(firmware)), "gsp_version.txt")
    if os.path.isfile(sidecar):
        try:
            with open(sidecar, "r", encoding="utf-8", errors="replace") as fh:
                text = "".join(ch for ch in fh.read() if ch.isprintable()).strip()
            if text:
                return text, "gsp_version.txt sidecar"
        except OSError:
            pass
    parent = os.path.basename(os.path.dirname(os.path.abspath(firmware)))
    if parent and VERSION_RE.fullmatch(parent):
        return parent, "parent directory name"
    stem = os.path.splitext(os.path.basename(firmware))[0]
    return (stem or None), "file stem"


def _target_brief(path: str, label: str | None = None) -> dict:
    info: dict = {"path": path, "name": os.path.basename(path)}
    version, version_source = extract_version(path, label)
    info["version"] = version
    info["version_source"] = version_source
    if os.path.isfile(path):
        info["sha256"] = sha256_file(path)
        info["size"] = os.path.getsize(path)
    else:
        info["error"] = "file not found"
    return info


# ---------------------------------------------------------------------------
# steps
# ---------------------------------------------------------------------------


def step_identity(options: TriageOptions, out_dir: str) -> dict:
    firmware = _target_brief(options.firmware, None)
    version, version_source = extract_version(options.firmware, options.version)
    firmware["version"] = version
    firmware["version_source"] = version_source
    if not os.path.isfile(options.firmware):
        return {
            "status": "error",
            "reason": f"firmware not found: {options.firmware}",
            "firmware": firmware,
        }
    result: dict = {"status": "ok", "firmware": firmware}
    try:
        from greeneye.elf import load_elf

        image = load_elf(options.firmware)
        result["elf"] = {
            "is_riscv": bool(image.is_riscv),
            "entry": image.entry,
            "min_vaddr": image.min_vaddr,
            "max_vaddr": image.max_vaddr,
            "segments": len(getattr(image, "segments", ()) or ()),
        }
    except Exception as exc:
        result["elf"] = {"unreadable": f"{type(exc).__name__}: {exc}"}
    if options.baseline:
        result["baseline"] = _target_brief(options.baseline, options.baseline_label)
    return result


def step_probe(options: TriageOptions, out_dir: str) -> dict:
    result = run_qemu_baseline(
        image=options.firmware, out_dir=out_dir, duration=options.qemu_duration
    )
    detail: dict = {
        "status": "ok" if result.get("ran") else "skipped",
        "reason": result.get("reason"),
        "rung": result.get("rung"),
        "stopped_pc": result.get("stopped_pc"),
        "first_fault": result.get("first_fault"),
        "first_touch_match": result.get("first_touch_match"),
        "fault_count_window": result.get("fault_count_window"),
        "device_map_entries": result.get("device_map_entries"),
        "wall_s": result.get("wall_s"),
        "sessions": [],
    }
    session = _abs(result.get("session"))
    if session:
        detail["sessions"].append({"name": "qemu-probe", "dir": session})
        detail["session_hashes"] = result.get("session_hashes") or session_hashes(session)
    return detail


def step_native_boot(options: TriageOptions, out_dir: str) -> dict:
    session_dir = os.path.join(out_dir, "sessions", "native-boot")
    _rm_tree(session_dir)
    argv = [
        sys.executable,
        "-m",
        "sim.gsp_boot",
        "--image",
        options.firmware,
        "--instructions",
        str(options.budget),
        "--record",
        session_dir,
        "--name",
        "triage-native-boot",
        "--json",
    ]
    try:
        proc = subprocess.run(
            argv,
            cwd=TOOL_ROOT,
            capture_output=True,
            text=True,
            timeout=options.step_timeout,
        )
    except subprocess.TimeoutExpired:
        return {"status": "error", "reason": f"native boot timed out after {options.step_timeout}s"}
    except OSError as exc:
        return {"status": "skipped", "reason": f"native boot ladder unavailable: {exc}"}

    stdout, stderr = proc.stdout or "", proc.stderr or ""
    payload = None
    if stdout.strip():
        try:
            payload = json.loads(stdout)
        except ValueError:
            payload = None
    if proc.returncode == 0 and isinstance(payload, dict):
        session = _abs(payload.get("session")) or session_dir
        return {
            "status": "ok",
            "rung": payload.get("rung"),
            "instructions_executed": payload.get("instructions_executed"),
            "cycles": payload.get("cycles"),
            "stop": payload.get("stop"),
            "first_unknown_mmio": payload.get("first_unknown_mmio"),
            "mmio_unknown": payload.get("mmio_unknown"),
            "device_accesses": payload.get("device_accesses"),
            "milestones": payload.get("milestones"),
            "sessions": [{"name": "native-boot", "dir": session}],
            "session_hashes": session_hashes(session),
        }
    reason = _first_line(stderr) or _first_line(stdout) or f"exit {proc.returncode}"
    status = "skipped" if _looks_unavailable(stderr + stdout) or proc.returncode == 2 else "error"
    if os.path.isdir(session_dir) and os.listdir(session_dir):
        return {
            "status": status,
            "reason": reason,
            "sessions": [{"name": "native-boot", "dir": session_dir}],
            "session_hashes": session_hashes(session_dir),
        }
    return {"status": status, "reason": reason}


def step_ab_eval(options: TriageOptions, out_dir: str) -> dict:
    if not os.path.isfile(options.firmware):
        return {"status": "skipped", "reason": f"firmware absent: {options.firmware}"}
    ab_dir = os.path.join(out_dir, "ab-eval")
    report = run_harness(
        image=options.firmware,
        out_dir=ab_dir,
        instructions=options.budget,
        qemu_duration=options.qemu_duration,
        repeat=2,
        features=False,
        run_qemu=True,
    )
    qemu = report.get("qemu") or {}
    native = report.get("native") or {}
    detail: dict = {
        "status": "ok" if native.get("ran") else "skipped",
        "reason": native.get("reason") if not native.get("ran") else None,
        "qemu_rung": qemu.get("rung"),
        "native_rung": native.get("rung"),
        "native_identical": native.get("identical"),
        "native_repeat": native.get("repeat"),
        "above_qemu_line": report.get("above_qemu_line"),
        "verdict": report.get("verdict"),
        "threshold_checks": report.get("threshold_checks"),
        "report_json": _abs(report.get("report_json")),
        "report_md": _abs(report.get("report_md")),
        "sessions": [],
    }
    qemu_session = _abs(qemu.get("session"))
    if qemu_session:
        detail["sessions"].append({"name": "ab-qemu", "dir": qemu_session})
    for index, run in enumerate(native.get("runs") or [], start=1):
        session = _abs(run.get("session")) or os.path.join(ab_dir, "sessions", f"native-{index}")
        detail["sessions"].append({"name": f"ab-native-{index}", "dir": session})
    return detail


def step_version_diff(options: TriageOptions, out_dir: str) -> dict:
    if not options.baseline:
        return {"status": "skipped", "reason": "no --baseline supplied"}
    if not os.path.isfile(options.baseline):
        return {"status": "skipped", "reason": f"baseline absent: {options.baseline}"}
    vd_dir = os.path.join(out_dir, "version-diff")
    old_version, _ = extract_version(options.baseline, options.baseline_label)
    report = run_version_diff(
        VersionDiffOptions(
            old=options.baseline,
            new=options.firmware,
            out=vd_dir,
            old_label=options.baseline_label or old_version,
            new_label=options.version,
            release=options.release,
            probe_instructions=max(1, min(options.budget, 2000)),
            behavior_duration=0.4,
            approval_token=options.approval_token,
        )
    )
    sections = {name: section.get("status") for name, section in report["sections"].items()}
    symbols = report["sections"].get("symbols") or {}
    detail: dict = {
        "status": "error" if "error" in sections.values() else "ok",
        "reason": None,
        "old_label": report.get("old", {}).get("label"),
        "new_label": report.get("new", {}).get("label"),
        "sections": sections,
        "sections_skipped": report.get("summary", {}).get("sections_skipped"),
        "sections_error": report.get("summary", {}).get("sections_error"),
        "report_json": os.path.join(vd_dir, "version-diff.json"),
        "report_md": os.path.join(vd_dir, "version-diff.md"),
        "image_changes": _image_changes(report),
        "symbol_changes": _symbol_changes(symbols),
        "sessions": [],
    }
    sessions_root = os.path.join(vd_dir, "sessions")
    if os.path.isdir(sessions_root):
        for name in sorted(os.listdir(sessions_root)):
            path = os.path.join(sessions_root, name)
            if os.path.isdir(path):
                detail["sessions"].append({"name": f"version-diff/{name}", "dir": path})
    return detail


def _image_changes(report: dict) -> dict:
    section = (report.get("sections") or {}).get("image") or {}
    return {
        "status": section.get("status"),
        "same": section.get("same"),
        "deltas": section.get("deltas"),
    }


def _symbol_changes(symbols: dict) -> dict:
    totals = symbols.get("totals") or {}
    return {
        "status": symbols.get("status"),
        "reason": symbols.get("reason"),
        "shift": symbols.get("shift"),
        "matched_named": symbols.get("matched_named"),
        "matched_shifted": symbols.get("matched_shifted"),
        "added": totals.get("added"),
        "removed": totals.get("removed"),
        "moved": totals.get("moved"),
        "resized_named": totals.get("resized_named"),
    }


def _spec_available(module: str) -> bool:
    try:
        return importlib.util.find_spec(module) is not None
    except (ImportError, ValueError):
        return False


def step_features(options: TriageOptions, out_dir: str) -> dict:
    checks: dict = {}
    notes: list[str] = []
    try:
        import greeneye as ge

        checks["greeneye_version"] = ge.__version__
        checks["hooks_watch_step"] = all(
            hasattr(ge.Session, name) for name in ("hook", "watch", "run", "step", "close")
        )
        checks["memory_struct"] = hasattr(ge, "Memory")
        checks["native_backend"] = _spec_available("greeneye.sim_backend")
    except Exception as exc:
        notes.append(f"greeneye import failed: {type(exc).__name__}: {exc}")
        checks.update({"greeneye_version": None, "hooks_watch_step": False, "memory_struct": False, "native_backend": False})
    checks["native_boot_ladder"] = os.path.isfile(os.path.join(TOOL_ROOT, "sim", "gsp_boot.py")) and os.path.isdir(
        os.path.join(TOOL_ROOT, "sim", "profiles")
    )
    checks["ab_harness"] = os.path.isfile(os.path.join(TOOL_ROOT, "harness", "sim_qemu_line.py"))
    checks["version_diff"] = os.path.isfile(os.path.join(TOOL_ROOT, "harness", "version_diff.py"))
    checks["tcg_trace_plugin"] = os.path.isfile(os.path.join(TOOL_ROOT, "qemu", "greeneye-trace.so"))
    checks["mcp_server"] = os.path.isfile(os.path.join(TOOL_ROOT, "plugins", "mcp_server.py"))
    checks["demo_firmware"] = os.path.isfile(DEMO_ELF)
    checks["qemu"] = qemu_available()
    unavailable = sorted(name for name, value in checks.items() if value is False)
    if unavailable:
        notes.append("unavailable: " + ", ".join(unavailable))
    return {"status": "ok", "checks": checks, "unavailable": unavailable, "notes": notes}


DEFAULT_RUNNERS = (
    ("identity", step_identity),
    ("probe", step_probe),
    ("native_boot", step_native_boot),
    ("ab_eval", step_ab_eval),
    ("version_diff", step_version_diff),
    ("features", step_features),
)


# ---------------------------------------------------------------------------
# readiness and report assembly
# ---------------------------------------------------------------------------


def support_readiness(report: dict) -> dict:
    """Blockers and next actions derived from the recorded step results."""
    steps = {step["name"]: step for step in report["steps"]}
    blockers: list[str] = []
    next_actions: list[str] = []
    identity = steps.get("identity", {})
    if identity.get("status") != "ok":
        blockers.append(f"firmware identity failed: {identity.get('reason') or 'unreadable'}")
    probe = steps.get("probe", {})
    if probe.get("status") == "error":
        blockers.append(f"QEMU probe failed: {probe.get('reason')}")
    native = steps.get("native_boot", {})
    native_rung = native.get("rung")
    native_ok = native.get("status") == "ok"
    if not native_ok:
        next_actions.append(
            "native boot ladder unavailable/again incomplete "
            f"({native.get('reason') or native.get('status')}): the S1 device model/profile is the enabler"
        )
    elif rung_ordinal(native_rung) is not None and rung_ordinal(native_rung) >= rung_ordinal(OBSERVABLE_RUNG):
        next_actions.append(
            f"native ladder is at {native_rung} (runtime-observable): inspect the milestone/stop with "
            "hooks and watchpoints, then route the finding through the accuracy loop"
        )
    else:
        next_actions.append(
            f"native ladder stops at {native_rung}: extend the device model past the next unknown MMIO "
            "before claiming runtime observation"
        )
    ab = steps.get("ab_eval", {})
    if ab.get("status") == "ok" and not ab.get("above_qemu_line"):
        next_actions.append(
            f"native A/B is not yet above the QEMU line (verdict={ab.get('verdict')}): "
            "device model and documented shims gate the claim"
        )
    vd = steps.get("version_diff", {})
    baseline_supplied = bool(report.get("input", {}).get("baseline_supplied"))
    if vd.get("status") == "error":
        blockers.append(f"version diff errored: {vd.get('sections_error') or vd.get('reason')}")
    elif vd.get("status") == "ok" and baseline_supplied:
        skipped = vd.get("sections_skipped") or {}
        if "symbols" in skipped:
            next_actions.append(
                "symbol inventory skipped: supply Ghidra exports / symbol DBs so the handler "
                "inventory delta can gate the vgsp support scope"
            )
        if "corpus" in skipped:
            next_actions.append(
                "corpus diff skipped: supply --corpus-diff indexes for decompiled-change hints"
            )
    else:
        next_actions.append(
            "re-run with --baseline <previous.fwimage> for the symbol/behavior delta "
            "(the release-to-support gate needs the A/B)"
        )
    features = steps.get("features", {})
    for name in features.get("unavailable") or []:
        if name in ("qemu", "native_boot_ladder", "hooks_watch_step", "memory_struct"):
            next_actions.append(f"capability unavailable: {name}")
    if not blockers and native_ok and rung_ordinal(native_rung) is not None and rung_ordinal(native_rung) >= rung_ordinal(OBSERVABLE_RUNG):
        verdict = "sim-observable"
    elif not blockers:
        verdict = "first-touch-only"
    else:
        verdict = "blocked"
    return {
        "verdict": verdict,
        "baseline_compared": bool(baseline_supplied and vd.get("status") == "ok"),
        "observable_rung": OBSERVABLE_RUNG,
        "blockers": blockers,
        "next_actions": next_actions,
    }


def _collect_sessions(steps: list[dict], out_dir: str) -> list[dict]:
    rows: list[dict] = []
    seen: set[str] = set()
    for step in steps:
        for entry in step.get("sessions") or []:
            path = _abs(entry.get("dir"))
            if not path or path in seen:
                continue
            seen.add(path)
            hashes = step.get("session_hashes") or {}
            rows.append(
                {
                    "step": step["name"],
                    "name": entry.get("name") or step["name"],
                    "dir": _rel(path, out_dir),
                    "present": os.path.isdir(path),
                    "hashes": hashes or session_hashes(path),
                }
            )
    return rows


def render_markdown(report: dict) -> str:
    firmware = report["input"].get("firmware") or {}
    steps = report["steps"]
    readiness = report["support_readiness"]
    lines = [
        f"# New-version triage — {firmware.get('version') or firmware.get('name')}",
        "",
        "> Status: in development (docs/40). Simulator-only: no hardware, GPU, mdev,",
        "> kernel module or guest was used. Frozen thresholds were recorded before the run.",
        "",
        f"* format: `{report['format']}`",
        f"* generated: {report['generated_at']}",
        f"* wall: {report['wall_s']}s (budget {report['input'].get('budget')} instructions/attempt)",
        f"* RM610 gate: {report['rm610_gate']['status']}",
        "",
        "## Firmware",
        "",
        "| field | value |",
        "|---|---|",
        f"| path | `{firmware.get('path')}` |",
        f"| sha256 | `{firmware.get('sha256')}` |",
        f"| size | {firmware.get('size')} |",
        f"| version | {firmware.get('version')} ({firmware.get('version_source')}) |",
        f"| ELF | {json.dumps(report['input'].get('elf') or {}, sort_keys=True)} |",
    ]
    baseline = report["input"].get("baseline")
    if baseline:
        lines += [
            f"| baseline | `{baseline.get('name')}` {baseline.get('sha256')} |",
            f"| baseline version | {baseline.get('version')} ({baseline.get('version_source')}) |",
        ]
    lines += ["", "## Steps", "", "| step | status | wall_s | key result |", "|---|---|---|---|"]
    for step in steps:
        key = (
            step.get("reason")
            or step.get("verdict")
            or step.get("rung")
            or step.get("status")
        )
        lines.append(f"| {step['name']} | {step['status']} | {step.get('wall_s')} | {str(key)[:120]} |")
    lines += [
        "",
        "## Support readiness",
        "",
        f"* verdict: **{readiness['verdict']}**",
        f"* baseline compared: {readiness['baseline_compared']}",
    ]
    for label, items in (("blockers", readiness["blockers"]), ("next actions", readiness["next_actions"])):
        lines.append(f"* {label}:")
        if not items:
            lines.append("  * none")
        for item in items:
            lines.append(f"  * {item}")
    sessions = report.get("sessions") or []
    lines += ["", "## Sessions used", "", "| step | name | dir | session.json sha256 |", "|---|---|---|---|"]
    if not sessions:
        lines.append("| - | - | - | (none recorded) |")
    for row in sessions:
        sha = (row.get("hashes") or {}).get("session.json") or "-"
        lines.append(f"| {row['step']} | {row['name']} | `{row['dir']}` | `{sha}` |")
    checks = report["threshold_checks"]
    lines += [
        "",
        "## Value (frozen thresholds, docs/40 section 5)",
        "",
        "| metric | value |",
        "|---|---|",
        f"| triage wall_s | {checks['triage_wall_s']} |",
        f"| wall_s max | {checks['triage_wall_s_max']} |",
        f"| within max | {checks['wall_within_max']} |",
        f"| manual baseline_s | {checks['manual_baseline_s']} |",
        f"| speedup_x | {checks['speedup_x']} |",
        f"| beats manual baseline | {checks['beats_manual_baseline']} |",
        "",
        "## Artifacts",
        "",
    ]
    for key in ("report_json", "report_md"):
        lines.append(f"* {key}: `{report['artifacts'][key]}`")
    lines.append("")
    return "\n".join(lines)


def run_triage(options: TriageOptions, runners=None) -> dict:
    """Run every step (wrapped), write ``triage-report.json|.md``, return it."""
    check_rm610_gate(options.labels(), approval_token=options.approval_token)
    if not options.out:
        raise ValueError("--out is required unless --dry-run is used")
    if options.budget <= 0 or options.budget > MAX_INSTRUCTIONS:
        raise ValueError(f"--budget must be in 1..{MAX_INSTRUCTIONS}, got {options.budget}")
    out = os.path.abspath(options.out)
    ensure_dir(out)
    started = time.monotonic()
    steps: list[dict] = []
    for name, fn in runners or DEFAULT_RUNNERS:
        step_started = time.monotonic()
        try:
            detail = fn(options, out)
            if not isinstance(detail, dict):
                detail = {"status": "ok", "result": detail}
        except Exception as exc:
            detail = {"status": "error", "reason": f"{type(exc).__name__}: {exc}"}
        entry = {"name": name, "wall_s": round(time.monotonic() - step_started, 3)}
        entry.update(detail)
        steps.append(entry)
    wall_s = round(time.monotonic() - started, 3)
    report: dict = {
        "format": FORMAT,
        "tool": "harness/triage.py",
        "dry_run": False,
        "generated_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
        "rm610_gate": {"status": "passed", "labels": sorted(set(options.labels()))},
        "input": {
            "firmware": _target_brief(options.firmware, options.version),
            "baseline": _target_brief(options.baseline, options.baseline_label) if options.baseline else None,
            "baseline_supplied": options.baseline is not None,
            "budget": options.budget,
            "qemu_duration": options.qemu_duration,
            "triage_command": (
                "python3 harness/triage.py --firmware <firmware>"
                + (" --baseline <baseline>" if options.baseline else "")
                + (f" --version {options.version}" if options.version else "")
                + f" --budget {options.budget} --out <out>"
            ),
        },
        "thresholds": dict(FROZEN_THRESHOLDS),
        "steps": steps,
        "support_readiness": {},
        "wall_s": wall_s,
        "artifacts": {},
    }
    firmware = report["input"]["firmware"]
    if firmware.get("error"):
        report["input"]["firmware"]["status"] = "error"
    report["input"]["elf"] = next((s.get("elf") for s in steps if s["name"] == "identity"), None)
    report["errors"] = [step["name"] for step in steps if step["status"] == "error"]
    report["sessions"] = _collect_sessions(steps, out)
    report["support_readiness"] = support_readiness(report)
    checks = {
        "triage_wall_s": wall_s,
        "triage_wall_s_max": FROZEN_THRESHOLDS["triage_wall_s_max"],
        "wall_within_max": wall_s <= FROZEN_THRESHOLDS["triage_wall_s_max"],
        "manual_baseline_s": FROZEN_THRESHOLDS["manual_baseline_s"],
        "speedup_x": round(FROZEN_THRESHOLDS["manual_baseline_s"] / wall_s, 1) if wall_s > 0 else None,
        "beats_manual_baseline": wall_s < FROZEN_THRESHOLDS["manual_baseline_s"],
        "required_steps_missing": [
            name
            for name in FROZEN_THRESHOLDS["required_steps"]
            if steps_by_name(steps, name).get("status") not in ("ok", "skipped")
        ],
    }
    report["threshold_checks"] = checks
    artifacts = {
        "report_json": os.path.join(out, "triage-report.json"),
        "report_md": os.path.join(out, "triage-report.md"),
    }
    report["artifacts"] = artifacts
    write_json(artifacts["report_json"], report)
    write_text(artifacts["report_md"], render_markdown(report))
    return report


def steps_by_name(steps: list[dict], name: str) -> dict:
    return next((step for step in steps if step["name"] == name), {})


# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------


def build_plan(options: TriageOptions) -> dict:
    out = os.path.abspath(options.out) if options.out else "<out>"
    return {
        "format": PLAN_FORMAT,
        "tool": "harness/triage.py",
        "dry_run": True,
        "status": "planned",
        "rm610_gate": {"status": "passed", "labels": sorted(set(options.labels()))},
        "firmware": {"path": options.firmware, "version": options.version or "auto"},
        "baseline": options.baseline,
        "out": out,
        "budget": options.budget,
        "steps": [
            {"name": name, "budget": options.budget if name in ("native_boot", "ab_eval") else None}
            for name, _fn in DEFAULT_RUNNERS
        ],
        "side_effects": {
            "writes": [
                os.path.join(out, "triage-report.json"),
                os.path.join(out, "triage-report.md"),
                os.path.join(out, "sessions/"),
                os.path.join(out, "ab-eval/"),
                os.path.join(out, "version-diff/ (only with --baseline)"),
            ],
            "hardware": "none",
            "network": "none",
            "max_instructions_per_attempt": MAX_INSTRUCTIONS,
        },
    }


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(prog="risc-v-gsp-rm-debugger triage", description=__doc__.splitlines()[0])
    parser.add_argument("--firmware", required=True, help="new GSP firmware image (read-only)")
    parser.add_argument("--baseline", help="previous firmware image for the version diff (optional)")
    parser.add_argument("--version", help="version label (default: gsp_version.txt / dir / stem)")
    parser.add_argument("--baseline-label", help="version label recorded for the baseline")
    parser.add_argument("--out", help="report directory (required unless --dry-run)")
    parser.add_argument("--budget", type=int, default=DEFAULT_BUDGET,
                        help=f"instruction budget per native attempt (1..{MAX_INSTRUCTIONS})")
    parser.add_argument("--qemu-duration", type=float, default=DEFAULT_QEMU_DURATION_S,
                        help="QEMU probe wall-clock bound in seconds")
    parser.add_argument("--step-timeout", type=float, default=DEFAULT_STEP_TIMEOUT_S,
                        help="per-step wall-clock net in seconds")
    parser.add_argument("--release", help="release label; an rm610 release is gated by docs/23")
    parser.add_argument("--approval-token", help=f"explicit RM610 token ({RM610_TOKEN})")
    parser.add_argument("--dry-run", action="store_true",
                        help="print the plan and exit without opening targets or creating --out")
    return parser


def options_from_args(args: argparse.Namespace) -> TriageOptions:
    return TriageOptions(
        firmware=args.firmware,
        baseline=args.baseline,
        version=args.version,
        baseline_label=args.baseline_label,
        out=args.out,
        budget=args.budget,
        qemu_duration=args.qemu_duration,
        step_timeout=args.step_timeout,
        release=args.release,
        approval_token=args.approval_token,
        dry_run=args.dry_run,
    )


def main(argv: list[str] | None = None) -> int:
    parser = build_parser()
    args = parser.parse_args(argv)
    options = options_from_args(args)
    try:
        check_rm610_gate(options.labels(), approval_token=options.approval_token)
    except Rm610GateError as exc:
        print(f"[triage] RM610 gate refused: {exc}", file=sys.stderr)
        return 3
    if options.dry_run:
        plan = build_plan(options)
        print(json.dumps(plan, indent=2, sort_keys=True))
        print(f"[triage] dry-run: {len(plan['steps'])} steps planned; no targets opened, no --out created")
        return 0
    if not options.out:
        parser.error("--out is required unless --dry-run is used")
    if options.budget <= 0 or options.budget > MAX_INSTRUCTIONS:
        parser.error(f"--budget must be in 1..{MAX_INSTRUCTIONS}")
    if not os.path.isfile(options.firmware):
        print(f"[triage] skipped: firmware absent: {options.firmware}", file=sys.stderr)
        return 2
    try:
        report = run_triage(options)
    except Rm610GateError as exc:
        print(f"[triage] RM610 gate refused: {exc}", file=sys.stderr)
        return 3
    except ValueError as exc:
        parser.error(str(exc))
    except Exception as exc:
        print(f"[triage] error: {type(exc).__name__}: {exc}", file=sys.stderr)
        return 2
    for step in report["steps"]:
        reason = f" reason={step['reason']}" if step.get("reason") else ""
        print(f"[triage] {step['name']}: status={step['status']}{reason}")
    readiness = report["support_readiness"]
    print(f"[triage] readiness: {readiness['verdict']} (baseline_compared={readiness['baseline_compared']})")
    for action in readiness["next_actions"][:6]:
        print(f"[triage] next: {action}")
    checks = report["threshold_checks"]
    print(f"[triage] wall: {checks['triage_wall_s']}s vs manual baseline {checks['manual_baseline_s']}s "
          f"(speedup {checks['speedup_x']}x)")
    print(f"[triage] wrote {report['artifacts']['report_json']} + {report['artifacts']['report_md']}")
    return 1 if report["errors"] else 0


if __name__ == "__main__":  # pragma: no cover
    raise SystemExit(main())