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

#!/usr/bin/env python3
"""Driver-version port sweep (roadmap H4.1/R6, docs/28).

Given a version list (``--versions 540,550,560,570``) or a JSON config, find
each version's firmware target under ``--input-dir`` (driver runfiles/firmware
blobs already on disk -- this tool never downloads), then per version:

1. run ``harness/version_diff.py`` against a chosen baseline;
2. run ``harness/portability_matrix.py`` on the baseline + candidate pair;
3. run a bounded ``harness/mutation_harness.py`` smoke;
4. assemble a per-version evidence bundle and a cross-version summary
   (markdown + JSON) with the docs/06 M1-M6 skeleton.

``--dry-run`` is the default: it prints the plan (inputs found/missing, steps,
expected artifacts) and exits 0. ``--run`` executes, is resume-safe (a version
whose ``version.json`` already reports ok/partial is skipped unless --force)
and exits non-zero when any executed version has an errored step. The RM 610
gate from ``version_diff.py`` (docs/23) is enforced before any target is
opened, in dry-run too: versions >= 610 require ``--approval-token
RM610-APPROVED`` or ``GREENEYE_RM610_APPROVED=1``.

Known driver branch cadence (edit DEFAULT_VERSIONS or pass --versions/--config;
no image is ever invented or downloaded):

    branch   expected image
    540      earlier branch (cadence only)
    550      earlier branch (cadence only)
    560      earlier branch (cadence only)
    570      current anchor, e.g. 570.144
    580      expected cadence, no local image assumed
    610      RM 610 gate, refused without approval (docs/23)

Run from the tool root:

    python3 harness/port_driver_versions.py --input-dir ../../firmware/57X-GSP
    python3 harness/port_driver_versions.py --run --input-dir ../../firmware/57X-GSP \\
        --versions 540,550,560,570 --out dumps/sweep-570
"""
from __future__ import annotations

import argparse
import contextlib
import json
import os
import re
import signal
import subprocess
import sys
import threading
from dataclasses import dataclass, field

try:
    from .common import (
        DEMO_ELF,
        MAX_LOG_BYTES,
        ensure_dir,
        qemu_available,
        sha256_file,
        write_json,
        write_text,
    )
    from .mutation_harness import HarnessOptions, run_harness
    from .portability_matrix import (
        METRIC_DEFS,
        NO_REFERENCE,
        MatrixOptions,
        TargetSpec,
        render_markdown as render_matrix_markdown,
        run_matrix,
    )
    from .version_diff import (
        RM610_TOKEN,
        Rm610GateError,
        VersionDiffOptions,
        check_rm610_gate,
        is_rm610_label,
        run_version_diff,
    )
except ImportError:  # direct script execution (python3 harness/port_driver_versions.py)
    from common import (  # type: ignore[no-redef]
        DEMO_ELF,
        MAX_LOG_BYTES,
        ensure_dir,
        qemu_available,
        sha256_file,
        write_json,
        write_text,
    )
    from mutation_harness import HarnessOptions, run_harness  # type: ignore[no-redef]
    from portability_matrix import (  # type: ignore[no-redef]
        METRIC_DEFS,
        NO_REFERENCE,
        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,
        is_rm610_label,
        run_version_diff,
    )

FORMAT = "greeneye-port-sweep/1"
PLAN_FORMAT = "greeneye-port-sweep-plan/1"
VERSION_FORMAT = "greeneye-port-version/1"
BUNDLE_FORMAT = "greeneye-release-bundle/1"
CONFIG_FORMAT = "greeneye-port-sweep-config/1"

#: Known driver branch cadence; edit here or pass --versions/--config. No image
#: is downloaded by this tool: branches without local images become no-input.
DEFAULT_VERSIONS = ("540", "550", "560", "570")
RESUMABLE_STATUSES = ("ok", "partial")
TARGET_SUFFIXES = (".fwimage.bin", ".fwimage", ".elf", ".bin", ".run")
VERSION_TOKEN_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]*$")


# ---------------------------------------------------------------------------
# options + version helpers
# ---------------------------------------------------------------------------


@dataclass
class SweepTarget:
    version: str
    paths: list[str] = field(default_factory=list)
    kind: str = "none"  # fwimage | elf | raw | runfile | explicit | none
    source: str = "none"  # explicit | discovered | none
    matches: list[str] = field(default_factory=list)
    runfiles: list[str] = field(default_factory=list)
    notes: list[str] = field(default_factory=list)
    symbols: str | None = None
    ghidra_exports: str | None = None

    @property
    def path(self) -> str | None:
        return self.paths[0] if self.paths else None


@dataclass
class SweepOptions:
    versions: list[str] = field(default_factory=list)
    baseline: str | None = None
    input_dir: str | None = None
    out: str | None = None
    config: str | None = None
    targets: dict[str, list[str]] = field(default_factory=dict)
    symbols: dict[str, str] = field(default_factory=dict)
    ghidra_exports: dict[str, str] = field(default_factory=dict)
    run: bool = False
    force: bool = False
    extract_runfiles: 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
    mutation_kernel: str = DEMO_ELF
    mutation_iterations: int = 8
    mutation_seed: int = 7
    mutation_timeout: float = 0.25
    mutation_settle: float = 0.1
    allow_foreign_mutation: bool = False
    step_timeout: float = 600.0
    log_bytes: int = MAX_LOG_BYTES

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


