src/ui/fishbowl/session.py

"""``FishbowlSession`` — a thin wrapper over one live :class:`Conductor`.

The Fishbowl app shell holds one of these per browser session (in ``gr.State`` or
keyed by ``gr.Request.session_hash``).  This class owns no Gradio components; it is
plain Python that builds the engine exactly as the root ``app.py`` does — via
``default_registry()`` / ``build_scenario`` / ``governor_for`` / ``make_ledger`` —
and exposes the read surface ``view_model_at`` needs.

Everything stays offline-safe: with no API key the deterministic stub drives the
cast, so a session is reproducible on stage.
"""

from __future__ import annotations

from src import observability as obs
from src.core.conductor import Conductor
from src.core.ledger_factory import make_ledger
from src.core.leaderboard_store import build_entry, make_leaderboard_store
from src.core.manifest import AgentManifest
from src.core.registry import Registry, default_registry
from src.core.run_index import index_runs
from src.tools.builtins import default_tool_registry
from src.ui.fishbowl.view_model import view_model_at

# Preferred display order, mirroring root app.py.  Any other scenarios dropped into
# config/ follow in sorted order.
_PREFERRED = ["thousand-token-wood", "mystery-roots", "oracle-grove"]


def _ordered_names(registry: Registry) -> list[str]:
    return [n for n in _PREFERRED if n in registry.scenarios] + [
        n for n in sorted(registry.scenarios) if n not in _PREFERRED
    ]


def _cast_model(seat: dict | None) -> str | None:
    """The concrete winning model for one cast seat (ADR-0029 / ADR-0035).

    Prefers the explicit ``model_endpoint`` (the precise catalogue key an operator pinned —
    the sponsor-track receipt), then the router-resolved ``model`` stamped on ``run.started``
    so profile-bound agents (``model_endpoint`` is None) are still credited a real model.
    Either makes the run eligible for the Hall of Fame; without either it stays None."""
    seat = seat or {}
    return seat.get("model_endpoint") or seat.get("model")


def scenario_titles(registry: Registry | None = None) -> dict[str, str]:
    """Map display title → internal scenario name, in the app's preferred order."""
    registry = registry or default_registry()
    return {(registry.scenarios[n].title or n): n for n in _ordered_names(registry)}


