"""``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), )