c4e42cec0a
Split source and binary scan routing with merge-only hybrid support, binary provenance, and router-backed MCP scan contexts. Add async MCP scan jobs for long-running scans, detector precision fixes, benchmark scripts, docs, and integration coverage.
727 lines
28 KiB
Python
727 lines
28 KiB
Python
#!/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 <corpus>/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())
|