#!/usr/bin/env python3 """Run immutable paired Fold A/B measurements on source-project manifests. Each baseline/current pair uses the same filtered compile database and repetition index. Serial cold/warm cells exercise fresh and reused Fold project-index state. Fanout cells remain non-incremental because Fold intentionally serializes shared project-index updates; their warm cell records an excluded preflight process rather than pretending the host page cache can be flushed portably. """ from __future__ import annotations import argparse import hashlib import importlib import importlib.util import json import re import statistics import sys from dataclasses import dataclass from pathlib import Path from typing import Any REPO_ROOT = Path(__file__).resolve().parents[2] BENCHMARK_ROOT = REPO_ROOT / "scripts" / "benchmarks" if str(BENCHMARK_ROOT) not in sys.path: sys.path.insert(0, str(BENCHMARK_ROOT)) sha256_file = importlib.import_module("fold_eval.schema").sha256_file load_project_manifest = importlib.import_module( "fold_eval.project" ).load_project_manifest _PROJECT_EVAL_SPEC = importlib.util.spec_from_file_location( "fold_project_eval", BENCHMARK_ROOT / "run_project_eval.py" ) if _PROJECT_EVAL_SPEC is None or _PROJECT_EVAL_SPEC.loader is None: raise RuntimeError("cannot load run_project_eval.py") project_eval = importlib.util.module_from_spec(_PROJECT_EVAL_SPEC) _PROJECT_EVAL_SPEC.loader.exec_module(project_eval) DEFAULT_CURRENT = REPO_ROOT / "build" / "release" / "src" / "cli" / "fold" DEFAULT_MANIFESTS = [ REPO_ROOT / "tests" / "benchmarks" / "projects" / "fold_source.json", REPO_ROOT / "tests" / "benchmarks" / "projects" / "yams_source.json", ] @dataclass(frozen=True) class MatrixCell: execution: str cache_state: str processes: int profiled: bool incremental: bool cache_mechanism: str @property def key(self) -> str: return f"{self.execution}-{self.cache_state}" def matrix_cells(fanout_processes: int) -> list[MatrixCell]: if fanout_processes < 2: raise ValueError("fanout processes must be at least 2") return [ MatrixCell("serial", "cold", 1, True, True, "fold-project-index"), MatrixCell("serial", "warm", 1, True, True, "fold-project-index"), MatrixCell( "fanout", "cold", fanout_processes, False, False, "fresh-output-process" ), MatrixCell( "fanout", "warm", fanout_processes, False, False, "excluded-preflight-process", ), ] def ensure_empty_output(path: Path) -> None: path.mkdir(parents=True, exist_ok=True) if any(path.iterdir()): raise ValueError(f"A/B output directory must be empty: {path}") def file_sha256(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as stream: for block in iter(lambda: stream.read(1024 * 1024), b""): digest.update(block) return digest.hexdigest() def finding_equivalence(runs: list[dict[str, Any]]) -> dict[str, Any]: fingerprints = [str(run.get("findings", {}).get("fingerprint", "")) for run in runs] missing = sum(not fingerprint for fingerprint in fingerprints) distinct = sorted({fingerprint for fingerprint in fingerprints if fingerprint}) return { "equivalent": missing == 0 and len(distinct) <= 1, "missing_fingerprints": missing, "distinct_fingerprints": distinct, } def measured_number(value: Any, name: str) -> int | float: if isinstance(value, bool) or not isinstance(value, (int, float)): raise ValueError(f"{name} must be numeric") return value def repetition_number(value: Any) -> int: if isinstance(value, bool) or not isinstance(value, int) or value < 1: raise ValueError("pair repetition must be a positive integer") return value def paired_cell_summary( baseline_runs: list[dict[str, Any]], current_runs: list[dict[str, Any]] ) -> dict[str, Any]: if len(baseline_runs) != len(current_runs): raise ValueError("baseline/current repetition counts differ") pairs: list[dict[str, Any]] = [] for baseline, current in zip(baseline_runs, current_runs, strict=True): baseline_finding = baseline.get("findings", {}) current_finding = current.get("findings", {}) pairs.append( { "repetition": repetition_number(current.get("pair_repetition")), "finding_equivalent": baseline_finding.get("fingerprint") == current_finding.get("fingerprint"), "baseline_finding_count": baseline_finding.get("count", 0), "current_finding_count": current_finding.get("count", 0), "baseline_wall_seconds": baseline.get("wall_seconds", 0), "current_wall_seconds": current.get("wall_seconds", 0), } ) baseline_wall = [ measured_number(pair["baseline_wall_seconds"], "baseline wall seconds") for pair in pairs ] current_wall = [ measured_number(pair["current_wall_seconds"], "current wall seconds") for pair in pairs ] return { "pairs": pairs, "finding_equivalent": all(pair["finding_equivalent"] for pair in pairs), "baseline_wall_median_seconds": statistics.median(baseline_wall), "current_wall_median_seconds": statistics.median(current_wall), "paired_wall_delta_seconds": [ current - baseline for baseline, current in zip(baseline_wall, current_wall, strict=True) ], } def resolve_manifest_source( manifest_path: Path, work_dir: Path ) -> tuple[dict[str, Any], Path, Path, Path, list[Path]]: manifest = load_project_manifest(manifest_path) if manifest["lane"] != "source": raise ValueError(f"A/B harness supports source manifests only: {manifest_path}") corpus = manifest["corpus"] manifest_dir = manifest_path.parent project_root = project_eval.resolve(manifest_dir, str(corpus["path"])) database = project_eval.compile_db_path( project_eval.resolve(manifest_dir, str(corpus["compile_db"])) ) if not database.exists(): if corpus.get("missing_corpus") == "skip": raise FileNotFoundError(f"optional corpus missing: {database}") raise ValueError(f"compile database is missing: {database}") analysis = manifest.get("analysis", {}) include = ( re.compile(str(analysis["path_regex"])) if analysis.get("path_regex") else None ) exclude = re.compile( str(analysis.get("exclude_regex", project_eval.DEFAULT_EXCLUDE_REGEX)) ) filtered, tus = project_eval.filtered_compile_db( database, work_dir, include, exclude, project_eval.integer(analysis.get("max_tus", 0), name="analysis.max_tus"), ) return manifest, project_root, database, filtered, tus def run_cell_repetition( *, fold: Path, variant: str, project: str, manifest: dict[str, Any], project_root: Path, database: Path, filtered_db: Path, selected_translation_units: int, cell: MatrixCell, repetition: int, output_dir: Path, timeout: float, ) -> tuple[dict[str, Any], list[str], list[dict[str, Any]]]: output_dir.mkdir(parents=True, exist_ok=False) cache_dir = output_dir / "fold-cache" cache_dir.mkdir(parents=True) analysis = dict(manifest.get("analysis", {})) extra_args = [str(value) for value in analysis.get("extra_args", [])] if cell.incremental: extra_args.extend(["--incremental", "--cache-dir", str(cache_dir)]) analysis["extra_args"] = extra_args scan_settings = project_eval.source_scan_settings(analysis) target = { "project": project, "project_root": str(project_root), "compile_db": str(filtered_db), "input_fingerprint": sha256_file(filtered_db), "selected_tus": selected_translation_units, "manifest_configuration": project_eval.configuration_fingerprint( project_eval.manifest_configuration(manifest) ), } budgets = { "wall_seconds": timeout, "processes": cell.processes, **scan_settings, } def command_for(report: Path, profile: Path, _index: int) -> list[str]: return project_eval.build_source_command( fold, filtered_db, report, profile, analysis, scan_settings, processes=cell.processes, profiled=cell.profiled, ) warmups: list[dict[str, Any]] = [] failures: list[str] = [] if cell.cache_state == "warm": warmup_dir = output_dir / "warmup" warmup_dir.mkdir() warmups, warmup_failures = project_eval.run_lane( name="excluded-warmup", fold=fold, output_dir=warmup_dir, repeats=1, command_for=command_for, uri_base=database.parent, project_root=project_root, timeout=timeout, experiment_prefix=f"ab:{project}:{variant}:{cell.key}:r{repetition}", target=target, budgets=budgets, profiled=cell.profiled, ) failures.extend(f"warmup: {failure}" for failure in warmup_failures) measured_dir = output_dir / "measured" measured_dir.mkdir() runs, run_failures = project_eval.run_lane( name="measured", fold=fold, output_dir=measured_dir, repeats=1, command_for=command_for, uri_base=database.parent, project_root=project_root, timeout=timeout, experiment_prefix=f"ab:{project}:{variant}:{cell.key}:r{repetition}", target=target, budgets=budgets, profiled=cell.profiled, ) failures.extend(run_failures) run = runs[0] measured_log = measured_dir / "measured-1.log" try: log_text = measured_log.read_text(encoding="utf-8") except OSError: log_text = "" profile_scan = run.get("profile", {}).get("scan", {}) run.update( { "variant": variant, "project": project, "cell": cell.key, "pair_repetition": repetition, "cache_mechanism": cell.cache_mechanism, "coverage": { "selected_translation_units": selected_translation_units, "profile_reported_translation_units": profile_scan.get("num_tus"), "diagnostic_error_lines": sum( 1 for line in log_text.splitlines() if "error:" in line.lower() ), "process_status": ( "completed-clean" if run.get("returncode") == 0 else "completed-with-findings" if run.get("returncode") == 1 else "analyzer-failure" ), }, "run_artifact": str(Path(run["run_artifact"]).relative_to(output_dir)), } ) return run, failures, warmups def render_markdown(summary: dict[str, Any]) -> str: lines = ["# Fold paired A/B evaluation", ""] for project in summary["projects"]: lines.extend([f"## {project['project']}", ""]) if project.get("status") == "skipped": lines.extend([f"- Status: skipped ({project['reason']})", ""]) continue for key, comparison in project["comparisons"].items(): lines.append( f"- `{key}`: findings={'same' if comparison['finding_equivalent'] else 'DIFFER'}, " f"baseline={comparison['baseline_wall_median_seconds']:.3f}s, " f"current={comparison['current_wall_median_seconds']:.3f}s" ) lines.append("") if summary["failures"]: lines.extend( ["## Failures", "", *[f"- {item}" for item in summary["failures"]], ""] ) return "\n".join(lines) def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--baseline-fold", type=Path, required=True) parser.add_argument("--current-fold", type=Path, default=DEFAULT_CURRENT) parser.add_argument("--manifest", type=Path, action="append", default=[]) parser.add_argument("--out-dir", type=Path, required=True) parser.add_argument("--repeats", type=int, default=3) parser.add_argument("--fanout-processes", type=int, default=2) parser.add_argument("--timeout", type=float, default=600.0) parser.add_argument("--allow-finding-drift", action="store_true") args = parser.parse_args() if args.repeats < 2: print("--repeats must be at least 2 for variability evidence", file=sys.stderr) return 2 try: cells = matrix_cells(args.fanout_processes) out_dir = args.out_dir.expanduser().resolve() ensure_empty_output(out_dir) except ValueError as exc: print(exc, file=sys.stderr) return 2 binaries = { "baseline": args.baseline_fold.expanduser().resolve(), "current": args.current_fold.expanduser().resolve(), } for variant, binary in binaries.items(): if not binary.is_file(): print(f"{variant} Fold binary is missing: {binary}", file=sys.stderr) return 2 binary_hashes = {variant: file_sha256(path) for variant, path in binaries.items()} manifests = [ path.expanduser().resolve() for path in (args.manifest or DEFAULT_MANIFESTS) ] failures: list[str] = [] projects: list[dict[str, Any]] = [] for manifest_path in manifests: project_dir = out_dir / manifest_path.stem project_dir.mkdir() try: manifest, project_root, database, filtered_db, tus = ( resolve_manifest_source(manifest_path, project_dir) ) except FileNotFoundError as exc: projects.append( {"project": manifest_path.stem, "status": "skipped", "reason": str(exc)} ) continue except (OSError, ValueError, json.JSONDecodeError) as exc: failures.append(f"{manifest_path.name}: {exc}") projects.append( {"project": manifest_path.stem, "status": "failed", "reason": str(exc)} ) continue project = str(manifest["project"]) variant_runs: dict[str, dict[str, list[dict[str, Any]]]] = { variant: {cell.key: [] for cell in cells} for variant in binaries } warmup_records: list[dict[str, Any]] = [] for cell in cells: for repetition in range(1, args.repeats + 1): # Alternate variant order by repetition to reduce monotonic host # temperature bias while retaining exact pair indices. order = ( ["baseline", "current"] if repetition % 2 else ["current", "baseline"] ) for variant in order: run_dir = project_dir / variant / cell.key / f"repeat-{repetition}" run, run_failures, warmups = run_cell_repetition( fold=binaries[variant], variant=variant, project=project, manifest=manifest, project_root=project_root, database=database, filtered_db=filtered_db, selected_translation_units=len(tus), cell=cell, repetition=repetition, output_dir=run_dir, timeout=args.timeout, ) variant_runs[variant][cell.key].append(run) warmup_records.extend(warmups) failures.extend( f"{project}/{variant}/{cell.key}/r{repetition}: {failure}" for failure in run_failures ) comparisons: dict[str, Any] = {} for cell in cells: comparison = paired_cell_summary( variant_runs["baseline"][cell.key], variant_runs["current"][cell.key] ) comparisons[cell.key] = comparison if not comparison["finding_equivalent"] and not args.allow_finding_drift: failures.append( f"{project}/{cell.key}: baseline/current findings differ" ) for variant in binaries: all_measured = [ run for runs in variant_runs[variant].values() for run in runs ] equivalence = finding_equivalence(all_measured) if not equivalence["equivalent"] and not args.allow_finding_drift: failures.append( f"{project}/{variant}: matrix findings are not equivalent" ) projects.append( { "project": project, "status": "ok", "manifest": str(manifest_path), "manifest_sha256": file_sha256(manifest_path), "translation_units": len(tus), "matrix": [cell.__dict__ | {"key": cell.key} for cell in cells], "runs": variant_runs, "comparisons": comparisons, "within_variant_equivalence": { variant: finding_equivalence( [run for runs in variant_runs[variant].values() for run in runs] ) for variant in binaries }, "warmup_run_count": len(warmup_records), "warmup_run_artifacts": [ str(Path(run["run_artifact"]).relative_to(project_dir)) for run in warmup_records ], } ) for variant, path in binaries.items(): final_hash = file_sha256(path) if final_hash != binary_hashes[variant]: failures.append(f"{variant} Fold binary changed during the A/B run") summary = { "schema_version": 1, "status": "failed" if failures else "ok", "binaries": { variant: {"path": str(path), "sha256": binary_hashes[variant]} for variant, path in binaries.items() }, "pairing": { "repeats": args.repeats, "variant_order": "alternating-by-repetition", "fanout_processes": args.fanout_processes, "warmups_excluded": True, "fanout_incremental_disabled": True, }, "projects": projects, "failures": failures, } (out_dir / "summary.json").write_text( json.dumps(summary, indent=2, sort_keys=True) + "\n", encoding="utf-8" ) (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())