diff --git a/analysis/benchmark_report.py b/analysis/benchmark_report.py index fbf7592..0da7e5c 100644 --- a/analysis/benchmark_report.py +++ b/analysis/benchmark_report.py @@ -191,6 +191,13 @@ class ProgressMetricsSummary: failure_rate_p75: Optional[float] +@dataclass +class RelativeScoreSummary: + fuzzer: str + relscore: Optional[float] + relcov: Optional[float] + + def parse_optional_float(value: str | None) -> Optional[float]: if value is None: return None @@ -272,6 +279,66 @@ def load_progress_metrics_summary(path: Path) -> Dict[str, ProgressMetricsSummar return rows +def _find_column(fieldnames: List[str], candidates: List[str]) -> Optional[str]: + normalized = {name.strip().lower().replace("_", ""): name for name in fieldnames} + for candidate in candidates: + found = normalized.get(candidate.strip().lower().replace("_", "")) + if found is not None: + return found + return None + + +def load_relative_scores(path: Path) -> Dict[str, RelativeScoreSummary]: + if not path.exists(): + return {} + rows: Dict[str, RelativeScoreSummary] = {} + with path.open("r", newline="", encoding="utf-8") as handle: + reader = csv.DictReader(handle) + fieldnames = reader.fieldnames or [] + fuzzer_col = _find_column( + fieldnames, + ["fuzzer", "fuzzer_label", "approach", "tool", "name"], + ) + relscore_col = _find_column(fieldnames, ["relscore", "relscores", "relative_score"]) + relcov_col = _find_column( + fieldnames, + ["relcov", "relative_coverage", "coverage_score", "covered_edges"], + ) + if fuzzer_col is None: + die(f"missing fuzzer column in relative score CSV {path}") + if relscore_col is None and relcov_col is None: + die( + f"missing relscore/relscores or relcov/covered_edges column in " + f"relative score CSV {path}" + ) + + raw_rows = [row for row in reader] + coverage_values = [ + value + for value in ( + parse_optional_float(row.get(relcov_col)) if relcov_col else None + for row in raw_rows + ) + if value is not None and math.isfinite(value) + ] + max_coverage_value = max(coverage_values, default=0.0) + relcov_from_covered_edges = relcov_col == _find_column(fieldnames, ["covered_edges"]) + + for row in raw_rows: + fuzzer = str(row.get(fuzzer_col, "")).strip() + if not fuzzer: + continue + relcov = parse_optional_float(row.get(relcov_col)) if relcov_col else None + if relcov_from_covered_edges: + relcov = _relative_to_best(relcov, max_coverage_value) + rows[fuzzer] = RelativeScoreSummary( + fuzzer=fuzzer, + relscore=parse_optional_float(row.get(relscore_col)) if relscore_col else None, + relcov=relcov, + ) + return rows + + def load_metric_samples_csv(path: Path, value_columns: List[str]) -> pd.DataFrame: if not path.exists(): cols = ["fuzzer", "series_id", "time_hours", *value_columns] @@ -447,6 +514,185 @@ def append_progress_metrics_section( lines.append("") +def _relative_to_best(value: Optional[float], best: float) -> Optional[float]: + if value is None or not math.isfinite(value) or best <= 0: + return None + return value / best + + +def build_relative_score_summaries( + *, + metrics: List[FuzzerMetrics], + progress_metrics_by_fuzzer: Dict[str, ProgressMetricsSummary], + relative_scores_by_fuzzer: Dict[str, RelativeScoreSummary] | None = None, +) -> Dict[str, RelativeScoreSummary]: + summaries: Dict[str, RelativeScoreSummary] = { + name: RelativeScoreSummary( + fuzzer=score.fuzzer, + relscore=score.relscore, + relcov=score.relcov, + ) + for name, score in (relative_scores_by_fuzzer or {}).items() + } + + best_final = max((float(metric.final_p50) for metric in metrics), default=0.0) + best_cov = max( + ( + float(row.coverage_p50) + for row in progress_metrics_by_fuzzer.values() + if row.coverage_p50 is not None and math.isfinite(row.coverage_p50) + ), + default=0.0, + ) + + for metric in metrics: + current = summaries.get(metric.fuzzer) + relscore = current.relscore if current else None + relcov = current.relcov if current else None + if relscore is None: + relscore = _relative_to_best(float(metric.final_p50), best_final) + if relcov is None: + progress = progress_metrics_by_fuzzer.get(metric.fuzzer) + relcov = _relative_to_best(progress.coverage_p50 if progress else None, best_cov) + summaries[metric.fuzzer] = RelativeScoreSummary( + fuzzer=metric.fuzzer, + relscore=relscore, + relcov=relcov, + ) + + return summaries + + +def fmt_relative_score(value: Optional[float]) -> str: + if value is None or not math.isfinite(value): + return "n/a" + return f"{value:.3f}" + + +def append_relative_scoreboard( + lines: List[str], + relative_scores_by_fuzzer: Dict[str, RelativeScoreSummary], + fuzzer_order: List[str], +) -> None: + if not relative_scores_by_fuzzer: + return + + def sort_key(score: RelativeScoreSummary) -> Tuple[float, float, str]: + relscore = ( + score.relscore + if score.relscore is not None and math.isfinite(score.relscore) + else -1.0 + ) + relcov = ( + score.relcov + if score.relcov is not None and math.isfinite(score.relcov) + else -1.0 + ) + return (relscore, relcov, score.fuzzer) + + ordered = sorted(relative_scores_by_fuzzer.values(), key=sort_key, reverse=True) + best_relscore = max( + ( + score.relscore + for score in ordered + if score.relscore is not None and math.isfinite(score.relscore) + ), + default=None, + ) + best_relcov = max( + ( + score.relcov + for score in ordered + if score.relcov is not None and math.isfinite(score.relcov) + ), + default=None, + ) + relscore_leaders = [ + score.fuzzer + for score in ordered + if best_relscore is not None + and score.relscore is not None + and math.isclose(score.relscore, best_relscore) + ] + relcov_leaders = [ + score.fuzzer + for score in ordered + if best_relcov is not None + and score.relcov is not None + and math.isclose(score.relcov, best_relcov) + ] + + lines.append("## Fuzzer scoreboard (higher is better)") + lines.append( + "`relscore` orders approaches by coverage value: edges that fewer approaches reach " + "are worth more, and an approach gets more credit when it reaches those edges " + "consistently across non-empty trials. `relcov` is directional coverage overlap: " + "`relcov(A, B)` is the share of B's total reached coverage that A reaches. In this " + "scoreboard, higher values are better within each column; a normalized `relcov` of " + "1.000 means the largest observed coverage total for the selected report input." + ) + if relscore_leaders: + lines.append( + f"- Best relative score: **{', '.join(relscore_leaders)}** " + f"({fmt_relative_score(best_relscore)} relscore)" + ) + if relcov_leaders: + lines.append( + f"- Best coverage score: **{', '.join(relcov_leaders)}** " + f"({fmt_relative_score(best_relcov)} relcov)" + ) + lines.append("") + + order_index = {fuzzer: idx for idx, fuzzer in enumerate(fuzzer_order)} + ordered = sorted( + ordered, + key=lambda score: ( + -( + score.relscore + if score.relscore is not None and math.isfinite(score.relscore) + else -1.0 + ), + -( + score.relcov + if score.relcov is not None and math.isfinite(score.relcov) + else -1.0 + ), + order_index.get(score.fuzzer, len(order_index)), + score.fuzzer, + ), + ) + header = ["Rank", "Fuzzer", "relscore", "relcov", "Readout"] + lines.append("| " + " | ".join(header) + " |") + lines.append("|" + "|".join(["---"] * len(header)) + "|") + for rank, score in enumerate(ordered, start=1): + if score.relscore is not None and score.relcov is not None: + readout = ( + "best overall" + if score.fuzzer in relscore_leaders and score.fuzzer in relcov_leaders + else "compare both scores" + ) + elif score.relscore is not None: + readout = "relative score only" + elif score.relcov is not None: + readout = "coverage score only" + else: + readout = "no relative scores" + lines.append( + "| " + + " | ".join( + [ + str(rank), + score.fuzzer, + fmt_relative_score(score.relscore), + fmt_relative_score(score.relcov), + readout, + ] + ) + + " |" + ) + lines.append("") + + def nan_percentile_rows(arr: np.ndarray, percentile_value: float) -> np.ndarray: out = np.full(arr.shape[0], np.nan, dtype=float) for idx, row in enumerate(arr): @@ -1075,6 +1321,7 @@ def write_report( outpath: Path, throughput_by_fuzzer: Dict[str, ThroughputSummary] | None = None, progress_metrics_by_fuzzer: Dict[str, ProgressMetricsSummary] | None = None, + relative_scores_by_fuzzer: Dict[str, RelativeScoreSummary] | None = None, stat_results: Optional[List[PairwiseResult]] = None, stat_warnings: Optional[List[str]] = None, alpha: float = 0.05, @@ -1095,6 +1342,16 @@ def write_report( ) lines.append("") + append_relative_scoreboard( + lines, + build_relative_score_summaries( + metrics=metrics, + progress_metrics_by_fuzzer=progress_metrics_by_fuzzer or {}, + relative_scores_by_fuzzer=relative_scores_by_fuzzer or {}, + ), + fuzzer_order=[metric.fuzzer for metric in metrics], + ) + lines.append("## Bugs found at fixed time budgets (median [IQR])") header = ["Fuzzer", "Runs"] + [f"{t:g}h" for t in checkpoints] lines.append("| " + " | ".join(header) + " |") @@ -1288,6 +1545,12 @@ def main() -> int: default=None, help="Optional progress metrics samples CSV for time-series charts.", ) + parser.add_argument( + "--relative-scores-csv", + type=Path, + default=None, + help="Optional CSV with fuzzer plus relscore/relscores and relcov columns for the report scoreboard.", + ) parser.add_argument( "--additional-metrics-summary-csv", dest="progress_metrics_summary_csv", @@ -1316,6 +1579,11 @@ def main() -> int: if args.progress_metrics_summary_csv is not None else {} ) + relative_scores_by_fuzzer = ( + load_relative_scores(args.relative_scores_csv) + if args.relative_scores_csv is not None + else {} + ) throughput_samples_df = ( load_metric_samples_csv(args.throughput_samples_csv, THROUGHPUT_SAMPLE_VALUE_COLS) if args.throughput_samples_csv is not None @@ -1441,6 +1709,7 @@ def main() -> int: outpath=report_outdir / "REPORT.md", throughput_by_fuzzer=throughput_by_fuzzer, progress_metrics_by_fuzzer=progress_metrics_by_fuzzer, + relative_scores_by_fuzzer=relative_scores_by_fuzzer, stat_results=stat_results, stat_warnings=stat_warnings, ) diff --git a/analysis/tests/test_benchmark_report.py b/analysis/tests/test_benchmark_report.py index 4848b90..f72b058 100644 --- a/analysis/tests/test_benchmark_report.py +++ b/analysis/tests/test_benchmark_report.py @@ -192,6 +192,111 @@ def test_write_report_includes_progress_metrics_section(self): self.assertNotIn("Favored", report) self.assertNotIn("Failure-rate", report) + def test_write_report_includes_relative_scoreboard_from_metrics(self): + metrics = [ + _make_metrics("echidna", [8, 10], final_p50=9), + _make_metrics("medusa", [3, 5], final_p50=4), + ] + progress = { + "echidna": benchmark_report.ProgressMetricsSummary( + fuzzer="echidna", + runs=2, + seqps_runs=0, + coverage_runs=2, + corpus_runs=0, + favored_runs=0, + failure_rate_runs=0, + seqps_p50=None, + seqps_p25=None, + seqps_p75=None, + coverage_p50=300.0, + coverage_p25=280.0, + coverage_p75=320.0, + corpus_p50=None, + corpus_p25=None, + corpus_p75=None, + favored_p50=None, + favored_p25=None, + favored_p75=None, + failure_rate_p50=None, + failure_rate_p25=None, + failure_rate_p75=None, + ), + "medusa": benchmark_report.ProgressMetricsSummary( + fuzzer="medusa", + runs=2, + seqps_runs=0, + coverage_runs=2, + corpus_runs=0, + favored_runs=0, + failure_rate_runs=0, + seqps_p50=None, + seqps_p25=None, + seqps_p75=None, + coverage_p50=150.0, + coverage_p25=140.0, + coverage_p75=160.0, + corpus_p50=None, + corpus_p25=None, + corpus_p75=None, + favored_p50=None, + favored_p25=None, + favored_p75=None, + failure_rate_p50=None, + failure_rate_p25=None, + failure_rate_p75=None, + ), + } + + with tempfile.TemporaryDirectory() as tmp: + outpath = Path(tmp) / "REPORT.md" + benchmark_report.write_report( + metrics=metrics, + budget=1.0, + checkpoints=[1.0], + ks=[1], + outpath=outpath, + progress_metrics_by_fuzzer=progress, + ) + report = outpath.read_text(encoding="utf-8") + + self.assertIn("## Fuzzer scoreboard (higher is better)", report) + self.assertIn("`relscore` orders approaches by coverage value", report) + self.assertIn("`relcov` is directional coverage overlap", report) + self.assertIn("| 1 | echidna | 1.000 | 1.000 | best overall |", report) + self.assertIn("| 2 | medusa | 0.444 | 0.500 | compare both scores |", report) + + def test_relative_scores_csv_overrides_scoreboard_values(self): + with tempfile.TemporaryDirectory() as tmp: + tmp_dir = Path(tmp) + rel_csv = tmp_dir / "relative.csv" + rel_csv.write_text( + "\n".join( + [ + "approach,relscores,covered_edges", + "echidna,0.75,300", + "medusa,1.0,150", + ] + ) + + "\n", + encoding="utf-8", + ) + relative = benchmark_report.load_relative_scores(rel_csv) + outpath = tmp_dir / "REPORT.md" + benchmark_report.write_report( + metrics=[_make_metrics("echidna", [10]), _make_metrics("medusa", [1])], + budget=1.0, + checkpoints=[1.0], + ks=[1], + outpath=outpath, + relative_scores_by_fuzzer=relative, + ) + report = outpath.read_text(encoding="utf-8") + + self.assertIn("Best relative score: **medusa** (1.000 relscore)", report) + self.assertIn("Best coverage score: **echidna** (1.000 relcov)", report) + self.assertIn("| 1 | medusa | 1.000 | 0.500 | compare both scores |", report) + def test_cli_clamps_checkpoints_to_budget(self): with tempfile.TemporaryDirectory() as tmp: tmp_dir = Path(tmp)