def parse_versions(text: str) -> list[str]:
    """Parse ``540,550,560`` into a validated, de-duplicated order-preserving list."""
    versions: list[str] = []
    for token in str(text).replace(";", ",").split(","):
        value = token.strip()
        if not value:
            continue
        if not VERSION_TOKEN_RE.match(value):
            raise ValueError(f"invalid version token: {value!r}")
        if value not in versions:
            versions.append(value)
    return versions


def version_sort_key(value: str) -> tuple:
    parts = re.findall(r"\d+", value)
    return (tuple(int(part) for part in parts) if parts else (0,), value)


def version_ge_610(value: str) -> bool:
    match = re.match(r"\s*(\d+)", str(value))
    return bool(match) and int(match.group(1)) >= 610


def _is_numeric_firmware_match(version: str, text: str) -> bool:
    pattern = re.compile(r"(?<!\d)" + re.escape(version) + r"(?:\.\d+)*(?!\d)")
    return pattern.search(text) is not None


def _gate_version_labels(versions: list[str]) -> list[str]:
    """Numeric versions >= 610 must trigger the rm610-prefixed gate regex."""
    return [f"rm610-{value}" if version_ge_610(value) else value
            for value in versions if version_ge_610(value) or is_rm610_label(value)]


def gate_labels(versions: list[str], targets: dict[str, SweepTarget]) -> list[str]:
    labels = _gate_version_labels(versions)
    for target in targets.values():
        labels.append(target.version)
        labels.extend(os.path.basename(path) for path in target.paths)
        labels.extend(os.path.basename(path) for path in target.runfiles)
    return labels


# ---------------------------------------------------------------------------
# config loading
# ---------------------------------------------------------------------------


def load_config(path: str) -> dict:
    """Read a sweep config; strict enough that typos do not silently pass."""
    if not os.path.isfile(path):
        raise ValueError(f"config not found: {path}")
    try:
        with open(path, "r", encoding="utf-8") as fh:
            data = json.load(fh)
    except (OSError, json.JSONDecodeError) as exc:
        raise ValueError(f"config {path}: {type(exc).__name__}: {exc}") from exc
    if not isinstance(data, dict):
        raise ValueError(f"config {path}: expected a JSON object")
    known = {"format", "versions", "baseline", "input_dir", "targets", "symbols",
             "ghidra_exports", "notes"}
    unknown = sorted(set(data) - known)
    if unknown:
        raise ValueError(f"config {path}: unknown key(s): {', '.join(unknown)}")
    if "format" in data and data["format"] != CONFIG_FORMAT:
        raise ValueError(f"config {path}: unsupported format {data['format']!r}")
    if "versions" in data:
        if not isinstance(data["versions"], list):
            raise ValueError(f"config {path}: versions must be a list")
        data["versions"] = [str(value) for value in data["versions"]]
    for key in ("targets", "symbols", "ghidra_exports"):
        if key not in data:
            continue
        if not isinstance(data[key], dict):
            raise ValueError(f"config {path}: {key} must be an object")
        normalised: dict[str, object] = {}
        for version, value in data[key].items():
            if key == "targets":
                if isinstance(value, str):
                    value = [value]
                if not isinstance(value, list) or not all(isinstance(v, str) for v in value):
                    raise ValueError(f"config {path}: targets[{version}] must be a path or list of paths")
            elif not isinstance(value, str):
                raise ValueError(f"config {path}: {key}[{version}] must be a string path")
            normalised[str(version)] = value
        data[key] = normalised
    return data


def options_from_config(options: SweepOptions, config: dict) -> None:
    if config.get("versions"):
        options.versions = list(config["versions"])
    if config.get("baseline") and not options.baseline:
        options.baseline = str(config["baseline"])
    if config.get("input_dir") and not options.input_dir:
        options.input_dir = str(config["input_dir"])
    for version, paths in (config.get("targets") or {}).items():
        options.targets.setdefault(str(version), []).extend(paths)
    for version, path in (config.get("symbols") or {}).items():
        options.symbols.setdefault(str(version), path)
    for version, path in (config.get("ghidra_exports") or {}).items():
        options.ghidra_exports.setdefault(str(version), path)


# ---------------------------------------------------------------------------
# target discovery (local disk only)
# ---------------------------------------------------------------------------


def _kind_for(name: str) -> str | None:
    lower = name.lower()
    if not any(lower.endswith(suffix) for suffix in TARGET_SUFFIXES):
        return None
    if lower.endswith(".run"):
        return "runfile"
    if lower.endswith(".fwimage.bin") or lower.endswith(".fwimage"):
        return "fwimage"
    if lower.endswith(".elf"):
        return "elf"
    if lower.endswith(".bin"):
        return "raw"
    return None


