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

#!/usr/bin/env python3
"""New-device port scaffold (roadmap H4.1/R6, docs/07 P2, docs/22 §3, docs/28).

``--device <name>`` creates or uses a profile JSON holding the user-supplied
device facts (PCI IDs, BARs, FB geometry, engine inventory, interrupt topology,
ROM/VBIOS notes) and then drives, on the local QEMU backend only:

1. probe-first-touch on the device's firmware (first unmapped access, bounded);
2. a symbol inventory (ELF .symtab + optional Ghidra/risc-v-gsp-rm-debugger DB sources);
3. a portability matrix run of the device against a reference firmware;
4. a port checklist report against the docs/07 evidence checklist.

``--dry-run`` is the default: it prints the plan (profile action, missing
profile fields, steps, expected artifacts) and exits 0 without creating files.
``--run`` writes an evidence bundle under ``--out`` with a sha256 manifest.
The RM 610 gate from ``version_diff.py`` (docs/23) is enforced before any file
is opened, in dry-run too.

This tool never touches hardware: no /dev, no kernel module, no mdev, no
guest, no physical GPU. Physical validation is a separate supervised step
(docs/06-portability/hardware-software-matrix.md §5.1).

Run from the tool root:

    python3 harness/port_new_device.py --device rtx4090 --profile profiles/rtx4090.json
    python3 harness/port_new_device.py --run --device rtx4090 \\
        --firmware ../../firmware/57X-GSP/570.144/gsp-570.144.fwimage.bin \\
        --reference firmware/uproc_demo/build/uproc_demo.elf --out dumps/port-rtx4090
"""
from __future__ import annotations

import argparse
import json
import os
import re
import sys
from dataclasses import dataclass

try:
    from .common import (
        MAX_LOG_BYTES,
        SymbolDB,
        ensure_dir,
        qemu_available,
        sha256_file,
        write_json,
        write_text,
    )
    from .portability_matrix import (
        MatrixOptions,
        TargetSpec,
        render_markdown as render_matrix_markdown,
        run_matrix,
    )
    from .version_diff import (
        RM610_TOKEN,
        Rm610GateError,
        VersionDiffOptions,
        check_rm610_gate,
        load_any_target,
        load_symbol_sources,
        run_probe,
    )
except ImportError:  # direct script execution (python3 harness/port_new_device.py)
    from common import (  # type: ignore[no-redef]
        MAX_LOG_BYTES,
        SymbolDB,
        ensure_dir,
        qemu_available,
        sha256_file,
        write_json,
        write_text,
    )
    from portability_matrix import (  # type: ignore[no-redef]
        MatrixOptions,
        TargetSpec,
        render_markdown as render_matrix_markdown,
        run_matrix,
    )
    from version_diff import (  # type: ignore[no-redef]
        RM610_TOKEN,
        Rm610GateError,
        VersionDiffOptions,
        check_rm610_gate,
        load_any_target,
        load_symbol_sources,
        run_probe,
    )

FORMAT = "greeneye-device-port/1"
PLAN_FORMAT = "greeneye-device-port-plan/1"
PROFILE_FORMAT = "greeneye-device-profile/1"
BUNDLE_FORMAT = "greeneye-release-bundle/1"
DEVICE_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]*$")

#: Where each user-supplied profile section comes from (docs/22 §3.1).
PROFILE_HELP = {
    "pci": "lspci / vendor docs, config/hardware/*.yaml is a design target only",
    "bars": "guest lspci -vvv, host sysfs",
    "fb": "host oracle: rmtool mem stats --format=json",
    "engines": "host oracle: rmtool engine list --format=json; validate per driver profile",
    "interrupts": "MSI-X tables, guest lspci -vvv",
    "rom": "ROM/VBIOS read; laptop/Surface overrides",
    "firmware": "extracted GSP .fwimage or re-hosted ELF (user-supplied; never downloaded)",
    "pairing": "verify image family + FMC manifest + version string before debugging (docs/07 P2.2)",
    "physical": "supervised physical validation only; this tool is simulator-only (docs/22 §3.5)",
}


