fold/bench/run_quality_gate.py

#!/usr/bin/env python3
"""Run fold's optimization + correctness smoke gates.

This orchestrates the existing focused harnesses and writes one manifest with
paths to all raw artifacts. It is intentionally conservative: no downloads, no
large Juliet run unless requested, and curl is a small source/binary smoke.

Examples:
  python3 bench/run_quality_gate.py --skip-curl --skip-juliet
  python3 bench/run_quality_gate.py --juliet-cwes 476 369 --curl-source ../curl-source/curl
"""

from __future__ import annotations

import argparse
import json
import subprocess
import sys
import time
from pathlib import Path
from typing import Any

ROOT = Path(__file__).resolve().parents[1]
DEFAULT_FOLD = ROOT / "build" / "release" / "src" / "cli" / "fold"
sys.path.insert(0, str(ROOT / "scripts" / "benchmarks"))

from fold_eval.scoring import (
    negative_cases_from_universe,
    score_findings_by_rule,
    score_labeled_cases,
)


def now_stamp() -> str:
    return time.strftime("%Y-%m-%dT%H-%M-%SZ", time.gmtime())


def run_step(name: str, cmd: list[str], cwd: Path, timeout: int | None = None) -> dict[str, Any]:
    started = time.monotonic()
    proc = subprocess.run(cmd, cwd=cwd, text=True, capture_output=True, timeout=timeout)
    elapsed = time.monotonic() - started
    return {
        "name": name,
        "command": cmd,
        "exit_code": proc.returncode,
        "elapsed_s": elapsed,
        "stdout_tail": proc.stdout[-4000:],
        "stderr_tail": proc.stderr[-4000:],
    }


def load_json(path: Path) -> Any | None:
    try:
        return json.loads(path.read_text())
    except Exception:
        return None


def latest_child(path: Path, suffix: str) -> Path | None:
    if not path.exists():
        return None
    matches = sorted(p for p in path.iterdir() if p.is_dir() and p.name.endswith(suffix))
    return matches[-1] if matches else None


def summarize_security_bugs(run_dir: Path) -> dict[str, Any]:
    fold_json = load_json(run_dir / "fold.json") or {}
    gold = load_json(ROOT / "bench" / "gold" / "security-bugs.json") or {}
    positive = gold.get("expected", [])
    negative = gold.get("negative")
    if negative is None:
        negative = negative_cases_from_universe(positive, gold.get("universe", {}))
    score = score_labeled_cases(
        fold_json.get("findings", []), positive, negative
    )
    score["per_rule"] = score_findings_by_rule(
        fold_json.get("findings", []), positive, negative
    )
    return score


