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

#!/usr/bin/env python3
"""Multi-version portability orchestration (roadmap R6, H4.1).

Given one or more firmware ELF targets (a demo or re-hosted build, a raw
``.fwimage``, a future version), run the same deterministic scenario against
each on the QEMU backend:

1. bounded probe: relocate into virt RAM and run/step a fixed budget,
   capturing the first exception via ``parse_qemu_faults``;
2. symbol DB build: ELF symbols plus optional Ghidra exports / an existing
   green-eye JSON DB, saved per target;
3. scripted demo run with handler hooks and a control-block watchpoint,
   recorded as a normal green-eye session;
4. cross-target comparison with ``greeneye.analyze.compare_sessions`` plus
   explicit hit/watch/write profile deltas and the docs/06 M1-M6 skeleton
   (metrics without a reference capture are marked "not measured").

Outputs ``portability-report.json`` and ``portability-report.md`` under
``--out DIR`` with stable ordering. Unsupported targets are skipped with a
reason instead of aborting the matrix.

Run from the tool root:

    python3 harness/portability_matrix.py --target firmware/uproc_demo/build/uproc_demo.elf --out dumps/matrix
"""
from __future__ import annotations

import argparse
import hashlib
import os
import struct
from collections import Counter
from dataclasses import dataclass
from pathlib import Path

try:
    from .common import (
        LOAD_ADDR,
        MAX_LOG_BYTES,
        SymbolDB,
        counter_delta,
        ensure_dir,
        load_target,
        open_flat_session,
        qemu_available,
        read_log_window,
        sha256_file,
        write_json,
        write_text,
    )
except ImportError:  # direct script execution (python3 harness/portability_matrix.py)
    from common import (  # type: ignore[no-redef]
        LOAD_ADDR,
        MAX_LOG_BYTES,
        SymbolDB,
        counter_delta,
        ensure_dir,
        load_target,
        open_flat_session,
        qemu_available,
        read_log_window,
        sha256_file,
        write_json,
        write_text,
    )

from greeneye.analyze import compare_sessions, parse_qemu_faults  # noqa: E402
from greeneye.record import load_events  # noqa: E402

FORMAT = "greeneye-portability-matrix/1"
RPC_HANDLERS = ("rpc_ping", "rpc_echo", "rpc_fill", "rpc_sum", "rpc_state", "rpc_fault")
CTRL_SIZE = 400
CTRL_RPC_COUNT = 16
CTRL_STATE = 8
CTRL_HANDLER_COUNT = 328

METRIC_DEFS = (
    ("M1", "Inventory coverage"),
    ("M2", "Reply agreement"),
    ("M3", "Ordering agreement"),
    ("M4", "State agreement"),
    ("M5", "Error-path agreement"),
    ("M6", "Memory side-effects"),
)
NO_REFERENCE = "agreement is not measured: needs a reference capture (docs/06 5)"


@dataclass
class TargetSpec:
    path: str
    name: str


@dataclass
class MatrixOptions:
    out: str
    probe_instructions: int = 2000
    probe_duration: float = 0.5
    scenario_duration: float = 0.4
    max_stops: int = 5000
    max_rows: int = 40
    symbols_json: str | None = None
    ghidra_names: str | None = None
    ghidra_index: str | None = None
    ghidra_exports: str | None = None
    ghidra_callgraph: str | None = None
    ghidra_bias: int = 0
    scenario_script: str | None = None
    log_bytes: int = MAX_LOG_BYTES


def _skipped_probe(reason: str | None) -> dict:
    return {
        "status": "skipped",
        "reason": reason,
        "session": None,
        "instructions": None,
        "stopped_pc": None,
        "stopped_symbol": None,
        "fault_count": 0,
        "faults_capped": False,
        "first_fault": None,
    }


def _skipped_scenario(reason: str | None) -> dict:
    return {
        "status": "skipped",
        "reason": reason,
        "session": None,
        "hooks": [],
        "uart_banner": None,
        "ctrl": None,
        "snapshot": None,
        "event_counts": {},
        "bp": {},
        "watch": {},
        "writes": {},
        "fault_count": 0,
        "faults_capped": False,
        "first_fault": None,
    }


