#!/usr/bin/env python3 """Lane-generic project evaluation runner driven by a versioned manifest. One manifest declares corpus, lane (source | rust | binary), analysis configuration, and evaluation policy; this runner executes Fold, retains immutable run artifacts, checks finding stability and reliability budgets, and optionally scores against gold, reviewed dispositions, and a prior performance baseline. Metric, matching, and gate semantics come from ``fold_eval`` so they cannot drift between entry points. Literature grounding: ``bench/literature_baselines.json`` group ``profiling-eval-2026-07`` and ``docs/research/project-eval-scaffolding-2026-07-27.md``. Example: scripts/benchmarks/run_project_eval.py \ --manifest tests/benchmarks/projects/yams_source.json \ --out-dir /tmp/fold-yams-eval """ from __future__ import annotations import argparse import hashlib import importlib.util import json import os import re import subprocess import sys import tempfile from collections import Counter from collections.abc import Callable 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 CLASSIFICATION_THRESHOLD_METRICS, distribution_summary from fold_eval.performance import compare_performance_lanes, optimization_surface from fold_eval.project import ( load_project_manifest, manifest_configuration, ) 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.scoring import ( negative_cases_from_universe, score_findings_by_rule, score_grouped_findings, score_labeled_cases, ) 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" SUCCESS_EXIT_CODES = (0, 1) # fold returns 1 when findings are present. SCAN_METRICS = ( "total_wall_ms", "analyze_ms", "compile_db_load_ms", "serial_dispatch_ms", "function_fingerprint_ms", "project_index_load_ms", "project_index_update_ms", "call_graph_ms", "project_graph_collect_ms", "project_graph_merge_ms", "project_graph_fallback_ms", "project_index_contribution_update_ms", "project_index_snapshot_update_ms", "security_candidate_ms", "cross_tu_property_ms", "symbolic_summary_ms", "frontend_extraction_ms", "report_ms", "num_tus", "num_functions", "findings_before_filter", "findings_report_visible", "source_tus_selected", "source_tus_excluded", "source_tus_attempted", "source_tus_completed", "source_tus_timed_out", "source_tus_frontend_failed", "peak_rss_kb", ) DETECTOR_METRICS = ( "wall_ms", "z3_queries", "z3_timeouts", "findings_emitted", "functions_analyzed", "functions_skipped", ) # Manifest analysis keys -> fold CLI flags for the source lane. Defaults match # run_code_quality.py so a manifest with identical settings produces the same # analysis configuration fingerprint as the pinned Yams baseline. SOURCE_SETTING_FLAGS = ( ("complexity_threshold", "--complexity-threshold", 20), ("min_body_stmts", "--min-body-stmts", 4), ("clone_type", "--clone-type", 2), ("clone_similarity", "--clone-similarity", 0.7), ("clone_min_score", "--clone-min-score", 0.0), ("class_similarity_threshold", "--class-similarity-threshold", 0.7), ("budget_seconds", "--budget-seconds", 60), ("per_query_ms", "--per-query-ms", 2000), ("per_tu_timeout", "--per-tu-timeout", 60), ) DEFAULT_EXCLUDE_REGEX = r"/(third_party|subprojects|generated)/" def numeric(value: Any, *, name: str = "value") -> float: try: return float(value) except (TypeError, ValueError) as error: raise ValueError(f"{name} must be numeric") from error def integer(value: Any, *, name: str = "value") -> int: if isinstance(value, bool): raise ValueError(f"{name} must be an integer") try: return int(value) except (TypeError, ValueError) as error: raise ValueError(f"{name} must be an integer") from error def load_json_document(path: Path) -> Any: try: return json.loads(path.read_text(encoding="utf-8")) except OSError as error: raise ValueError(f"failed to read JSON file {path}: {error}") from error except json.JSONDecodeError as error: raise ValueError(f"invalid JSON file {path}: {error}") from error # --------------------------------------------------------------------------- # Corpus preparation # --------------------------------------------------------------------------- def resolve(base: Path, value: str) -> Path: candidate = Path(value).expanduser() return ( candidate.resolve() if candidate.is_absolute() else (base / candidate).resolve() ) def compile_db_path(value: Path) -> Path: return value / "compile_commands.json" if value.is_dir() else value 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) 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 = path.read_text(encoding="utf-8") loaded = ( load_json_document(path) 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)) # --------------------------------------------------------------------------- # Lane command construction # --------------------------------------------------------------------------- def source_scan_settings(analysis: dict[str, Any]) -> dict[str, int | float]: return { key: analysis.get(key, default) for key, _flag, default in SOURCE_SETTING_FLAGS } def rust_toolchain(analysis: dict[str, Any]) -> str | None: configured = str(analysis.get("rust_toolchain", "")).strip() if configured: return configured toolchain_file = REPO_ROOT / "tools" / "fold-rust" / "rust-toolchain.toml" if not toolchain_file.exists(): return None for line in toolchain_file.read_text(encoding="utf-8").splitlines(): match = re.match(r'\s*channel\s*=\s*["\']([^"\']+)["\']\s*$', line) if match: return match.group(1) return None def detector_args(command: list[str], analysis: dict[str, Any]) -> None: detectors = [str(item) for item in analysis.get("detectors", [])] scopes = [str(item) for item in analysis.get("scopes", [])] if detectors: command.extend(["--detectors", ",".join(detectors)]) else: for scope in scopes: command.extend(["--scope", scope]) def build_source_command( fold: Path, database: Path, report_path: Path, profile_path: Path, analysis: dict[str, Any], scan_settings: dict[str, int | float], *, processes: int, profiled: bool, ) -> list[str]: command = [ str(fold), "--quiet", "scan", "--mode", "source", "--compile-db", str(database), "--format", "json", "--output", str(report_path), "--processes", str(processes), ] for key, flag, _default in SOURCE_SETTING_FLAGS: command.extend([flag, str(scan_settings[key])]) detector_args(command, analysis) command.extend(str(arg) for arg in analysis.get("extra_args", [])) if profiled: command.extend(["--profile", "--profile-output", str(profile_path)]) return command def build_rust_command( fold: Path, corpus: Path, report_path: Path, profile_path: Path, analysis: dict[str, Any], *, processes: int, profiled: bool, ) -> list[str]: command = [ str(fold), "--quiet", "scan", "--mode=rust", str(corpus), "--format", "json", "--output", str(report_path), "--processes", str(processes), ] detector_args(command, analysis) command.extend(str(arg) for arg in analysis.get("extra_args", [])) if profiled: command.extend(["--profile", "--profile-output", str(profile_path)]) return command def build_binary_command( fold: Path, binary: Path, report_path: Path, profile_path: Path, analysis: dict[str, Any], *, profiled: bool, ) -> list[str]: command = [ str(fold), "--quiet", "scan", "--mode", "binary", "--binary", str(binary), "--format", "json", "--output", str(report_path), ] detector_args(command, analysis) command.extend(str(arg) for arg in analysis.get("extra_args", [])) if profiled: command.extend(["--profile", "--profile-output", str(profile_path)]) return command # --------------------------------------------------------------------------- # Lane execution # --------------------------------------------------------------------------- def scan_operational_evidence(output: str) -> dict[str, Any]: """Classify frontend coverage failures without treating them as findings.""" categories: dict[str, list[tuple[str, str]]] = { "missing_dependencies": [], "tu_timeouts": [], "tu_crashes": [], "frontend_failures": [], } for raw_line in output.splitlines(): line = raw_line.strip() lowered = line.lower() worker_tu = ( line.split("while processing TU:", 1)[1].strip() if "while processing TU:" in line else "" ) compiler_match = re.match( r"^(.*?)(?::[0-9]+:[0-9]+)?:\s*(?:fatal )?error:", line ) compiler_tu = compiler_match.group(1) if compiler_match else "" if ("fatal error:" in lowered or "error:" in lowered) and ( "file not found" in lowered or "no such file or directory" in lowered ): categories["missing_dependencies"].append((compiler_tu, line)) if "while processing tu:" in lowered and ( "killed by signal 14" in lowered or "timed out" in lowered ): categories["tu_timeouts"].append((worker_tu, line)) elif "while processing tu:" in lowered and "killed by signal" in lowered: categories["tu_crashes"].append((worker_tu, line)) if "while processing tu:" in lowered and "exited with status" in lowered: categories["frontend_failures"].append((worker_tu, line)) evidence: dict[str, Any] = {} for name, entries in categories.items(): affected_tus = sorted({tu for tu, _line in entries if tu}) evidence[name] = { "count": len(affected_tus) if affected_tus else len(entries), "event_count": len(entries), "affected_tus": affected_tus, "examples": [line for _tu, line in entries[:8]], } return evidence def run_lane( *, name: str, fold: Path, output_dir: Path, repeats: int, command_for: Callable[[Path, Path, int], list[str]], uri_base: Path, project_root: Path, timeout: float, experiment_prefix: str, target: dict[str, Any], budgets: dict[str, Any], profiled: bool, env: dict[str, str] | None = None, ) -> tuple[list[dict[str, Any]], list[str]]: """Execute one lane ``repeats`` times and retain per-run artifacts.""" runs: list[dict[str, Any]] = [] failures: list[str] = [] fold_fingerprint = sha256_file(fold) fold_version = executable_version(fold) 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" for stale_path in (report_path, profile_path, log_path, artifact_path): stale_path.unlink(missing_ok=True) command = command_for(report_path, profile_path, index) started_at = utc_now() proc = common.run_measured(command, REPO_ROOT, env=env, timeout_seconds=timeout) ended_at = utc_now() log_path.write_text(proc.output, encoding="utf-8") operational_evidence = scan_operational_evidence(proc.output) affected_tus = sorted( { tu for evidence in operational_evidence.values() for tu in evidence["affected_tus"] } ) selected_tus = integer(target.get("selected_tus", 0), name="selected TUs") coverage_reconciliation = { "selected_tus": selected_tus, "failed_tus": len(affected_tus), "completed_tus_lower_bound": max(0, selected_tus - len(affected_tus)), "affected_tus": affected_tus, } run: dict[str, Any] = { "lane": name, "iteration": index + 1, "returncode": proc.returncode, "output_bytes": len(proc.output.encode("utf-8")), "operational_evidence": operational_evidence, "coverage_reconciliation": coverage_reconciliation, **proc.metrics(), } for category, evidence in operational_evidence.items(): if evidence["count"]: failures.append( f"{name} run {index + 1} {category.replace('_', ' ')}: " f"{evidence['count']}" ) if proc.returncode not in SUCCESS_EXIT_CODES: 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: if not profile_path.exists(): raise ValueError("profile output is missing") run["profile"] = json.loads( profile_path.read_text(encoding="utf-8") ) scan = run["profile"].get("scan", {}) if not isinstance(scan, dict): raise ValueError("profile scan metrics are missing") profile_selected = integer( scan.get("source_tus_selected"), name="profile selected TUs" ) profile_attempted = integer( scan.get("source_tus_attempted"), name="profile attempted TUs" ) profile_completed = integer( scan.get("source_tus_completed"), name="profile completed TUs" ) profile_timed_out = integer( scan.get("source_tus_timed_out"), name="profile timed-out TUs" ) profile_frontend_failed = integer( scan.get("source_tus_frontend_failed"), name="profile frontend-failed TUs", ) run["coverage_reconciliation"] = { "selected_tus": profile_selected, "attempted_tus": profile_attempted, "completed_tus": profile_completed, "timed_out_tus": profile_timed_out, "frontend_failed_tus": profile_frontend_failed, "affected_tus": affected_tus, } if ( profile_attempted != profile_selected or profile_completed != profile_attempted or profile_timed_out != 0 or profile_frontend_failed != 0 ): failures.append( f"{name} run {index + 1} incomplete profile coverage: " f"selected={profile_selected} attempted={profile_attempted} " f"completed={profile_completed} timed_out={profile_timed_out} " f"frontend_failed={profile_frontend_failed}" ) 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"{experiment_prefix}:{name}:{index + 1}", case_id="real-project-evaluation", tool={ "name": "fold", "path": str(fold), "version": fold_version, "sha256": fold_fingerprint, }, target=target, command=command, started_at=started_at, ended_at=ended_at, exit_status=proc.returncode, budgets=budgets, artifacts=artifacts, measurements=proc.metrics(), observations={ **run.get("findings", {}), "operational_evidence": operational_evidence, "coverage_reconciliation": run["coverage_reconciliation"], }, ) 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 aggregate_profiles(profiles: list[dict[str, Any]]) -> dict[str, Any]: scan = { name: distribution_summary( [ numeric(profile.get("scan", {}).get(name, 0), name=f"scan.{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 str(detector.get("name", "")) == name ] detectors[name] = { metric: distribution_summary( [ numeric(row.get(metric, 0), name=f"detector.{name}.{metric}") for row in rows ] ) for metric in DETECTOR_METRICS } return {"scan": scan, "detectors": detectors} # --------------------------------------------------------------------------- # Baseline comparison # --------------------------------------------------------------------------- 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_value = numeric(baseline_rss, name="baseline peak RSS") if baseline_value > 0: current_value = numeric(current_rss, name="current peak RSS") rss_delta = 100.0 * (current_value - baseline_value) / baseline_value if rss_delta > max_rss_percent: failures.append( f"peak RSS regressed {rss_delta:.1f}% (limit {max_rss_percent:.1f}%)" ) return failures def gold_threshold_failures( score: dict[str, Any], thresholds: dict[str, Any] ) -> list[str]: failures: list[str] = [] for metric in CLASSIFICATION_THRESHOLD_METRICS: if metric not in thresholds: continue observed = numeric(score.get(metric, 0.0), name=f"gold {metric}") minimum = numeric(thresholds[metric], name=f"gold {metric} threshold") if observed < minimum: failures.append( f"gold {metric} {observed:.3f} below threshold {minimum:.3f}" ) return failures def per_rule_gold_threshold_failures( scores: dict[str, dict[str, Any]], thresholds: dict[str, Any] ) -> list[str]: failures: list[str] = [] for rule, rule_thresholds in sorted(thresholds.items()): if not isinstance(rule_thresholds, dict): continue score = scores.get(rule, {}) for metric in CLASSIFICATION_THRESHOLD_METRICS: if metric not in rule_thresholds: continue observed = numeric(score.get(metric, 0.0), name=f"{rule} {metric}") minimum = numeric( rule_thresholds[metric], name=f"{rule} {metric} threshold" ) if observed < minimum: failures.append( f"rule {rule} {metric} {observed:.3f} below threshold {minimum:.3f}" ) return failures def stability_by_rule(finding_rows: list[dict[str, Any]]) -> dict[str, dict[str, Any]]: """Return deterministic stability summaries for every emitted rule.""" rules = sorted( { str(record.get("rule_id", "")) for row in finding_rows for record in row.get("records", []) if record.get("rule_id") } ) return { rule: stability_summary( [ [ str(record.get("finding_id", "")) for record in row.get("records", []) if record.get("rule_id") == rule ] for row in finding_rows ] ) for rule in rules } def stability_budget_failures( by_rule: dict[str, dict[str, Any]], policy: dict[str, Any], experimental_rules: set[str], ) -> list[str]: failures: list[str] = [] max_churn = policy.get("max_churn") min_jaccard = policy.get("min_pairwise_jaccard") for rule, stability in sorted(by_rule.items()): if ( rule.removeprefix("fold.") in experimental_rules or rule in experimental_rules ): continue churn = integer(stability.get("churn_count", 0), name=f"{rule} churn") median = numeric( stability.get("pairwise_jaccard", {}).get("median", 1.0), name=f"{rule} pairwise Jaccard", ) if max_churn is not None and churn > integer(max_churn, name="max churn"): failures.append(f"rule {rule} churn {churn} exceeds budget {max_churn}") if min_jaccard is not None: minimum = numeric(min_jaccard, name="minimum pairwise Jaccard") if median < minimum: failures.append( f"rule {rule} pairwise Jaccard {median:.3f} " f"below budget {minimum:.3f}" ) return failures # --------------------------------------------------------------------------- # Reporting # --------------------------------------------------------------------------- def render_markdown(summary: dict[str, Any]) -> str: reliability = summary["reliability"] stability = reliability.get("stability", {}) performance = summary["performance"] surface = summary["optimization_surface"] manifest = summary["manifest"] lines = [ "# Fold project evaluation", "", f"- Project: `{manifest['project']}` (lane `{manifest['lane']}`)", f"- Status: **{summary['status']}**", f"- Manifest fingerprint: `{manifest['configuration_fingerprint']}`", 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: `" f"{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']} |" ) per_binary = stability.get("per_binary") if isinstance(per_binary, dict): lines.extend( [ "", "## Per-binary stability", "", "| Binary lane | Stable | Findings |", "| --- | --- | ---: |", ] ) for name, item in sorted(per_binary.items()): if not isinstance(item, dict): continue counts = item.get("per_run_counts", []) findings = counts[0] if counts else 0 lines.append(f"| `{name}` | {item.get('exact', False)} | {findings} |") if surface["ranked_source_phases"]: lines.extend( [ "", "## 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']}% |" ) gold = summary.get("gold") if isinstance(gold, dict): lines.extend( [ "", "## Gold scoring", "", f"- TP/FP/FN: `{gold['true_positive']}/{gold['false_positive']}" f"/{gold['false_negative']}`", f"- Precision: `{gold['precision']}`", f"- Recall: `{gold['recall']}`", f"- F1: `{gold['f1']}`", ] ) 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"- Review coverage: `{change_time['review_coverage']}`", f"- Alert density: `{change_time['alert_density_per_kloc']}` per KLOC", f"- New alert density: " f"`{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" # --------------------------------------------------------------------------- # Main # --------------------------------------------------------------------------- def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--manifest", type=Path, required=True) parser.add_argument("--fold", type=Path, default=DEFAULT_FOLD) parser.add_argument( "--out-dir", type=Path, default=Path(tempfile.gettempdir()) / "fold-project-eval", ) parser.add_argument("--timeout", type=float, default=600.0) parser.add_argument("--changed-files", type=Path) parser.add_argument("--max-new-findings", type=int, default=0) parser.add_argument( "--allow-finding-drift", action="store_true", help="Also implied by evaluation.allow_finding_drift in the manifest.", ) args = parser.parse_args() fold = args.fold.expanduser().resolve() if not fold.exists(): print(f"fold binary not found: {fold}", file=sys.stderr) return 2 manifest_path = args.manifest.expanduser().resolve() try: manifest = load_project_manifest(manifest_path) except (OSError, ValueError, json.JSONDecodeError) as exc: print(f"{exc}", file=sys.stderr) return 2 lane = str(manifest["lane"]) manifest_dir = manifest_path.parent corpus = manifest.get("corpus", {}) analysis = manifest.get("analysis", {}) evaluation = manifest.get("evaluation", {}) missing_policy = str(corpus.get("missing_corpus", "fail")) # --- corpus resolution ------------------------------------------------- project_root = ( resolve(manifest_dir, str(corpus["path"])) if corpus.get("path") else None ) binaries: list[Path] = [] database: Path | None = None if lane == "source": database = compile_db_path(resolve(manifest_dir, str(corpus["compile_db"]))) if project_root is None: for parent in (database.parent, *database.parents): if (parent / ".git").exists(): project_root = parent break if project_root is None: project_root = database.parent missing = [] if database.exists() else [database] elif lane == "rust": missing = [] if project_root and project_root.is_dir() else [project_root] else: binaries = [ resolve(manifest_dir, str(item)) for item in corpus.get("binaries", []) ] missing = [binary for binary in binaries if not binary.exists()] if project_root is None: project_root = REPO_ROOT if missing: rendered = ", ".join(str(item) for item in missing) if missing_policy == "skip": print(f"missing corpus, skipping per manifest policy: {rendered}") return 0 print(f"missing corpus: {rendered}", file=sys.stderr) return 2 if project_root is None: raise ValueError("project root could not be resolved") args.out_dir.mkdir(parents=True, exist_ok=True) repeats = integer(evaluation.get("repeats", 2), name="evaluation.repeats") processes = integer(analysis.get("processes", 1), name="analysis.processes") allow_drift = bool(evaluation.get("allow_finding_drift", False)) or ( args.allow_finding_drift ) scan_settings = source_scan_settings(analysis) effective_configuration = manifest_configuration(manifest) if lane == "source": effective_configuration["exclude_regex"] = str( analysis.get("exclude_regex", DEFAULT_EXCLUDE_REGEX) ) effective_configuration["source_settings"] = scan_settings if lane == "rust": effective_configuration["rust_toolchain"] = rust_toolchain(analysis) or "" analysis_fingerprint = configuration_fingerprint(effective_configuration) # --- lane execution ------------------------------------------------------ failures: list[str] = [] all_runs: list[dict[str, Any]] = [] profile_runs: list[dict[str, Any]] = [] target: dict[str, Any] = {} tus: list[Path] = [] experiment_prefix = f"project-eval:{analysis_fingerprint[:12]}" try: if lane == "source": if database is None: raise ValueError("source lane requires a compile database") include = ( re.compile(str(analysis["path_regex"])) if analysis.get("path_regex") else None ) exclude = re.compile( str(analysis.get("exclude_regex", DEFAULT_EXCLUDE_REGEX)) ) filtered_db, tus = filtered_compile_db( database, args.out_dir, include, exclude, integer(analysis.get("max_tus", 0), name="analysis.max_tus"), ) input_fingerprint = sha256_file(filtered_db) target = { "project_root": str(project_root), "compile_db": str(filtered_db), "input_fingerprint": input_fingerprint, "selected_tus": len(tus), } budgets = { "wall_seconds": args.timeout, "processes": 1, **scan_settings, } profile_runs, failures = run_lane( name="profiled-serial", fold=fold, output_dir=args.out_dir, repeats=repeats, command_for=lambda report, profile, _i: build_source_command( fold, filtered_db, report, profile, analysis, scan_settings, processes=1, profiled=True, ), uri_base=database.parent, project_root=project_root, timeout=args.timeout, experiment_prefix=experiment_prefix, target=target, budgets=budgets, profiled=True, ) all_runs.extend(profile_runs) if processes > 1: throughput_runs, lane_failures = run_lane( name="throughput", fold=fold, output_dir=args.out_dir, repeats=repeats, command_for=lambda report, profile, _i: build_source_command( fold, filtered_db, report, profile, analysis, scan_settings, processes=processes, profiled=False, ), uri_base=database.parent, project_root=project_root, timeout=args.timeout, experiment_prefix=experiment_prefix, target=target, budgets={**budgets, "processes": processes}, profiled=False, ) failures.extend(lane_failures) all_runs.extend(throughput_runs) elif lane == "rust": input_fingerprint = configuration_fingerprint( {"corpus": str(project_root), "lane": "rust"} ) target = { "project_root": str(project_root), "input_fingerprint": input_fingerprint, } rust_env = os.environ.copy() if toolchain := rust_toolchain(analysis): rust_env["RUSTUP_TOOLCHAIN"] = toolchain profile_runs, failures = run_lane( name="profiled-rust", fold=fold, output_dir=args.out_dir, repeats=repeats, command_for=lambda report, profile, _i: build_rust_command( fold, project_root, report, profile, analysis, processes=processes, profiled=True, ), uri_base=project_root, project_root=project_root, timeout=args.timeout, experiment_prefix=experiment_prefix, target=target, budgets={"wall_seconds": args.timeout, "processes": processes}, profiled=True, env=rust_env, ) all_runs.extend(profile_runs) else: # binary lane for binary in binaries: slug = hashlib.sha256(str(binary).encode()).hexdigest()[:8] input_fingerprint = sha256_file(binary) target = { "binary": str(binary), "input_fingerprint": input_fingerprint, } binary_runs, lane_failures = run_lane( name=f"binary-{binary.stem}-{slug}", fold=fold, output_dir=args.out_dir, repeats=repeats, command_for=lambda report, profile, _i, b=binary: ( build_binary_command( fold, b, report, profile, analysis, profiled=True ) ), uri_base=binary.parent, project_root=project_root, timeout=args.timeout, experiment_prefix=experiment_prefix, target=target, budgets={"wall_seconds": args.timeout}, profiled=True, ) failures.extend(lane_failures) all_runs.extend(binary_runs) profile_runs = all_runs except (OSError, ValueError, json.JSONDecodeError, re.error) as exc: print(f"benchmark setup failed: {exc}", file=sys.stderr) return 2 # --- reliability and stability ------------------------------------------ finding_rows = [run["findings"] for run in all_runs if "findings" in run] if lane == "binary": runs_by_binary: dict[str, list[dict[str, Any]]] = {} for run in all_runs: if "findings" in run: runs_by_binary.setdefault(str(run["lane"]), []).append(run["findings"]) per_binary_stability = { name: stability_summary([row.get("identity_keys", []) for row in rows]) for name, rows in sorted(runs_by_binary.items()) } stable = len(finding_rows) == len(all_runs) and all( row["exact"] for row in per_binary_stability.values() ) finding_stability: dict[str, Any] = { "exact": stable, "per_binary": per_binary_stability, } first_findings_by_binary = { name: rows[0] for name, rows in sorted(runs_by_binary.items()) if rows } first_findings = next(iter(first_findings_by_binary.values()), {}) finding_count = sum( integer(row.get("count", 0), name="binary finding count") for row in first_findings_by_binary.values() ) by_rule_counter: Counter[str] = Counter() for row in first_findings_by_binary.values(): by_rule_counter.update(row.get("by_rule", {})) finding_fingerprint = configuration_fingerprint( { name: row.get("fingerprint", "") for name, row in first_findings_by_binary.items() } ) else: finding_stability = stability_summary( [row.get("identity_keys", []) for row in finding_rows] ) stable = finding_stability["exact"] and len(finding_rows) == len(all_runs) first_findings = finding_rows[0] if finding_rows else {} finding_count = integer(first_findings.get("count", 0), name="finding count") by_rule_counter = Counter(first_findings.get("by_rule", {})) finding_fingerprint = str(first_findings.get("fingerprint", "")) if not stable: failures.append( "finding fingerprints differ across repetitions or execution lanes" ) per_rule_stability = stability_by_rule(finding_rows) stability_policy = evaluation.get("stability", {}) if isinstance(stability_policy, dict): failures.extend( stability_budget_failures( per_rule_stability, stability_policy, {str(rule) for rule in evaluation.get("experimental_rules", [])}, ) ) missing_backends = sum( integer(row["missing_backend"], name="missing backend count") for row in finding_rows ) if missing_backends: failures.append( f"reports contain {missing_backends} findings without backend provenance" ) # Locationless results are fatal on source/rust where every finding must # carry a file+line; on the binary lane they are identity-bearing heuristic # observations (artifact+function+kind) and are reported, not failed. missing_locations = sum( integer(row["missing_location"], name="missing location count") for row in finding_rows ) if missing_locations and lane != "binary": failures.append(f"reports contain {missing_locations} locationless findings") 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() ) z3_budget = integer( evaluation.get("z3_timeout_budget", 0), name="evaluation.z3_timeout_budget" ) if z3_timeouts > z3_budget: failures.append(f"Z3 timeouts {z3_timeouts} exceed budget {z3_budget}") if "min_functions" in evaluation: functions = numeric( profile_summary["scan"]["num_functions"].get("median", 0), name="median function count", ) functions = max( [ functions, *( numeric( metrics["functions_analyzed"].get("median", 0), name="analyzed function count", ) for metrics in profile_summary["detectors"].values() ), ] ) minimum = integer(evaluation["min_functions"], name="evaluation.min_functions") if functions < minimum: failures.append( f"functions analyzed {functions:g} below required minimum {minimum}" ) absolute_budgets = evaluation.get("budgets", {}) if isinstance(absolute_budgets, dict): median_wall = numeric( distribution_summary( [ numeric(run["wall_seconds"], name="wall seconds") for run in profile_runs ] ).get("median", 0.0), name="median wall seconds", ) median_output = numeric( distribution_summary( [ numeric(run["output_bytes"], name="output bytes") for run in profile_runs ] ).get("median", 0.0), name="median output bytes", ) peak_rss_mb = ( numeric( profile_summary["scan"]["peak_rss_kb"].get("median", 0.0), name="median peak RSS", ) / 1024.0 ) frontend_ms = numeric( profile_summary["scan"]["frontend_extraction_ms"].get("median", 0.0), name="median frontend extraction time", ) limits = ( ("max_wall_seconds", median_wall, "wall seconds"), ("max_peak_rss_mb", peak_rss_mb, "peak RSS MiB"), ("max_frontend_extraction_ms", frontend_ms, "frontend extraction ms"), ("max_output_bytes", median_output, "output bytes"), ) for field, observed, label in limits: if field not in absolute_budgets: continue limit = numeric(absolute_budgets[field], name=f"evaluation.budgets.{field}") if observed > limit: failures.append(f"{label} {observed:.3f} exceeds budget {limit:.3f}") target_paths = [project_path(path, project_root) for path in tus] if lane == "source": target_hash = hashlib.sha256("\n".join(target_paths).encode()).hexdigest() elif lane == "rust": target_hash = configuration_fingerprint( {"lane": lane, "corpus": str(project_root)} ) else: target_hash = configuration_fingerprint( { "lane": lane, "binaries": [ {"path": str(binary), "sha256": sha256_file(binary)} for binary in binaries ], } ) revision = project_revision(project_root) 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, } } throughput_runs = [run for run in all_runs if run["lane"] == "throughput"] if throughput_runs: performance["throughput"] = { "processes": 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 ] ), } operational_by_run = [ { "lane": run["lane"], "iteration": run["iteration"], "reconciliation": run.get("coverage_reconciliation", {}), **run.get("operational_evidence", {}), } for run in all_runs ] operational_totals = { category: sum( integer( run.get("operational_evidence", {}).get(category, {}).get("count", 0), name=f"operational evidence {category}", ) for run in all_runs ) for category in ( "missing_dependencies", "tu_timeouts", "tu_crashes", "frontend_failures", ) } summary: dict[str, Any] = { "schema_version": 2, "status": "ok", "manifest": { "path": str(manifest_path), "project": str(manifest["project"]), "lane": lane, "configuration_fingerprint": analysis_fingerprint, "configuration": effective_configuration, }, "target": { "project_root": str(project_root), "lane": lane, "selected_tus": len(tus), "tu_set_fingerprint": target_hash, "revision": revision, "paths": target_paths, "binaries": [str(binary) for binary in binaries], }, "configuration": { **effective_configuration, "analysis_fingerprint": analysis_fingerprint, "repeats": repeats, "throughput_processes": processes if throughput_runs else 1, }, "reliability": { "stable_findings": stable, "finding_count": finding_count, "finding_fingerprint": finding_fingerprint, "stability": finding_stability, "per_rule_stability": per_rule_stability, "by_rule": dict(sorted(by_rule_counter.items())), "missing_location": missing_locations, "z3_timeouts": z3_timeouts, "operational_coverage": { "totals": operational_totals, "by_run": operational_by_run, }, "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 ], } # --- optional gold scoring ------------------------------------------------ if evaluation.get("gold"): gold_path = resolve(manifest_dir, str(evaluation["gold"])) try: gold_document = json.loads(gold_path.read_text(encoding="utf-8")) negative_entries: list[dict[str, Any]] | None = None gold_entries = gold_document if isinstance(gold_document, dict): gold_entries = gold_document.get( "cases", gold_document.get("gold", gold_document.get("expected", [])), ) if "negative" in gold_document: negative_entries = gold_document["negative"] elif "universe" in gold_document: negative_entries = negative_cases_from_universe( gold_entries, gold_document["universe"] ) if not isinstance(gold_entries, list): raise ValueError("gold file must contain a JSON array of cases") if negative_entries is not None and not isinstance(negative_entries, list): raise ValueError("gold negative cases must be a JSON array") score_input = [ { "id": record.get("finding_id", ""), "kind": str(record.get("rule_id", "")).removeprefix("fold."), "file": record.get("uri", ""), } for record in first_findings.get("records", []) ] summary["gold"] = ( score_labeled_cases(score_input, gold_entries, negative_entries) if negative_entries is not None else score_grouped_findings(score_input, gold_entries) ) summary["gold"]["per_rule"] = score_findings_by_rule( score_input, gold_entries, negative_entries ) thresholds = evaluation.get("gold_thresholds") if isinstance(thresholds, dict): failures.extend(gold_threshold_failures(summary["gold"], thresholds)) per_rule_thresholds = evaluation.get("per_rule_gold_thresholds") if isinstance(per_rule_thresholds, dict): failures.extend( per_rule_gold_threshold_failures( summary["gold"]["per_rule"], per_rule_thresholds ) ) except (OSError, ValueError, json.JSONDecodeError) as exc: failures.append(f"invalid gold file: {exc}") # --- optional disposition compare ------------------------------------------ changed_files = load_changed_files(args.changed_files, project_root) if args.changed_files and not evaluation.get("dispositions"): failures.append("--changed-files requires evaluation.dispositions") if evaluation.get("dispositions"): dispositions_path = resolve(manifest_dir, str(evaluation["dispositions"])) try: dispositions = json.loads(dispositions_path.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"] != str(manifest["project"]): failures.append("disposition baseline targets a different project") if lane == "source": if disposition_target["tu_set_fingerprint"] != target_hash: failures.append( "disposition baseline uses a different translation-unit set" ) # Source disposition baselines pin the run_code_quality analysis # configuration fingerprint; a manifest with matching settings # reproduces it through the same canonical shape. cq_configuration = { "scopes": effective_configuration["scopes"], "detectors": ( ",".join(effective_configuration["detectors"]) or None ), "scan_settings": scan_settings, } if disposition_target[ "configuration_fingerprint" ] != configuration_fingerprint(cq_configuration): 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"] > evaluation.get( "max_new_findings", args.max_new_findings ): failures.append( f"new findings {change_time['counts']['new']} exceed limit " f"{evaluation.get('max_new_findings', args.max_new_findings)}" ) except (OSError, ValueError, KeyError, json.JSONDecodeError) as exc: failures.append(f"invalid disposition baseline: {exc}") # --- optional baseline compare --------------------------------------------- if evaluation.get("baseline"): baseline_path = resolve(manifest_dir, str(evaluation["baseline"])) try: baseline = json.loads(baseline_path.read_text(encoding="utf-8")) if ( baseline.get("target", {}).get("tu_set_fingerprint") != summary["target"]["tu_set_fingerprint"] ): failures.append("baseline uses a different target set") if ( baseline.get("manifest", {}).get("configuration_fingerprint") != summary["manifest"]["configuration_fingerprint"] ): failures.append("baseline uses a different analysis configuration") baseline_fingerprint = baseline.get("reliability", {}).get( "finding_fingerprint" ) if ( not allow_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=numeric( evaluation.get("wall_regression_pct", 20.0), name="evaluation.wall_regression_pct", ), ) summary["baseline_comparison"] = baseline_comparison failures.extend(wall_failures) failures.extend( rss_regression( summary, baseline, numeric( evaluation.get("rss_regression_pct", 20.0), name="evaluation.rss_regression_pct", ), ) ) except (OSError, ValueError, json.JSONDecodeError) as exc: failures.append(f"invalid baseline: {exc}") 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())