#!/usr/bin/env python3 """Benchmark Fold's structural code-quality lane on a real compile database. The runner deliberately separates a serial profiled scan from an optional throughput scan. It records performance, validates report invariants, and requires finding fingerprints to remain stable across repetitions and lanes. Example (Yams): scripts/benchmarks/run_code_quality.py \ --compile-db ../yams/build/release/compile_commands.json \ --project-root ../yams --path-regex '/src/metadata/' --max-tus 12 """ from __future__ import annotations import argparse import hashlib import importlib.util import json import re import subprocess import sys from pathlib import Path from typing import Any from fold_eval.dispositions import ( compare_dispositions, configuration_fingerprint, validate_disposition_baseline, ) from fold_eval.metrics import distribution_summary from fold_eval.performance import ( SOURCE_PHASE_METRICS, compare_performance_lanes, optimization_surface, ) from fold_eval.sarif import summarize_sarif_findings from fold_eval.schema import ( artifact_record, build_run_artifact, executable_version, sha256_file, utc_now, validate_run_artifact, write_json, ) from fold_eval.stability import stability_summary REPO_ROOT = Path(__file__).resolve().parents[2] RUNNER_COMMON_PATH = REPO_ROOT / "scripts" / "benchmarks" / "runner_common.py" _spec = importlib.util.spec_from_file_location("runner_common", RUNNER_COMMON_PATH) if _spec is None or _spec.loader is None: raise ImportError(f"failed to load {RUNNER_COMMON_PATH}") common = importlib.util.module_from_spec(_spec) _spec.loader.exec_module(common) DEFAULT_FOLD = REPO_ROOT / "build" / "release" / "src" / "cli" / "fold" SCAN_METRICS = ( "total_wall_ms", "analyze_ms", "compile_db_load_ms", "serial_dispatch_ms", *SOURCE_PHASE_METRICS, "frontend_extraction_ms", "report_ms", "peak_rss_kb", ) DETECTOR_METRICS = ( "wall_ms", "z3_queries", "z3_timeouts", "findings_emitted", "functions_analyzed", "functions_skipped", ) SCAN_SETTING_FLAGS = ( ("complexity_threshold", "--complexity-threshold"), ("min_body_stmts", "--min-body-stmts"), ("clone_type", "--clone-type"), ("clone_similarity", "--clone-similarity"), ("clone_min_score", "--clone-min-score"), ("class_similarity_threshold", "--class-similarity-threshold"), ("budget_seconds", "--budget-seconds"), ("per_query_ms", "--per-query-ms"), ("per_tu_timeout", "--per-tu-timeout"), ) def compile_db_path(value: Path) -> Path: path = value.expanduser().resolve() return path / "compile_commands.json" if path.is_dir() else path def discover_project_root(database: Path) -> Path: for parent in (database.parent, *database.parents): if (parent / ".git").exists(): return parent return database.parent def numeric(value: Any, *, name: str) -> float: try: return float(value) except (TypeError, ValueError) as exc: raise ValueError(f"{name} must be numeric") from exc def integer(value: Any, *, name: str) -> int: try: return int(value) except (TypeError, ValueError) as exc: raise ValueError(f"{name} must be an integer") from exc def load_text_document(path: Path, *, name: str) -> str: try: return path.read_text(encoding="utf-8") except OSError as exc: raise ValueError(f"cannot read {name} {path}: {exc}") from exc def load_json_document(path: Path, *, name: str) -> Any: text = load_text_document(path, name=name) try: return json.loads(text) except json.JSONDecodeError as exc: raise ValueError(f"invalid JSON in {name} {path}: {exc}") from exc def entry_file(entry: dict[str, Any]) -> Path: file_path = Path(str(entry.get("file", ""))) directory = Path(str(entry.get("directory", "."))) return ( (directory / file_path).resolve() if not file_path.is_absolute() else file_path.resolve() ) def filtered_compile_db( source: Path, output: Path, include: re.Pattern[str] | None, exclude: re.Pattern[str] | None, max_tus: int, ) -> tuple[Path, list[Path]]: loaded = load_json_document(source, name="compile database") if not isinstance(loaded, list): raise ValueError(f"compile database must contain a JSON array: {source}") selected: list[tuple[str, dict[str, Any], Path]] = [] seen: set[str] = set() for raw in loaded: if not isinstance(raw, dict): continue path = entry_file(raw) key = str(path) if key in seen or (include and include.search(key) is None): continue if exclude and exclude.search(key) is not None: continue seen.add(key) selected.append((key, raw, path)) selected.sort(key=lambda item: item[0]) if max_tus > 0: selected = selected[:max_tus] if not selected: raise ValueError("compile-database filters selected no translation units") database = output / "compile_commands.filtered.json" database.write_text( json.dumps([item[1] for item in selected], indent=2) + "\n", encoding="utf-8", ) return database, [item[2] for item in selected] def project_path(path: Path, project_root: Path) -> str: try: return path.resolve().relative_to(project_root).as_posix() except ValueError: return path.resolve().as_posix() def project_revision(project_root: Path) -> str: proc = subprocess.run( ["git", "-C", str(project_root), "rev-parse", "HEAD"], text=True, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, ) return proc.stdout.strip() if proc.returncode == 0 else "unknown" def source_kloc(paths: list[Path]) -> float: lines = 0 for path in paths: try: with path.open("r", encoding="utf-8", errors="replace") as stream: lines += sum(1 for _ in stream) except OSError: continue return lines / 1000.0 def density_scope_paths( tus: list[Path], findings: list[dict[str, Any]], changed_files: list[str], project_root: Path, ) -> list[Path]: if changed_files: return sorted({(project_root / item).resolve() for item in changed_files}) paths = {path.resolve() for path in tus} for finding in findings: uri = str(finding.get("uri", "")) if not uri: continue path = Path(uri) paths.add( path.resolve() if path.is_absolute() else (project_root / path).resolve() ) return sorted(paths) def load_changed_files(path: Path | None, project_root: Path) -> list[str]: if path is None: return [] text = load_text_document(path, name="changed-files input") loaded = ( load_json_document(path, name="changed-files input") if text.lstrip().startswith("[") else [line.strip() for line in text.splitlines() if line.strip()] ) if not isinstance(loaded, list) or not all( isinstance(item, str) for item in loaded ): raise ValueError( "changed-files must be a JSON string array or one path per line" ) normalized: list[str] = [] for item in loaded: candidate = Path(item) normalized.append( project_path(candidate, project_root) if candidate.is_absolute() else candidate.as_posix().removeprefix("./") ) return sorted(set(normalized)) def aggregate_profiles(profiles: list[dict[str, Any]]) -> dict[str, Any]: scan = { name: distribution_summary( [ numeric( profile.get("scan", {}).get(name, 0), name=f"scan metric {name}" ) for profile in profiles ] ) for name in SCAN_METRICS } detector_names = sorted( { str(detector.get("name", "")) for profile in profiles for detector in profile.get("detectors", []) if isinstance(detector, dict) } ) detectors: dict[str, Any] = {} for name in detector_names: rows = [ detector for profile in profiles for detector in profile.get("detectors", []) if isinstance(detector, dict) and detector.get("name") == name ] detectors[name] = { metric: distribution_summary( [ numeric(row.get(metric, 0), name=f"detector metric {metric}") for row in rows ] ) for metric in DETECTOR_METRICS } return {"scan": scan, "detectors": detectors} def run_lane( *, name: str, fold: Path, database: Path, uri_base: Path, project_root: Path, output_dir: Path, repeats: int, processes: int, scopes: list[str], detectors: str | None, scan_settings: dict[str, int | float], timeout: float, profiled: bool, ) -> tuple[list[dict[str, Any]], list[str]]: runs: list[dict[str, Any]] = [] failures: list[str] = [] fold_fingerprint = sha256_file(fold) input_fingerprint = sha256_file(database) for index in range(repeats): report_path = output_dir / f"{name}-{index + 1}.sarif.json" profile_path = output_dir / f"{name}-{index + 1}.profile.json" log_path = output_dir / f"{name}-{index + 1}.log" artifact_path = output_dir / f"{name}-{index + 1}.run.json" command = [ str(fold), "--quiet", "scan", "--mode", "source", "--compile-db", str(database), "--format", "json", "--output", str(report_path), "--processes", str(processes), ] for setting, flag in SCAN_SETTING_FLAGS: command.extend([flag, str(scan_settings[setting])]) if detectors: command.extend(["--detectors", detectors]) else: for scope in scopes: command.extend(["--scope", scope]) if profiled: command.extend(["--profile", "--profile-output", str(profile_path)]) started_at = utc_now() proc = common.run_measured(command, REPO_ROOT, timeout_seconds=timeout) ended_at = utc_now() log_path.write_text(proc.output, encoding="utf-8") run: dict[str, Any] = { "lane": name, "iteration": index + 1, "returncode": proc.returncode, **proc.metrics(), } if proc.returncode not in (0, 1): failures.append(f"{name} run {index + 1} exited {proc.returncode}") else: try: report = json.loads(report_path.read_text(encoding="utf-8")) run["findings"] = summarize_sarif_findings( report, project_root, uri_base ) if profiled: run["profile"] = json.loads( profile_path.read_text(encoding="utf-8") ) except (OSError, ValueError, json.JSONDecodeError) as exc: failures.append( f"{name} run {index + 1} produced invalid artifacts: {exc}" ) artifacts = [artifact_record("log", log_path, "text/plain")] if report_path.exists(): artifacts.append( artifact_record("sarif-report", report_path, "application/sarif+json") ) if profiled and profile_path.exists(): artifacts.append( artifact_record("profile", profile_path, "application/json") ) run_artifact = build_run_artifact( experiment_id=(f"code-quality:{input_fingerprint[:12]}:{name}:{index + 1}"), case_id="real-project-code-quality", tool={ "name": "fold", "path": str(fold), "version": executable_version(fold), "sha256": fold_fingerprint, }, target={ "project_root": str(project_root), "compile_db": str(database), "input_fingerprint": input_fingerprint, }, command=command, started_at=started_at, ended_at=ended_at, exit_status=proc.returncode, budgets={ "wall_seconds": timeout, "processes": processes, **scan_settings, }, artifacts=artifacts, measurements=proc.metrics(), observations=run.get("findings", {}), ) artifact_errors = validate_run_artifact(run_artifact) if artifact_errors: failures.extend( f"{name} run {index + 1} artifact: {error}" for error in artifact_errors ) write_json(artifact_path, run_artifact) run["run_artifact"] = str(artifact_path) runs.append(run) return runs, failures def rss_regression( current: dict[str, Any], baseline: dict[str, Any], max_rss_percent: float ) -> list[str]: failures: list[str] = [] current_rss = ( current.get("performance", {}) .get("profiled_serial", {}) .get("profile", {}) .get("scan", {}) .get("peak_rss_kb", {}) .get("median", 0) ) baseline_rss = ( baseline.get("performance", {}) .get("profiled_serial", {}) .get("profile", {}) .get("scan", {}) .get("peak_rss_kb", {}) .get("median", 0) ) baseline_rss_value = numeric(baseline_rss, name="baseline peak RSS") current_rss_value = numeric(current_rss, name="current peak RSS") if baseline_rss_value > 0: rss_delta = ( 100.0 * (current_rss_value - baseline_rss_value) / baseline_rss_value ) if rss_delta > max_rss_percent: failures.append( f"peak RSS regressed {rss_delta:.1f}% (limit {max_rss_percent:.1f}%)" ) return failures def render_markdown(summary: dict[str, Any]) -> str: reliability = summary["reliability"] stability = reliability.get("stability", {}) performance = summary["performance"] surface = summary["optimization_surface"] lines = [ "# Fold code-quality benchmark", "", f"- Status: **{summary['status']}**", f"- Translation units: `{summary['target']['selected_tus']}`", f"- Stable findings: `{reliability['stable_findings']}`", f"- Finding count: `{reliability.get('finding_count', 0)}`", f"- Finding fingerprint: `{reliability.get('finding_fingerprint', '')}`", f"- Finding intersection/union: " f"`{stability.get('intersection_count', 0)}/{stability.get('union_count', 0)}`", f"- Finding churn: `{stability.get('churn_count', 0)}`", f"- Pairwise Jaccard median: " f"`{stability.get('pairwise_jaccard', {}).get('median', 0.0)}`", f"- Profiled median: `{performance['profiled_serial']['wall_seconds']['median']} s`", f"- Peak RSS: `{surface['peak_rss_mb']} MiB`", "", "## Ranked optimization surface", "", "| Detector | Median ms | Analyze share | Z3 queries | Timeouts |", "| --- | ---: | ---: | ---: | ---: |", ] for row in surface["ranked_detectors"]: lines.append( f"| `{row['detector']}` | {row['wall_ms']} | {row['analyze_share_pct']}% | " f"{row['z3_queries']} | {row['z3_timeouts']} |" ) lines.extend( [ "", "## Source pipeline phases", "", "| Phase | Median ms | Analyze share |", "| --- | ---: | ---: |", ] ) for row in surface["ranked_source_phases"]: lines.append( f"| `{row['phase']}` | {row['wall_ms']} | {row['analyze_share_pct']}% |" ) lines.extend( [ "", f"Shared frontend/orchestration remainder: " f"`{surface['shared_frontend_and_orchestration_ms']} ms`; " f"known source phases: `{surface['known_source_phase_ms']} ms`; " f"still unattributed: " f"`{surface['unattributed_frontend_and_orchestration_ms']} ms`.", ] ) change_time = summary.get("change_time") if isinstance(change_time, dict): counts = change_time["counts"] lines.extend( [ "", "## Reviewed change-time findings", "", f"- New: `{counts['new']}`", f"- Persisting: `{counts['persisting']}`", f"- Resolved: `{counts['resolved']}`", f"- New in changed files: `{counts['new_in_changed_files']}`", f"- Review coverage: `{change_time['review_coverage']}`", f"- Density scope: `{change_time['scope_kloc']}` KLOC " f"(`{change_time['changed_scope']}`)", f"- Alert density: `{change_time['alert_density_per_kloc']}` per KLOC", f"- New alert density: `{change_time['new_alert_density_per_kloc']}` per KLOC", ] ) if reliability["failures"]: lines.extend(["", "## Failures", ""]) lines.extend(f"- {failure}" for failure in reliability["failures"]) return "\n".join(lines) + "\n" def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--fold", type=Path, default=DEFAULT_FOLD) parser.add_argument("--compile-db", type=Path, required=True) parser.add_argument("--project-root", type=Path) parser.add_argument("--out-dir", type=Path, default=Path("/tmp/fold-code-quality")) parser.add_argument("--path-regex") parser.add_argument( "--exclude-regex", default=r"/(third_party|subprojects|generated)/" ) parser.add_argument("--max-tus", type=int, default=20) parser.add_argument("--repeats", type=int, default=2) parser.add_argument("--processes", type=int, default=4) parser.add_argument("--scopes", default="quality,redundancy") parser.add_argument( "--detectors", help="Explicit detector list instead of --scopes" ) parser.add_argument("--complexity-threshold", type=int, default=20) parser.add_argument("--min-body-stmts", type=int, default=4) parser.add_argument("--clone-type", type=int, default=2) parser.add_argument("--clone-similarity", type=float, default=0.7) parser.add_argument("--clone-min-score", type=float, default=0.0) parser.add_argument("--class-similarity-threshold", type=float, default=0.7) parser.add_argument("--budget-seconds", type=int, default=60) parser.add_argument("--per-query-ms", type=int, default=2000) parser.add_argument("--per-tu-timeout", type=int, default=60) parser.add_argument("--timeout", type=float, default=600.0) parser.add_argument("--max-z3-timeouts", type=int, default=0) parser.add_argument("--baseline", type=Path) parser.add_argument( "--dispositions", type=Path, help="Reviewed finding baseline for new/persisting/resolved reporting", ) parser.add_argument( "--changed-files", type=Path, help="JSON array or newline list of project-relative changed files", ) parser.add_argument("--max-new-findings", type=int, default=0) parser.add_argument("--max-wall-regression-pct", type=float, default=20.0) parser.add_argument("--max-rss-regression-pct", type=float, default=20.0) parser.add_argument("--allow-finding-drift", action="store_true") args = parser.parse_args() fold = args.fold.expanduser().resolve() database = compile_db_path(args.compile_db) if not fold.exists() or not database.exists(): missing = fold if not fold.exists() else database print(f"required path not found: {missing}", file=sys.stderr) return 2 if ( args.repeats < 1 or args.processes < 1 or args.max_tus < 0 or args.max_new_findings < 0 ): print( "repeats/processes must be positive and limits non-negative", file=sys.stderr, ) return 2 project_root = ( args.project_root.expanduser().resolve() if args.project_root else discover_project_root(database) ) args.out_dir.mkdir(parents=True, exist_ok=True) try: filtered_db, tus = filtered_compile_db( database, args.out_dir, re.compile(args.path_regex) if args.path_regex else None, re.compile(args.exclude_regex) if args.exclude_regex else None, args.max_tus, ) changed_files = load_changed_files(args.changed_files, project_root) except (OSError, ValueError, json.JSONDecodeError, re.error) as exc: print(f"benchmark setup failed: {exc}", file=sys.stderr) return 2 scopes = [scope.strip() for scope in args.scopes.split(",") if scope.strip()] scan_settings: dict[str, int | float] = { "complexity_threshold": args.complexity_threshold, "min_body_stmts": args.min_body_stmts, "clone_type": args.clone_type, "clone_similarity": args.clone_similarity, "clone_min_score": args.clone_min_score, "class_similarity_threshold": args.class_similarity_threshold, "budget_seconds": args.budget_seconds, "per_query_ms": args.per_query_ms, "per_tu_timeout": args.per_tu_timeout, } profile_runs, failures = run_lane( name="profiled-serial", fold=fold, database=filtered_db, uri_base=database.parent, project_root=project_root, output_dir=args.out_dir, repeats=args.repeats, processes=1, scopes=scopes, detectors=args.detectors, scan_settings=scan_settings, timeout=args.timeout, profiled=True, ) throughput_runs: list[dict[str, Any]] = [] if args.processes > 1: throughput_runs, lane_failures = run_lane( name="throughput", fold=fold, database=filtered_db, uri_base=database.parent, project_root=project_root, output_dir=args.out_dir, repeats=args.repeats, processes=args.processes, scopes=scopes, detectors=args.detectors, scan_settings=scan_settings, timeout=args.timeout, profiled=False, ) failures.extend(lane_failures) all_runs = profile_runs + throughput_runs finding_rows = [run["findings"] for run in all_runs if "findings" in run] finding_stability = stability_summary( [row.get("identity_keys", []) for row in finding_rows] ) stable = finding_stability["exact"] and len(finding_rows) == len(all_runs) if not stable: failures.append( "finding fingerprints differ across repetitions or execution lanes" ) missing_locations = sum( integer(row["missing_location"], name="missing-location count") for row in finding_rows ) missing_backends = sum( integer(row["missing_backend"], name="missing-backend count") for row in finding_rows ) if missing_locations: failures.append(f"reports contain {missing_locations} locationless findings") if missing_backends: failures.append( f"reports contain {missing_backends} findings without backend provenance" ) profiles = [run["profile"] for run in profile_runs if "profile" in run] profile_summary = aggregate_profiles(profiles) z3_timeouts = sum( integer(metrics["z3_timeouts"]["max"], name="Z3 timeout count") for metrics in profile_summary["detectors"].values() ) if z3_timeouts > args.max_z3_timeouts: failures.append( f"Z3 timeouts {z3_timeouts} exceed limit {args.max_z3_timeouts}" ) target_paths = [project_path(path, project_root) for path in tus] target_hash = hashlib.sha256("\n".join(target_paths).encode()).hexdigest() revision = project_revision(project_root) first_findings = finding_rows[0] if finding_rows else {} performance: dict[str, Any] = { "profiled_serial": { "wall_samples_seconds": [ numeric(run["wall_seconds"], name="profiled wall seconds") for run in profile_runs ], "wall_seconds": distribution_summary( [ numeric(run["wall_seconds"], name="profiled wall seconds") for run in profile_runs ] ), "profile": profile_summary, } } if throughput_runs: performance["throughput"] = { "processes": args.processes, "wall_samples_seconds": [ numeric(run["wall_seconds"], name="throughput wall seconds") for run in throughput_runs ], "wall_seconds": distribution_summary( [ numeric(run["wall_seconds"], name="throughput wall seconds") for run in throughput_runs ] ), } analysis_configuration = { "scopes": [] if args.detectors else scopes, "detectors": args.detectors, "scan_settings": scan_settings, } analysis_fingerprint = configuration_fingerprint(analysis_configuration) summary: dict[str, Any] = { "schema_version": 2, "status": "ok", "target": { "project_root": str(project_root), "compile_db": str(database), "selected_compile_db": str(filtered_db), "selected_tus": len(tus), "tu_set_fingerprint": target_hash, "revision": revision, "paths": target_paths, }, "configuration": { **analysis_configuration, "analysis_fingerprint": analysis_fingerprint, "repeats": args.repeats, "throughput_processes": args.processes, }, "reliability": { "stable_findings": stable, "finding_count": first_findings.get("count", 0), "finding_fingerprint": first_findings.get("fingerprint", ""), "stability": finding_stability, "by_rule": first_findings.get("by_rule", {}), "z3_timeouts": z3_timeouts, "failures": failures, }, "performance": performance, "optimization_surface": optimization_surface(profile_summary), "run_artifacts": [ run["run_artifact"] for run in all_runs if "run_artifact" in run ], } if args.changed_files and not args.dispositions: failures.append("--changed-files requires --dispositions") if args.dispositions: try: dispositions = json.loads(args.dispositions.read_text(encoding="utf-8")) disposition_errors = validate_disposition_baseline(dispositions) if disposition_errors: raise ValueError("; ".join(disposition_errors)) disposition_target = dispositions["target"] if dispositions["target_name"] != project_root.name: failures.append("disposition baseline targets a different project") if disposition_target["tu_set_fingerprint"] != target_hash: failures.append( "disposition baseline uses a different translation-unit set" ) if disposition_target["configuration_fingerprint"] != analysis_fingerprint: failures.append( "disposition baseline uses a different analysis configuration" ) change_time = compare_dispositions( first_findings.get("records", []), dispositions, changed_files=changed_files, scope_kloc=source_kloc( density_scope_paths( tus, first_findings.get("records", []), changed_files, project_root, ) ), ) change_time["baseline_revision"] = disposition_target["revision"] change_time["current_revision"] = revision change_time["reviewed_at"] = dispositions["reviewed_at"] change_time["reviewer"] = dispositions["reviewer"] summary["change_time"] = change_time if change_time["identity_mismatches"]: failures.append("persisting finding IDs changed rule identity") if change_time["counts"]["new"] > args.max_new_findings: failures.append( f"new findings {change_time['counts']['new']} exceed limit " f"{args.max_new_findings}" ) except (OSError, ValueError, KeyError, json.JSONDecodeError) as exc: failures.append(f"invalid disposition baseline: {exc}") if args.baseline: baseline = load_json_document(args.baseline, name="performance baseline") if ( baseline.get("target", {}).get("tu_set_fingerprint") != summary["target"]["tu_set_fingerprint"] ): failures.append("baseline uses a different translation-unit set") comparable_configuration = ( "scopes", "detectors", "scan_settings", "throughput_processes", ) if any( baseline.get("configuration", {}).get(key) != summary["configuration"].get(key) for key in comparable_configuration ): failures.append("baseline uses a different analysis configuration") baseline_fingerprint = baseline.get("reliability", {}).get( "finding_fingerprint" ) if ( not args.allow_finding_drift and baseline_fingerprint and baseline_fingerprint != summary["reliability"]["finding_fingerprint"] ): failures.append("finding fingerprint differs from baseline") baseline_comparison, wall_failures = compare_performance_lanes( summary, baseline, max_wall_percent=args.max_wall_regression_pct, ) summary["baseline_comparison"] = baseline_comparison failures.extend(wall_failures) failures.extend(rss_regression(summary, baseline, args.max_rss_regression_pct)) summary["status"] = "failed" if failures else "ok" (args.out_dir / "summary.json").write_text( json.dumps(summary, indent=2, sort_keys=True) + "\n", encoding="utf-8" ) (args.out_dir / "summary.md").write_text(render_markdown(summary), encoding="utf-8") print(json.dumps(summary, indent=2, sort_keys=True)) return 1 if failures else 0 if __name__ == "__main__": raise SystemExit(main())