# ---------------------------------------------------------------------------
# per-target steps
# ---------------------------------------------------------------------------


def build_symbols(spec: TargetSpec, bias: int, out_dir: str, options: MatrixOptions):
    """ELF symbols + optional Ghidra exports / JSON DB, saved per target."""
    db = SymbolDB()
    db.load_elf(spec.path, bias=bias, source="elf")
    sources = ["elf"]
    if options.symbols_json:
        if not os.path.isfile(options.symbols_json):
            raise FileNotFoundError(f"symbol DB not found: {options.symbols_json}")
        db.load_json(options.symbols_json)
        sources.append(os.path.basename(options.symbols_json))
    ghidra_inputs = any(
        (
            options.ghidra_names,
            options.ghidra_index,
            options.ghidra_exports,
            options.ghidra_callgraph,
        )
    )
    if ghidra_inputs:
        from greeneye.ghidra import build_symbol_db

        gdb = build_symbol_db(
            names_tsv=options.ghidra_names,
            index_tsv=options.ghidra_index,
            exports_tsv=options.ghidra_exports,
            callgraph_tsv=options.ghidra_callgraph,
            bias=options.ghidra_bias,
        )
        for symbol in gdb:
            db.add(symbol)
        sources.append("ghidra")
    rel = os.path.join("symbols", f"{spec.name}.json")
    out_path = os.path.join(out_dir, rel)
    ensure_dir(os.path.dirname(out_path))
    db.save_json(
        out_path,
        meta={"bias": bias, "sources": sources, "firmware": os.path.basename(spec.path)},
    )
    return db, sources, rel


def run_probe(spec: TargetSpec, image, db: SymbolDB, out_dir: str, options: MatrixOptions, qemu_ok: bool) -> dict:
    if not qemu_ok:
        return _skipped_probe("qemu-system-riscv64 not found")
    result = _skipped_probe(None)
    result["instructions"] = options.probe_instructions
    session_rel = os.path.join("sessions", f"{spec.name}-probe")
    try:
        with open_flat_session(
            image,
            spec.path,
            record=os.path.join(out_dir, session_rel),
            name=f"{spec.name}-probe",
            symbols=db,
            log_faults=True,
        ) as fs:
            session = fs.session
            if options.probe_instructions > 0:
                session.step(options.probe_instructions)
            else:
                session.run(duration=options.probe_duration)
            result["stopped_pc"] = session.pc
            symbol = session.symbols.resolve(session.pc)
            result["stopped_symbol"] = symbol.name if symbol else None
            text, capped = read_log_window(fs.target.qemu_log, 0, options.log_bytes)
            faults = parse_qemu_faults(text)
            result["fault_count"] = len(faults)
            result["faults_capped"] = capped
            result["first_fault"] = faults[0] if faults else None
            result["session"] = session_rel
            if text:
                fs.session.recorder.meta["qemu_log_blob"] = fs.session.recorder.blob("qemu.log", text.encode("utf-8"))
                fs.session.recorder.meta["faults_capped"] = capped
            result["status"] = "ok"
    except Exception as exc:  # a failed step must not abort the matrix
        result["status"] = "error"
        result["reason"] = f"{type(exc).__name__}: {exc}"
    return result


def install_scenario(session) -> list[str]:
    hooks: list[str] = []
    for name in RPC_HANDLERS:
        symbol = session.symbols.get(name)
        if symbol is None:
            continue
        session.hook(symbol.addr, on_enter=lambda hit: None, listener=f"matrix:{name}")
        hooks.append(name)
    return hooks


def read_ctrl(session, symbol) -> dict | None:
    if symbol is None:
        return None
    raw = session.memory.read(symbol.addr, CTRL_SIZE)
    u32 = lambda offset: struct.unpack_from("<I", raw, offset)[0]  # noqa: E731
    u64 = lambda offset: struct.unpack_from("<Q", raw, offset)[0]  # noqa: E731
    return {
        "addr": f"{symbol.addr:#x}",
        "magic": u32(0),
        "version": u32(4),
        "state": u32(CTRL_STATE),
        "rpc_count": u64(CTRL_RPC_COUNT),
        "last_rpc": u64(24),
        "result": u64(64),
        "handler_count": u64(CTRL_HANDLER_COUNT),
    }


