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

#!/usr/bin/env python3
"""Release triage orchestration for green-eye (planned docs/21).

Given an old and a new firmware target it runs, in order:

1. ``harness/version_diff.py``      -- image/symbol/corpus/probe/behavior diff;
2. ``harness/portability_matrix.py`` -- probe + scenario + cross-target deltas;
3. ``harness/mutation_harness.py``   -- a small deterministic mutation smoke.

Every step writes under ``--out`` and the whole directory is hashed into
``bundle-manifest.json`` (the evidence bundle). A failing step is recorded and
the remaining steps still run; the process exits non-zero only when a step
reports an error.

``--dry-run`` prints the plan and exits 0 without opening targets, and the
RM 610 gate (docs/23) is enforced before any target file is read. Nothing is
ever downloaded.

Run from the tool root:

    python3 harness/ci_release.py --old old.elf --new new.elf --out dumps/release
    python3 harness/ci_release.py --old old.elf --new new.elf --dry-run
"""
from __future__ import annotations

import argparse
import json
import os
import sys
from dataclasses import dataclass
from pathlib import Path

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

FORMAT = "greeneye-ci-release/1"
PLAN_FORMAT = "greeneye-ci-release-plan/1"
BUNDLE_FORMAT = "greeneye-release-bundle/1"


@dataclass
class ReleaseOptions:
    old: str
    new: str
    out: str | None = None
    old_label: str | None = None
    new_label: str | None = None
    release: str | None = None
    approval_token: str | None = None
    old_symbols: str | None = None
    new_symbols: str | None = None
    matrix_symbols: str | None = None
    old_ghidra_names: str | None = None
    new_ghidra_names: str | None = None
    old_ghidra_index: str | None = None
    new_ghidra_index: str | None = None
    old_ghidra_exports: str | None = None
    new_ghidra_exports: str | None = None
    old_ghidra_bias: int = 0
    new_ghidra_bias: int = 0
    corpus_diff: tuple[str, str] | None = None
    corpus_report: str | None = None
    stimulus: str | None = None
    probe_instructions: int = 2000
    behavior_duration: float = 0.4
    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
    dry_run: bool = False

    def labels(self) -> list[str]:
        return [
            value
            for value in (
                self.release,
                self.old_label,
                self.new_label,
                os.path.basename(self.old),
                os.path.basename(self.new),
                os.path.basename(self.mutation_kernel) if self.mutation_kernel else None,
            )
            if value
        ]


def _version_diff_options(options: ReleaseOptions, out_dir: str) -> VersionDiffOptions:
    return VersionDiffOptions(
        old=options.old,
        new=options.new,
        out=out_dir,
        old_label=options.old_label,
        new_label=options.new_label,
        release=options.release,
        old_symbols=options.old_symbols,
        new_symbols=options.new_symbols,
        old_ghidra_names=options.old_ghidra_names,
        new_ghidra_names=options.new_ghidra_names,
        old_ghidra_index=options.old_ghidra_index,
        new_ghidra_index=options.new_ghidra_index,
        old_ghidra_exports=options.old_ghidra_exports,
        new_ghidra_exports=options.new_ghidra_exports,
        old_ghidra_bias=options.old_ghidra_bias,
        new_ghidra_bias=options.new_ghidra_bias,
        corpus_diff=options.corpus_diff,
        corpus_report=options.corpus_report,
        stimulus=options.stimulus,
        probe_instructions=options.probe_instructions,
        behavior_duration=options.behavior_duration,
        max_rows=options.max_rows,
        approval_token=options.approval_token,
    )


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