def discover_version(input_dir: str, version: str) -> SweepTarget:
    """Find the best local firmware target for one version under ``input_dir``."""
    target = SweepTarget(version=version)
    if not input_dir:
        target.notes.append("no --input-dir supplied and no explicit target")
        return target
    if not os.path.isdir(input_dir):
        target.notes.append(f"input directory not found: {input_dir}")
        return target
    ranked: list[tuple[int, int, str, str]] = []
    for root, dirs, names in os.walk(input_dir):
        dirs.sort()
        for name in sorted(names):
            kind = _kind_for(name)
            if kind is None:
                continue
            path = os.path.join(root, name)
            rel = os.path.relpath(path, input_dir)
            parent = os.path.basename(root)
            if name and _is_numeric_firmware_match(version, name):
                scope = 0
            elif parent and _is_numeric_firmware_match(version, parent):
                scope = 1
            elif _is_numeric_firmware_match(version, rel):
                scope = 2
            else:
                continue
            kind_rank = {"fwimage": 0, "elf": 1, "raw": 2, "runfile": 3}[kind]
            ranked.append((kind_rank, int(scope), path, kind))
    ranked.sort()
    for _, _, path, kind in ranked:
        if kind == "runfile":
            target.runfiles.append(path)
        else:
            target.paths.append(path)
            target.matches.append(path)
            if target.kind == "none":
                target.kind = kind
                target.source = "discovered"
    if not target.paths and target.runfiles:
        target.notes.append(
            "driver runfile(s) present but not extracted; pass --extract-runfiles "
            "(local extraction only, nothing is downloaded) or extract with the "
            "driver runfile's own --extract-only"
        )
    if not target.paths and not target.runfiles:
        target.notes.append(f"no local image matching {version!r} under {input_dir}")
    return target


def resolve_targets(options: SweepOptions) -> dict[str, SweepTarget]:
    """Resolve every requested version plus an out-of-list baseline when given."""
    versions = list(options.versions)
    if options.baseline and options.baseline not in versions:
        versions.append(options.baseline)
    resolved: dict[str, SweepTarget] = {}
    for version in versions:
        explicit = [path for path in options.targets.get(version, [])]
        target = SweepTarget(version=version)
        target.symbols = options.symbols.get(version)
        target.ghidra_exports = options.ghidra_exports.get(version)
        if explicit:
            target.source = "explicit"
            for path in explicit:
                if os.path.isfile(path):
                    target.paths.append(os.path.abspath(path))
                    target.matches.append(path)
                else:
                    target.notes.append(f"explicit target not found: {path}")
            if target.paths:
                target.kind = _kind_for(os.path.basename(target.paths[0])) or "explicit"
        else:
            target = discover_version(options.input_dir or "", version)
            target.symbols = options.symbols.get(version)
            target.ghidra_exports = options.ghidra_exports.get(version)
        resolved[version] = target
    return resolved


def choose_baseline(options: SweepOptions, resolved: dict[str, SweepTarget]) -> str | None:
    if options.baseline:
        return options.baseline
    found = [version for version in options.versions if resolved.get(version) and resolved[version].paths]
    return min(found, key=version_sort_key) if found else None


# ---------------------------------------------------------------------------
# bounded steps
# ---------------------------------------------------------------------------


@contextlib.contextmanager
def _deadline(seconds: float, label: str):
    """Wall-clock safety net around one step (main thread only)."""
    usable = (
        seconds > 0
        and hasattr(signal, "SIGALRM")
        and threading.current_thread() is threading.main_thread()
    )
    if not usable:
        yield
        return

    def _raise(signum, frame):  # pragma: no cover - signal trampoline
        raise TimeoutError(f"{label} exceeded the {seconds:g}s step budget")

    previous = signal.signal(signal.SIGALRM, _raise)
    previous_timer = signal.setitimer(signal.ITIMER_REAL, seconds)
    try:
        yield
    finally:
        signal.setitimer(signal.ITIMER_REAL, *previous_timer)
        signal.signal(signal.SIGALRM, previous)


def _run_step(name: str, fn, *args, timeout: float = 0.0, **kwargs) -> dict:
    try:
        with _deadline(timeout, name):
            detail = fn(*args, **kwargs)
    except TimeoutError as exc:
        return {"name": name, "status": "error", "reason": str(exc), "timeout": True}
    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


# ---------------------------------------------------------------------------
# per-version steps
# ---------------------------------------------------------------------------


def step_version_diff(version: str, baseline: str, target: SweepTarget, baseline_target: SweepTarget,
                      options: SweepOptions, out_dir: str) -> dict:
    if target.path is None:
        return {"status": "skipped", "reason": "no target"}
    if baseline_target.path is None:
        return {"status": "error", "reason": f"baseline {baseline!r} has no target"}
    report = run_version_diff(
        VersionDiffOptions(
            old=baseline_target.path,
            new=target.path,
            out=out_dir,
            old_label=baseline,
            new_label=version,
            old_symbols=baseline_target.symbols,
            new_symbols=target.symbols,
            old_ghidra_exports=baseline_target.ghidra_exports,
            new_ghidra_exports=target.ghidra_exports,
            probe_instructions=options.probe_instructions,
            probe_duration=options.probe_duration,
            behavior_duration=options.scenario_duration,
            max_stops=options.max_stops,
            max_rows=options.max_rows,
            log_bytes=options.log_bytes,
            approval_token=options.approval_token,
        )
    )
    sections = {name: section["status"] for name, section in report["sections"].items()}
    behavior = report["sections"]["behavior"]
    metrics = behavior.get("metrics") or []
    if not metrics:
        metrics = [
            {"id": metric_id, "name": name, "status": "not measured", "reason": NO_REFERENCE,
             "a_observed": None, "b_observed": None}
            for metric_id, name in METRIC_DEFS
        ]
    return {
        "status": "error" if "error" in sections.values() else "ok",
        "report": "version-diff/version-diff.json",
        "markdown": "version-diff/version-diff.md",
        "sections": sections,
        "all_zero": behavior.get("all_zero"),
        "metrics": metrics,
    }


