e30764fb19
Add negative_cases_from_universe to expand an explicit file-by-rule universe into labeled negatives (enabling TP/FP/TN/FN, not just TP/FP), plus project-eval harness updates in run_project_eval and the score/score_fp/quality-gate helpers.
261 lines
11 KiB
Python
261 lines
11 KiB
Python
"""Versioned project-evaluation manifest contract.
|
|
|
|
A project manifest declares *what to run* for the lane-generic
|
|
``run_project_eval.py`` runner: corpus, lane, detector scope, budgets, and
|
|
the optional evaluation inputs (gold, dispositions, perf baseline). Runners
|
|
execute manifests; this module owns validation and configuration
|
|
fingerprinting so those semantics do not 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``.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from collections.abc import Mapping
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from .dispositions import configuration_fingerprint
|
|
from .metrics import CLASSIFICATION_THRESHOLD_METRICS
|
|
|
|
PROJECT_MANIFEST_SCHEMA_VERSION = 1
|
|
LANES = frozenset({"source", "rust", "binary"})
|
|
MISSING_CORPUS_POLICIES = frozenset({"fail", "skip"})
|
|
|
|
|
|
def _nonempty_string(value: Any) -> bool:
|
|
return isinstance(value, str) and bool(value.strip())
|
|
|
|
|
|
def _nonnegative_number(value: Any) -> bool:
|
|
return (
|
|
isinstance(value, (int, float)) and not isinstance(value, bool) and value >= 0
|
|
)
|
|
|
|
|
|
def _positive_int(value: Any) -> bool:
|
|
return isinstance(value, int) and not isinstance(value, bool) and value >= 1
|
|
|
|
|
|
def _string_list(value: Any, *, allow_empty: bool = True) -> bool:
|
|
if not isinstance(value, list):
|
|
return False
|
|
if not allow_empty and not value:
|
|
return False
|
|
return all(_nonempty_string(item) for item in value)
|
|
|
|
|
|
def validate_project_manifest(document: Mapping[str, Any]) -> list[str]:
|
|
"""Return a list of validation errors (empty when the manifest is valid)."""
|
|
|
|
errors: list[str] = []
|
|
if document.get("schema_version") != PROJECT_MANIFEST_SCHEMA_VERSION:
|
|
errors.append(
|
|
"unsupported project manifest schema_version: "
|
|
f"{document.get('schema_version')}"
|
|
)
|
|
if not _nonempty_string(document.get("project")):
|
|
errors.append("project must be a non-empty string")
|
|
lane = document.get("lane")
|
|
if lane not in LANES:
|
|
errors.append(f"lane must be one of {sorted(LANES)}")
|
|
|
|
corpus = document.get("corpus")
|
|
if not isinstance(corpus, dict):
|
|
errors.append("corpus must be an object")
|
|
corpus = {}
|
|
if lane == "source" and not _nonempty_string(corpus.get("compile_db")):
|
|
errors.append("corpus.compile_db is required for the source lane")
|
|
if lane == "rust" and not _nonempty_string(corpus.get("path")):
|
|
errors.append("corpus.path is required for the rust lane")
|
|
if lane == "binary" and not _string_list(
|
|
corpus.get("binaries", []), allow_empty=False
|
|
):
|
|
errors.append(
|
|
"corpus.binaries must be a non-empty string list for the binary lane"
|
|
)
|
|
missing_corpus = corpus.get("missing_corpus", "fail")
|
|
if missing_corpus not in MISSING_CORPUS_POLICIES:
|
|
errors.append(
|
|
f"corpus.missing_corpus must be one of {sorted(MISSING_CORPUS_POLICIES)}"
|
|
)
|
|
|
|
analysis = document.get("analysis", {})
|
|
if not isinstance(analysis, dict):
|
|
errors.append("analysis must be an object when present")
|
|
analysis = {}
|
|
for field in ("detectors", "scopes", "extra_args"):
|
|
if field in analysis and not _string_list(analysis[field]):
|
|
errors.append(f"analysis.{field} must be a list of non-empty strings")
|
|
if "detectors" in analysis and "scopes" in analysis:
|
|
if analysis["detectors"] and analysis["scopes"]:
|
|
errors.append("analysis must not set both detectors and scopes")
|
|
for field in ("path_regex", "exclude_regex"):
|
|
if field in analysis and not isinstance(analysis[field], str):
|
|
errors.append(f"analysis.{field} must be a string")
|
|
if "rust_toolchain" in analysis and not _nonempty_string(
|
|
analysis["rust_toolchain"]
|
|
):
|
|
errors.append("analysis.rust_toolchain must be a non-empty string")
|
|
if "max_tus" in analysis and not (
|
|
isinstance(analysis["max_tus"], int)
|
|
and not isinstance(analysis["max_tus"], bool)
|
|
and analysis["max_tus"] >= 0
|
|
):
|
|
errors.append("analysis.max_tus must be an integer >= 0")
|
|
if "processes" in analysis and not _positive_int(analysis["processes"]):
|
|
errors.append("analysis.processes must be an integer >= 1")
|
|
for field in ("budget_seconds", "per_query_ms", "per_tu_timeout"):
|
|
if field in analysis and not _nonnegative_number(analysis[field]):
|
|
errors.append(f"analysis.{field} must be a number >= 0")
|
|
|
|
evaluation = document.get("evaluation", {})
|
|
if not isinstance(evaluation, dict):
|
|
errors.append("evaluation must be an object when present")
|
|
evaluation = {}
|
|
if "repeats" in evaluation and not _positive_int(evaluation["repeats"]):
|
|
errors.append("evaluation.repeats must be an integer >= 1")
|
|
for field in ("gold", "dispositions", "baseline"):
|
|
if field in evaluation and not _nonempty_string(evaluation[field]):
|
|
errors.append(f"evaluation.{field} must be a non-empty string")
|
|
for field in ("wall_regression_pct", "rss_regression_pct"):
|
|
if field in evaluation and not _nonnegative_number(evaluation[field]):
|
|
errors.append(f"evaluation.{field} must be a number >= 0")
|
|
gold_thresholds = evaluation.get("gold_thresholds")
|
|
if gold_thresholds is not None:
|
|
if "gold" not in evaluation:
|
|
errors.append("evaluation.gold_thresholds requires evaluation.gold")
|
|
if not isinstance(gold_thresholds, dict):
|
|
errors.append("evaluation.gold_thresholds must be an object")
|
|
else:
|
|
for key in CLASSIFICATION_THRESHOLD_METRICS:
|
|
if key not in gold_thresholds:
|
|
continue
|
|
value = gold_thresholds[key]
|
|
if not _nonnegative_number(value) or float(value) > 1.0:
|
|
errors.append(
|
|
f"evaluation.gold_thresholds.{key} must be a number in [0, 1]"
|
|
)
|
|
per_rule_thresholds = evaluation.get("per_rule_gold_thresholds")
|
|
if per_rule_thresholds is not None:
|
|
if "gold" not in evaluation:
|
|
errors.append(
|
|
"evaluation.per_rule_gold_thresholds requires evaluation.gold"
|
|
)
|
|
if not isinstance(per_rule_thresholds, dict):
|
|
errors.append("evaluation.per_rule_gold_thresholds must be an object")
|
|
else:
|
|
for rule, thresholds in per_rule_thresholds.items():
|
|
if not _nonempty_string(rule) or not isinstance(thresholds, dict):
|
|
errors.append(
|
|
"evaluation.per_rule_gold_thresholds must map rule names to objects"
|
|
)
|
|
continue
|
|
for key in CLASSIFICATION_THRESHOLD_METRICS:
|
|
if key not in thresholds:
|
|
continue
|
|
value = thresholds[key]
|
|
if not _nonnegative_number(value) or float(value) > 1.0:
|
|
errors.append(
|
|
f"evaluation.per_rule_gold_thresholds.{rule}.{key} "
|
|
"must be a number in [0, 1]"
|
|
)
|
|
if "experimental_rules" in evaluation and not _string_list(
|
|
evaluation["experimental_rules"]
|
|
):
|
|
errors.append(
|
|
"evaluation.experimental_rules must be a list of non-empty strings"
|
|
)
|
|
stability = evaluation.get("stability")
|
|
if stability is not None:
|
|
if not isinstance(stability, dict):
|
|
errors.append("evaluation.stability must be an object")
|
|
else:
|
|
if "max_churn" in stability and not (
|
|
isinstance(stability["max_churn"], int)
|
|
and not isinstance(stability["max_churn"], bool)
|
|
and stability["max_churn"] >= 0
|
|
):
|
|
errors.append("evaluation.stability.max_churn must be an integer >= 0")
|
|
if "min_pairwise_jaccard" in stability:
|
|
value = stability["min_pairwise_jaccard"]
|
|
if not _nonnegative_number(value) or float(value) > 1.0:
|
|
errors.append(
|
|
"evaluation.stability.min_pairwise_jaccard must be a number in [0, 1]"
|
|
)
|
|
budgets = evaluation.get("budgets")
|
|
if budgets is not None:
|
|
if not isinstance(budgets, dict):
|
|
errors.append("evaluation.budgets must be an object")
|
|
else:
|
|
for field in (
|
|
"max_wall_seconds",
|
|
"max_peak_rss_mb",
|
|
"max_frontend_extraction_ms",
|
|
"max_output_bytes",
|
|
):
|
|
if field in budgets and not _nonnegative_number(budgets[field]):
|
|
errors.append(f"evaluation.budgets.{field} must be a number >= 0")
|
|
if "z3_timeout_budget" in evaluation and not (
|
|
isinstance(evaluation["z3_timeout_budget"], int)
|
|
and not isinstance(evaluation["z3_timeout_budget"], bool)
|
|
and evaluation["z3_timeout_budget"] >= 0
|
|
):
|
|
errors.append("evaluation.z3_timeout_budget must be an integer >= 0")
|
|
if "min_functions" in evaluation and not (
|
|
isinstance(evaluation["min_functions"], int)
|
|
and not isinstance(evaluation["min_functions"], bool)
|
|
and evaluation["min_functions"] >= 0
|
|
):
|
|
errors.append("evaluation.min_functions must be an integer >= 0")
|
|
if "allow_finding_drift" in evaluation and not isinstance(
|
|
evaluation["allow_finding_drift"], bool
|
|
):
|
|
errors.append("evaluation.allow_finding_drift must be a boolean")
|
|
return errors
|
|
|
|
|
|
def manifest_configuration(manifest: Mapping[str, Any]) -> dict[str, Any]:
|
|
"""Extract the analysis-relevant configuration used for fingerprinting.
|
|
|
|
The fingerprint identifies the analysis configuration, not the evaluation
|
|
policy: two runs with different repeats/baselines but identical analysis
|
|
settings share a configuration fingerprint.
|
|
"""
|
|
|
|
analysis = manifest.get("analysis", {})
|
|
return {
|
|
"lane": manifest.get("lane", ""),
|
|
"detectors": sorted(analysis.get("detectors", [])),
|
|
"scopes": sorted(analysis.get("scopes", [])),
|
|
"path_regex": analysis.get("path_regex", ""),
|
|
"exclude_regex": analysis.get("exclude_regex", ""),
|
|
"max_tus": int(analysis.get("max_tus", 0)),
|
|
"processes": int(analysis.get("processes", 1)),
|
|
"budget_seconds": analysis.get("budget_seconds", 0),
|
|
"per_query_ms": analysis.get("per_query_ms", 0),
|
|
"per_tu_timeout": analysis.get("per_tu_timeout", 0),
|
|
"rust_toolchain": analysis.get("rust_toolchain", ""),
|
|
"extra_args": list(analysis.get("extra_args", [])),
|
|
}
|
|
|
|
|
|
def manifest_fingerprint(manifest: Mapping[str, Any]) -> str:
|
|
"""Stable SHA-256 over the analysis configuration of a manifest."""
|
|
|
|
return configuration_fingerprint(manifest_configuration(manifest))
|
|
|
|
|
|
def load_project_manifest(path: Path) -> dict[str, Any]:
|
|
"""Load and validate a project manifest; raise ValueError on any error."""
|
|
|
|
document = json.loads(Path(path).read_text(encoding="utf-8"))
|
|
if not isinstance(document, dict):
|
|
raise ValueError(f"project manifest must be a JSON object: {path}")
|
|
errors = validate_project_manifest(document)
|
|
if errors:
|
|
raise ValueError(f"invalid project manifest {path}: " + "; ".join(errors))
|
|
return document
|