#!/usr/bin/env python3 """Run profiled project evaluations and summarize optimization evidence. This wrapper keeps profiling, external-process failures, and report-size/token accounting in one reproducible artifact. It delegates the actual scanner workload to ``run_project_eval.py`` so optimization measurements retain the same corpus, stability, and correctness gates as ordinary project evaluation. """ from __future__ import annotations import argparse import importlib import json import subprocess import sys import tempfile 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)) artifact_token_efficiency = importlib.import_module( "fold_eval.token_efficiency" ).artifact_token_efficiency DEFAULT_FOLD = REPO_ROOT / "build" / "release" / "src" / "cli" / "fold" DEFAULT_MANIFEST = REPO_ROOT / "tests" / "benchmarks" / "projects" / "yams_source.json" PROJECT_EVAL = REPO_ROOT / "scripts" / "benchmarks" / "run_project_eval.py" def read_json(path: Path) -> dict[str, Any]: try: loaded = json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError) as exc: raise ValueError(f"invalid JSON artifact {path}: {exc}") from exc if not isinstance(loaded, dict): raise ValueError(f"JSON artifact must be an object: {path}") return loaded def file_size(path: Path | None) -> int: if path is None or not path.is_file(): return 0 try: return path.stat().st_size except OSError: return 0 def artifact_records(run_dir: Path) -> list[dict[str, int]]: records: list[dict[str, int]] = [] for artifact_path in sorted(run_dir.glob("*.run.json")): artifact = read_json(artifact_path) observations = artifact.get("observations", {}) artifacts = artifact.get("artifacts", []) if not isinstance(observations, dict) or not isinstance(artifacts, list): raise ValueError(f"invalid run artifact shape: {artifact_path}") paths = { str(item.get("role", "")): Path(str(item.get("path", ""))) for item in artifacts if isinstance(item, dict) and item.get("path") } count = observations.get("count", 0) if isinstance(count, bool) or not isinstance(count, int) or count < 0: raise ValueError(f"invalid finding count in {artifact_path}") records.append( { "finding_count": count, "output_bytes": file_size(paths.get("log")), "report_bytes": file_size(paths.get("sarif-report")), } ) return records def run_evaluation( manifest: Path, fold: Path, run_dir: Path, timeout_seconds: float ) -> dict[str, Any]: command = [ sys.executable, str(PROJECT_EVAL), "--manifest", str(manifest), "--fold", str(fold), "--out-dir", str(run_dir), "--timeout", str(timeout_seconds), ] try: completed = subprocess.run( command, cwd=REPO_ROOT, text=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, timeout=timeout_seconds + 30.0, ) except subprocess.TimeoutExpired as exc: output = exc.output if isinstance(exc.output, str) else "" return { "command": command, "returncode": 124, "output": output, "timed_out": True, } except OSError as exc: return { "command": command, "returncode": 127, "output": f"failed to launch project evaluation: {exc}", "launch_error": True, } return { "command": command, "returncode": completed.returncode, "output": completed.stdout, } def render_markdown(summary: dict[str, Any]) -> str: lines = ["# Fold optimization harness", ""] for result in summary["evaluations"]: token = result.get("token_efficiency", {}).get("sarif_report", {}) estimate = token.get("estimated_tokens", {}).get("median", 0) density = token.get("findings_per_1k_estimated_tokens", {}).get("median", 0) lines.extend( [ f"## {result['manifest']}", "", f"- Status: **{result['status']}**", f"- Profiled median: `{result.get('profiled_median_seconds', 0)} s`", f"- Peak RSS: `{result.get('peak_rss_mb', 0)} MiB`", f"- SARIF estimate: `{estimate}` tokens (`ceil(utf8_bytes/4)`)", f"- Findings per 1K SARIF estimate: `{density}`", "", ] ) return "\n".join(lines) def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--fold", type=Path, default=DEFAULT_FOLD) parser.add_argument("--manifest", type=Path, action="append", default=[]) parser.add_argument("--out-dir", type=Path, default=Path("/tmp/fold-optimization")) parser.add_argument("--timeout", type=float, default=600.0) parser.add_argument("--max-report-tokens", type=int) args = parser.parse_args() manifests = args.manifest or [DEFAULT_MANIFEST] out_dir = args.out_dir.expanduser().resolve() out_dir.mkdir(parents=True, exist_ok=True) evaluations: list[dict[str, Any]] = [] failures: list[str] = [] for manifest in manifests: resolved_manifest = manifest.expanduser().resolve() run_dir = Path( tempfile.mkdtemp(prefix=f"{resolved_manifest.stem}-", dir=out_dir) ) execution = run_evaluation( resolved_manifest, args.fold.expanduser().resolve(), run_dir, args.timeout ) result: dict[str, Any] = { "manifest": str(resolved_manifest), "run_dir": str(run_dir), **execution, } if execution["returncode"] not in (0, 1): result["status"] = "failed" failures.append( f"{resolved_manifest.name} exited {execution['returncode']}" ) else: try: project_summary = read_json(run_dir / "summary.json") token_efficiency = artifact_token_efficiency(artifact_records(run_dir)) surface = project_summary.get("optimization_surface", {}) performance = project_summary.get("performance", {}).get( "profiled_serial", {} ) result.update( { "status": str(project_summary.get("status", "failed")), "profiled_median_seconds": performance.get( "wall_seconds", {} ).get("median", 0), "peak_rss_mb": surface.get("peak_rss_mb", 0), "optimization_surface": surface, "token_efficiency": token_efficiency, } ) estimate = token_efficiency["sarif_report"]["estimated_tokens"][ "median" ] if ( args.max_report_tokens is not None and estimate > args.max_report_tokens ): result["status"] = "failed" failures.append( f"{resolved_manifest.name} SARIF estimate {estimate} exceeds {args.max_report_tokens}" ) if result["status"] != "ok": failures.append( f"{resolved_manifest.name} project evaluation failed" ) except ValueError as exc: result["status"] = "failed" failures.append(f"{resolved_manifest.name}: {exc}") evaluations.append(result) summary = { "schema_version": 1, "status": "failed" if failures else "ok", "evaluations": evaluations, "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())