b0b51f1e71
- fail closed on MCP input and compile database policy - add shared scan catalogs, finding queries, and tiered gates
231 lines
8.4 KiB
Python
231 lines
8.4 KiB
Python
#!/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())
|