#!/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"(? 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 "" 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 && 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())