def profile_from_events(events: list[dict]) -> dict:
    counts: Counter = Counter()
    bp: Counter = Counter()
    watch: Counter = Counter()
    writes: Counter = Counter()
    for event in events:
        kind = event.get("type", "?")
        counts[kind] += 1
        if kind == "bp_hit":
            bp[event.get("symbol") or f"{event.get('pc', 0):#x}"] += 1
        elif kind == "watch_hit":
            watch[f"{event.get('addr', 0):#x}:{event.get('kind') or 'w'}"] += 1
        elif kind == "mem_write":
            writes[f"{(event.get('addr', 0) >> 12) << 12:#x}"] += 1
    return {
        "event_counts": dict(sorted(counts.items())),
        "bp": {key: bp[key] for key in sorted(bp)},
        "watch": {key: watch[key] for key in sorted(watch)},
        "writes": {key: writes[key] for key in sorted(writes)},
    }


def run_scenario(spec: TargetSpec, image, db: SymbolDB, out_dir: str, options: MatrixOptions, qemu_ok: bool) -> dict:
    if not qemu_ok:
        return _skipped_scenario("qemu-system-riscv64 not found")
    result = _skipped_scenario(None)
    session_rel = os.path.join("sessions", f"{spec.name}-scenario")
    try:
        with open_flat_session(
            image,
            spec.path,
            record=os.path.join(out_dir, session_rel),
            name=f"{spec.name}-scenario",
            symbols=db,
            log_faults=True,
        ) as fs:
            session = fs.session
            result["hooks"] = install_scenario(session)
            ctrl_symbol = session.symbols.get("gsp_ctrl")
            ctrl_addr = ctrl_symbol.addr if ctrl_symbol else None
            if ctrl_addr is not None:
                session.watch(ctrl_addr + CTRL_RPC_COUNT, size=8, kind="w", listener="matrix:rpc_count")
            if options.scenario_script:
                import greeneye as ge

                script_path = Path(options.scenario_script)
                namespace = {
                    "ge": ge,
                    "greeneye": ge,
                    "session": session,
                    "s": session,
                    "memory": session.memory,
                    "symbols": session.symbols,
                    "recorder": session.recorder,
                }
                exec(compile(script_path.read_text(encoding="utf-8"), str(script_path), "exec"), namespace)
            else:
                session.run(duration=options.scenario_duration, max_stops=options.max_stops)
                if ctrl_addr is not None:
                    session.memory.write_u64(ctrl_addr + 32, 1)  # PING
                    session.memory.write_u64(ctrl_addr + 40, 0x1234)
                    session.run(duration=options.scenario_duration, max_stops=options.max_stops)
            result["ctrl"] = read_ctrl(session, ctrl_symbol)
            if ctrl_addr is not None:
                data = session.snapshot(ctrl_addr, CTRL_SIZE, name="ctrl")
                result["snapshot"] = {"label": "ctrl", "size": len(data), "sha256": hashlib.sha256(data).hexdigest()}
            uart = fs.target.read_uart()
            result["uart_banner"] = uart.strip().splitlines()[0] if uart.strip() else None
            text, capped = read_log_window(fs.target.qemu_log, 0, options.log_bytes)
            faults = parse_qemu_faults(text)
            result["fault_count"] = len(faults)
            result["faults_capped"] = capped
            result["first_fault"] = faults[0] if faults else None
        result.update(profile_from_events(load_events(os.path.join(out_dir, session_rel))))
        result["session"] = session_rel
        result["status"] = "ok"
    except Exception as exc:  # a failed step must not abort the matrix
        result["status"] = "error"
        result["reason"] = f"{type(exc).__name__}: {exc}"
    return result


# ---------------------------------------------------------------------------
# metrics + comparisons
# ---------------------------------------------------------------------------


