diff options
Diffstat (limited to 'debug/risc-v-gsp-rm-vm/harness/triage.py')
| -rw-r--r-- | debug/risc-v-gsp-rm-vm/harness/triage.py | 875 |
1 files changed, 875 insertions, 0 deletions
diff --git a/debug/risc-v-gsp-rm-vm/harness/triage.py b/debug/risc-v-gsp-rm-vm/harness/triage.py new file mode 100644 index 0000000..2175e48 --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/harness/triage.py @@ -0,0 +1,875 @@ +#!/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()) |