"""Leaderboard store — a dedicated, durable table for competitive run results. Unlike :mod:`src.core.leaderboard` (which *aggregates* results), this module owns the **persistence** of one row per decided competitive run. It is deliberately **detached from the event ledger**: the append-only ``events`` table stays the source of truth for *what happened* in a run (every utterance, thought, verdict — the full logs), while this ``leaderboard_entries`` table is a separate, denormalised record of *who won* — a materialised scoreboard the Hall of Fame reads without ever folding the event log. Why a separate table (not a pure projection) --------------------------------------------- A run's result is written **once**, at finish, only when it is genuinely decided — so the Hall of Fame is a cheap ``SELECT`` over a small table rather than an O(all-events) fold of the whole ledger on every page load. The two tables share one database (the same ``DATABASE_URL`` / Postgres instance) but never share rows: ``events`` is the trace, ``leaderboard_entries`` is the scoreboard. The ``run_id`` on each entry links a row back to its full trace (``ledger.events_for_run(run_id)``) for replay. The write gate (the operator's explicit requirement) ---------------------------------------------------- A row is recorded **only if** the run is finished AND a winner with a concrete winning model is selected. Unfinished runs, abandoned runs (a budget close with no verdict), and runs whose winner has no bound model endpoint never produce a row — see :func:`build_entry` and the call site in ``FishbowlSession.finalize``. SQLAlchemy is imported lazily inside ``LeaderboardStore.__init__`` so importing this module — and therefore ``src.core.*`` — never requires SQLAlchemy or a DB driver. The :class:`LeaderboardEntry` Pydantic model is import-clean and is reused by both the store (persistence) and :mod:`src.core.leaderboard` (aggregation). """ from __future__ import annotations import json import os from datetime import datetime, timezone from pathlib import Path from typing import TYPE_CHECKING, Any from pydantic import BaseModel, ConfigDict, Field from src.core.run_index import CastBinding, RunSummary if TYPE_CHECKING: # types only — never imported at runtime on the offline path from sqlalchemy import Engine, MetaData, Table from src.scenarios.base import CompetitionConfig # ── the row model (shared by the store and the aggregations) ───────────────────── class LeaderboardEntry(BaseModel): """One decided competitive run — the durable scoreboard row. Carries everything the Hall of Fame needs without touching the event ledger: the cast→model bindings, the winner and the model(s) behind it, the competition shape (``kind`` / ``teams`` / ``symmetric_seats`` — needed for per-seat fairness), and the run's cost/timing. ``run_id`` links the row back to its full trace for replay. """ model_config = ConfigDict(extra="forbid") run_id: str session_id: str | None = None scenario: str = "" seed: str = "" # Competition shape (mirrors CompetitionConfig — needed for seat/fairness rollups). competition_kind: str = "none" teams: dict[str, list[str]] | None = None symmetric_seats: list[str] | None = None cast: dict[str, CastBinding] = Field(default_factory=dict) winner: str | None = None winner_kind: str | None = None winning_model: str | None = None winning_models: list[str] = Field(default_factory=list) reason: str | None = None turns: int = 0 tokens: int = 0 started_at: datetime | None = None finished_at: datetime | None = None recorded_at: datetime | None = None """When this scoreboard row was written (insert time).""" def build_entry(summary: RunSummary, competition: "CompetitionConfig | None") -> LeaderboardEntry | None: """Build a :class:`LeaderboardEntry` from a finished run, or ``None`` if not eligible. The single eligibility gate (the operator's requirement): the run must be **finished** (``finished_at`` set), have a **winner**, carry at least one **winning model**, and be **competitive** (``competition.kind != "none"``). Returns ``None`` otherwise, so the caller simply doesn't record a row. ``summary`` is the :class:`~src.core.run_index.RunSummary` folded from the run's events (it already resolves winner / winner_kind / winning_model(s) / cast / turns / tokens); ``competition`` is the scenario's config (for the seat/team shape). This function adds no new attribution — it only decides eligibility and copies the fields across. """ kind = getattr(competition, "kind", "none") or "none" winning_models = [m for m in (summary.winning_models or []) if m] if summary.winning_model and summary.winning_model not in winning_models: winning_models.append(summary.winning_model) finished = summary.finished_at is not None if not finished or not summary.winner or not winning_models or kind == "none": return None return LeaderboardEntry( run_id=summary.run_id, session_id=summary.session_id, scenario=summary.scenario, seed=summary.seed, competition_kind=kind, teams=getattr(competition, "teams", None), symmetric_seats=getattr(competition, "symmetric_seats", None), cast=dict(summary.cast), winner=summary.winner, winner_kind=summary.winner_kind, winning_model=summary.winning_model, winning_models=winning_models, reason=summary.reason, turns=summary.turns, tokens=summary.tokens, started_at=summary.started_at, finished_at=summary.finished_at, ) # ── the durable store (mirrors SqlAlchemyLedger's lazy-SQLAlchemy pattern) ──────── def _normalise_url(url: str | Path) -> str: """Accept a SQLAlchemy URL or a bare filesystem path (mirrors the ledger).""" text = str(url) if "://" in text: return text if text == ":memory:": return "sqlite://" return f"sqlite:///{text}" class LeaderboardStore: """Durable store for :class:`LeaderboardEntry` rows in a dedicated table. Backed by SQLAlchemy (Postgres or SQLite), exactly like ``SqlAlchemyLedger`` — but it owns the ``leaderboard_entries`` table, never ``events``. ``record`` is an idempotent upsert keyed on ``run_id`` (a re-finalised run — e.g. a budget close later superseded by a verdict — replaces its row rather than duplicating it). """ def __init__(self, url: str | Path = ":memory:") -> None: # Lazy import: keeps src.core.* importable without SQLAlchemy installed. from sqlalchemy import ( Column, DateTime, Integer, MetaData, String, Table, Text, create_engine, ) self._url = _normalise_url(url) self._engine: Engine = create_engine(self._url, pool_pre_ping=True) self._metadata: MetaData = MetaData() self._table: Table = Table( "leaderboard_entries", self._metadata, Column("offset", Integer, primary_key=True, autoincrement=True), Column("run_id", String(64), unique=True, nullable=False, index=True), Column("session_id", String, nullable=True, index=True), Column("scenario", String, nullable=False, index=True), Column("seed", String, nullable=False, server_default=""), Column("competition_kind", String, nullable=False, server_default="none"), Column("teams", Text, nullable=True), Column("symmetric_seats", Text, nullable=True), Column("cast", Text, nullable=False, server_default="{}"), Column("winner", String, nullable=True), Column("winner_kind", String, nullable=True), Column("winning_model", String, nullable=True), Column("winning_models", Text, nullable=False, server_default="[]"), Column("reason", String, nullable=True), Column("turns", Integer, nullable=False, server_default="0"), Column("tokens", Integer, nullable=False, server_default="0"), Column("started_at", DateTime(timezone=True), nullable=True), Column("finished_at", DateTime(timezone=True), nullable=True), Column("recorded_at", DateTime(timezone=True), nullable=False), ) self._metadata.create_all(self._engine) # ── write ────────────────────────────────────────────────────────────────── def record(self, entry: LeaderboardEntry) -> LeaderboardEntry: """Persist *entry*, upserting on ``run_id`` (idempotent — never duplicates a run). Stamps ``recorded_at`` if unset. A pre-existing row for the same ``run_id`` is replaced (delete-then-insert in one transaction), so a corrective re-finalise keeps exactly one, current scoreboard row per run. """ recorded = entry.model_copy(update={"recorded_at": entry.recorded_at or datetime.now(timezone.utc)}) values = self._to_row(recorded) with self._engine.begin() as conn: conn.execute(self._table.delete().where(self._table.c.run_id == recorded.run_id)) conn.execute(self._table.insert().values(**values)) return recorded # ── read ─────────────────────────────────────────────────────────────────── def entries(self) -> list[LeaderboardEntry]: """Every recorded entry, newest finish first (``run_id`` breaks ties).""" from sqlalchemy import select t = self._table stmt = select(t).order_by(t.c.offset) with self._engine.connect() as conn: rows = conn.execute(stmt).mappings().all() out = [self._row_to_entry(row) for row in rows] out.sort(key=lambda e: (e.finished_at is None, _neg_ts(e.finished_at), e.run_id)) return out def entries_for_scenario(self, scenario: str) -> list[LeaderboardEntry]: """The recorded entries for one scenario (indexed query), newest finish first.""" from sqlalchemy import select t = self._table stmt = select(t).where(t.c.scenario == scenario).order_by(t.c.offset) with self._engine.connect() as conn: rows = conn.execute(stmt).mappings().all() out = [self._row_to_entry(row) for row in rows] out.sort(key=lambda e: (e.finished_at is None, _neg_ts(e.finished_at), e.run_id)) return out def close(self) -> None: self._engine.dispose() # ── internal (de)serialisation ─────────────────────────────────────────────── @staticmethod def _to_row(entry: LeaderboardEntry) -> dict[str, Any]: return { "run_id": entry.run_id, "session_id": entry.session_id, "scenario": entry.scenario, "seed": entry.seed, "competition_kind": entry.competition_kind, "teams": json.dumps(entry.teams) if entry.teams is not None else None, "symmetric_seats": (json.dumps(entry.symmetric_seats) if entry.symmetric_seats is not None else None), "cast": json.dumps({name: binding.model_dump() for name, binding in entry.cast.items()}), "winner": entry.winner, "winner_kind": entry.winner_kind, "winning_model": entry.winning_model, "winning_models": json.dumps(list(entry.winning_models)), "reason": entry.reason, "turns": int(entry.turns), "tokens": int(entry.tokens), "started_at": _aware(entry.started_at), "finished_at": _aware(entry.finished_at), "recorded_at": _aware(entry.recorded_at) or datetime.now(timezone.utc), } @staticmethod def _row_to_entry(row: Any) -> LeaderboardEntry: cast_raw = _loads(row["cast"], {}) cast = {name: CastBinding(**(binding or {})) for name, binding in cast_raw.items()} return LeaderboardEntry( run_id=row["run_id"], session_id=row.get("session_id"), scenario=row["scenario"], seed=row["seed"] or "", competition_kind=row["competition_kind"] or "none", teams=_loads(row.get("teams"), None), symmetric_seats=_loads(row.get("symmetric_seats"), None), cast=cast, winner=row.get("winner"), winner_kind=row.get("winner_kind"), winning_model=row.get("winning_model"), winning_models=_loads(row.get("winning_models"), []) or [], reason=row.get("reason"), turns=row["turns"] or 0, tokens=row["tokens"] or 0, started_at=_aware(_parse_dt(row.get("started_at"))), finished_at=_aware(_parse_dt(row.get("finished_at"))), recorded_at=_aware(_parse_dt(row.get("recorded_at"))), ) # ── factory (mirrors src.core.ledger_factory — same DATABASE_URL, separate table) ─ # Memoise one store per resolved URL. This is what makes the no-key, in-memory # (``sqlite://``) stage demo work end-to-end within a process: the write at finalize and # the Hall-of-Fame read share the *same* engine (a fresh in-memory engine would be a # different, empty database). For a file/Postgres URL it just avoids re-opening engines. _STORES: dict[str, LeaderboardStore] = {} def make_leaderboard_store(url: str | None = None) -> LeaderboardStore: """Construct (or reuse) the leaderboard store for the configured database. Resolves the same ``DATABASE_URL`` as the event ledger (a separate *table* in the same database), so no extra configuration is needed. *url* overrides it (tests pass ``"sqlite://"``). Raises :class:`RuntimeError` when neither is set — like the ledger, the durable store is not optional. """ resolved = url or os.getenv("DATABASE_URL") if not resolved: raise RuntimeError( "DATABASE_URL is required for the leaderboard store — it lives in the same " "database as the event ledger (a separate table), so set DATABASE_URL or pass " "an explicit url to make_leaderboard_store()." ) from src.core.ledger_factory import _normalize_db_url key = _normalize_db_url(resolved) store = _STORES.get(key) if store is None: store = LeaderboardStore(key) _STORES[key] = store return store def _reset_store_cache() -> None: """Drop memoised stores (test hook — production never needs this).""" for store in _STORES.values(): try: store.close() except Exception: # pragma: no cover - best-effort cleanup pass _STORES.clear() # ── tiny datetime helpers (mirror the ledger's tz coercion) ────────────────────── def _aware(dt: datetime | None) -> datetime | None: """Coerce a datetime to tz-aware UTC; pass ``None`` through (SQLite drops tzinfo).""" if dt is None: return None if dt.tzinfo is None: return dt.replace(tzinfo=timezone.utc) return dt def _parse_dt(value: Any) -> datetime | None: if value is None or isinstance(value, datetime): return value try: return datetime.fromisoformat(str(value)) except (ValueError, TypeError): # pragma: no cover - defensive return None def _neg_ts(value: datetime | None) -> float: """Negated POSIX timestamp for descending sort; ``0.0`` for ``None`` (sorted last).""" return -value.timestamp() if value is not None else 0.0 def _loads(value: Any, default: Any) -> Any: """JSON-decode a stored text column, tolerating already-decoded values / nulls.""" if value is None: return default if isinstance(value, (dict, list)): return value try: return json.loads(value) except (ValueError, TypeError): # pragma: no cover - defensive return default __all__ = [ "LeaderboardEntry", "LeaderboardStore", "build_entry", "make_leaderboard_store", ]