def metric_rows(probe: dict, scenario: dict) -> list[dict]:
    """docs/06 M1-M6 skeleton; observed data is local, agreement stays unmeasured."""
    bp = scenario.get("bp") or {}
    ctrl = scenario.get("ctrl") or {}
    snapshot = scenario.get("snapshot") or {}
    probe_fault = (probe.get("first_fault") or {}).get("desc")
    scenario_fault = (scenario.get("first_fault") or {}).get("desc")
    rows = [
        {
            "id": "M1",
            "name": "Inventory coverage",
            "status": "not measured",
            "reason": NO_REFERENCE,
            "observed": {
                "rpc_handlers_hit": sorted(k for k in bp if k in RPC_HANDLERS),
                "other_hit_sites": sorted(k for k in bp if k not in RPC_HANDLERS),
                "declared_handlers": ctrl.get("handler_count"),
            },
        },
        {
            "id": "M2",
            "name": "Reply agreement",
            "status": "not measured",
            "reason": NO_REFERENCE,
            "observed": {"note": "reply fields are not decoded by the builtin scenario", "result": ctrl.get("result")},
        },
        {
            "id": "M3",
            "name": "Ordering agreement",
            "status": "not measured",
            "reason": NO_REFERENCE,
            "observed": {"events": sum((scenario.get("event_counts") or {}).values())},
        },
        {
            "id": "M4",
            "name": "State agreement",
            "status": "not measured",
            "reason": NO_REFERENCE,
            "observed": {"final_state": ctrl.get("state"), "rpc_state_hits": bp.get("rpc_state", 0)},
        },
        {
            "id": "M5",
            "name": "Error-path agreement",
            "status": "not measured",
            "reason": NO_REFERENCE,
            "observed": {
                "probe_first_fault": probe_fault,
                "scenario_first_fault": scenario_fault,
                "scenario_fault_count": scenario.get("fault_count", 0),
            },
        },
        {
            "id": "M6",
            "name": "Memory side-effects",
            "status": "not measured",
            "reason": NO_REFERENCE,
            "observed": {
                "written_pages": len(scenario.get("writes") or {}),
                "ctrl_snapshot_sha256": snapshot.get("sha256"),
            },
        },
    ]
    return rows


def compare_targets(a: dict, b: dict, out_dir: str, options: MatrixOptions) -> dict:
    entry = {
        "a": a["name"],
        "b": b["name"],
        "status": "ok",
        "reason": None,
        "compare_report": None,
        "deltas": {"bp": [], "watch": [], "writes": []},
        "all_zero": True,
        "metrics": [],
    }
    a_scenario = a.get("scenario") or {}
    b_scenario = b.get("scenario") or {}
    if a_scenario.get("status") != "ok" or b_scenario.get("status") != "ok":
        entry["status"] = "skipped"
        entry["reason"] = "a scenario session is not available on both targets"
        return entry
    a_dir = os.path.join(out_dir, a_scenario.get("session") or "")
    b_dir = os.path.join(out_dir, b_scenario.get("session") or "")
    try:
        report = compare_sessions(a_dir, b_dir, max_rows=options.max_rows)
        rel = os.path.join("compare", f"{a['name']}--{b['name']}.md")
        write_text(os.path.join(out_dir, rel), report)
        entry["compare_report"] = rel
    except Exception as exc:
        entry["reason"] = f"{type(exc).__name__}: {exc}"
    deltas = {"bp": [], "watch": [], "writes": []}
    for name in ("bp", "watch", "writes"):
        deltas[name] = counter_delta(a_scenario.get(name) or {}, b_scenario.get(name) or {})
    entry["deltas"] = deltas
    entry["all_zero"] = all(row["delta"] == 0 for rows in deltas.values() for row in rows)
    a_metrics = {row["id"]: row for row in a.get("metrics") or []}
    b_metrics = {row["id"]: row for row in b.get("metrics") or []}
    entry["metrics"] = [
        {
            "id": metric_id,
            "name": name,
            "status": "not measured",
            "reason": NO_REFERENCE,
            "a_observed": (a_metrics.get(metric_id) or {}).get("observed"),
            "b_observed": (b_metrics.get(metric_id) or {}).get("observed"),
        }
        for metric_id, name in METRIC_DEFS
    ]
    return entry


# ---------------------------------------------------------------------------
# report assembly + rendering
# ---------------------------------------------------------------------------