@dataclass
class DeviceOptions:
    device: str
    profile: str | None = None
    firmware: str | None = None
    reference: str | None = None
    symbols: str | None = None
    ghidra_exports: str | None = None
    ghidra_bias: int = 0
    out: str | None = None
    run: bool = False
    approval_token: str | None = None
    probe_instructions: int = 2000
    probe_duration: float = 0.5
    scenario_duration: float = 0.4
    max_stops: int = 5000
    max_rows: int = 40
    step_timeout: float = 600.0
    log_bytes: int = MAX_LOG_BYTES

    @property
    def dry_run(self) -> bool:
        return not self.run


# ---------------------------------------------------------------------------
# profile
# ---------------------------------------------------------------------------


def scaffold_profile(name: str) -> dict:
    """Empty user-supplied profile; every field documents where it comes from."""
    return {
        "format": PROFILE_FORMAT,
        "name": name,
        "vendor": "",
        "device": "",
        "pci": {"vendor_id": "0x10de", "device_id": "", "subsystem_vendor_id": "",
                "subsystem_device_id": ""},
        "bars": [],
        "fb": {"vram_mib": None, "bar1_mib": None, "ecc": None, "notes": ""},
        "engines": [],
        "interrupts": {"msix_vectors": None, "notes": ""},
        "rom": {"vblos_sha256": "", "notes": ""},
        "firmware": "",
        "firmware_sha256": "",
        "firmware_family": "",
        "firmware_version": "",
        "reference": "",
        "physical_validation": {"status": "pending", "date": "", "by": "", "evidence": "", "notes": ""},
        "notes": "",
        "where_to_fill": dict(PROFILE_HELP),
        "references": [
            "docs/07-PORTABILITY-PLAYBOOKS.md P2",
            "docs/22-PORTING-CUDA-AND-DEVICES.md 3",
            "docs/06-portability/hardware-software-matrix.md 5.1",
        ],
    }


def load_profile(path: str) -> tuple[dict, bool]:
    """Return ``(profile, created)``; an unreadable profile is a clean error."""
    if not os.path.isfile(path):
        return {}, False
    with open(path, "r", encoding="utf-8") as fh:
        data = json.load(fh)
    if not isinstance(data, dict):
        raise ValueError(f"profile {path}: expected a JSON object")
    return data, True


def profile_missing_fields(profile: dict) -> list[str]:
    missing: list[str] = []
    pci = profile.get("pci") or {}
    if not any(str(pci.get(key, "")).strip() for key in ("vendor_id", "device_id", "subsystem_device_id")):
        missing.append("pci")
    if not profile.get("bars"):
        missing.append("bars")
    fb = profile.get("fb") or {}
    if not any(fb.get(key) not in (None, "", 0) for key in ("vram_mib", "bar1_mib", "notes")):
        missing.append("fb")
    if not profile.get("engines"):
        missing.append("engines")
    interrupts = profile.get("interrupts") or {}
    if not any(interrupts.get(key) not in (None, "", 0) for key in ("msix_vectors", "notes")):
        missing.append("interrupts")
    rom = profile.get("rom") or {}
    if not any(str(rom.get(key, "")).strip() for key in ("vblos_sha256", "notes")):
        missing.append("rom")
    return missing


def profile_labels(profile: dict, device: str, profile_path: str | None) -> list[str]:
    labels = [device, os.path.basename(profile_path or "")]
    for key in ("name", "device", "firmware", "reference", "firmware_family", "firmware_version"):
        value = profile.get(key)
        if isinstance(value, str) and value:
            labels.append(os.path.basename(value))
    return labels


def merge_profile(profile: dict, options: DeviceOptions) -> dict:
    merged = json.loads(json.dumps(profile))
    if options.firmware:
        merged["firmware"] = options.firmware
    if options.reference:
        merged["reference"] = options.reference
    return merged


# ---------------------------------------------------------------------------
# plan (dry run)
# ---------------------------------------------------------------------------