def build_plan(options: ReleaseOptions) -> dict:
    out = os.path.abspath(options.out) if options.out else "<out>"
    vd_out = os.path.join(out, "version-diff")
    matrix_out = os.path.join(out, "portability-matrix")
    mutation_out = os.path.join(out, "mutation-smoke")
    vd_argv = [sys.executable, "harness/version_diff.py", "--old", options.old, "--new", options.new,
               "--out", vd_out, "--probe-instructions", str(options.probe_instructions),
               "--behavior-duration", str(options.behavior_duration), "--max-rows", str(options.max_rows)]
    if options.old_label:
        vd_argv += ["--old-label", options.old_label]
    if options.new_label:
        vd_argv += ["--new-label", options.new_label]
    if options.release:
        vd_argv += ["--release", options.release]
    if options.old_symbols:
        vd_argv += ["--old-symbols", options.old_symbols]
    if options.new_symbols:
        vd_argv += ["--new-symbols", options.new_symbols]
    if options.stimulus:
        vd_argv += ["--stimulus", options.stimulus]
    matrix_argv = [sys.executable, "harness/portability_matrix.py", "--target", options.old,
                   "--target", options.new, "--out", matrix_out,
                   "--probe-instructions", str(options.probe_instructions),
                   "--scenario-duration", str(options.behavior_duration),
                   "--max-rows", str(options.max_rows)]
    mutation_argv = [sys.executable, "harness/mutation_harness.py", "--kernel", options.mutation_kernel,
                     "--out", mutation_out, "--iterations", str(options.mutation_iterations),
                     "--seed", str(options.mutation_seed), "--timeout", str(options.mutation_timeout),
                     "--settle", str(options.mutation_settle)]
    if options.allow_foreign_mutation:
        mutation_argv.append("--allow-foreign-elf")
    return {
        "format": PLAN_FORMAT,
        "tool": "harness/ci_release.py",
        "dry_run": True,
        "status": "planned",
        "rm610_gate": {"status": "passed", "labels": sorted(set(options.labels()))},
        "old": {"label": options.old_label or Path(options.old).stem, "path": options.old},
        "new": {"label": options.new_label or Path(options.new).stem, "path": options.new},
        "out": out,
        "steps": [
            {"name": "version_diff", "out": vd_out, "argv": vd_argv},
            {"name": "portability_matrix", "out": matrix_out, "argv": matrix_argv},
            {"name": "mutation_smoke", "out": mutation_out, "argv": mutation_argv},
        ],
        "manifest": os.path.join(out, "bundle-manifest.json"),
        "release_report": os.path.join(out, "release-report.json"),
    }


# ---------------------------------------------------------------------------
# steps
# ---------------------------------------------------------------------------


def step_version_diff(options: ReleaseOptions, out_dir: str) -> dict:
    report = run_version_diff(_version_diff_options(options, out_dir))
    sections = {name: section["status"] for name, section in report["sections"].items()}
    status = "error" if "error" in sections.values() else "ok"
    return {
        "status": status,
        "out": os.path.basename(out_dir),
        "report": "version-diff.json",
        "sections": sections,
        "summary": report["summary"],
    }


def step_portability_matrix(options: ReleaseOptions, out_dir: str) -> dict:
    matrix = run_matrix(
        [TargetSpec(path=os.path.abspath(options.old), name=Path(options.old).stem),
         TargetSpec(path=os.path.abspath(options.new), name=Path(options.new).stem)],
        MatrixOptions(
            out=out_dir,
            probe_instructions=options.probe_instructions,
            scenario_duration=options.behavior_duration,
            max_rows=options.max_rows,
            symbols_json=options.matrix_symbols,
        ),
    )
    ensure_dir(out_dir)
    write_json(os.path.join(out_dir, "portability-report.json"), matrix)
    write_text(
        os.path.join(out_dir, "portability-report.md"),
        render_markdown(matrix, MatrixOptions(out=out_dir, max_rows=options.max_rows)),
    )
    statuses = [target["status"] for target in matrix["targets"]]
    status = "error" if "error" in statuses else "ok"
    return {
        "status": status,
        "report": "portability-report.json",
        "targets": {target["name"]: target["status"] for target in matrix["targets"]},
        "comparisons": [
            {"a": comp["a"], "b": comp["b"], "status": comp["status"], "all_zero": comp["all_zero"]}
            for comp in matrix["comparisons"]
        ],
    }