def analyze_target(spec: TargetSpec, options: MatrixOptions, qemu_ok: bool) -> dict:
    target = {
        "name": spec.name,
        "path": spec.path,
        "sha256": "",
        "kind": "elf",
        "entry": None,
        "status": "skipped",
        "skip_reason": None,
        "probe": _skipped_probe(None),
        "symbols": {"count": 0, "sources": [], "path": None},
        "scenario": _skipped_scenario(None),
        "metrics": metric_rows({}, {}),
    }
    if not os.path.isfile(spec.path):
        target["skip_reason"] = "file not found"
        return target
    target["sha256"] = sha256_file(spec.path)
    try:
        image = load_target(spec.path)
    except Exception as exc:
        target["kind"] = "unknown"
        target["skip_reason"] = f"{type(exc).__name__}: {exc}"
        return target
    target["entry"] = f"{image.entry:#x}"
    bias = LOAD_ADDR - image.min_vaddr
    db = SymbolDB()
    try:
        db, sources, sym_rel = build_symbols(spec, bias, options.out, options)
        target["symbols"] = {"count": len(db), "sources": sources, "path": sym_rel}
    except Exception as exc:
        target["symbols"] = {"count": 0, "sources": [], "path": None, "reason": f"{type(exc).__name__}: {exc}"}
        db.load_elf(spec.path, bias=bias, source="elf")
        target["symbols"]["count"] = len(db)
        target["symbols"]["sources"] = ["elf"]
    target["probe"] = run_probe(spec, image, db, options.out, options, qemu_ok)
    target["scenario"] = run_scenario(spec, image, db, options.out, options, qemu_ok)
    target["metrics"] = metric_rows(target["probe"], target["scenario"])
    statuses = {target["probe"]["status"], target["scenario"]["status"]}
    if statuses == {"ok"}:
        target["status"] = "ok"
    elif "ok" in statuses:
        target["status"] = "partial"
    elif statuses == {"skipped"}:
        target["status"] = "skipped"
        target["skip_reason"] = target["probe"].get("reason") or target["scenario"].get("reason")
    else:
        target["status"] = "error"
        target["skip_reason"] = target["probe"].get("reason") or target["scenario"].get("reason")
    return target


def run_matrix(specs: list[TargetSpec], options: MatrixOptions) -> dict:
    ensure_dir(options.out)
    qemu_ok = qemu_available()
    targets = [analyze_target(spec, options, qemu_ok) for spec in specs]
    comparisons: list[dict] = []
    for i in range(len(targets)):
        for j in range(i + 1, len(targets)):
            comparisons.append(compare_targets(targets[i], targets[j], options.out, options))
    report = {
        "format": FORMAT,
        "tool": "harness/portability_matrix.py",
        "backend": "qemu",
        "qemu_available": qemu_ok,
        "options": {
            "probe_instructions": options.probe_instructions,
            "probe_duration": options.probe_duration,
            "scenario_duration": options.scenario_duration,
            "scenario_script": options.scenario_script,
            "max_rows": options.max_rows,
            "symbols_json": options.symbols_json,
            "ghidra_bias": options.ghidra_bias,
        },
        "targets": targets,
        "comparisons": comparisons,
        "skipped": [
            {"name": target["name"], "reason": target["skip_reason"]}
            for target in targets
            if target["status"] == "skipped"
        ],
    }
    return report


def _table(headers: list[str], rows: list[list]) -> list[str]:
    lines = ["| " + " | ".join(headers) + " |", "|" + "|".join("---" for _ in headers) + "|"]
    for row in rows:
        lines.append("| " + " | ".join(str(cell) for cell in row) + " |")
    return lines


def _profile_tables(title: str, profile: dict, max_rows: int) -> list[str]:
    lines = [f"**{title}**", ""]
    if not profile:
        return lines + ["_none observed_", ""]
    rows = [[f"`{key}`", count] for key, count in sorted(profile.items(), key=lambda kv: (-int(kv[1]), str(kv[0])))[:max_rows]]
    lines += _table(["key", "hits"], rows) + [""]
    return lines