def _expected_artifacts(out: str, device: str) -> list[str]:
    base = os.path.join(out, device)
    return [
        os.path.join(base, "device-port.json"),
        os.path.join(base, "device-port.md"),
        os.path.join(base, "port-checklist.json"),
        os.path.join(base, "port-checklist.md"),
        os.path.join(base, "symbols", f"{device}.json"),
        os.path.join(base, "sessions"),
        os.path.join(base, "portability-matrix", "portability-report.json"),
        os.path.join(base, "portability-matrix", "portability-report.md"),
        os.path.join(base, "bundle-manifest.json"),
    ]


def build_plan(options: DeviceOptions) -> dict:
    profile_path = options.profile or (os.path.join(options.out or "<out>", "profiles",
                                                     f"{options.device}.json"))
    profile, created = load_profile(profile_path) if os.path.isfile(profile_path) else ({}, False)
    merged = merge_profile(profile if created else scaffold_profile(options.device), options)
    firmware = merged.get("firmware") or ""
    reference = merged.get("reference") or ""
    check_rm610_gate(
        profile_labels(profile if created else {}, options.device, profile_path)
        + [os.path.basename(firmware), os.path.basename(reference)],
        approval_token=options.approval_token,
    )
    out = os.path.abspath(options.out) if options.out else "<out>"
    return {
        "format": PLAN_FORMAT,
        "tool": "harness/port_new_device.py",
        "dry_run": True,
        "rm610_gate": {"status": "passed", "labels": sorted(set(
            profile_labels(profile if created else {}, options.device, profile_path)
            + [os.path.basename(firmware), os.path.basename(reference)]
        ))},
        "device": options.device,
        "profile": {
            "path": profile_path,
            "action": "use" if created else "would-create-scaffold",
            "missing_fields": profile_missing_fields(merged),
            "where_to_fill": PROFILE_HELP,
        },
        "firmware": {"path": firmware or None, "found": bool(firmware) and os.path.isfile(firmware)},
        "reference": {"path": reference or None,
                      "found": bool(reference) and os.path.isfile(reference)},
        "out": out,
        "steps": [
            {"name": "probe_first_touch", "target": firmware or None,
             "out": os.path.join(out, options.device, "sessions", f"{options.device}-probe")},
            {"name": "symbol_inventory", "sources": [firmware or None, options.symbols]},
            {"name": "portability_matrix", "targets": [reference or None, firmware or None]},
            {"name": "port_checklist", "out": os.path.join(out, options.device, "port-checklist.json")},
        ],
        "expected_artifacts": _expected_artifacts(out, options.device),
        "interpretation": (
            "dry-run: fill the profile fields, then pass --run; physical validation "
            "is a separate supervised step and is never done by this tool"
        ),
    }


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


def _run_step(name: str, fn, *args, **kwargs) -> dict:
    try:
        detail = fn(*args, **kwargs)
    except Exception as exc:  # a failed step must not hide the others
        return {"name": name, "status": "error", "reason": f"{type(exc).__name__}: {exc}"}
    entry = {"name": name}
    entry.update(detail)
    return entry


def step_probe(loaded, db: SymbolDB, options: DeviceOptions, out_dir: str, qemu_ok: bool) -> dict:
    vd = VersionDiffOptions(
        old=loaded.path,
        new=loaded.path,
        out=out_dir,
        probe_instructions=options.probe_instructions,
        probe_duration=options.probe_duration,
        log_bytes=options.log_bytes,
        approval_token=options.approval_token,
    )
    result = run_probe(loaded, db, vd, out_dir, qemu_ok)
    return {
        "status": result["status"],
        "reason": result.get("reason"),
        "session": result.get("session"),
        "stopped_pc": result.get("stopped_pc"),
        "stopped_symbol": result.get("stopped_symbol"),
        "fault_count": result.get("fault_count"),
        "first_fault": result.get("first_fault"),
    }


def step_symbols(loaded, db: SymbolDB, sources: list[str], notes: list[str],
                 options: DeviceOptions, out_dir: str) -> dict:
    records = [
        {"name": symbol.name, "addr": symbol.addr, "size": symbol.size, "source": symbol.source}
        for symbol in db
    ]
    rel = None
    if records:
        rel = os.path.join("symbols", f"{options.device}.json")
        path = os.path.join(out_dir, rel)
        ensure_dir(os.path.dirname(path))
        db.save_json(path, meta={
            "device": options.device,
            "firmware": os.path.basename(loaded.path),
            "bias": loaded.bias,
            "sources": sources,
            "notes": notes,
        })
    return {
        "status": "ok" if records else "skipped",
        "reason": None if records else "no symbols: no ELF .symtab and no Ghidra/JSON sources supplied",
        "count": len(records),
        "sources": sources,
        "notes": notes,
        "db": rel,
    }


