af66058f3a
Backlog #5 (operator-confirmed). chromium_source.json and sage_rust.json were validated but unwired. This adds: - reviewed dispositions (5 parallel review subagents): chromium 11 (10 FP/1 accepted-risk — deliberate modular arithmetic in base/file_path.cc); sage 2714 (9 actionable uncontained path-join reads in skill install/daemon library/model-downloader; 890 FP; 1815 accepted-risk quality bulk) - wall/RSS baselines for both corpora Measured both corpora produce finding drift run-to-run (chromium Z3 fingerprint drifts; sage +5 findings) and chromium's wall swings up to ~2x under load — so they are wired as STRESS/DIVERSITY corpora: a new additive evaluation.max_new_findings override lets a manifest declare its finding-drift tolerance (CLI default unchanged for fold/yams), allow_finding_drift is on, and chromium's wall/RSS thresholds are relaxed. Final: both evals pass. Also fixes a pre-existing /tmp-literal default (now platform tempdir) in run_project_eval.py.
1618 lines
60 KiB
Python
1618 lines
60 KiB
Python
#!/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())
|