def render_markdown(report: dict, options: MatrixOptions) -> str:
    lines = [
        "# green-eye portability delta report",
        "",
        "> TOOL IN DEVELOPMENT - NOT READY YET! Generated by `harness/portability_matrix.py`.",
        "",
        f"* format: `{report['format']}` (green-eye {_greeneye_version()})",
        f"* backend: `{report['backend']}` (qemu available: {report['qemu_available']})",
        f"* targets: {len(report['targets'])}",
        "",
        "## Targets",
        "",
    ]
    lines += _table(
        ["target", "status", "sha256", "probe", "symbols", "scenario events"],
        [
            [
                f"`{target['name']}`",
                target["status"],
                f"`{(target['sha256'] or '')[:16]}`",
                f"{target['probe']['status']} ({target['probe']['fault_count']} faults)",
                target["symbols"]["count"],
                sum((target["scenario"].get("event_counts") or {}).values()),
            ]
            for target in report["targets"]
        ],
    )
    lines += [""]

    for target in report["targets"]:
        lines += [f"### {target['name']}", ""]
        lines += [f"* path: `{target['path']}`", f"* entry: `{target['entry']}`", f"* sha256: `{target['sha256']}`"]
        if target["skip_reason"]:
            lines += [f"* note: {target['skip_reason']}"]
        lines += [""]

        probe = target["probe"]
        lines += ["#### Probe", ""]
        fault = probe.get("first_fault")
        stopped_pc = probe.get("stopped_pc")
        probe_rows = [
            ["status", probe["status"]],
            ["instructions", probe.get("instructions")],
            ["stopped pc", f"`{stopped_pc:#x}`" if stopped_pc is not None else None],
            ["stopped symbol", probe.get("stopped_symbol")],
            ["fault count", probe.get("fault_count")],
            ["faults capped", probe.get("faults_capped")],
        ]
        if fault:
            probe_rows.append(
                ["first fault", f"cause {fault.get('cause')} {fault.get('desc')} @ epc `{fault.get('epc', 0):#x}`"]
            )
        if probe.get("reason"):
            probe_rows.append(["reason", probe["reason"]])
        lines += _table(["field", "value"], probe_rows) + [""]

        lines += ["#### Symbols", ""]
        lines += _table(
            ["count", "sources", "path"],
            [[target["symbols"]["count"], ", ".join(target["symbols"]["sources"]), target["symbols"]["path"]]],
        ) + [""]

        scenario = target["scenario"]
        lines += ["#### Scenario", ""]
        lines += _table(
            ["status", "session", "hooks", "uart banner", "faults"],
            [[
                scenario["status"],
                f"`{scenario.get('session')}`",
                ", ".join(scenario.get("hooks") or []),
                scenario.get("uart_banner"),
                scenario.get("fault_count"),
            ]],
        ) + [""]
        lines += _profile_tables("Breakpoint hits", scenario.get("bp") or {}, options.max_rows)
        lines += _profile_tables("Watchpoint hits", scenario.get("watch") or {}, options.max_rows)
        lines += _profile_tables("Written pages", scenario.get("writes") or {}, options.max_rows)

        lines += ["#### M1-M6", ""]
        lines += [f"> {NO_REFERENCE}", ""]
        lines += _table(
            ["#", "metric", "status", "observed"],
            [
                [row["id"], row["name"], row["status"], _short_json(row["observed"])]
                for row in target["metrics"]
            ],
        ) + [""]

    lines += ["## Cross-target comparisons", ""]
    if not report["comparisons"]:
        lines += ["_single target: no comparison_", ""]
    for comp in report["comparisons"]:
        lines += [f"### {comp['a']} vs {comp['b']}", ""]
        lines += [f"* status: {comp['status']}", f"* zero deltas: {'yes' if comp['all_zero'] else 'no'}"]
        if comp.get("compare_report"):
            lines += [f"* compare report: `{comp['compare_report']}`"]
        if comp.get("reason"):
            lines += [f"* note: {comp['reason']}"]
        lines += [""]
        for name in ("bp", "watch", "writes"):
            rows = [row for row in comp["deltas"][name] if row["delta"] != 0]
            rows.sort(key=lambda row: (-abs(row["delta"]), row["key"]))
            lines += [f"**{name} delta (nonzero, first {options.max_rows})**", ""]
            if rows:
                lines += _table(
                    ["key", comp["a"], comp["b"], "delta"],
                    [[f"`{row['key']}`", row["a"], row["b"], row["delta"]] for row in rows[: options.max_rows]],
                ) + [""]
            else:
                lines += ["_no deltas_", ""]
        lines += ["**M1-M6 (not measured: reference capture required)**", ""]
        lines += _table(
            ["#", "metric", "A observed", "B observed"],
            [
                [row["id"], row["name"], _short_json(row["a_observed"]), _short_json(row["b_observed"])]
                for row in comp["metrics"]
            ],
        ) + [""]

    if report["skipped"]:
        lines += ["## Skipped targets", ""]
        lines += _table(["target", "reason"], [[f"`{item['name']}`", item["reason"]] for item in report["skipped"]])
        lines += [""]
    return "\n".join(lines)


