#!/usr/bin/env python3 """A/B benchmark fold with and without symbolic summaries. The harness is intentionally small and reproducible: it builds a temporary compile_commands.json for a corpus, runs `fold scan` in profile mode for each variant, captures SARIF/profile/log artifacts, and writes a compact delta report. Default corpus: bench/corpora/interproc, which exercises guarded helper chains and cross-file callsites without requiring downloads. """ from __future__ import annotations import argparse import datetime as dt import json import os import shlex import subprocess import sys import time from pathlib import Path from statistics import mean from typing import Any REPO_ROOT = Path(__file__).resolve().parent.parent DEFAULT_FOLD = REPO_ROOT / "build" / "release" / "src" / "cli" / "fold" DEFAULT_CORPUS = REPO_ROOT / "bench" / "corpora" / "interproc" RESULTS_ROOT = REPO_ROOT / "bench" / "results" LLVM_BIN = Path(os.environ.get("LLVM_BIN", "/opt/homebrew/opt/llvm@22/bin")) def now_iso() -> str: return dt.datetime.now(dt.UTC).strftime("%Y-%m-%dT%H-%M-%SZ") def tool_path(name: str) -> str: candidate = LLVM_BIN / name return str(candidate if candidate.exists() else name) def write_compile_db( corpus_dir: Path, db_path: Path, c_standard: str, cxx_standard: str ) -> list[dict[str, str]]: include_dirs: set[str] = set() for path in sorted(corpus_dir.rglob("*")): if path.is_dir() and path.name == "testcasesupport": include_dirs.add(str(path)) if path.is_dir() and any(path.glob("*.h")): include_dirs.add(str(path)) include_args = [arg for inc in sorted(include_dirs) for arg in ("-I", inc)] entries: list[dict[str, str]] = [] for src in sorted(corpus_dir.rglob("*.c")): command = [tool_path("clang"), f"-std={c_standard}", *include_args, "-c", str(src)] entries.append({ "directory": str(corpus_dir), "command": shlex.join(command), "file": str(src), }) for src in sorted(corpus_dir.rglob("*.cpp")): command = [tool_path("clang++"), f"-std={cxx_standard}", *include_args, "-c", str(src)] entries.append({ "directory": str(corpus_dir), "command": shlex.join(command), "file": str(src), }) db_path.write_text(json.dumps(entries, indent=2) + "\n") return entries def normalize_kind(rule_id: str) -> str: kind = rule_id.removeprefix("fold.") return { "dead-branch": "dead", "duplicate-class": "class-dup", }.get(kind, kind) def parse_sarif(path: Path) -> list[dict[str, Any]]: if not path.exists(): return [] doc = json.loads(path.read_text()) findings: list[dict[str, Any]] = [] for run in doc.get("runs", []): for result in run.get("results", []): kind = normalize_kind(str(result.get("ruleId", "fold.unknown"))) fid = str(result.get("properties", {}).get("id", "")) message = str(result.get("message", {}).get("text", "")) locations = [] for loc in result.get("locations", []): physical = loc.get("physicalLocation", {}) artifact = physical.get("artifactLocation", {}) region = physical.get("region", {}) locations.append({ "file": artifact.get("uri", ""), "line": int(region.get("startLine", 0) or 0), "column": int(region.get("startColumn", 0) or 0), }) findings.append({ "kind": kind, "id": fid, "message": message, "locations": locations, }) return sorted(findings, key=finding_key) def finding_key(finding: dict[str, Any]) -> tuple[Any, ...]: locs = finding.get("locations") or [] loc_key = tuple( (str(loc.get("file", "")), int(loc.get("line", 0) or 0), int(loc.get("column", 0) or 0)) for loc in locs ) return (finding.get("kind", ""), finding.get("id", ""), loc_key) def compact_finding_key(finding: dict[str, Any]) -> str: kind, fid, loc_key = finding_key(finding) loc_text = ";".join(f"{Path(file).name}:{line}:{col}" for file, line, col in loc_key) return f"{kind}|{fid}|{loc_text}" def detector_totals(profile: dict[str, Any]) -> dict[str, int]: totals = { "wall_ms": 0, "z3_queries": 0, "z3_sat": 0, "z3_unsat": 0, "z3_unknown": 0, "z3_timeouts": 0, "abstract_sat": 0, "abstract_unsat": 0, "states_merged": 0, "states_kept_separate": 0, "findings_emitted": 0, "functions_analyzed": 0, "functions_skipped": 0, } for detector in profile.get("detectors", []): for key in totals: totals[key] += int(detector.get(key, 0) or 0) return totals def profile_summary(path: Path) -> dict[str, Any]: if not path.exists(): return {"scan": {}, "detectors_total": detector_totals({}), "detectors": []} profile = json.loads(path.read_text()) return { "scan": profile.get("scan", {}), "detectors_total": detector_totals(profile), "detectors": profile.get("detectors", []), } def suffix_match(path: str, expected: str) -> bool: return path.replace("\\", "/").endswith(expected.replace("\\", "/")) def score_against_gold( findings: list[dict[str, Any]], gold_path: Path | None, score_mode: str ) -> dict[str, Any] | None: if gold_path is None or not gold_path.exists(): return None gold = json.loads(gold_path.read_text()) expected = gold.get("expected", []) if score_mode == "file": return score_file_level(findings, expected, gold.get("name", gold_path.stem)) matched_gold: set[int] = set() matched_findings: set[int] = set() for gi, item in enumerate(expected): kind = item.get("kind", "any") files = [str(file) for file in item.get("files", [])] for fi, finding in enumerate(findings): if kind != "any" and finding.get("kind") != kind: continue finding_files = [str(loc.get("file", "")) for loc in finding.get("locations", [])] if all(any(suffix_match(file, expected_file) for file in finding_files) for expected_file in files): matched_gold.add(gi) matched_findings.add(fi) break tp = len(matched_gold) fp = max(0, len(findings) - len(matched_findings)) fn = len(expected) - tp precision = tp / (tp + fp) if tp + fp else 0.0 recall = tp / len(expected) if expected else 0.0 f1 = 2 * precision * recall / (precision + recall) if precision + recall else 0.0 return { "gold_name": gold.get("name", gold_path.stem), "true_positive": tp, "false_positive": fp, "false_negative": fn, "gold_total": len(expected), "precision": precision, "recall": recall, "f1": f1, "matched_gold_indices": sorted(matched_gold), "missed_gold_indices": sorted(set(range(len(expected))) - matched_gold), } def score_file_level( findings: list[dict[str, Any]], expected: list[dict[str, Any]], gold_name: str ) -> dict[str, Any]: finding_groups: set[tuple[str, str]] = set() for finding in findings: kind = str(finding.get("kind", "unknown")) for loc in finding.get("locations", []): file = str(loc.get("file", "")) if file: finding_groups.add((kind, file)) matched_gold: set[int] = set() matched_groups: set[tuple[str, str]] = set() for gi, item in enumerate(expected): kind = str(item.get("kind", "any")) files = [str(file) for file in item.get("files", [])] for group_kind, group_file in finding_groups: if kind != "any" and group_kind != kind: continue if any(suffix_match(group_file, expected_file) for expected_file in files): matched_gold.add(gi) matched_groups.add((group_kind, group_file)) break tp = len(matched_gold) fp = max(0, len(finding_groups) - len(matched_groups)) fn = len(expected) - tp precision = tp / (tp + fp) if tp + fp else 0.0 recall = tp / len(expected) if expected else 0.0 f1 = 2 * precision * recall / (precision + recall) if precision + recall else 0.0 return { "gold_name": gold_name, "true_positive": tp, "false_positive": fp, "false_negative": fn, "gold_total": len(expected), "finding_groups": len(finding_groups), "precision": precision, "recall": recall, "f1": f1, "matched_gold_indices": sorted(matched_gold), "missed_gold_indices": sorted(set(range(len(expected))) - matched_gold), } def variant_args(variant: str) -> list[str]: if variant == "baseline": return [] if variant == "summaries": return ["--build-summaries"] raise ValueError(f"unknown variant: {variant}") def run_variant( fold_bin: Path, db_path: Path, out_dir: Path, detectors: list[str], variant: str, repeat: int, processes: int, exclude_globs: list[str], extra_args: list[str], gold_path: Path | None, score_mode: str, ) -> dict[str, Any]: run_dir = out_dir / variant / f"repeat-{repeat}" run_dir.mkdir(parents=True, exist_ok=True) sarif_path = run_dir / "fold.sarif" profile_path = run_dir / "profile.json" stdout_path = run_dir / "stdout.log" stderr_path = run_dir / "stderr.log" command = [ str(fold_bin), "scan", "--compile-db", str(db_path), "--format", "sarif", "--output", str(sarif_path), "--min-body-stmts", "1", "--processes", str(processes), "--profile", "--profile-output", str(profile_path), ] for detector in detectors: command.extend(["--detectors", detector]) for pattern in exclude_globs: command.extend(["--exclude-glob", pattern]) command.extend(variant_args(variant)) command.extend(extra_args) env = os.environ.copy() env["PATH"] = f"{LLVM_BIN}:{env.get('PATH', '')}" started = time.monotonic() proc = subprocess.run(command, capture_output=True, text=True, env=env) elapsed_s = time.monotonic() - started stdout_path.write_text(proc.stdout) stderr_path.write_text(proc.stderr) findings = parse_sarif(sarif_path) score = score_against_gold(findings, gold_path, score_mode) payload = { "variant": variant, "repeat": repeat, "command": command, "elapsed_s": elapsed_s, "exit_code": proc.returncode, "findings_count": len(findings), "findings": findings, "finding_keys": [compact_finding_key(finding) for finding in findings], "profile": profile_summary(profile_path), "score": score, "artifacts": { "sarif": str(sarif_path), "profile": str(profile_path), "stdout": str(stdout_path), "stderr": str(stderr_path), }, } (run_dir / "result.json").write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n") if proc.returncode not in (0, 1): raise RuntimeError( f"fold scan failed for {variant} repeat {repeat} with rc={proc.returncode}; see {stderr_path}" ) return payload def refresh_result_from_artifacts( payload: dict[str, Any], gold_path: Path | None, score_mode: str ) -> dict[str, Any]: raw_artifacts = payload.get("artifacts") artifacts: dict[str, Any] = raw_artifacts if isinstance(raw_artifacts, dict) else {} sarif = artifacts.get("sarif") if not sarif: return payload sarif_path = Path(str(sarif)) if not sarif_path.exists(): return payload findings = parse_sarif(sarif_path) payload["findings_count"] = len(findings) payload["findings"] = findings payload["finding_keys"] = [compact_finding_key(finding) for finding in findings] payload["score"] = score_against_gold(findings, gold_path, score_mode) return payload def average_runs(runs: list[dict[str, Any]]) -> dict[str, Any]: if not runs: return {} total_keys = [ "wall_ms", "z3_queries", "z3_sat", "z3_unsat", "z3_unknown", "z3_timeouts", "abstract_sat", "abstract_unsat", "states_merged", "states_kept_separate", "findings_emitted", ] profile_avg = { key: mean(run["profile"]["detectors_total"].get(key, 0) for run in runs) for key in total_keys } scan_avg = { key: mean(run["profile"].get("scan", {}).get(key, 0) for run in runs) for key in ["total_wall_ms", "parse_ms", "analyze_ms", "report_ms", "num_tus"] } score = runs[-1].get("score") return { "variant": runs[-1]["variant"], "repeats": len(runs), "elapsed_s_avg": mean(run["elapsed_s"] for run in runs), "findings_count": runs[-1]["findings_count"], "finding_keys": runs[-1]["finding_keys"], "profile_avg": profile_avg, "scan_avg": scan_avg, "score": score, } def pct_delta(new: float, old: float) -> str: if old == 0: return "n/a" return f"{((new - old) / old) * 100:+.1f}%" def signed_int(value: int) -> str: return f"{value:+d}" def safe_ratio(new: float, old: float) -> float | None: if old == 0: return None return new / old def seconds_per_tp(elapsed_delta_s: float, tp_delta: int) -> float | None: if tp_delta <= 0 or elapsed_delta_s <= 0: return None return elapsed_delta_s / tp_delta def matched_indices(score: dict[str, Any]) -> set[int]: if isinstance(score.get("matched_gold_indices"), list): return {int(idx) for idx in score.get("matched_gold_indices", [])} gold_total = int(score.get("gold_total", 0) or 0) missed = {int(idx) for idx in score.get("missed_gold_indices", []) or []} return set(range(gold_total)) - missed def signal_recommendation(signal: dict[str, Any]) -> str: tp_delta = int(signal.get("tp_delta", 0) or 0) fp_delta = int(signal.get("fp_delta", 0) or 0) elapsed_delta_s = float(signal.get("elapsed_delta_s", 0.0) or 0.0) f1_delta = float(signal.get("f1_delta", 0.0) or 0.0) if tp_delta > 0: return "preserve finding gain; optimize summary precompute/caching around this path" if tp_delta < 0: return "fix recall regression before treating this as an optimization win" if fp_delta > 0 and f1_delta < 0: return "precision-only regression; tighten feasibility before adding more search" if elapsed_delta_s > 10: return "no gold gain; reduce implementation overhead or make this path lazy" return "neutral; keep as regression guard" def compute_optimization_signals(variants: dict[str, dict[str, Any]]) -> dict[str, dict[str, Any]]: baseline = variants.get("baseline") if not baseline: return {} base_score = baseline.get("score") or {} base_profile = baseline.get("profile_avg") or {} base_keys = set(baseline.get("finding_keys", [])) signals: dict[str, dict[str, Any]] = {} for name, data in variants.items(): if name == "baseline": continue score = data.get("score") or {} profile = data.get("profile_avg") or {} keys = set(data.get("finding_keys", [])) elapsed_delta_s = float(data.get("elapsed_s_avg", 0.0) or 0.0) - float( baseline.get("elapsed_s_avg", 0.0) or 0.0 ) tp_delta = int(score.get("true_positive", 0) or 0) - int( base_score.get("true_positive", 0) or 0 ) base_matched = matched_indices(base_score) variant_matched = matched_indices(score) signal = { "variant": name, "added_findings": len(keys - base_keys), "removed_findings": len(base_keys - keys), "finding_delta": int(data.get("findings_count", 0) or 0) - int(baseline.get("findings_count", 0) or 0), "tp_delta": tp_delta, "fp_delta": int(score.get("false_positive", 0) or 0) - int(base_score.get("false_positive", 0) or 0), "fn_delta": int(score.get("false_negative", 0) or 0) - int(base_score.get("false_negative", 0) or 0), "recall_delta": float(score.get("recall", 0.0) or 0.0) - float(base_score.get("recall", 0.0) or 0.0), "f1_delta": float(score.get("f1", 0.0) or 0.0) - float(base_score.get("f1", 0.0) or 0.0), "elapsed_delta_s": elapsed_delta_s, "elapsed_ratio": safe_ratio( float(data.get("elapsed_s_avg", 0.0) or 0.0), float(baseline.get("elapsed_s_avg", 0.0) or 0.0), ), "detector_wall_delta_ms": float(profile.get("wall_ms", 0.0) or 0.0) - float(base_profile.get("wall_ms", 0.0) or 0.0), "z3_query_delta": float(profile.get("z3_queries", 0.0) or 0.0) - float(base_profile.get("z3_queries", 0.0) or 0.0), "seconds_per_added_tp": seconds_per_tp(elapsed_delta_s, tp_delta), "gained_gold_indices": sorted(variant_matched - base_matched), "lost_gold_indices": sorted(base_matched - variant_matched), } signal["recommendation"] = signal_recommendation(signal) signals[name] = signal return signals def render_markdown(summary: dict[str, Any]) -> str: variants = summary["variant_summaries"] baseline = variants.get("baseline") max_delta_details = int(summary.get("max_delta_details", 50) or 0) optimization_signals = summary.get("optimization_signals") or compute_optimization_signals(variants) lines = [ f"# Summary A/B Benchmark: {summary['name']}", "", f"Corpus: `{summary['corpus']}`", f"Detectors: `{', '.join(summary['detectors'])}`", f"Repeats: `{summary['repeats']}`", f"Processes: `{summary['processes']}`", f"Score Mode: `{summary['score_mode']}`", "", "| Variant | Findings | Elapsed Avg | Profile Wall Avg | Z3 Queries | Z3 SAT | Z3 UNSAT | Abstract SAT | Abstract UNSAT | Unknown | States Merged | States Split |", "|---------|----------|-------------|------------------|------------|--------|----------|--------------|----------------|---------|---------------|--------------|", ] for name, data in variants.items(): profile = data["profile_avg"] lines.append( f"| {name} | {data['findings_count']} | {data['elapsed_s_avg']:.3f}s | " f"{profile['wall_ms']:.1f}ms | {profile['z3_queries']:.1f} | {profile['z3_sat']:.1f} | " f"{profile['z3_unsat']:.1f} | {profile['abstract_sat']:.1f} | " f"{profile['abstract_unsat']:.1f} | {profile['z3_unknown']:.1f} | " f"{profile['states_merged']:.1f} | {profile['states_kept_separate']:.1f} |" ) if baseline: lines.extend(["", "## Delta vs Baseline", ""]) base_keys = set(baseline["finding_keys"]) base_profile = baseline["profile_avg"] delta_details = [] lines.append("| Variant | Added Findings | Removed Findings | Query Delta | Wall Delta |") lines.append("|---------|----------------|------------------|-------------|------------|") for name, data in variants.items(): if name == "baseline": continue keys = set(data["finding_keys"]) added = sorted(keys - base_keys) removed = sorted(base_keys - keys) delta_details.append((name, added, removed)) query_delta = pct_delta(data["profile_avg"]["z3_queries"], base_profile["z3_queries"]) wall_delta = pct_delta(data["profile_avg"]["wall_ms"], base_profile["wall_ms"]) lines.append( f"| {name} | {len(added)} | {len(removed)} | {query_delta} | {wall_delta} |" ) for name, added, removed in delta_details: if not added and not removed: continue lines.extend(["", f"### {name} Finding Changes", ""]) if added: lines.append("Added:") shown = added[:max_delta_details] lines.extend(f"- `{key}`" for key in shown) if len(added) > len(shown): lines.append(f"- ... {len(added) - len(shown)} more omitted") if removed: lines.append("Removed:") shown = removed[:max_delta_details] lines.extend(f"- `{key}`" for key in shown) if len(removed) > len(shown): lines.append(f"- ... {len(removed) - len(shown)} more omitted") if optimization_signals: lines.extend(["", "## Optimization Signal", ""]) lines.append( "| Variant | TP Delta | FP Delta | Recall Delta | F1 Delta | Extra Time | Sec / Added TP | Recommendation |" ) lines.append( "|---------|----------|----------|--------------|----------|------------|----------------|----------------|" ) for name, signal in optimization_signals.items(): sec_per_tp = signal.get("seconds_per_added_tp") sec_per_tp_text = "n/a" if sec_per_tp is None else f"{float(sec_per_tp):.1f}s" lines.append( f"| {name} | {signed_int(int(signal['tp_delta']))} | " f"{signed_int(int(signal['fp_delta']))} | " f"{float(signal['recall_delta']):+.1%} | {float(signal['f1_delta']):+.1%} | " f"{float(signal['elapsed_delta_s']):+.1f}s | {sec_per_tp_text} | " f"{signal['recommendation']} |" ) if any(data.get("score") for data in variants.values()): lines.extend(["", "## Gold Score", ""]) lines.append("| Variant | TP | FP | FN | Precision | Recall | F1 |") lines.append("|---------|----|----|----|-----------|--------|----|") for name, data in variants.items(): score = data.get("score") if not score: continue lines.append( f"| {name} | {score['true_positive']} | {score['false_positive']} | " f"{score['false_negative']} | {score['precision']:.1%} | " f"{score['recall']:.1%} | {score['f1']:.1%} |" ) if baseline and baseline.get("score"): base_score = baseline["score"] lines.extend(["", "## Gold Delta vs Baseline", ""]) lines.append("| Variant | TP Delta | FP Delta | FN Delta | Recall Delta | F1 Delta |") lines.append("|---------|----------|----------|----------|--------------|----------|") for name, data in variants.items(): if name == "baseline": continue score = data.get("score") if not score: continue lines.append( f"| {name} | " f"{signed_int(score['true_positive'] - base_score['true_positive'])} | " f"{signed_int(score['false_positive'] - base_score['false_positive'])} | " f"{signed_int(score['false_negative'] - base_score['false_negative'])} | " f"{score['recall'] - base_score['recall']:+.1%} | " f"{score['f1'] - base_score['f1']:+.1%} |" ) lines.append("") return "\n".join(lines) def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--corpus", default=str(DEFAULT_CORPUS)) parser.add_argument("--name", default="summary-ab") parser.add_argument("--gold", default=None, help="Optional gold JSON; defaults to /gold.json when present") parser.add_argument("--fold-bin", default=str(DEFAULT_FOLD)) parser.add_argument("--out-root", default=str(RESULTS_ROOT)) parser.add_argument("--out-dir", default=None, help="Existing or new exact output directory") parser.add_argument("--detectors", default="null-deref,div-by-zero,oob,int-overflow") parser.add_argument("--variants", default="baseline,summaries") parser.add_argument("--repeats", type=int, default=1) parser.add_argument("--reuse-existing", action="store_true", help="Reuse existing repeat result.json files") parser.add_argument("--processes", type=int, default=1) parser.add_argument("--c-standard", default="c99") parser.add_argument("--cxx-standard", default="c++23") parser.add_argument("--exclude-glob", action="append", default=[]) parser.add_argument("--score-mode", choices=("finding", "file"), default="finding") parser.add_argument("--max-delta-details", type=int, default=50) parser.add_argument("--extra-arg", action="append", default=[], help="Extra argument passed to fold scan") args = parser.parse_args() corpus = Path(args.corpus).resolve() if not corpus.is_dir(): print(f"corpus path is not a directory: {corpus}", file=sys.stderr) return 2 fold_bin = Path(args.fold_bin).resolve() if not fold_bin.exists(): print(f"fold binary not found: {fold_bin}", file=sys.stderr) return 2 detectors = [detector.strip() for detector in args.detectors.split(",") if detector.strip()] variants = [variant.strip() for variant in args.variants.split(",") if variant.strip()] if args.repeats <= 0: print("--repeats must be positive", file=sys.stderr) return 2 if args.processes <= 0: print("--processes must be positive", file=sys.stderr) return 2 gold_path = Path(args.gold).resolve() if args.gold else corpus / "gold.json" if not gold_path.exists(): gold_path = None stamp = now_iso() out_dir = Path(args.out_dir).resolve() if args.out_dir else Path(args.out_root).resolve() / f"{stamp}_{args.name}" out_dir.mkdir(parents=True, exist_ok=True) db_path = out_dir / "compile_commands.json" entries = write_compile_db(corpus, db_path, args.c_standard, args.cxx_standard) runs_by_variant: dict[str, list[dict[str, Any]]] = {variant: [] for variant in variants} for variant in variants: for repeat in range(1, args.repeats + 1): existing_result = out_dir / variant / f"repeat-{repeat}" / "result.json" if args.reuse_existing and existing_result.exists(): print(f"reusing {variant} repeat {repeat}/{args.repeats}") payload = refresh_result_from_artifacts( json.loads(existing_result.read_text()), gold_path, args.score_mode ) existing_result.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n") runs_by_variant[variant].append(payload) else: print(f"running {variant} repeat {repeat}/{args.repeats}") runs_by_variant[variant].append( run_variant( fold_bin, db_path, out_dir, detectors, variant, repeat, args.processes, args.exclude_glob, args.extra_arg, gold_path, args.score_mode, ) ) variant_summaries = {variant: average_runs(runs) for variant, runs in runs_by_variant.items()} optimization_signals = compute_optimization_signals(variant_summaries) summary = { "name": args.name, "timestamp": stamp, "corpus": str(corpus), "compile_db": str(db_path), "source_entries": len(entries), "fold_bin": str(fold_bin), "detectors": detectors, "variants": variants, "repeats": args.repeats, "processes": args.processes, "c_standard": args.c_standard, "cxx_standard": args.cxx_standard, "exclude_globs": args.exclude_glob, "score_mode": args.score_mode, "max_delta_details": args.max_delta_details, "gold": str(gold_path) if gold_path else None, "runs": runs_by_variant, "variant_summaries": variant_summaries, "optimization_signals": optimization_signals, } (out_dir / "summary.json").write_text(json.dumps(summary, indent=2, sort_keys=True) + "\n") report = render_markdown(summary) (out_dir / "summary.md").write_text(report) print(report) print(f"wrote {out_dir}") return 0 if __name__ == "__main__": raise SystemExit(main())