def step_matrix(firmware: str, reference: str | None, options: DeviceOptions, out_dir: str) -> dict:
    specs = []
    if reference and os.path.isfile(reference):
        specs.append(TargetSpec(path=os.path.abspath(reference), name="reference"))
    specs.append(TargetSpec(path=os.path.abspath(firmware), name=options.device))
    matrix = run_matrix(
        specs,
        MatrixOptions(
            out=out_dir,
            probe_instructions=options.probe_instructions,
            probe_duration=options.probe_duration,
            scenario_duration=options.scenario_duration,
            max_stops=options.max_stops,
            max_rows=options.max_rows,
            log_bytes=options.log_bytes,
        ),
    )
    write_json(os.path.join(out_dir, "portability-report.json"), matrix)
    write_text(
        os.path.join(out_dir, "portability-report.md"),
        render_matrix_markdown(matrix, MatrixOptions(out=out_dir, max_rows=options.max_rows)),
    )
    statuses = {row["name"]: row["status"] for row in matrix["targets"]}
    return {
        "status": "error" if "error" in statuses.values() else "ok",
        "report": "portability-matrix/portability-report.json",
        "targets": statuses,
        "comparisons": [
            {"a": row["a"], "b": row["b"], "status": row["status"], "all_zero": row["all_zero"]}
            for row in matrix["comparisons"]
        ],
    }


def build_checklist(profile: dict, firmware: str, reference: str | None, results: dict) -> dict:
    """docs/07 evidence checklist + docs/22 §3.1 profile items."""
    missing = profile_missing_fields(profile)
    pairing_ok = bool(str(profile.get("firmware_family", "")).strip()
                      and str(profile.get("firmware_version", "")).strip())
    physical = profile.get("physical_validation") or {}
    physical_status = str(physical.get("status", "pending")).lower()
    probe = results.get("probe") or {}
    symbols = results.get("symbols") or {}
    matrix = results.get("matrix") or {}
    items = [
        {
            "id": "profile",
            "item": "device profile fields recorded (PCI IDs, BARs, FB, engines, interrupts, ROM)",
            "status": "ok" if not missing else "pending",
            "detail": "missing: " + ", ".join(missing) if missing else "all profile sections populated",
        },
        {
            "id": "pairing",
            "item": "firmware hash + family/version pairing recorded before debugging (docs/07 P2.2)",
            "status": "ok" if pairing_ok else "pending",
            "detail": f"family={profile.get('firmware_family') or '-'} version={profile.get('firmware_version') or '-'}",
        },
        {
            "id": "hash",
            "item": "firmware sha256 recorded",
            "status": "ok" if firmware and os.path.isfile(firmware) else "missing",
            "detail": sha256_file(firmware) if firmware and os.path.isfile(firmware) else "firmware not found",
        },
        {
            "id": "probe",
            "item": "probe first-touch captured and explained as profile vs logic",
            "status": "ok" if probe.get("status") == "ok" else "pending",
            "detail": f"stopped_pc={probe.get('stopped_pc')} faults={probe.get('fault_count')}",
        },
        {
            "id": "symbols",
            "item": "symbol inventory built with provenance",
            "status": "ok" if symbols.get("status") == "ok" else "pending",
            "detail": f"count={symbols.get('count')} sources={symbols.get('sources')}",
        },
        {
            "id": "matrix",
            "item": "portability matrix run vs the reference firmware",
            "status": "ok" if matrix.get("status") == "ok" and reference else "pending",
            "detail": f"targets={matrix.get('targets')}",
        },
        {
            "id": "docs07",
            "item": "docs/07 evidence checklist attachable (hash, probe, symbols, sessions, compare, M1-M6)",
            "status": "ok" if firmware else "missing",
            "detail": "bundle-manifest.json hashes every artifact",
        },
        {
            "id": "physical",
            "item": "physical/supervised validation done (outside this simulator-only tool)",
            "status": "ok" if physical_status in ("passed", "ok", "done") else "pending",
            "detail": f"status={physical_status}; docs/22 §3.5: simulator-only evidence never justifies the support claim",
        },
    ]
    required_pending = [row["id"] for row in items
                        if row["status"] != "ok" and row["id"] != "physical"]
    if required_pending:
        recommendation = "hold: " + ", ".join(required_pending) + " not complete"
    elif physical_status in ("passed", "ok", "done"):
        recommendation = "go: simulator evidence complete and physical validation recorded"
    else:
        recommendation = "simulator-evidence-complete; physical validation pending (supervised step)"
    return {
        "format": "greeneye-device-checklist/1",
        "tool": "harness/port_new_device.py",
        "device": profile.get("name") or "-",
        "firmware": firmware,
        "reference": reference,
        "profile_missing_fields": missing,
        "items": items,
        "recommendation": recommendation,
        "safety": "simulator-only: no /dev, no kernel module, no mdev, no guest, no physical GPU",
    }