class FishbowlSession:
    """Owns one live ``Conductor`` for a chosen scenario and feeds the presenter."""

    def __init__(self, scenario_name: str, *, registry: Registry | None = None, tools=None) -> None:
        self._registry = registry or default_registry()
        self._scenario_name = scenario_name
        tools = tools if tools is not None else default_tool_registry()
        scenario = self._registry.build_scenario(scenario_name, tools=tools)
        self.conductor = Conductor(
            scenario,
            governor=self._registry.governor_for(scenario_name),
            ledger=make_ledger(),
        )
        # The leaderboard's dedicated scoreboard table (ADR-0035) — a *separate* table in
        # the same database as the event ledger, never the events log.  Built defensively
        # so a store/config hiccup can never break a live run; a None store just means no
        # scoreboard row is recorded.
        try:
            self._leaderboard = make_leaderboard_store()
        except Exception:  # pragma: no cover - no store configured (defensive)
            self._leaderboard = None
        obs.log(
            "session.created",
            scenario=scenario_name,
            ledger=type(self.conductor.ledger).__name__,
            cast=len(scenario.agents),
        )

    # ── lifecycle ───────────────────────────────────────────────────────────────

    def reset(self, seed: str = "", *, session_id: str | None = None) -> None:
        obs.log("session.reset", scenario=self._scenario_name, seed=seed or self.scenario.default_seed)
        self.conductor.reset(seed or self.scenario.default_seed, session_id=session_id)

    def step(self, n_ticks: int = 1) -> None:
        with obs.span("session.step", **{"session.n_ticks": n_ticks}):
            obs.log("session.step", n_ticks=n_ticks)
            self.conductor.step(n_ticks)
            self._finalize_if_verdict()

    def step_one(self) -> bool:
        """Advance a single agent (streaming): one event per call so the Show reveals
        each mind the moment it responds, not after the whole turn finishes."""
        with obs.span("session.step", **{"session.streaming": True}):
            advanced = self.conductor.step_one()
            self._finalize_if_verdict()
            return advanced

    def _finalize_if_verdict(self) -> None:
        """Close the run the instant the judge rules — from *any* drive path.

        The judge can land a ``judge.verdict`` during normal play (``step`` / ``step_one``,
        autoplay or a manual step) or on a forced curtain call.  Whichever it is, the winner
        must be attributed (``run.finished``) and the Hall of Fame row written as soon as the
        verdict exists — so the scoreboard fills the moment the winner is revealed in the UI,
        not on some later tick that may never come.  Idempotent: a no-op once the run is
        already closed (``finalize`` itself is idempotent too)."""
        if self.has_verdict() and not self.is_finalized():
            self.finalize("verdict")

    def inject(self, text: str, label: str | None = None) -> None:
        obs.log("session.inject", label=label or "", chars=len(text))
        obs.log("session.inject.text", level="debug", label=label or "", text=text)
        self.conductor.inject_user_event(text, label=label)
        self.conductor.step()
        self._finalize_if_verdict()

    # ── read surface (feeds view_model_at) ────────────────────────────────────────

    @property
    def scenario(self):
        return self.conductor.scenario

    # Marks the live, generative session apart from a read-only ``ReplaySession``;
    # the autoplay loop checks this so loading a past run never spends tokens.
    replay = False

    @property
    def events(self):
        # Run-scoped: the ledger is a shared store of *every* run (ADR-0009), so the
        # Show must only ever see the current run's events — otherwise scenario B's
        # stage would replay scenario A's discussion.  Every read below (head,
        # snapshot, scrubber) flows from this, so scoping here scopes the whole Show.
        return self.conductor.ledger.events_for_run(self.conductor.run_id)

    @property
    def head(self) -> int:
        """The generation-head: number of events in *this run* so far."""
        return len(self.events)

    def has_verdict(self) -> bool:
        """True once a ``judge.verdict`` event sits in *this run* — the show resolved.

        Run-scoped (ADR-0009): the ledger is a shared, append-only store of every run,
        so we only consult the current run's events.  The Fishbowl autoplay loop calls
        this to auto-pause the timer when the Judge has ruled, so the curtain falls on
        its own (no extra token spend)."""
        return any(e.kind == "judge.verdict" for e in self.conductor.ledger.events_for_run(self.conductor.run_id))

    def is_finalized(self) -> bool:
        """True once this run has been closed with a ``run.finished`` event.

        The autoplay loop consults this so it can close a run whose judge ruled on its own
        during a normal tick (writing ``run.finished`` and the Hall of Fame row) without
        re-finalising on every subsequent tick."""
        return any(e.kind == "run.finished" for e in self.conductor.ledger.events_for_run(self.conductor.run_id))

    def peek_next_actor_name(self) -> str | None:
        """Best-effort name of whoever the next :meth:`step_one` will run.

        Feeds the Show's "who's thinking…" hint while a model call is in flight; a pure
        read that never advances the run.  Returns ``None`` when nothing is queued."""
        return self.conductor.peek_next_actor_name()

    def has_judge(self) -> bool:
        """True when this cast has a judge that can be asked to rule.

        The Show enables "Start judging" only when this holds, and the autoplay loop
        consults it before bringing on the judge at a budget/turn limit (a cast with
        no judge halts visibly instead)."""
        return any(getattr(agent.manifest, "role", None) == "judge" for agent in self.scenario.agents)

    def force_verdict(self) -> bool:
        """Curtain call: silence the cast and have the judge rule on the whole run.

        Delegates to :meth:`Conductor.force_verdict` (which reads every event of this
        run and lands a ``judge.verdict`` even on a spent budget), then closes the run
        with a ``run.finished`` so the winner is attributed (ADR-0029).  Returns True
        when a verdict landed, False when the cast has no judge.  Idempotent — a second
        call after a verdict is a no-op that returns True."""
        with obs.span("session.force_verdict", **{"session.scenario": self._scenario_name}):
            obs.log("session.force_verdict", scenario=self._scenario_name)
            verdict = self.conductor.force_verdict()
            if verdict is not None:
                self.finalize("verdict")
            return verdict is not None

    def finalize(self, reason: str) -> None:
        """Close the current run with a ``run.finished`` event (idempotent-safe).

        On a verdict we derive ``winner`` from the judge's ruling and resolve its kind
        (ADR-0029): a cast agent name → ``winner_kind: "agent"`` with that agent's
        ``winning_model``; a team label → ``winner_kind: "team"`` with every member's
        endpoint in ``winning_models`` (``winning_model`` left ``None``, never guessed).
        All fall back to ``None`` / empty when unknown."""
        winner: str | None = None
        winner_kind: str | None = None
        winning_model: str | None = None
        winning_models: list[str] = []
        run_events = self.conductor.ledger.events_for_run(self.conductor.run_id)
        if reason == "verdict":
            verdict = next((e for e in reversed(run_events) if e.kind == "judge.verdict"), None)
            if verdict is not None:
                winner = verdict.payload.get("winner") or None
            if winner:
                started = next((e for e in run_events if e.kind == "run.started"), None)
                cast = (started.payload.get("cast") or {}) if started is not None else {}
                scenario = self._registry.scenarios.get(self._scenario_name)
                teams = getattr(getattr(scenario, "competition", None), "teams", None) or {}
                if winner in cast:
                    winner_kind = "agent"
                    winning_model = _cast_model(cast.get(winner))
                    winning_models = [winning_model] if winning_model else []
                elif winner in teams:
                    winner_kind = "team"
                    winning_models = [
                        endpoint for member in teams[winner] if (endpoint := _cast_model(cast.get(member)))
                    ]
        self.conductor.finalize(
            reason,
            winner=winner,
            winner_kind=winner_kind,
            winning_model=winning_model,
            winning_models=winning_models,
        )
        self._record_leaderboard()

    def _record_leaderboard(self) -> None:
        """Write this run's scoreboard row to the dedicated leaderboard table (ADR-0035).

        Detached from the event ledger: the run's events stay the source of truth for the
        trace, while this persists one denormalised result row to ``leaderboard_entries``
        — but only when :func:`build_entry` deems the run eligible (finished, a winner, a
        concrete winning model, and a competitive scenario).  Idempotent via the store's
        upsert-on-``run_id``, so a verdict that supersedes a budget close replaces the row.

        Fully defensive: any failure is logged and swallowed so a leaderboard hiccup never
        breaks the show."""
        store = getattr(self, "_leaderboard", None)
        if store is None:
            obs.log("leaderboard.no_store", level="warning", run_id=self.conductor.run_id, scenario=self._scenario_name)
            return
        try:
            run_events = self.conductor.ledger.events_for_run(self.conductor.run_id)
            summary = next((s for s in index_runs(run_events) if s.run_id == self.conductor.run_id), None)
            if summary is None:
                return
            scenario = self._registry.scenarios.get(self._scenario_name)
            entry = build_entry(summary, getattr(scenario, "competition", None))
            if entry is None:
                # Not eligible (unfinished, no winner, no winning model, or kind:none) — say
                # *which*, so an empty Hall of Fame can be diagnosed instead of guessed at.
                obs.log(
                    "leaderboard.skipped",
                    level="debug",
                    run_id=self.conductor.run_id,
                    scenario=self._scenario_name,
                    finished=summary.finished_at is not None,
                    winner=summary.winner,
                    winning_models=summary.winning_models,
                )
                return
            store.record(entry)
            obs.log("leaderboard.recorded", run_id=entry.run_id, scenario=entry.scenario, winner=entry.winner)
        except Exception as exc:  # never let a scoreboard write break a run — but make it visible
            obs.log(
                "leaderboard.record_failed",
                level="warning",
                run_id=self.conductor.run_id,
                scenario=self._scenario_name,
                error=str(exc),
                error_type=type(exc).__name__,
            )

    @property
    def cast(self) -> list[AgentManifest]:
        return [agent.manifest for agent in self.scenario.agents]

    @property
    def governor(self):
        return self.conductor.governor

    @property
    def scenario_name(self) -> str:
        return self._scenario_name

    @property
    def goal(self) -> str:
        return self.scenario.goal

    @property
    def token_ceiling(self) -> int | None:
        return self.governor.max_total_tokens

    @property
    def max_rounds(self) -> int:
        return self.governor.max_turns

    @property
    def autoplay_tick_cap(self) -> int:
        """Hard backstop on consecutive autoplay generations, derived from the budget.

        The governor (max_turns / max_total_tokens / max_total_calls) is the real bound
        on how long a show runs — this cap only exists to stop an *unbounded* loop if
        those are misconfigured. We size it just above the total-call ceiling so a
        legitimately long show (a late Judge verdict) always resolves on a real budget
        bound with a meaningful reason, never on this arbitrary backstop."""
        return max(120, int(getattr(self.governor, "max_total_calls", 0) or 0) + 10)

    # ── snapshot ──────────────────────────────────────────────────────────────────

    def snapshot(self, k: int | None = None) -> dict:
        """Build the Show's view-model at step *k* (defaults to the head)."""
        events = self.events
        return view_model_at(
            events,
            k if k is not None else len(events),
            self.cast,
            scenario_name=self.scenario_name,
            goal=self.goal,
            governor=self.governor,
            token_ceiling=self.token_ceiling,
            max_rounds=self.max_rounds,
        )


