"""Load leaderboard entries from liderboard/results.""" from __future__ import annotations import json from pathlib import Path from typing import Any from src.bench_config import load_bench_config from src.export_results import ( _cost_from_payload, _recorded_cost_from_payload, _total_agent_steps_from_payload, raw_results_path_for, ui_classes_from_payload, ) from src.i18n import t from src.models import LeaderboardEntry, reconcile_success_metrics from src.token_display import reconcile_entry_usage_totals, token_usage_from_run _RAW_CACHE: dict[str, tuple[float, dict[str, Any]]] = {} def space_root(root: Path | None = None) -> Path: return root or Path(__file__).resolve().parent.parent def results_dir(root: Path | None = None) -> Path: return space_root(root) / "results" def _resolve_raw_path(entry: LeaderboardEntry, root: Path | None = None) -> Path | None: base = space_root(root) if entry.raw_results_path: raw_path = base / entry.raw_results_path if raw_path.is_file(): return raw_path fallback = raw_results_path_for(entry, base) if fallback.is_file(): return fallback return None def load_raw_results(entry: LeaderboardEntry, root: Path | None = None) -> dict[str, Any] | None: raw_path = _resolve_raw_path(entry, root) if raw_path is None: return None cache_key = str(raw_path.resolve()) try: mtime = raw_path.stat().st_mtime except OSError: mtime = 0.0 cached = _RAW_CACHE.get(cache_key) if cached is not None and cached[0] == mtime: return cached[1] payload = json.loads(raw_path.read_text(encoding="utf-8")) _RAW_CACHE[cache_key] = (mtime, payload) return payload def _enrich_entry_from_raw(entry: LeaderboardEntry, raw: dict[str, Any]) -> None: run = raw.get("run") or {} if not entry.ui_classes: entry.ui_classes = ui_classes_from_payload(raw) if not entry.token_usage_by_model: entry.token_usage_by_model = token_usage_from_run(run) if entry.total_cost_usd is None or entry.total_cost_usd <= 0: recorded = _recorded_cost_from_payload(raw) if recorded is not None: total_cost, avg_cost = recorded else: total_cost, avg_cost = _cost_from_payload(raw) if total_cost is not None and total_cost > 0: entry.total_cost_usd = total_cost entry.avg_cost_per_task_usd = avg_cost if entry.total_tokens is None and isinstance(run.get("total_tokens"), (int, float)): entry.total_tokens = int(run["total_tokens"]) if entry.avg_tokens_per_task is None and isinstance(run.get("avg_tokens_per_task"), (int, float)): entry.avg_tokens_per_task = float(run["avg_tokens_per_task"]) if entry.total_duration_seconds is None and isinstance(run.get("total_duration_seconds"), (int, float)): entry.total_duration_seconds = float(run["total_duration_seconds"]) if entry.total_agent_steps is None: total_steps = _total_agent_steps_from_payload(raw) if total_steps is not None: entry.total_agent_steps = total_steps slots = run.get("ouroboros_model_slots") if isinstance(slots, dict) and slots and not entry.ouroboros_model_slots: entry.ouroboros_model_slots = { str(slot): [str(model) for model in models] for slot, models in slots.items() if isinstance(models, list) and models } reconcile_entry_usage_totals(entry) def _conditions_passed(test: dict[str, Any]) -> bool: if "success" in test: return bool(test.get("success")) dab = test.get("dab_check") if isinstance(test.get("dab_check"), dict) else {} if "all_passed" in dab: return bool(dab.get("all_passed")) return False def apply_task_recount(entry: LeaderboardEntry, raw: dict[str, Any]) -> None: """Count tasks the model finished with every condition passed. Success in the table is the condition check alone. If that stored count disagrees with ``tests[].success``, the table uses this recount. """ tests = [row for row in (raw.get("tests") or []) if isinstance(row, dict)] if not tests: return passed = 0 strict = 0 for test in tests: ok = _conditions_passed(test) if ok: passed += 1 if ok and bool(test.get("agent_is_done")): strict += 1 entry.strict_passed_tasks = strict if passed != entry.passed_tasks: entry.passed_tasks = passed reconcile_success_metrics(entry) def load_entry_file(path: Path) -> LeaderboardEntry | None: payload = json.loads(path.read_text(encoding="utf-8")) if "metrics" not in payload: return None entry = LeaderboardEntry.from_json(payload) if not entry.raw_results_path: raw_path = raw_results_path_for(entry, space_root(path.parent.parent)) if raw_path.is_file(): entry.raw_results_path = str(raw_path.relative_to(space_root(path.parent.parent)).as_posix()) needs_raw = ( not entry.ui_classes or not entry.token_usage_by_model or entry.total_cost_usd is None or entry.total_cost_usd <= 0 or entry.total_tokens is None or entry.avg_tokens_per_task is None or entry.total_duration_seconds is None or entry.total_agent_steps is None or (entry.harness.startswith("ouroboros-full") and not entry.ouroboros_model_slots) ) raw = load_raw_results(entry, space_root(path.parent.parent)) if needs_raw and raw: _enrich_entry_from_raw(entry, raw) else: reconcile_entry_usage_totals(entry) if raw: apply_task_recount(entry, raw) return entry def load_entries(results_path: Path | None = None) -> list[LeaderboardEntry]: root = results_path or results_dir() if not root.is_dir(): return [] entries: list[LeaderboardEntry] = [] for path in sorted(root.glob("*.json")): entry = load_entry_file(path) if entry is not None: entries.append(entry) return sorted( entries, key=lambda entry: entry.success_rate if entry.success_rate is not None else -1, reverse=True, ) def entry_label(entry: LeaderboardEntry) -> str: return f"{entry.model} · {entry.harness}" def entry_choices(entries: list[LeaderboardEntry]) -> list[tuple[str, str]]: return [(entry_label(entry), entry.entry_id) for entry in entries] def entry_by_id(entries: list[LeaderboardEntry], entry_id: str) -> LeaderboardEntry | None: for entry in entries: if entry.entry_id == entry_id: return entry return None FILTER_ALL = "all" INPUT_FILTER_ORDER = ("text", "text+image", "image") def filter_entries( entries: list[LeaderboardEntry], *, input_id: str | None = FILTER_ALL, harness_id: str | None = FILTER_ALL, ) -> list[LeaderboardEntry]: filtered = entries if input_id and input_id != FILTER_ALL: filtered = [entry for entry in filtered if entry.input_modality == input_id] if harness_id and harness_id != FILTER_ALL: filtered = [entry for entry in filtered if entry.harness == harness_id] return filtered def input_filter_choices(entries: list[LeaderboardEntry]) -> list[tuple[str, str]]: present = {entry.input_modality for entry in entries if entry.input_modality not in {"", "—"}} ordered = [name for name in INPUT_FILTER_ORDER if name in present] ordered.extend(sorted(present - set(INPUT_FILTER_ORDER))) return [(t("all_inputs"), FILTER_ALL), *[(name, name) for name in ordered]] def harness_filter_choices(entries: list[LeaderboardEntry]) -> list[tuple[str, str]]: names = sorted({entry.harness for entry in entries if entry.harness}) return [(t("all_harnesses"), FILTER_ALL), *[(name, name) for name in names]] def sort_entries(entries: list[LeaderboardEntry], view_id: str) -> list[LeaderboardEntry]: config = load_bench_config() view = next((row for row in config["views"] if row["id"] == view_id), config["views"][0]) key = view["sort_key"] reverse = not view.get("ascending", False) def sort_value(entry: LeaderboardEntry) -> float: value = entry.metric(key) if value is None: return float("inf") if view.get("ascending") else -1.0 return float(value) return sorted(entries, key=sort_value, reverse=reverse)