def render_checklist_markdown(checklist: dict) -> str:
    lines = [
        f"# risc-v-gsp-rm-debugger port checklist: {checklist['device']}",
        "",
        "> TOOL IN DEVELOPMENT - NOT READY YET! Generated by `harness/port_new_device.py`.",
        "",
        f"* firmware: `{checklist['firmware']}`",
        f"* reference: `{checklist['reference']}`",
        f"* recommendation: **{checklist['recommendation']}**",
        f"* safety: {checklist['safety']}",
        "",
        "| id | item | status | detail |",
        "|---|---|---|---|",
    ]
    for row in checklist["items"]:
        lines.append(f"| {row['id']} | {row['item']} | {row['status']} | {row['detail']} |")
    lines += ["", "Evidence: `port-checklist.json`, `bundle-manifest.json`, `portability-matrix/`.", ""]
    return "\n".join(lines)


def build_bundle(out_dir: str) -> dict:
    files: list[dict] = []
    for root, dirs, names in os.walk(out_dir):
        dirs.sort()
        for name in sorted(names):
            path = os.path.join(root, name)
            rel = os.path.relpath(path, out_dir)
            if rel == "bundle-manifest.json":
                continue
            files.append({"path": rel, "bytes": os.path.getsize(path), "sha256": sha256_file(path)})
    files.sort(key=lambda row: row["path"])
    return {"format": BUNDLE_FORMAT, "tool": "harness/port_new_device.py", "files": files}


