debug: green-eye porting automation - new-device and driver-version port scripts, simulator packaging, porting docs; control-discovery fix

1 files changed, 1155 insertions, 0 deletions

diff --git a/debug/risc-v-gsp-rm-vm/harness/port_driver_versions.py b/debug/risc-v-gsp-rm-vm/harness/port_driver_versions.py
new file mode 100644
index 0000000..e751d42
--- /dev/null
+++ b/debug/risc-v-gsp-rm-vm/harness/port_driver_versions.py
@@ -0,0 +1,1155 @@
+#!/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())