hack-house/hh/scripts/bench/olympics/scoring.py

"""Deterministic scoring for one (team, event).

M1 covers the verifiable-first layer of SPEC §9: correctness gates everything,
speed and cost are normalized deterministically, quality/collaboration are left
to the judge (M4) and reported as ``None`` here. The composite uses a weight
*profile* (same idea as ``bench/workflows.json``) so re-ranking never re-runs a
model.

Single-sample M1 note: with one implementation attempt, ``pass@1`` is 0/1 and
``pass^k`` (worst-case reliability) equals it; both fields are emitted so the
schema is stable when M2 adds repeated samples.
"""

from __future__ import annotations

from dataclasses import dataclass, field

# Default weight profile. Quality/collaboration weights are reserved for the
# judge layer (M4); with judge scores absent they contribute 0 and the profile
# is renormalized over the available axes so M1 scores stay in [0, 1].
PROFILES: dict[str, dict[str, float]] = {
    "balanced": {"correctness": 0.6, "speed": 0.2, "quality": 0.1,
                 "collaboration": 0.1},
    "correctness": {"correctness": 0.9, "speed": 0.1, "quality": 0.0,
                    "collaboration": 0.0},
    "speed": {"correctness": 0.5, "speed": 0.5, "quality": 0.0,
              "collaboration": 0.0},
}


@dataclass
class EventOutcome:
    """What the loop produces for one team at one event (deterministic facts)."""
    team: str
    challenge: str
    submitted: bool
    correct: bool                 # hidden tests all passed
    rounds: int                   # deliberate/implement rounds consumed
    wall_clock_s: float
    tokens: dict = field(default_factory=lambda: {"in": 0, "out": 0})
    public_passed: bool = False   # public tests green before SUBMIT
    rounds_to_green: int | None = None  # round at which public went green
    penalties: list = field(default_factory=list)  # [{"kind","detail"}]
    quality: float | None = None        # filled by judge (M4)
    collaboration: float | None = None  # filled by judge (M4)
    # bridge diagnostics: separate a broken product path from a model error, and
    # record the model's true ability via an un-stripped control generation.
    defect_class: str | None = None      # e.g. "product-path-defect:IndentationError"
    model_baseline_correct: bool | None = None


def _speed_score(o: EventOutcome, budget: dict) -> float:
    """1.0 for instant green, decaying toward 0 as rounds approach the cap.
    Only meaningful if the team got correct code; uncorrect teams get 0."""
    if not o.correct:
        return 0.0
    cap = max(1, int(budget.get("max_rounds", 6)))
    used = o.rounds_to_green if o.rounds_to_green is not None else o.rounds
    used = max(1, min(used, cap))
    return round(1.0 - (used - 1) / cap, 4)


def _penalty_total(o: EventOutcome) -> float:
    # each penalty shaves a flat slice; capped so a score never goes negative.
    return min(0.5, 0.1 * len(o.penalties))


def score_event(o: EventOutcome, *, profile: str = "balanced",
                budget: dict | None = None) -> dict:
    budget = budget or {}
    w = PROFILES.get(profile, PROFILES["balanced"])
    axes = {
        "correctness": 1.0 if o.correct else 0.0,
        "speed": _speed_score(o, budget),
        "quality": o.quality,            # may be None (no judge yet)
        "collaboration": o.collaboration,
    }
    # renormalize weights over axes that actually have a value this run
    active = {k: w[k] for k, v in axes.items() if v is not None and w.get(k, 0)}
    wsum = sum(active.values()) or 1.0
    composite = sum(active[k] / wsum * axes[k] for k in active)
    composite = round(max(0.0, composite - _penalty_total(o)), 4)

    tok = o.tokens.get("in", 0) + o.tokens.get("out", 0)
    return {
        "team": o.team, "challenge": o.challenge,
        "submitted": o.submitted, "correct": o.correct,
        "composite": composite,
        "axes": {k: (round(v, 4) if v is not None else None)
                 for k, v in axes.items()},
        "pass@1": 1.0 if o.correct else 0.0,
        "pass^1": 1.0 if o.correct else 0.0,   # worst-case == pass@1 at n=1
        "rounds": o.rounds, "rounds_to_green": o.rounds_to_green,
        "wall_clock_s": round(o.wall_clock_s, 1),
        "tokens": dict(o.tokens), "total_tokens": tok,
        "cost_normalized": round(composite / tok, 8) if tok else None,
        "penalties": list(o.penalties),
        "defect_class": o.defect_class,
        "model_baseline_correct": o.model_baseline_correct,
        "profile": profile,
    }


def print_score(s: dict) -> None:
    print("=" * 72)
    print(f"score · team={s['team']} · challenge={s['challenge']} "
          f"· profile={s['profile']}")
    print("-" * 72)
    g = "✓" if s["correct"] else "✗"
    green = f"(green@{s['rounds_to_green']})" if s["rounds_to_green"] else ""
    print(f"  correct={g}  composite={s['composite']:.3f}  "
          f"pass@1={s['pass@1']:.0f}  rounds={s['rounds']}{green}"
          f"  {s['wall_clock_s']:.1f}s")
    print("  axes: " + "  ".join(
        f"{k}={'—' if v is None else f'{v:.2f}'}" for k, v in s["axes"].items()))
    print(f"  tokens={s['total_tokens']}  "
          f"cost_norm={s['cost_normalized']}")
    if s.get("defect_class") or s.get("model_baseline_correct") is not None:
        base = s.get("model_baseline_correct")
        base_str = "—" if base is None else ("✓" if base else "✗")
        print(f"  defect={s.get('defect_class') or 'none'}  "
              f"model_baseline={base_str}")
    if s["penalties"]:
        print("  penalties:")
        for p in s["penalties"]:
            print(f"    - {p}")
    print("=" * 72)