def _short_json(value, limit: int = 72) -> str:
    import json

    if value is None:
        return "-"
    text = json.dumps(value, sort_keys=True)
    return f"`{text}`" if len(text) <= limit else f"`{text[:limit - 1]}...`"


def _greeneye_version() -> str:
    try:
        import greeneye

        return greeneye.__version__
    except Exception:  # pragma: no cover
        return "unknown"


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


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(prog="portability_matrix", description=__doc__.splitlines()[0])
    parser.add_argument("targets", nargs="*", help="firmware ELF paths")
    parser.add_argument("--target", action="append", default=[], help="firmware ELF path (repeatable)")
    parser.add_argument("--out", required=True, help="output directory for the report, sessions and symbols")
    parser.add_argument("--symbols", dest="symbols_json", help="existing greeneye symbol DB JSON")
    parser.add_argument("--ghidra-names", help="Ghidra names.tsv export")
    parser.add_argument("--ghidra-index", help="Ghidra decompiled_index.tsv export")
    parser.add_argument("--ghidra-exports", help="Ghidra exports.tsv")
    parser.add_argument("--ghidra-callgraph", help="Ghidra callgraph.tsv")
    parser.add_argument("--ghidra-bias", type=lambda value: int(value, 0), default=0)
    parser.add_argument("--probe-instructions", type=int, default=2000, help="probe step budget (0 = wall-clock run)")
    parser.add_argument("--probe-duration", type=float, default=0.5, help="probe wall budget when --probe-instructions=0")
    parser.add_argument("--scenario-duration", type=float, default=0.4, help="budget for each scenario phase")
    parser.add_argument("--max-stops", type=int, default=5000, help="safety bound on serviced stops per scenario phase")
    parser.add_argument("--max-rows", type=int, default=40, help="table row cap in reports")
    parser.add_argument("--scenario-script", help="replace the builtin scenario with a green-eye script")
    return parser


def main(argv: list[str] | None = None) -> int:
    parser = build_parser()
    args = parser.parse_args(argv)
    targets = list(args.targets) + list(args.target)
    if not targets:
        parser.error("at least one target is required (positional or --target)")
    options = MatrixOptions(
        out=os.path.abspath(args.out),
        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,
        symbols_json=args.symbols_json,
        ghidra_names=args.ghidra_names,
        ghidra_index=args.ghidra_index,
        ghidra_exports=args.ghidra_exports,
        ghidra_callgraph=args.ghidra_callgraph,
        ghidra_bias=args.ghidra_bias,
        scenario_script=args.scenario_script,
    )
    specs = _name_specs(targets)
    report = run_matrix(specs, options)
    json_path = write_json(os.path.join(options.out, "portability-report.json"), report)
    md_path = write_text(os.path.join(options.out, "portability-report.md"), render_markdown(report, options))
    print(f"[matrix] {len(specs)} target(s), out={options.out}")
    for target in report["targets"]:
        print(f"[matrix] {target['name']}: status={target['status']} probe={target['probe']['status']} "
              f"scenario={target['scenario']['status']}")
    print(f"[matrix] wrote {json_path}")
    print(f"[matrix] wrote {md_path}")
    return 0


def _name_specs(paths: list[str]) -> list[TargetSpec]:
    seen: dict[str, int] = {}
    specs: list[TargetSpec] = []
    for path in paths:
        base = Path(path).stem
        seen[base] = seen.get(base, 0) + 1
        name = base if seen[base] == 1 else f"{base}-{seen[base]}"
        specs.append(TargetSpec(path=os.path.abspath(path), name=name))
    return specs


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