diff options
Diffstat (limited to 'debug/risc-v-gsp-rm-vm/harness/port_driver_versions.py')
| -rw-r--r-- | debug/risc-v-gsp-rm-vm/harness/port_driver_versions.py | 1155 |
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()) |