def main() -> int:
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--fold", default=str(DEFAULT_FOLD))
    ap.add_argument("--out-dir", default=None)
    ap.add_argument("--micro-size", type=int, default=24)
    ap.add_argument("--micro-repeats", type=int, default=2)
    ap.add_argument("--skip-micro", action="store_true")
    ap.add_argument("--skip-source-structure", action="store_true")
    ap.add_argument("--skip-structural-quality", action="store_true")
    ap.add_argument("--skip-interproc", action="store_true")
    ap.add_argument("--skip-security-bugs", action="store_true")
    ap.add_argument("--skip-curl", action="store_true")
    ap.add_argument("--curl-source", default=str(ROOT.parent / "curl-source" / "curl"))
    ap.add_argument("--curl-binary", default="/usr/bin/curl")
    ap.add_argument("--skip-juliet", action="store_true")
    ap.add_argument("--juliet-cwes", nargs="+", default=["476", "369"])
    ap.add_argument("--juliet-timeout", type=int, default=300)
    ap.add_argument(
        "--project-manifest",
        action="append",
        default=[],
        help="Run a manifest-driven project evaluation (repeatable; opt-in).",
    )
    args = ap.parse_args()

    out_dir = Path(args.out_dir) if args.out_dir else ROOT / "bench" / "results" / f"{now_stamp()}_quality-gate"
    out_dir.mkdir(parents=True, exist_ok=True)
    fold = Path(args.fold)

    manifest: dict[str, Any] = {"fold": str(fold), "out_dir": str(out_dir), "steps": []}

    def step(name: str, cmd: list[str], timeout: int | None = None) -> dict[str, Any]:
        result = run_step(name, cmd, ROOT, timeout)
        (out_dir / f"{name}.stdout.log").write_text(result.pop("stdout_tail"))
        (out_dir / f"{name}.stderr.log").write_text(result.pop("stderr_tail"))
        manifest["steps"].append(result)
        if result["exit_code"] != 0:
            print(f"{name} failed rc={result['exit_code']}", file=sys.stderr)
        return result

    if not args.skip_source_structure:
        step(
            "source_structure",
            [sys.executable, "scripts/check_source_structure.py", "--json"],
        )

    if args.project_manifest:
        project_evaluations: list[dict[str, str]] = []
        for raw_manifest in args.project_manifest:
            project_manifest = Path(raw_manifest)
            project_dir = out_dir / f"project-{project_manifest.stem}"
            step(
                f"project_{project_manifest.stem}",
                [
                    sys.executable,
                    "scripts/benchmarks/run_project_eval.py",
                    "--manifest",
                    str(project_manifest),
                    "--fold",
                    str(fold),
                    "--out-dir",
                    str(project_dir),
                ],
            )
            project_evaluations.append(
                {
                    "manifest": str(project_manifest),
                    "summary": str(project_dir / "summary.json"),
                }
            )
        manifest["project_evaluations"] = project_evaluations

    if not args.skip_micro:
        micro_dir = out_dir / "micro"
        step(
            "micro",
            [sys.executable, "bench/run_micro_opt.py", "--case", "all", "--size", str(args.micro_size),
             "--repeats", str(args.micro_repeats), "--fold", str(fold), "--out-dir", str(micro_dir)],
        )
        manifest["micro"] = {"summary": str(micro_dir / "summary.json")}

    if not args.skip_structural_quality:
        structural_dir = out_dir / "structural-quality"
        step(
            "structural_quality",
            [
                sys.executable,
                "scripts/benchmarks/run_source_goldens.py",
                "--fold",
                str(fold),
                "--output-dir",
                str(structural_dir),
                "tests/benchmarks/source/code_quality_structural.json",
            ],
        )
        manifest["structural_quality"] = {
            "report": str(structural_dir / "code_quality_structural.json")
        }

    if not args.skip_interproc:
        interproc_dir = out_dir / "interproc-summary-ab"
        step(
            "interproc",
            [sys.executable, "bench/run_summary_ab.py", "--name", "interproc-quality-gate",
             "--repeats", "1", "--processes", "1", "--fold-bin", str(fold),
             "--detectors", "null-deref,div-by-zero,oob,int-overflow", "--out-dir", str(interproc_dir)],
        )
        manifest["interproc"] = {"summary": str(interproc_dir / "summary.json")}

    if not args.skip_security_bugs:
        security_root = out_dir / "security-bugs"
        step(
            "security_bugs_run",
            [sys.executable, "bench/run_bench.py", "--corpus", "bench/corpora/security-bugs",
             "--name", "security-bugs-quality-gate", "--tools", "fold", "--fold-bin", str(fold),
             "--detectors", "null-deref,div-by-zero,int-overflow,oob,dead", "--processes", "1",
             "--out-root", str(security_root)],
        )
        run_dir = latest_child(security_root, "_security-bugs-quality-gate")
        if run_dir:
            step(
                "security_bugs_score",
                [sys.executable, "bench/score.py", "--run-dir", str(run_dir),
                 "--gold", "bench/gold/security-bugs.json"],
            )
            manifest["security_bugs"] = {
                "run_dir": str(run_dir),
                "score": summarize_security_bugs(run_dir),
            }

    if not args.skip_curl:
        curl_source = Path(args.curl_source)
        if curl_source.exists():
            curl_dir = out_dir / "curl-smoke"
            step(
                "curl_smoke",
                [sys.executable, "bench/run_curl_ab.py", "--fold", str(fold),
                 "--curl-source", str(curl_source), "--curl-binary", args.curl_binary,
                 "--source-detectors", "null-deref,div-by-zero,int-overflow,oob",
                 "--source-max-files", "2", "--source-path-regex", "/(lib|src)/",
                 "--source-budget-seconds", "10", "--source-per-query-ms", "250",
                 "--source-per-tu-timeout", "20", "--binary-max-functions", "16",
                 "--processes", "1", "--out-dir", str(curl_dir)],
                timeout=180,
            )
            manifest["curl"] = {"summary": str(curl_dir / "summary.json")}
        else:
            manifest["curl"] = {"skipped": f"curl source not found: {curl_source}"}

    if not args.skip_juliet:
        juliet_dir = out_dir / "juliet"
        step(
            "juliet",
            [sys.executable, "bench/run_fold_on_juliet.py", "--cwes", *args.juliet_cwes,
             "--fold-bin", str(fold), "--processes", "1", "--out-root", str(juliet_dir)],
            timeout=args.juliet_timeout,
        )
        manifest["juliet"] = {"out_root": str(juliet_dir), "cwes": args.juliet_cwes}

    (out_dir / "manifest.json").write_text(json.dumps(manifest, indent=2, sort_keys=True) + "\n")
    print(f"wrote {out_dir / 'manifest.json'}")
    failed = [s for s in manifest["steps"] if s["exit_code"] != 0]
    return 1 if failed else 0


if __name__ == "__main__":
    raise SystemExit(main())