def run_device_port(options: DeviceOptions) -> dict:
    if options.out is None:
        raise ValueError("--out is required unless --dry-run is used")
    out_root = os.path.abspath(options.out)
    device_dir = os.path.join(out_root, options.device)
    profile_path = options.profile or os.path.join(out_root, "profiles", f"{options.device}.json")
    profile, created = load_profile(profile_path)
    created_now = not created
    if not created:
        profile = scaffold_profile(options.device)
    profile = merge_profile(profile, options)
    labels = profile_labels(profile, options.device, profile_path)
    firmware = profile.get("firmware") or ""
    reference = profile.get("reference") or ""
    check_rm610_gate(labels + [os.path.basename(firmware), os.path.basename(reference)],
                     approval_token=options.approval_token)
    if created_now:
        ensure_dir(os.path.dirname(profile_path))
        write_json(profile_path, profile)
    if not firmware:
        raise ValueError(
            "no device firmware: pass --firmware PATH or set \"firmware\" in the profile "
            f"({profile_path}); see docs/07 P2"
        )
    loaded = load_any_target(firmware)
    if not loaded.ok:
        raise ValueError(f"firmware not usable: {loaded.path}: {loaded.reason or loaded.kind}")
    if created_now:
        profile["firmware_sha256"] = loaded.sha256
        write_json(profile_path, profile)

    ensure_dir(device_dir)
    db, sources, notes = load_symbol_sources(
        loaded,
        symbols_json=options.symbols,
        ghidra_exports=options.ghidra_exports,
        ghidra_bias=options.ghidra_bias,
    )
    qemu_ok = qemu_available()
    results: dict = {}
    steps = [
        _run_step("probe_first_touch", step_probe, loaded, db, options, device_dir, qemu_ok),
        _run_step("symbol_inventory", step_symbols, loaded, db, sources, notes, options, device_dir),
        _run_step("portability_matrix", step_matrix, firmware, reference, options,
                  os.path.join(device_dir, "portability-matrix")),
    ]
    results["probe"] = steps[0]
    results["symbols"] = steps[1]
    results["matrix"] = steps[2]
    checklist = build_checklist(profile, firmware, reference, results)
    write_json(os.path.join(device_dir, "port-checklist.json"), checklist)
    write_text(os.path.join(device_dir, "port-checklist.md"), render_checklist_markdown(checklist))
    checklist_step = {
        "name": "port_checklist",
        "status": "ok",
        "report": "port-checklist.json",
        "recommendation": checklist["recommendation"],
    }
    steps.append(checklist_step)
    failed = [step["name"] for step in steps if step["status"] == "error"]
    report = {
        "format": FORMAT,
        "tool": "harness/port_new_device.py",
        "dry_run": False,
        "rm610_gate": {"status": "passed", "labels": sorted(set(labels))},
        "device": options.device,
        "profile": {
            "path": os.path.relpath(profile_path, out_root),
            "created": created_now,
            "missing_fields": profile_missing_fields(profile),
        },
        "firmware": {"path": firmware, "kind": loaded.kind, "sha256": loaded.sha256},
        "reference": {"path": reference, "found": bool(reference) and os.path.isfile(reference)},
        "qemu_available": qemu_ok,
        "steps": steps,
        "checklist": checklist,
        "failed": failed,
        "manifest": "bundle-manifest.json",
    }
    write_json(os.path.join(device_dir, "device-port.json"), report)
    write_text(os.path.join(device_dir, "device-port.md"), render_report_markdown(report))
    write_json(os.path.join(device_dir, "bundle-manifest.json"), build_bundle(device_dir))
    return report


def render_report_markdown(report: dict) -> str:
    lines = [
        f"# risc-v-gsp-rm-debugger device port: {report['device']}",
        "",
        "> TOOL IN DEVELOPMENT - NOT READY YET! Generated by `harness/port_new_device.py`.",
        "",
        f"* format: `{report['format']}`",
        f"* firmware: `{report['firmware']['path']}` ({report['firmware']['kind']}, "
        f"sha256 `{report['firmware']['sha256'][:16]}`)",
        f"* reference: `{report['reference']['path'] or '-'}`",
        f"* qemu available: {report['qemu_available']}",
        "",
        "| step | status | report | reason |",
        "|---|---|---|---|",
    ]
    for step in report["steps"]:
        lines.append(f"| {step['name']} | {step['status']} | {step.get('report') or '-'} "
                     f"| {step.get('reason') or '-'} |")
    lines += [
        "",
        f"* checklist recommendation: **{report['checklist']['recommendation']}**",
        f"* manifest: `{report['manifest']}`",
        "",
    ]
    return "\n".join(lines)


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