def step_portability_matrix(version: str, baseline: str, target: SweepTarget, baseline_target: SweepTarget,
                            options: SweepOptions, out_dir: str) -> dict:
    if target.path is None:
        return {"status": "skipped", "reason": "no target"}
    if baseline_target.path is None:
        return {"status": "error", "reason": f"baseline {baseline!r} has no target"}
    matrix = run_matrix(
        [TargetSpec(path=baseline_target.path, name=baseline),
         TargetSpec(path=target.path, name=version)],
        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"]}
    comparisons = [
        {"a": row["a"], "b": row["b"], "status": row["status"], "all_zero": row["all_zero"]}
        for row in matrix["comparisons"]
    ]
    return {
        "status": "error" if "error" in statuses.values() else "ok",
        "report": "portability-matrix/portability-report.json",
        "markdown": "portability-matrix/portability-report.md",
        "targets": statuses,
        "comparisons": comparisons,
    }


def step_mutation_smoke(target: SweepTarget, options: SweepOptions, out_dir: str) -> dict:
    if not qemu_available():
        return {"status": "skipped", "reason": "qemu-system-riscv64 not found"}
    kernel = options.mutation_kernel
    if not os.path.isfile(kernel):
        return {"status": "skipped", "reason": f"mutation kernel not found: {kernel}"}
    findings = run_harness(
        HarnessOptions(
            kernel=os.path.abspath(kernel),
            out=out_dir,
            iterations=options.mutation_iterations,
            seed=options.mutation_seed,
            timeout=options.mutation_timeout,
            settle=options.mutation_settle,
            allow_foreign_elf=options.allow_foreign_mutation,
        )
    )
    outcomes: dict[str, int] = {}
    for finding in findings:
        outcomes[finding["outcome"]] = outcomes.get(finding["outcome"], 0) + 1
    return {
        "status": "ok",
        "report": "mutation-smoke/findings.jsonl",
        "markdown": "mutation-smoke/summary.md",
        "iterations": len(findings),
        "outcomes": dict(sorted(outcomes.items())),
    }


def version_status(steps: list[dict]) -> str:
    statuses = [step["status"] for step in steps]
    if "error" in statuses:
        return "error"
    if all(status == "ok" for status in statuses):
        return "ok"
    return "partial"


def build_bundle(out_dir: str, tool: 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": tool, "files": files}


def completed_status(version_dir: str) -> str | None:
    """Return the recorded status of a finished version (resume support)."""
    path = os.path.join(version_dir, "version.json")
    if not os.path.isfile(path):
        return None
    try:
        with open(path, "r", encoding="utf-8") as fh:
            data = json.load(fh)
    except (OSError, json.JSONDecodeError):
        return None
    status = data.get("status")
    return status if status in RESUMABLE_STATUSES else None


# ---------------------------------------------------------------------------
# sweep execution
# ---------------------------------------------------------------------------


def run_version(version: str, baseline: str, target: SweepTarget, baseline_target: SweepTarget,
                options: SweepOptions, out_root: str) -> dict:
    version_dir = os.path.join(out_root, version)
    entry: dict = {
        "format": VERSION_FORMAT,
        "tool": "harness/port_driver_versions.py",
        "version": version,
        "baseline": baseline,
        "target": target.path,
        "target_kind": target.kind,
        "target_source": target.source,
        "target_matches": target.matches,
        "runfiles": target.runfiles,
        "notes": target.notes,
        "symbols": target.symbols,
        "steps": [],
        "metrics": [],
        "status": "no-input",
    }
    if target.path is None:
        entry["reason"] = (
            "no local firmware target found; "
            + ("; ".join(target.notes) if target.notes else "supply --input-dir or --target VERSION=PATH")
        )
        ensure_dir(version_dir)
        write_json(os.path.join(version_dir, "version.json"), entry)
        write_text(os.path.join(version_dir, "version.md"), render_version_markdown(entry))
        return entry

    steps = [
        _run_step("version_diff", step_version_diff, version, baseline, target, baseline_target,
                  options, os.path.join(version_dir, "version-diff"), timeout=options.step_timeout),
        _run_step("portability_matrix", step_portability_matrix, version, baseline, target,
                  baseline_target, options, os.path.join(version_dir, "portability-matrix"),
                  timeout=options.step_timeout),
        _run_step("mutation_smoke", step_mutation_smoke, target, options,
                  os.path.join(version_dir, "mutation-smoke"), timeout=options.step_timeout),
    ]
    entry["steps"] = steps
    entry["status"] = version_status(steps)
    for step in steps:
        if step["name"] == "version_diff":
            entry["metrics"] = step.get("metrics") or []
            entry["all_zero"] = step.get("all_zero")
    entry["target_sha256"] = sha256_file(target.path)
    entry["evidence"] = {"manifest": os.path.join(version, "bundle-manifest.json")}
    ensure_dir(version_dir)
    write_json(os.path.join(version_dir, "version.json"), entry)
    write_text(os.path.join(version_dir, "version.md"), render_version_markdown(entry))
    manifest = build_bundle(version_dir, "harness/port_driver_versions.py")
    write_json(os.path.join(version_dir, "bundle-manifest.json"), manifest)
    return entry


def run_sweep(options: SweepOptions) -> dict:
    check_rm610_gate(
        _gate_version_labels(options.versions),
        approval_token=options.approval_token,
    )
    resolved = resolve_targets(options)
    baseline = choose_baseline(options, resolved)
    labels = gate_labels(list(options.versions), resolved)
    check_rm610_gate(labels, approval_token=options.approval_token)
    if not options.out:
        raise ValueError("--out is required unless --dry-run is used")
    out_root = os.path.abspath(options.out)
    ensure_dir(out_root)
    versions: list[dict] = []
    for version in options.versions:
        target = resolved[version]
        version_dir = os.path.join(out_root, version)
        previous = None if options.force else completed_status(version_dir)
        if previous is not None:
            versions.append({
                "format": VERSION_FORMAT,
                "tool": "harness/port_driver_versions.py",
                "version": version,
                "baseline": baseline,
                "status": previous,
                "resumed": True,
                "target": target.path,
                "evidence": {"manifest": os.path.relpath(
                    os.path.join(version_dir, "bundle-manifest.json"), out_root)},
            })
            continue
        if target.path is None and options.extract_runfiles and target.runfiles:
            extract_dir = os.path.join(version_dir, "runfile-extract")
            try:
                target = extract_runfiles(version, target, extract_dir, options)
            except Exception as exc:
                entry = {
                    "format": VERSION_FORMAT,
                    "tool": "harness/port_driver_versions.py",
                    "version": version,
                    "baseline": baseline,
                    "target": None,
                    "steps": [],
                    "metrics": [],
                    "status": "error",
                    "reason": f"runfile extraction failed: {type(exc).__name__}: {exc}",
                }
                ensure_dir(version_dir)
                write_json(os.path.join(version_dir, "version.json"), entry)
                write_text(os.path.join(version_dir, "version.md"), render_version_markdown(entry))
                versions.append(entry)
                continue
        baseline_target = resolved.get(baseline) if baseline else None
        if baseline_target is None:
            entry = {
                "format": VERSION_FORMAT,
                "tool": "harness/port_driver_versions.py",
                "version": version,
                "baseline": None,
                "target": target.path,
                "target_kind": target.kind,
                "target_source": target.source,
                "steps": [],
                "metrics": [],
                "status": "error",
                "reason": "no baseline target: pass --baseline or ensure one version has an image",
            }
            ensure_dir(version_dir)
            write_json(os.path.join(version_dir, "version.json"), entry)
            write_text(os.path.join(version_dir, "version.md"), render_version_markdown(entry))
            versions.append(entry)
            continue
        versions.append(run_version(version, baseline, target, baseline_target, options, out_root))

    failed = [row["version"] for row in versions if row["status"] == "error"]
    not_run = [row["version"] for row in versions if row["status"] == "no-input"]
    ran = len(versions) - len(not_run)
    if failed:
        interpretation = (
            "fail: " + ", ".join(failed)
            + " had an errored step; inspect each version.json and the step reports"
        )
    elif ran == 0:
        interpretation = (
            "not-run: no version had a local target; nothing was executed and no evidence was produced"
        )
    else:
        interpretation = (
            "pass: every executed version completed with no errored step; "
            "skipped steps and resume decisions are listed per version"
        )
    report = {
        "format": FORMAT,
        "tool": "harness/port_driver_versions.py",
        "dry_run": False,
        "rm610_gate": {"status": "passed", "labels": sorted(set(labels))},
        "baseline": {"version": baseline,
                     "path": resolved[baseline].path if baseline and baseline in resolved else None},
        "input_dir": options.input_dir,
        "versions": versions,
        "ran": ran,
        "failed": failed,
        "not_run": not_run,
        "interpretation": interpretation,
        "summary": "sweep-summary.json",
        "markdown": "sweep-summary.md",
        "manifest": "bundle-manifest.json",
    }
    write_json(os.path.join(out_root, "sweep-summary.json"), report)
    write_text(os.path.join(out_root, "sweep-summary.md"), render_sweep_markdown(report))
    manifest = build_bundle(out_root, "harness/port_driver_versions.py")
    write_json(os.path.join(out_root, "bundle-manifest.json"), manifest)
    return report


def extract_runfiles(version: str, target: SweepTarget, extract_dir: str, options: SweepOptions) -> SweepTarget:
    """Extract local driver runfile(s) with their own --extract-only (no download)."""
    ensure_dir(extract_dir)
    for runfile in target.runfiles:
        argv = ["sh", os.path.abspath(runfile), "--extract-only", "--target", extract_dir]
        subprocess.run(argv, check=True, timeout=options.step_timeout,
                       stdout=subprocess.DEVNULL, stderr=subprocess.PIPE)
    discovered = discover_version(extract_dir, version)
    discovered.symbols = target.symbols
    discovered.ghidra_exports = target.ghidra_exports
    discovered.notes = target.notes + [f"extracted from {len(target.runfiles)} runfile(s)"]
    discovered.runfiles = target.runfiles
    return discovered


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


def _expected_artifacts(out: str, version: str) -> list[str]:
    base = os.path.join(out, version)
    return [
        os.path.join(base, "version.json"),
        os.path.join(base, "version.md"),
        os.path.join(base, "bundle-manifest.json"),
        os.path.join(base, "version-diff", "version-diff.json"),
        os.path.join(base, "version-diff", "version-diff.md"),
        os.path.join(base, "portability-matrix", "portability-report.json"),
        os.path.join(base, "portability-matrix", "portability-report.md"),
        os.path.join(base, "mutation-smoke", "findings.jsonl"),
        os.path.join(base, "mutation-smoke", "summary.md"),
    ]


def plan_sweep(options: SweepOptions) -> dict:
    check_rm610_gate(
        _gate_version_labels(options.versions),
        approval_token=options.approval_token,
    )
    resolved = resolve_targets(options)
    baseline = choose_baseline(options, resolved)
    labels = gate_labels(list(options.versions), resolved)
    check_rm610_gate(labels, approval_token=options.approval_token)
    out = os.path.abspath(options.out) if options.out else "<out>"
    versions = []
    found = missing = 0
    for version in options.versions:
        target = resolved[version]
        entry = {
            "version": version,
            "status": "found" if target.path else ("runfile" if target.runfiles else "no-input"),
            "target": target.path,
            "kind": target.kind,
            "source": target.source,
            "matches": target.matches,
            "runfiles": target.runfiles,
            "notes": target.notes,
            "baseline": version == baseline,
            "symbols": target.symbols,
            "steps": [
                {"name": "version_diff", "out": os.path.join(out, version, "version-diff"),
                 "old": resolved[baseline].path if baseline and baseline in resolved else None,
                 "new": target.path},
                {"name": "portability_matrix", "out": os.path.join(out, version, "portability-matrix"),
                 "targets": [baseline, version] if baseline else [version]},
                {"name": "mutation_smoke", "out": os.path.join(out, version, "mutation-smoke"),
                 "kernel": options.mutation_kernel},
            ],
            "expected_artifacts": _expected_artifacts(out, version),
        }
        if options.extract_runfiles and target.runfiles and not target.path:
            entry["extract"] = [
                {"argv": ["sh", os.path.abspath(runfile), "--extract-only", "--target",
                          os.path.join(out, version, "runfile-extract")]}
                for runfile in target.runfiles
            ]
        if target.path:
            found += 1
        else:
            missing += 1
        versions.append(entry)
    return {
        "format": PLAN_FORMAT,
        "tool": "harness/port_driver_versions.py",
        "dry_run": True,
        "rm610_gate": {"status": "passed", "labels": sorted(set(labels))},
        "input_dir": options.input_dir,
        "out": out,
        "baseline": baseline,
        "versions": versions,
        "found": found,
        "missing": missing,
        "expected_artifacts": [
            os.path.join(out, "sweep-summary.json"),
            os.path.join(out, "sweep-summary.md"),
            os.path.join(out, "bundle-manifest.json"),
        ],
        "interpretation": (
            f"dry-run: {found} input(s) found, {missing} missing; "
            "pass --run to execute, --force to re-run completed versions"
        ),
    }


# ---------------------------------------------------------------------------
# rendering
# ---------------------------------------------------------------------------


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 render_version_markdown(entry: dict) -> str:
    lines = [
        f"# green-eye port version {entry['version']}",
        "",
        "> TOOL IN DEVELOPMENT - NOT READY YET! Generated by `harness/port_driver_versions.py`.",
        "",
        f"* status: **{entry['status']}**",
        f"* baseline: `{entry.get('baseline')}`",
        f"* target: `{entry.get('target')}`",
    ]
    if entry.get("target_sha256"):
        lines.append(f"* target sha256: `{entry['target_sha256']}`")
    if entry.get("reason"):
        lines.append(f"* reason: {entry['reason']}")
    lines += ["", "## Steps", ""]
    lines += _table(
        ["step", "status", "report", "reason"],
        [[step["name"], step["status"], step.get("report") or "-", step.get("reason") or "-"]
         for step in entry.get("steps") or []] or [["-", "-", "-", "not run"]],
    )
    lines += ["", "## M1-M6 (docs/06)", "", f"> {NO_REFERENCE}", ""]
    lines += _table(
        ["#", "metric", "status", "observed"],
        [[row["id"], row["name"], row["status"],
          json.dumps({"a": row.get("a_observed"), "b": row.get("b_observed")}, sort_keys=True)]
         for row in entry.get("metrics") or []],
    )
    lines += [""]
    return "\n".join(lines)


def render_sweep_markdown(report: dict) -> str:
    lines = [
        "# green-eye driver-version port sweep",
        "",
        "> TOOL IN DEVELOPMENT - NOT READY YET! Generated by `harness/port_driver_versions.py`.",
        "",
        f"* format: `{report['format']}`",
        f"* baseline: `{report['baseline']['version']}` (`{report['baseline']['path']}`)",
        f"* input dir: `{report.get('input_dir')}`",
        f"* ran: {report['ran']}, failed: {len(report['failed'])}, no input: {len(report['not_run'])}",
        "",
        f"> **{report['interpretation']}**",
        "",
        "## Versions",
        "",
    ]
    lines += _table(
        ["version", "status", "target", "version-diff", "matrix", "mutation", "evidence"],
        [
            [
                row["version"],
                row["status"] + (" (resumed)" if row.get("resumed") else ""),
                f"`{(row.get('target') or '-')}`",
                _step_status(row, "version_diff"),
                _step_status(row, "portability_matrix"),
                _step_status(row, "mutation_smoke"),
                row.get("evidence", {}).get("manifest", "-"),
            ]
            for row in report["versions"]
        ],
    )
    lines += ["", "## M1-M6 across versions", "", f"> {NO_REFERENCE}", ""]
    metric_ids = [metric_id for metric_id, _ in METRIC_DEFS]
    lines += _table(
        ["version"] + metric_ids,
        [
            [row["version"]] + [
                next((m["status"] for m in row.get("metrics") or [] if m["id"] == metric_id), "-")
                for metric_id in metric_ids
            ]
            for row in report["versions"]
        ],
    )
    lines += ["", "## Failed / not run", ""]
    if not report["failed"] and not report["not_run"]:
        lines += ["_none_", ""]
    for version in report["failed"]:
        lines += [f"* failed `{version}`: see `{version}/version.json` and the step reports"]
    for version in report["not_run"]:
        row = next(item for item in report["versions"] if item["version"] == version)
        lines += [f"* no input `{version}`: {row.get('reason', '-')}"]
    lines += [
        "",
        "## Evidence",
        "",
        f"* sweep summary: `{report['summary']}` / `{report['markdown']}`",
        f"* bundle manifest: `{report['manifest']}` (sha256 of every artifact)",
        "* per-version tree: `version.json`, `version.md`, `version-diff/`, "
        "`portability-matrix/`, `mutation-smoke/`, `bundle-manifest.json`",
        "",
    ]
    return "\n".join(lines)


def _step_status(row: dict, name: str) -> str:
    for step in row.get("steps") or []:
        if step["name"] == name:
            return step["status"]
    return "-"


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


def _parse_mapping(value: str, flag: str) -> tuple[str, str]:
    if "=" not in value:
        raise argparse.ArgumentTypeError(f"{flag} expects VERSION=PATH, got {value!r}")
    version, path = value.split("=", 1)
    version, path = version.strip(), path.strip()
    if not version or not path:
        raise argparse.ArgumentTypeError(f"{flag} expects VERSION=PATH, got {value!r}")
    return version, path


_HELP_EPILOG = """\
invocation:
  dry-run first (default; reads input dirs only, never touches a target):
    python3 harness/port_driver_versions.py --input-dir ../../firmware/57X-GSP
  execute, resume-safe:
    python3 harness/port_driver_versions.py --run --input-dir ../../firmware/57X-GSP \\
        --versions 540,550,560,570 --out dumps/sweep-570
  explicit local targets (no discovery):
    python3 harness/port_driver_versions.py --run --versions 570.144 \\
        --target 570.144=../../firmware/57X-GSP/570.144/gsp-570.144.fwimage.bin \\
        --baseline 570.144 --out dumps/sweep-570

cron (weekly, resume-safe; nothing is downloaded):
  17 3 * * 1 cd <tool-root> && python3 harness/port_driver_versions.py --run \\
      --input-dir ../../firmware/57X-GSP --versions 540,550,560,570 \\
      --out dumps/sweep-570 >> dumps/sweep-570/cron.log 2>&1

agent (MCP, docs/20): read the dry-run plan, then call --run; read
sweep-summary.json and each bundle-manifest.json instead of recomputing.
RM 610 (docs/23): versions >= 610 require --approval-token RM610-APPROVED or
GREENEYE_RM610_APPROVED=1; the gate fires before any file is opened.
"""


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        prog="port_driver_versions",
        description=__doc__.splitlines()[0],
        epilog=_HELP_EPILOG,
        formatter_class=argparse.RawDescriptionHelpFormatter,
    )
    parser.add_argument("--versions", help="comma-separated version list (default: known cadence: "
                                           + ",".join(DEFAULT_VERSIONS) + ")")
    parser.add_argument("--config", help="JSON sweep config (versions, baseline, input_dir, targets, symbols)")
    parser.add_argument("--baseline", help="baseline version for version_diff (default: earliest found)")
    parser.add_argument("--input-dir", help="directory to discover local firmware/driver images in (recursive)")
    parser.add_argument("--target", action="append", default=[], metavar="VERSION=PATH",
                        help="explicit local target for a version (repeatable; skips discovery)")
    parser.add_argument("--symbols", action="append", default=[], metavar="VERSION=PATH",
                        help="green-eye symbol DB JSON for a version (repeatable)")
    parser.add_argument("--ghidra-exports", action="append", default=[], metavar="VERSION=PATH",
                        help="Ghidra exports TSV for a version (repeatable)")
    parser.add_argument("--out", help="sweep output directory (required for --run)")
    parser.add_argument("--run", dest="run", action="store_true",
                        help="execute the sweep (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("--force", action="store_true",
                        help="re-run versions whose version.json already reports ok/partial")
    parser.add_argument("--extract-runfiles", action="store_true",
                        help="locally extract driver runfiles with --extract-only (never downloads)")
    parser.add_argument("--approval-token", help=f"explicit RM610 token ({RM610_TOKEN})")
    parser.add_argument("--probe-instructions", type=int, default=2000,
                        help="probe step budget per target (0 = wall-clock --probe-duration)")
    parser.add_argument("--probe-duration", type=float, default=0.5)
    parser.add_argument("--scenario-duration", type=float, default=0.4,
                        help="budget per version_diff/matrix scenario phase")
    parser.add_argument("--max-stops", type=int, default=5000)
    parser.add_argument("--max-rows", type=int, default=40)
    parser.add_argument("--mutation-kernel", default=DEMO_ELF,
                        help="kernel for the mutation smoke (default: bundled demo)")
    parser.add_argument("--mutation-iterations", type=int, default=8)
    parser.add_argument("--mutation-seed", type=int, default=7)
    parser.add_argument("--mutation-timeout", type=float, default=0.25)
    parser.add_argument("--mutation-settle", type=float, default=0.1)
    parser.add_argument("--allow-foreign-mutation", action="store_true",
                        help="permit a non-demo mutation kernel (QEMU only)")
    parser.add_argument("--step-timeout", type=float, default=600.0,
                        help="wall-clock safety net per step, seconds (0 disables)")
    parser.add_argument("--log-bytes", type=int, default=MAX_LOG_BYTES)
    return parser


def options_from_args(args: argparse.Namespace) -> SweepOptions:
    options = SweepOptions()
    if args.config:
        options.config = args.config
        options_from_config(options, load_config(args.config))
    if args.versions:
        options.versions = parse_versions(args.versions)
    if not options.versions:
        options.versions = list(DEFAULT_VERSIONS)
    options.baseline = args.baseline
    options.input_dir = args.input_dir
    options.out = args.out
    for value in args.target:
        version, path = _parse_mapping(value, "--target")
        options.targets.setdefault(version, []).append(path)
    for value in args.symbols:
        version, path = _parse_mapping(value, "--symbols")
        options.symbols[version] = path
    for value in args.ghidra_exports:
        version, path = _parse_mapping(value, "--ghidra-exports")
        options.ghidra_exports[version] = path
    options.run = args.run
    options.force = args.force
    options.extract_runfiles = args.extract_runfiles
    options.approval_token = args.approval_token
    options.probe_instructions = args.probe_instructions
    options.probe_duration = args.probe_duration
    options.scenario_duration = args.scenario_duration
    options.max_stops = args.max_stops
    options.max_rows = args.max_rows
    options.mutation_kernel = args.mutation_kernel
    options.mutation_iterations = args.mutation_iterations
    options.mutation_seed = args.mutation_seed
    options.mutation_timeout = args.mutation_timeout
    options.mutation_settle = args.mutation_settle
    options.allow_foreign_mutation = args.allow_foreign_mutation
    options.step_timeout = args.step_timeout
    options.log_bytes = args.log_bytes
    return options


def main(argv: list[str] | None = None) -> int:
    parser = build_parser()
    args = parser.parse_args(argv)
    try:
        options = options_from_args(args)
    except (ValueError, argparse.ArgumentTypeError) as exc:
        parser.error(str(exc))
    try:
        check_rm610_gate(
            _gate_version_labels(options.versions),
            approval_token=options.approval_token,
        )
        check_rm610_gate(
            gate_labels(options.versions, resolve_targets(options)),
            approval_token=options.approval_token,
        )
    except Rm610GateError as exc:
        print(f"[sweep] RM610 gate refused: {exc}", file=sys.stderr)
        return 3
    if options.dry_run:
        try:
            plan = plan_sweep(options)
        except (ValueError, FileNotFoundError) as exc:
            print(f"[sweep] error: {exc}", file=sys.stderr)
            return 2
        print(json.dumps(plan, indent=2, sort_keys=True))
        print(f"[sweep] dry-run: {plan['found']} input(s) found, {plan['missing']} missing; no target touched")
        return 0
    if not options.out:
        parser.error("--out is required unless --dry-run is used")
    try:
        report = run_sweep(options)
    except (ValueError, FileNotFoundError) as exc:
        print(f"[sweep] error: {exc}", file=sys.stderr)
        return 2
    except Rm610GateError as exc:  # pragma: no cover - main checks first
        print(f"[sweep] RM610 gate refused: {exc}", file=sys.stderr)
        return 3
    for row in report["versions"]:
        reason = f" reason={row['reason']}" if row.get("reason") else ""
        resumed = " resumed" if row.get("resumed") else ""
        print(f"[sweep] {row['version']}: status={row['status']}{resumed}{reason}")
    print(f"[sweep] {report['interpretation']}")
    print(f"[sweep] wrote {os.path.join(os.path.abspath(options.out), 'sweep-summary.json')}")
    print(f"[sweep] wrote {os.path.join(os.path.abspath(options.out), 'bundle-manifest.json')}")
    return 1 if report["failed"] or report["ran"] == 0 else 0


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