class ReplaySession:
    """A read-only view over one *past* run — the Archive's "Load" target.

    It exposes the exact read surface the Show renders (``events`` / ``head`` /
    ``snapshot`` / ``has_verdict``) so the transport's scrubber and replay just work,
    but it owns no live ``Conductor``: ``step``/``step_one``/``inject`` are no-ops and
    ``replay`` is ``True``, so autoplay never generates (no token spend) on a load.

    The fixed event list is the run's own slice (``events_for_run(run_id)``); the cast
    cards / meters bounds are rebuilt from that run's scenario via the registry, while
    the discussion itself is replayed verbatim from the events.
    """

    replay = True

    def __init__(
        self,
        *,
        run_id: str,
        events: tuple,
        scenario_name: str,
        registry: Registry | None = None,
        tools=None,
    ) -> None:
        self.run_id = run_id
        self._events = tuple(events)
        self._scenario_name = scenario_name
        registry = registry or default_registry()
        tools = tools if tools is not None else default_tool_registry()
        scenario = registry.build_scenario(scenario_name, tools=tools)
        self._scenario = scenario
        self._cast = [agent.manifest for agent in scenario.agents]
        self._governor = registry.governor_for(scenario_name)
        obs.log("session.replay", scenario=scenario_name, run_id=run_id, events=len(self._events))

    # ── read surface (mirrors FishbowlSession) ────────────────────────────────────

    @property
    def events(self):
        return self._events

    @property
    def head(self) -> int:
        return len(self._events)

    @property
    def scenario_name(self) -> str:
        return self._scenario_name

    @property
    def goal(self) -> str:
        return self._scenario.goal

    @property
    def cast(self) -> list[AgentManifest]:
        return self._cast

    def has_verdict(self) -> bool:
        return any(e.kind == "judge.verdict" for e in self._events)

    def has_judge(self) -> bool:
        # A replay owns no live engine, so it can never *call* a judge — but the
        # recorded run may already carry its verdict.  Report on the recording.
        return self.has_verdict()

    def peek_next_actor_name(self) -> str | None:  # pragma: no cover - inert by design
        # A replay never generates, so nobody is ever "thinking" — no hint to show.
        return None

    def force_verdict(self) -> bool:  # pragma: no cover - inert by design
        # A replay is read-only: nothing to judge, the recording stands as-is.
        return self.has_verdict()

    @property
    def autoplay_tick_cap(self) -> int:
        return self.head

    # ── inert lifecycle (a replay never generates) ────────────────────────────────

    def reset(self, *_args, **_kwargs) -> None:  # pragma: no cover - inert by design
        return None

    def step(self, *_args, **_kwargs) -> None:
        return None

    def step_one(self, *_args, **_kwargs) -> bool:
        return False

    def inject(self, *_args, **_kwargs) -> None:
        return None

    # ── snapshot ──────────────────────────────────────────────────────────────────

    def snapshot(self, k: int | None = None) -> dict:
        events = self._events
        return view_model_at(
            events,
            k if k is not None else len(events),
            self._cast,
            scenario_name=self._scenario_name,
            goal=self.goal,
            governor=None,  # no live governor on a replay — meters show recorded text only
            token_ceiling=getattr(self._governor, "max_total_tokens", None),
            max_rounds=getattr(self._governor, "max_turns", None),
        )