_HELP_EPILOG = """\
invocation:
  dry-run first (default; nothing is created or touched):
    python3 harness/port_new_device.py --device rtx4090 --profile profiles/rtx4090.json
  execute (scaffolds the profile if absent; simulator only):
    python3 harness/port_new_device.py --run --device rtx4090 \\
        --firmware ../../firmware/57X-GSP/570.144/gsp-570.144.fwimage.bin \\
        --reference firmware/uproc_demo/build/uproc_demo.elf --out dumps/port-rtx4090

agent (MCP, docs/20): read the dry-run plan, fill the profile, then --run;
read device-port.json and port-checklist.json from the bundle.
RM 610 (docs/23): rm610-labelled device/firmware labels are refused without
--approval-token RM610-APPROVED or GREENEYE_RM610_APPROVED=1.
Safety: no /dev, no kernel module, no mdev, no guest, no physical GPU.
"""


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        prog="port_new_device",
        description=__doc__.splitlines()[0],
        epilog=_HELP_EPILOG,
        formatter_class=argparse.RawDescriptionHelpFormatter,
    )
    parser.add_argument("--device", required=True, help="device name (profile key, e.g. rtx4090)")
    parser.add_argument("--profile", help="profile JSON path (default: <out>/profiles/<device>.json)")
    parser.add_argument("--firmware", help="device GSP firmware image or re-hosted ELF (overrides profile)")
    parser.add_argument("--reference", help="reference firmware for the portability matrix")
    parser.add_argument("--symbols", help="risc-v-gsp-rm-debugger symbol DB JSON for the device firmware")
    parser.add_argument("--ghidra-exports", help="Ghidra exports TSV for the device firmware")
    parser.add_argument("--ghidra-bias", type=lambda value: int(value, 0), default=0)
    parser.add_argument("--out", help="evidence bundle directory (required for --run)")
    parser.add_argument("--run", dest="run", action="store_true",
                        help="execute (default is the dry-run plan)")
    parser.add_argument("--dry-run", dest="run", action="store_false",
                        help="print the plan and exit 0 (default)")
    parser.set_defaults(run=False)
    parser.add_argument("--approval-token", help=f"explicit RM610 token ({RM610_TOKEN})")
    parser.add_argument("--probe-instructions", type=int, default=2000)
    parser.add_argument("--probe-duration", type=float, default=0.5)
    parser.add_argument("--scenario-duration", type=float, default=0.4)
    parser.add_argument("--max-stops", type=int, default=5000)
    parser.add_argument("--max-rows", type=int, default=40)
    parser.add_argument("--step-timeout", type=float, default=600.0)
    parser.add_argument("--log-bytes", type=int, default=MAX_LOG_BYTES)
    return parser


def options_from_args(args: argparse.Namespace) -> DeviceOptions:
    if not DEVICE_RE.match(args.device):
        raise ValueError(f"invalid device name: {args.device!r}")
    return DeviceOptions(
        device=args.device,
        profile=args.profile,
        firmware=args.firmware,
        reference=args.reference,
        symbols=args.symbols,
        ghidra_exports=args.ghidra_exports,
        ghidra_bias=args.ghidra_bias,
        out=args.out,
        run=args.run,
        approval_token=args.approval_token,
        probe_instructions=args.probe_instructions,
        probe_duration=args.probe_duration,
        scenario_duration=args.scenario_duration,
        max_stops=args.max_stops,
        max_rows=args.max_rows,
        step_timeout=args.step_timeout,
        log_bytes=args.log_bytes,
    )


def main(argv: list[str] | None = None) -> int:
    parser = build_parser()
    args = parser.parse_args(argv)
    try:
        options = options_from_args(args)
    except ValueError as exc:
        parser.error(str(exc))
    try:
        check_rm610_gate(
            [options.device, os.path.basename(options.profile or ""),
             os.path.basename(options.firmware or ""), os.path.basename(options.reference or "")],
            approval_token=options.approval_token,
        )
        if options.dry_run:
            plan = build_plan(options)
            print(json.dumps(plan, indent=2, sort_keys=True))
            print(f"[device] dry-run: profile {plan['profile']['action']}, "
                  f"firmware found={plan['firmware']['found']}; nothing created or touched")
            return 0
        report = run_device_port(options)
    except Rm610GateError as exc:
        print(f"[device] RM610 gate refused: {exc}", file=sys.stderr)
        return 3
    except (ValueError, FileNotFoundError, json.JSONDecodeError) as exc:
        print(f"[device] error: {exc}", file=sys.stderr)
        return 2
    for step in report["steps"]:
        reason = f" reason={step['reason']}" if step.get("reason") else ""
        print(f"[device] {step['name']}: status={step['status']}{reason}")
    print(f"[device] checklist: {report['checklist']['recommendation']}")
    print(f"[device] wrote {os.path.join(os.path.abspath(options.out), options.device, 'device-port.json')}")
    print(f"[device] wrote {os.path.join(os.path.abspath(options.out), options.device, 'bundle-manifest.json')}")
    return 1 if report["failed"] else 0


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