captchaforge 0.2.39

#!/usr/bin/env python3
"""Diff two captchaforge bench JSON reports + refuse promotion when
any vendor regressed by more than --max-regression-pp percentage
points.

The captchaforge-bench reporter writes `bench-results-report.json`;
operators run the bench against the OLD model, swap the trained
weights into Ollama, run the bench again, then compare. This
script automates the comparison and exits non-zero on regression
so CI can gate model promotion.

Usage:
    bench_compare.py baseline.json candidate.json \\
                     [--max-regression-pp 2.0] \\
                     [--require-improvement-pp 1.0]
"""
from __future__ import annotations

import argparse
import json
import logging
import sys
from pathlib import Path

LOG = logging.getLogger("captchaforge.bench_compare")


def parse_args(argv: list[str]) -> argparse.Namespace:
    p = argparse.ArgumentParser(
        prog="bench_compare.py",
        description="Diff two captchaforge bench JSON reports.",
        formatter_class=argparse.ArgumentDefaultsHelpFormatter,
    )
    p.add_argument("baseline", type=Path, help="OLD bench report JSON.")
    p.add_argument("candidate", type=Path, help="NEW bench report JSON.")
    p.add_argument(
        "--max-regression-pp",
        type=float,
        default=2.0,
        help="Refuse promotion when any vendor regresses by more than this many "
        "percentage points.",
    )
    p.add_argument(
        "--require-improvement-pp",
        type=float,
        default=0.0,
        help="Require the AGGREGATE pass-rate to improve by at least this many "
        "percentage points. 0 = any improvement (or even no change) passes; "
        "1.0 = at least 1pp uplift required.",
    )
    p.add_argument(
        "--vendor-key",
        default="fixture",
        help="JSON key naming each fixture (default: 'fixture').",
    )
    p.add_argument(
        "--rate-key",
        default="success_rate_pct",
        help="JSON key carrying the per-fixture success rate (default: 'success_rate_pct').",
    )
    return p.parse_args(argv)


def load_rates(path: Path, vendor_key: str, rate_key: str) -> dict[str, float]:
    """Read a bench JSON report into a {vendor: success_rate_pct} dict."""
    raw = json.loads(path.read_text(encoding="utf-8"))
    # captchaforge-bench's report shape: top-level dict with a
    # `suites` array, each suite carries `results` per fixture.
    out: dict[str, float] = {}
    suites = raw.get("suites") or raw.get("results") or [raw]
    for suite in suites:
        results = suite.get("results") or suite.get("fixtures") or []
        for r in results:
            vendor = r.get(vendor_key)
            rate = r.get(rate_key)
            if vendor is None or rate is None:
                continue
            out[str(vendor)] = float(rate)
    return out


def main(argv: list[str]) -> int:
    args = parse_args(argv)
    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s %(levelname)s %(name)s %(message)s",
    )

    baseline = load_rates(args.baseline, args.vendor_key, args.rate_key)
    candidate = load_rates(args.candidate, args.vendor_key, args.rate_key)
    LOG.info("baseline: %d fixtures", len(baseline))
    LOG.info("candidate: %d fixtures", len(candidate))

    common = sorted(set(baseline) & set(candidate))
    only_in_base = sorted(set(baseline) - set(candidate))
    only_in_cand = sorted(set(candidate) - set(baseline))

    if only_in_base:
        LOG.warning("fixtures missing from candidate: %s", only_in_base)
    if only_in_cand:
        LOG.info("new fixtures in candidate: %s", only_in_cand)

    regressions: list[tuple[str, float, float, float]] = []
    improvements: list[tuple[str, float, float, float]] = []
    for fix in common:
        delta = candidate[fix] - baseline[fix]
        if delta < 0:
            regressions.append((fix, baseline[fix], candidate[fix], delta))
        elif delta > 0:
            improvements.append((fix, baseline[fix], candidate[fix], delta))

    print(f"{'FIXTURE':<40} {'BEFORE':>8} {'AFTER':>8} {'DELTA':>8}")
    print("-" * 68)
    for fix in common:
        before = baseline[fix]
        after = candidate[fix]
        delta = after - before
        marker = "✗" if delta < -args.max_regression_pp else (
            "✓" if delta > 0 else " "
        )
        print(f"{marker} {fix:<38} {before:>7.1f}% {after:>7.1f}% {delta:+7.1f}pp")

    aggregate_before = (
        sum(baseline.values()) / len(baseline) if baseline else 0.0
    )
    aggregate_after = (
        sum(candidate.values()) / len(candidate) if candidate else 0.0
    )
    aggregate_delta = aggregate_after - aggregate_before
    print("-" * 68)
    print(
        f"  AGGREGATE                              "
        f"{aggregate_before:>7.1f}% {aggregate_after:>7.1f}% {aggregate_delta:+7.1f}pp"
    )

    bad_regressions = [r for r in regressions if r[3] < -args.max_regression_pp]
    if bad_regressions:
        print()
        print("REGRESSIONS exceeding --max-regression-pp:")
        for fix, before, after, delta in bad_regressions:
            print(f"  - {fix}: {before:.1f}% -> {after:.1f}% ({delta:+.1f}pp)")

    if bad_regressions:
        LOG.error(
            "%d vendor(s) regressed by >%.1fpp, refusing promotion",
            len(bad_regressions),
            args.max_regression_pp,
        )
        return 2

    if aggregate_delta < args.require_improvement_pp:
        LOG.error(
            "aggregate pass-rate moved %+.1fpp; --require-improvement-pp=%.1f not met",
            aggregate_delta,
            args.require_improvement_pp,
        )
        return 3

    LOG.info(
        "promotion OK: aggregate %+.1fpp, %d improvements, %d acceptable regressions",
        aggregate_delta,
        len(improvements),
        len(regressions) - len(bad_regressions),
    )
    return 0


if __name__ == "__main__":
    sys.exit(main(sys.argv[1:]))