def step_mutation_smoke(options: ReleaseOptions, out_dir: str) -> dict:
    if not qemu_available():
        return {"status": "skipped", "reason": "qemu-system-riscv64 not found"}
    if not os.path.isfile(options.mutation_kernel):
        return {"status": "skipped", "reason": f"mutation kernel not found: {options.mutation_kernel}"}
    harness = HarnessOptions(
        kernel=os.path.abspath(options.mutation_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,
    )
    findings = run_harness(harness)
    outcomes: dict[str, int] = {}
    for finding in findings:
        outcomes[finding["outcome"]] = outcomes.get(finding["outcome"], 0) + 1
    return {
        "status": "ok",
        "report": "findings.jsonl",
        "iterations": len(findings),
        "outcomes": dict(sorted(outcomes.items())),
    }


def _run_step(name: str, fn, *args) -> dict:
    try:
        detail = fn(*args)
    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


# ---------------------------------------------------------------------------
# bundle
# ---------------------------------------------------------------------------


def build_manifest(out_dir: 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": "harness/ci_release.py", "files": files}


def run_release(options: ReleaseOptions) -> dict:
    check_rm610_gate(options.labels(), approval_token=options.approval_token)
    if not options.out:
        raise ValueError("--out is required unless --dry-run is used")
    out = os.path.abspath(options.out)
    ensure_dir(out)
    steps = [
        _run_step("version_diff", step_version_diff, options, os.path.join(out, "version-diff")),
        _run_step("portability_matrix", step_portability_matrix, options,
                  os.path.join(out, "portability-matrix")),
        _run_step("mutation_smoke", step_mutation_smoke, options, os.path.join(out, "mutation-smoke")),
    ]
    report = {
        "format": FORMAT,
        "tool": "harness/ci_release.py",
        "dry_run": False,
        "rm610_gate": {"status": "passed", "labels": sorted(set(options.labels()))},
        "old": {"label": options.old_label or Path(options.old).stem, "path": os.path.basename(options.old)},
        "new": {"label": options.new_label or Path(options.new).stem, "path": os.path.basename(options.new)},
        "steps": steps,
        "errors": [step["name"] for step in steps if step["status"] == "error"],
        "manifest": "bundle-manifest.json",
    }
    write_json(os.path.join(out, "release-report.json"), report)
    manifest = build_manifest(out)
    write_json(os.path.join(out, "bundle-manifest.json"), manifest)
    return report


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


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(prog="ci_release", description=__doc__.splitlines()[0])
    parser.add_argument("--old", required=True, help="old target (RISC-V ELF or raw .fwimage)")
    parser.add_argument("--new", required=True, help="new target (RISC-V ELF or raw .fwimage)")
    parser.add_argument("--out", help="evidence bundle directory (required unless --dry-run)")
    parser.add_argument("--dry-run", action="store_true", help="print the plan and exit without touching targets")
    parser.add_argument("--old-label", help="report label for the old target")
    parser.add_argument("--new-label", help="report label for the new target")
    parser.add_argument("--release", help="release label; an rm610 release is gated by docs/23")
    parser.add_argument("--approval-token", help=f"explicit RM610 token ({RM610_TOKEN})")
    parser.add_argument("--old-symbols", help="green-eye symbol DB JSON for the old target")
    parser.add_argument("--new-symbols", help="green-eye symbol DB JSON for the new target")
    parser.add_argument("--matrix-symbols", help="one symbol DB JSON applied to both matrix targets")
    parser.add_argument("--old-ghidra-names", help="Ghidra names.tsv for the old target")
    parser.add_argument("--new-ghidra-names", help="Ghidra names.tsv for the new target")
    parser.add_argument("--old-ghidra-index", help="Ghidra decompiled_index.tsv for the old target")
    parser.add_argument("--new-ghidra-index", help="Ghidra decompiled_index.tsv for the new target")
    parser.add_argument("--old-ghidra-exports", help="Ghidra exports.tsv / 3-column export for the old target")
    parser.add_argument("--new-ghidra-exports", help="Ghidra exports.tsv / 3-column export for the new target")
    parser.add_argument("--old-ghidra-bias", type=lambda value: int(value, 0), default=0)
    parser.add_argument("--new-ghidra-bias", type=lambda value: int(value, 0), default=0)
    parser.add_argument("--corpus-diff", nargs=2, metavar=("OLD_TSV", "NEW_TSV"),
                        help="diff two decompiled_index.tsv corpus exports")
    parser.add_argument("--corpus-report", help="attach a previously produced corpus diff report")
    parser.add_argument("--stimulus", help="green-eye script replacing the builtin demo scenario")
    parser.add_argument("--probe-instructions", type=int, default=2000)
    parser.add_argument("--behavior-duration", type=float, default=0.4)
    parser.add_argument("--max-rows", type=int, default=40)
    parser.add_argument("--mutation-kernel", default=DEMO_ELF, help="kernel for the mutation smoke (default: 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)")
    return parser


def options_from_args(args: argparse.Namespace) -> ReleaseOptions:
    return ReleaseOptions(
        old=args.old,
        new=args.new,
        out=args.out,
        old_label=args.old_label,
        new_label=args.new_label,
        release=args.release,
        approval_token=args.approval_token,
        old_symbols=args.old_symbols,
        new_symbols=args.new_symbols,
        matrix_symbols=args.matrix_symbols,
        old_ghidra_names=args.old_ghidra_names,
        new_ghidra_names=args.new_ghidra_names,
        old_ghidra_index=args.old_ghidra_index,
        new_ghidra_index=args.new_ghidra_index,
        old_ghidra_exports=args.old_ghidra_exports,
        new_ghidra_exports=args.new_ghidra_exports,
        old_ghidra_bias=args.old_ghidra_bias,
        new_ghidra_bias=args.new_ghidra_bias,
        corpus_diff=tuple(args.corpus_diff) if args.corpus_diff else None,  # type: ignore[arg-type]
        corpus_report=args.corpus_report,
        stimulus=args.stimulus,
        probe_instructions=args.probe_instructions,
        behavior_duration=args.behavior_duration,
        max_rows=args.max_rows,
        mutation_kernel=args.mutation_kernel,
        mutation_iterations=args.mutation_iterations,
        mutation_seed=args.mutation_seed,
        mutation_timeout=args.mutation_timeout,
        mutation_settle=args.mutation_settle,
        allow_foreign_mutation=args.allow_foreign_mutation,
        dry_run=args.dry_run,
    )


def main(argv: list[str] | None = None) -> int:
    parser = build_parser()
    args = parser.parse_args(argv)
    options = options_from_args(args)
    try:
        check_rm610_gate(options.labels(), approval_token=options.approval_token)
    except Rm610GateError as exc:
        print(f"[release] RM610 gate refused: {exc}", file=sys.stderr)
        return 3
    if options.dry_run:
        plan = build_plan(options)
        print(json.dumps(plan, indent=2, sort_keys=True))
        print(f"[release] dry-run: {len(plan['steps'])} steps planned, no targets touched")
        return 0
    if not options.out:
        parser.error("--out is required unless --dry-run is used")
    try:
        report = run_release(options)
    except ValueError as exc:
        parser.error(str(exc))
    except Exception as exc:
        print(f"[release] error: {type(exc).__name__}: {exc}", file=sys.stderr)
        return 2
    for step in report["steps"]:
        reason = f" reason={step['reason']}" if step.get("reason") else ""
        print(f"[release] {step['name']}: status={step['status']}{reason}")
    print(f"[release] wrote {os.path.join(os.path.abspath(options.out), 'release-report.json')}")
    print(f"[release] wrote {os.path.join(os.path.abspath(options.out), 'bundle-manifest.json')}")
    return 1 if report["errors"] else 0


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