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