"""FISHBOWL — the Gradio app shell (Unit 9: integrator + root shim).
Two tabs — **The Lab** (compose a run) and **The Show** (watch the say-vs-think
MindCard replay) — wrapped in the CRT theater chrome. This module owns the app shell
and *all* cross-module wiring: it holds the per-user session in ``gr.State``, drives the
hybrid transport (scrub-back = pure prefix view; play-at-head = step the Conductor), and
composes the Show's stage/feed/meters/verdict HTML from the render units.
Every sibling unit (theme, render.*, session, show, lab) is imported **defensively**:
this worker runs in its own worktree before the leaf modules land, so each import falls
back to a friendly placeholder. The shell therefore builds and launches standalone, and
upgrades itself the moment the real modules are merged — no edits required here.
Offline-first: no API key is needed; the deterministic stub keeps the demo reproducible.
"""
from __future__ import annotations
import os
import socket
import uuid
# Import `spaces` before anything that could initialise CUDA: on ZeroGPU the package
# raises if torch's CUDA is touched first (our torch is lazy via mem0, so this stays
# first by construction). Off ZeroGPU the package is fully effect-free.
import spaces
import gradio as gr
# ── engine read surface (public; never mutated here) ───────────────────────────
from src.core.conductor import Conductor
from src.core.governor import BudgetExceeded
from src.core.ledger_factory import make_ledger
from src.core.registry import default_registry
from src.tools.builtins import default_tool_registry
from src.ui.fishbowl.adapter import scenario_voice
from src.ui.fishbowl.view_model import view_model_at
try: # archive: "my past sessions" read layer over the run-history ledger (ADR-0026)
from src.ui.fishbowl.archive import list_runs, load_replay, run_card_label
except Exception: # pragma: no cover - degrade gracefully if the archive unit is absent
def list_runs(*_args, **_kwargs):
return []
def load_replay(*_args, **_kwargs):
return None
def run_card_label(_summary):
return "▶ (run)"
try: # about: a compact project tour — what it is, the architecture, the submission links
from src.ui.fishbowl.about import build_about
except Exception: # pragma: no cover - degrade gracefully if the unit is absent
def build_about() -> dict:
gr.HTML('
About — coming soon.
')
return {}
try: # hall of fame: the competitive high-score board (Workstream 6.2 + 6.3)
from src.ui.fishbowl.hall_of_fame import build_hall_of_fame, wire_sessions_render
except Exception: # pragma: no cover - degrade gracefully if the unit is absent
def build_hall_of_fame() -> dict:
gr.HTML('
🏆 Hall of Fame — coming soon.
')
return {}
def wire_sessions_render(*_args, **_kwargs) -> None:
return None
# ── loop-safety backstop ────────────────────────────────────────────────────────
# Belt-and-suspenders against a runaway autoplay loop: even when the governor never
# trips (e.g. a generous budget) the timer halts after this many consecutive auto-ticks
# per play session. The user explicitly asked for "no infinite, token-burning loop".
_MAX_AUTO_TICKS = 40
# ── defensive imports of sibling units ─────────────────────────────────────────
# Each block degrades to a placeholder so the shell runs before the leaf modules land.
try: # theme: CSS + + a gr.Theme
from src.ui.fishbowl.theme import FISHBOWL_HEAD, FishbowlTheme, load_css
except Exception: # pragma: no cover - exercised only before the theme unit lands
def load_css() -> str:
# Minimal CRT-ish palette so the placeholder shell is legible.
return """
:root { --bg:#070a06; --ink:#cfe8c0; --lime:#8fe36a; --cyan:#5fd0d0;
--violet:#b59cff; --amber:#e3c14c; --coral:#e3786a; }
body { background: var(--bg); color: var(--ink);
font-family: 'IBM Plex Mono', ui-monospace, monospace; }
footer { display: none !important; }
.fishbowl-topbar { display:flex; align-items:baseline; gap:12px;
padding:10px 16px; border-bottom:1px solid #2e3d25; }
.fishbowl-topbar .logo { color: var(--lime); font-weight:800; letter-spacing:.08em; }
.fishbowl-topbar .sub { color: var(--cyan); font-style:italic; opacity:.8; }
.crt-bg, .crt-grid, .crt-scan, .crt-vignette {
position: fixed; inset: 0; pointer-events: none; z-index: 0; }
.crt-vignette { box-shadow: inset 0 0 220px rgba(0,0,0,.8); }
.fishbowl-placeholder { padding: 24px; color: var(--ink);
border:1px dashed #2e3d25; border-radius:10px; line-height:1.6; }
"""
FISHBOWL_HEAD = (
''
''
)
FishbowlTheme = None # launch() tolerates theme=None
try: # render: the HTML pieces of the Show
from src.ui.fishbowl.render.mindcard import render_mindcard
except Exception: # pragma: no cover
def render_mindcard(card, *, mind_reader: bool = False) -> str:
name = card.get("name", "?")
said = card.get("said") or "…"
body = said
if mind_reader and card.get("thought"):
body += f" “{card['thought']}”"
return f"
{name} {body}
"
try:
# render_constellation lives in render.stage alongside render_split.
from src.ui.fishbowl.render.stage import render_constellation
except Exception: # pragma: no cover
def render_constellation(vm, cards_html_by_id) -> str:
cards = "".join(cards_html_by_id.values())
return f"
{cards}
"
try:
from src.ui.fishbowl.render.stage import render_split
except Exception: # pragma: no cover
def render_split(vm) -> str:
rows = "".join(
f"
{c.get('name', '?')}
{c.get('said') or ''}
"
f"
{c.get('thought') or ''}
"
for c in vm.get("cast", [])
)
return f"
{rows}
"
try:
from src.ui.fishbowl.render.feed import render_feed
except Exception: # pragma: no cover
def render_feed(vm, *, mind_reader: bool = False) -> str:
items = []
for it in vm.get("feed", []):
kind = it.get("kind", "")
if kind == "say":
line = f"{it.get('agent', '?')}: {it.get('said') or ''}"
if mind_reader and it.get("thought"):
line += f" ({it['thought']})"
elif kind == "narrate":
line = f"{it.get('text', '')}"
elif kind == "poke":
line = f"[{it.get('label', 'POKE')}] {it.get('text', '')}"
elif kind == "verdict":
line = f"VERDICT: {it.get('text', '')}"
else:
line = it.get("text", "")
items.append(f"
{line}
")
return f"
{''.join(items)}
"
try:
from src.ui.fishbowl.render.meters import render_meters
except Exception: # pragma: no cover
def render_meters(vm) -> str:
return (
"
"
)
try:
# render_verdict lives in render.meters alongside render_meters.
from src.ui.fishbowl.render.meters import render_verdict
except Exception: # pragma: no cover
def render_verdict(vm) -> str:
v = vm.get("verdict")
if not v:
return ""
return f"
{v.get('text', '')}
"
try:
# The cheerful champion celebration — a modular overlay, empty unless a winner was
# declared, so it composes into the verdict pane without touching no-winner scenarios.
from src.ui.fishbowl.render.winner import render_winner
except Exception: # pragma: no cover
def render_winner(vm) -> str:
return ""
try: # session: the transport wrapper over a Conductor
from src.ui.fishbowl.session import FishbowlSession
except Exception: # pragma: no cover - fallback session keeps the shell live
class FishbowlSession:
"""Minimal transport over a Conductor (real session unit supersedes this).
Mirrors the contract the integrator wires against: ``reset(seed)``,
``step()`` (generate at the head), ``inject(text, label=...)``, and
``snapshot(k)`` (a pure prefix view via ``view_model_at``)."""
def __init__(self, scenario_name: str, *, registry=None, tools=None) -> None:
reg = registry or default_registry()
tool_reg = tools if tools is not None else default_tool_registry()
self.conductor = Conductor(
reg.build_scenario(scenario_name, tools=tool_reg),
governor=reg.governor_for(scenario_name),
ledger=make_ledger(),
)
self.scenario_name = scenario_name
@property
def head(self) -> int:
return len(self.conductor.ledger.events)
def has_verdict(self) -> bool:
# Mirrors the real session's loop-safety helper so on_tick never crashes.
return any(getattr(e, "kind", None) == "judge.verdict" for e in self.conductor.ledger.events)
def has_judge(self) -> bool:
return any(
getattr(getattr(a, "manifest", None), "role", None) == "judge" for a in self.conductor.scenario.agents
)
def force_verdict(self) -> bool:
return self.conductor.force_verdict() is not None
def reset(self, seed: str) -> None:
self.conductor.reset(seed or self.conductor.scenario.default_seed)
def step(self) -> None:
self.conductor.step()
def step_one(self) -> bool:
return self.conductor.step_one()
def inject(self, text: str, *, label: str | None = None) -> None:
if text and text.strip():
self.conductor.inject_user_event(text.strip(), label=label)
self.conductor.step()
def snapshot(self, k: int | None = None) -> dict:
events = self.conductor.ledger.events
kk = self.head if k is None else max(0, min(int(k), len(events)))
scn = self.conductor.scenario
# view_model_at wants AgentManifests; agents carry theirs on .manifest.
cast = [getattr(a, "manifest", a) for a in scn.agents]
return view_model_at(
events,
kk,
cast,
scenario_name=self.scenario_name,
goal=getattr(scn, "goal", ""),
governor=self.conductor.governor,
)
try: # the two tab bodies (leaf units own their internals)
from src.ui.fishbowl.show import build_show
except Exception: # pragma: no cover
def build_show():
gr.HTML(
"
The Show stage loads here once the show unit lands.
"
)
return {}
try:
from src.ui.fishbowl.lab import build_lab
except Exception: # pragma: no cover
def build_lab():
gr.HTML(
"
The Lab composer loads here once the lab unit lands.
"
)
return {}
def build_telemetry() -> None:
"""The Telemetry tab: structured log feed + metric charts + per-trace timeline.
Reads live from the in-memory telemetry store (ADR-0024). The 3s auto-tick is
*non-blocking and idle-cheap*: it gates on the store's revision and returns
``gr.skip()`` (no recompute, no repaint, ``show_progress='hidden'`` so no overlay)
whenever nothing was recorded since the last paint. The filter dropdowns and ⟳
Refresh force a repaint; "show more traces" pages older traces into the timeline.
Clicking a span reveals the prompt + memory that agent saw.
"""
try:
from src.ui.fishbowl.render import telemetry as t
except Exception: # pragma: no cover - degrade to a friendly placeholder
gr.HTML("
Telemetry panel unavailable.
")
return
gr.HTML(f"")
kpi = gr.Markdown(t.kpi_markdown())
with gr.Row():
level_dd = gr.Dropdown(t.LEVELS, value="all", label="Level", scale=1)
layer_dd = gr.Dropdown(t.LAYERS, value="all", label="Layer", scale=1)
refresh = gr.Button("⟳ Refresh", scale=1)
with gr.Row():
calls_plot = gr.BarPlot(t.calls_frame(), x="metric", y="count", title="Activity", scale=1)
tokens_plot = gr.BarPlot(t.tokens_frame(), x="kind", y="tokens", title="Tokens", scale=1)
latency_plot = gr.LinePlot(
t.latency_frame(), x="n", y="seconds", color="agent", title="Agent-turn latency (s)", scale=1
)
feed = gr.Dataframe(
headers=["time", "level", "agent/turn", "event", "detail"],
value=t.log_rows(),
wrap=True,
label="Structured log feed",
row_count=(12, "dynamic"),
column_count=(5, "fixed"),
)
_traces0, _shown0, _total0 = t.render_traces(t.DEFAULT_TRACES)
traces = gr.HTML(_traces0)
load_more = gr.Button(t.more_label(_shown0, _total0), interactive=_total0 > _shown0, size="sm")
# Per-user view state: the signature of the last paint (so an idle tick can skip the
# work) and how many traces the timeline currently shows ("show more" grows it).
sig_state = gr.State(None)
trace_limit_state = gr.State(t.DEFAULT_TRACES)
timer = gr.Timer(3.0)
outs = [kpi, feed, calls_plot, tokens_plot, latency_plot, traces, load_more]
def _values(level, layer, trace_limit):
"""Recompute all panel outputs in one store pass (single counter read)."""
snap = t.snapshot(level, layer, trace_limit)
more = gr.update(
value=t.more_label(snap["trace_shown"], snap["trace_total"]),
interactive=snap["trace_total"] > snap["trace_shown"],
)
return (snap["kpi"], snap["feed"], snap["calls"], snap["tokens"], snap["latency"], snap["traces"], more)
def _sig(level, layer, trace_limit):
# Read the revision BEFORE rendering: if data lands mid-render the stored sig
# stays behind, so the next tick repaints to catch up (never a missed update).
return (t.revision(), level, layer, int(trace_limit))
def _tick(level, layer, trace_limit, last_sig):
"""Auto-refresh: repaint only when the store advanced or a filter/limit changed."""
sig = _sig(level, layer, trace_limit)
if sig == last_sig:
return (gr.skip(),) * (len(outs) + 1) # idle → leave every output (and sig) as-is
return _values(level, layer, trace_limit) + (sig,)
def _force(level, layer, trace_limit):
"""Manual Refresh / filter change: always repaint with the current filters."""
sig = _sig(level, layer, trace_limit)
return _values(level, layer, trace_limit) + (sig,)
def _load_more(level, layer, trace_limit):
"""Grow the timeline by one page and repaint (carries the new limit back to state)."""
new_limit = int(trace_limit) + t.TRACE_PAGE
sig = _sig(level, layer, new_limit)
return _values(level, layer, new_limit) + (sig, new_limit)
# Auto-tick is hidden (no progress overlay → no 3s flicker); user-driven actions show
# a minimal indicator so the click registers.
timer.tick(_tick, [level_dd, layer_dd, trace_limit_state, sig_state], outs + [sig_state], show_progress="hidden")
refresh.click(_force, [level_dd, layer_dd, trace_limit_state], outs + [sig_state], show_progress="minimal")
level_dd.change(_force, [level_dd, layer_dd, trace_limit_state], outs + [sig_state], show_progress="minimal")
layer_dd.change(_force, [level_dd, layer_dd, trace_limit_state], outs + [sig_state], show_progress="minimal")
load_more.click(
_load_more,
[level_dd, layer_dd, trace_limit_state],
outs + [sig_state, trace_limit_state],
show_progress="minimal",
)
# ── scenario registry (assembled from config/, not hardcoded) ───────────────────
_registry = default_registry()
_tools = default_tool_registry()
_PREFERRED = [
"thousand-token-wood",
"mystery-roots",
"oracle-grove",
"the-steeped",
"debate-duel",
"twenty-sprouts",
"beat-battle",
]
_names = [n for n in _PREFERRED if n in _registry.scenarios] + [
n for n in sorted(_registry.scenarios) if n not in _PREFERRED
]
# display title -> internal scenario name
SCENARIOS: dict[str, str] = {(_registry.scenarios[n].title or n): n for n in _names}
_DEFAULT_TITLE = next(iter(SCENARIOS), "")
# Transport speeds → gr.Timer interval (seconds). Keys mirror the speed radio.
SPEEDS: dict[str, float] = {"live": 3.0, "1×": 1.9, "fast": 1.0}
# ── session lifecycle (a fresh Conductor + session per Summon) ──────────────────
def _new_session(scenario_name: str) -> FishbowlSession:
# The session unit builds its own Conductor from the scenario name.
return FishbowlSession(scenario_name, registry=_registry, tools=_tools)
def _empty_vm() -> dict:
"""A safe, empty view-model so the Show renders before the first Summon."""
return {"step": 0, "total": 0, "cast": [], "feed": [], "verdict": None, "rounds": 1, "tokens": 0}
# ── HTML composition (this worker owns this) ────────────────────────────────────
def _fishbowl(inner: str, *, role: str) -> str:
"""Wrap a rendered pane in the ``.fishbowl`` scope root.
Every theater rule in ``assets/styles.css`` is scoped under ``.fishbowl`` so it wins
over Gradio's own cascade; the ``gr.HTML`` islands have no such ancestor on their own,
so without this wrapper the MindCards/feed/meters render as unstyled stacked text. The
``role`` adds a stable hook (e.g. ``fb-stage``) for layout rules that target a pane.
"""
if not inner:
return ""
return f'
{inner}
'
def _thinking_strip(label: str) -> str:
"""A broadcast-style "lower third" overlaid on the stage while a model call runs.
LLM calls take seconds; without a cue the theater reads as frozen. This strip names
who we're waiting on ("scene-whisperer is thinking…") with a pulsing orb + animated
ellipsis, anchored to the *always-visible* stage so it never hides under the feed's
scroll. Pure string; rendered only while a generation is in flight."""
import html as _html
text = _html.escape(label or "the cast is thinking")
return (
'
'
''
f'{text}'
''
"
"
)
def render_show_html(
vm: dict, *, layout: str = "constellation", mind_reader: bool = False, thinking: str | None = None
) -> tuple[str, str, str, str]:
"""Compose the Show's four HTML panes from a view-model snapshot.
Returns ``(stage, feed, meters, verdict)``. The stage honours the layout radio:
*constellation* (MindCards), *split* (omniscient table), or *feed* (feed-only). Each
pane is wrapped in a ``.fishbowl`` scope root so the theater stylesheet applies. When
``thinking`` is set, a "who's thinking…" strip is overlaid on the stage pane."""
cards_html_by_id: dict[str, str] = {}
for card in vm.get("cast", []):
cards_html_by_id[card.get("id", card.get("name", ""))] = render_mindcard(card, mind_reader=mind_reader)
if layout == "split":
stage = render_split(vm)
elif layout == "feed":
stage = render_feed(vm, mind_reader=mind_reader)
else: # constellation (default)
stage = render_constellation(vm, cards_html_by_id)
if thinking:
# Prepend so the badge is the FIRST child of the stage pane: with position:sticky
# it then pins to the top and stays in view without any scrolling.
stage = _thinking_strip(thinking) + stage
feed = render_feed(vm, mind_reader=mind_reader)
meters = render_meters(vm)
# The sober ruling (banner) and its cheerful counterpart (champion celebration) share
# the verdict pane; render_winner is "" unless the run crowned someone, so no-winner
# scenarios get the banner alone.
verdict = render_verdict(vm) + render_winner(vm)
return (
_fishbowl(stage, role=f"fb-stage fb-{layout}"),
_fishbowl(feed, role="fb-feed"),
_fishbowl(meters, role="fb-meters"),
_fishbowl(verdict, role="fb-verdict"),
)
_FILE_PREFIX = "/file="
def _media_local_path(src: str | None, kind: str) -> str | None:
"""Turn a feed media ``src`` into something a native gr.Image/gr.Audio can load.
Live refs are ``/file=`` — strip the prefix to the real filesystem path
(Gradio 5+ serves it under ``/gradio_api/file=`` automatically, so the hand-built
``/file=`` URL the HTML feed used would 404). Offline/stub refs are ``data:`` URIs,
which native components can't take directly: decode once to a stable, content-hashed
file under the media dir (idempotent across ticks → no churn or flicker). An http(s)
URL passes through untouched."""
if not src:
return None
if src.startswith(_FILE_PREFIX):
return src[len(_FILE_PREFIX) :]
if src.startswith("data:"):
return _data_uri_to_file(src, kind)
return src
def _data_uri_to_file(src: str, kind: str) -> str | None:
"""Decode a ``data:;base64,`` URI to a content-hashed file; path or None."""
import base64
import hashlib
from src.media.inference import media_output_dir
try:
header, payload = src.split(",", 1)
mime = header[5:].split(";")[0]
ext = {
"image/png": "png",
"image/jpeg": "jpg",
"image/webp": "webp",
"audio/wav": "wav",
"audio/x-wav": "wav",
"audio/mpeg": "mp3",
}.get(mime, "bin")
raw = base64.b64decode(payload)
out_dir = media_output_dir() / "_native"
out_dir.mkdir(parents=True, exist_ok=True)
path = out_dir / f"{kind}-{hashlib.sha1(raw).hexdigest()[:16]}.{ext}"
if not path.exists():
path.write_bytes(raw)
return str(path)
except Exception: # noqa: BLE001 — media is garnish; never break a render
return None
def _rafters_cards(vm: dict) -> list[dict]:
"""Every illustrated/voiced ``commentate`` beat in this view, as navigable cards.
Each card is ``{"image": path|None, "audio": path|None, "caption": text}`` with media
refs already resolved to native-component-loadable filesystem paths. Drives the
"FROM THE RAFTERS" gallery under the chat (a ``@gr.render`` over a cards ``gr.State``),
so the critic's whole run of beats is browsable, not just the latest."""
import html as _html
cards: list[dict] = []
for item in vm.get("feed") or []:
if item.get("kind") != "commentate":
continue
img = _media_local_path((item.get("image") or {}).get("src"), "img")
audio = _media_local_path((item.get("audio") or {}).get("src"), "tts")
if not (img or audio):
continue # a text-only beat has no card in the gallery
cards.append({"image": img, "audio": audio, "caption": _html.escape(item.get("text") or "")})
return cards
def _render_at(
session: FishbowlSession | None, k: int, *, layout: str, mind_reader: bool, thinking: str | None = None
) -> tuple[str, str, str, str]:
"""Render the Show at play-head *k* (pure prefix view), or empty if no session.
Returns the four HTML panes (stage, feed, meters, verdict). The "FROM THE RAFTERS"
media gallery is driven separately (a cards ``gr.State`` + ``@gr.render``), NOT here —
so the high-frequency autoplay re-render never resets the native audio player mid-load.
``thinking`` (a "who's thinking…" label) overlays the stage while a model call runs."""
vm = session.snapshot(k) if session is not None else _empty_vm()
return render_show_html(vm, layout=layout, mind_reader=mind_reader, thinking=thinking)
def advance_one_tick(session: FishbowlSession | None, k: int, ticks: int, *, max_auto_ticks: int = _MAX_AUTO_TICKS):
"""Decide one autoplay tick without touching Gradio — the loop-safety core.
Returns ``(new_k, new_ticks, stop_reason)``. ``stop_reason`` is ``None`` while the
show should keep playing and a human-readable string once autoplay must halt — on a
verdict at the head (the show resolved), a tripped governor budget, or the
``max_auto_ticks`` backstop. Replaying the existing prefix (k < head) is free and
never counts toward the backstop; only *generating* ticks do. When a live run's judge
has ruled it also closes the run (idempotent ``finalize("verdict")``) so the winner is
recorded — the one non-Gradio side effect. This is the function the timer handler and
the loop-safety tests both drive."""
k = int(k or 0)
ticks = int(ticks or 0)
if session is None:
return 0, 0, None
if getattr(session, "replay", False):
# A loaded past run: replay forward through the recorded prefix, then stop.
# It owns no live engine, so generating is impossible — never spend a tick here.
if k < session.head:
return k + 1, ticks, None
return k, ticks, "end of session — replay complete"
if session.has_verdict():
# The judge ruled (here or on an earlier tick during normal play). Close the run so
# the winner is attributed (run.finished) and the Hall of Fame row is written — the
# force_verdict() curtain-call path only fires at a budget/turn cap, so a naturally
# reached verdict would otherwise never finalize (ADR-0029 / ADR-0035). Idempotent.
if not session.is_finalized():
session.finalize("verdict")
return k, ticks, "verdict reached — the show resolved"
if k < session.head:
return k + 1, ticks, None # replay forward through the existing prefix
if ticks >= max_auto_ticks:
return k, ticks, f"autoplay tick cap {max_auto_ticks} reached"
try:
session.step_one() # at the head → generate ONE agent (stream it, don't wait for the turn)
except BudgetExceeded as exc:
return session.head, ticks, (getattr(exc, "reason", None) or str(exc))
return session.head, ticks + 1, None
def _stopped_banner_html(reason: str) -> str:
"""A ``⛔ STOPPED`` banner reusing the verdict-banner chrome (assets/styles.css).
Surfaced in the verdict pane when the governor trips a budget bound or the
autoplay backstop fires, so the run halts visibly instead of crashing the
Gradio callback or burning tokens in a loop."""
import html as _html
text = _html.escape(reason or "budget exceeded")
return _fishbowl(
'
'
'
⏹ The show has ended
'
f'
{text}
'
"
",
role="fb-verdict",
)
# ── CRT theater chrome ──────────────────────────────────────────────────────────
def _live_chip() -> str:
"""The topbar status chip — computed from the per-backend live-credential gates.
Shows ``● LIVE · MODAL`` / ``● LIVE · HUGGING FACE`` (or ``MODAL+HF`` when both are
configured, with the pulsing dot) naming which inference backend(s) can drive the
cast, else ``OFFLINE · STUB`` so the demo is honest about which path is live.
Offline-first: with no env vars the stub label shows."""
from src.models import inference
configured = inference.configured_backends()
if configured:
name = " + ".join(inference.backend_label(b).upper() for b in configured)
return f'● LIVE · {name}'
return 'OFFLINE · STUB'
def _topbar_html() -> str:
return f"""
◉ FISHBOWLa fishbowl of minds you can read
{_live_chip()}
small minds · one ledger · ≤ 32B
"""
_TOPBAR_HTML = _topbar_html()
# The CSS expects these overlay layers (ui/raw/Fishbowl.html).
_CRT_BG_HTML = ''
_CRT_FG_HTML = ''
# ── app builder ─────────────────────────────────────────────────────────────────
def build_app() -> gr.Blocks:
"""Build the two-tab FISHBOWL theater (defensive about absent leaf modules)."""
# Theme/css/head are applied at launch() time (Gradio 6 moved them off Blocks).
with gr.Blocks(title="FISHBOWL — a fishbowl of minds you can read") as demo:
# CRT background layers (behind everything).
gr.HTML(_CRT_BG_HTML)
# Branding top bar.
gr.HTML(_TOPBAR_HTML)
# ── per-user state ──────────────────────────────────────────────────────
session_state = gr.State(None) # FishbowlSession | None
k_state = gr.State(0) # play-head
scenario_state = gr.State(_DEFAULT_TITLE)
mind_reader_state = gr.State(False)
layout_state = gr.State("constellation")
blank_state = gr.State("") # stand-in input when a leaf widget is absent
stopped_state = gr.State(False) # set once the run halts (budget/backstop/verdict)
tick_count_state = gr.State(0) # consecutive autoplay ticks (the 40-tick backstop)
play_active_state = gr.State(False) # is autoplay running? (drives the Play/Pause label)
# Per-user session id: resolved from the browser's localStorage on load (see
# _SESSION_ID_JS) so it survives reloads. Stamped onto run.started so the
# Archive can list "my sessions only"; a hidden carrier, never shown.
session_id_box = gr.Textbox(value="", visible=False, elem_id="fb-session-id")
# Populated after the tabs build, then read at gr.render runtime (post-load) so
# the Lab's Archive drawer can target the Show's panes it lists into.
archive_refs: dict = {}
with gr.Tabs() as tabs:
with gr.Tab("The Lab", id="lab"):
lab_handles = build_lab()
_build_archive_drawer(
scenario_handle=(lab_handles or {}).get("scenario"),
session_id_box=session_id_box,
refs=archive_refs,
tabs=tabs,
states={
"session": session_state,
"k": k_state,
"scenario": scenario_state,
"layout": layout_state,
"mind": mind_reader_state,
"stopped": stopped_state,
"ticks": tick_count_state,
},
)
with gr.Tab("The Show", id="show"):
show_handles = build_show()
with gr.Tab("Telemetry", id="telemetry"):
build_telemetry()
with gr.Tab("🏆 Hall of Fame", id="hall"):
hall_handles = build_hall_of_fame()
with gr.Tab("About", id="about"):
build_about()
# CRT foreground layers (scanlines + vignette, above content, click-through).
gr.HTML(_CRT_FG_HTML)
# The Archive's gr.render runs client-side after build; by then show_handles
# exists, so it reads the live panes through this ref.
archive_refs["show_handles"] = show_handles or {}
# The Hall of Fame's Replay buttons target the Show panes; wire them after all
# tabs build (mirrors the Archive's deferred show_handles trick at :644).
wire_sessions_render(
hall_handles or {},
refs=archive_refs,
tabs=tabs,
states={
"session": session_state,
"k": k_state,
"scenario": scenario_state,
"layout": layout_state,
"mind": mind_reader_state,
"stopped": stopped_state,
"ticks": tick_count_state,
},
)
_wire(
tabs=tabs,
lab_handles=lab_handles or {},
show_handles=show_handles or {},
session_state=session_state,
k_state=k_state,
scenario_state=scenario_state,
mind_reader_state=mind_reader_state,
layout_state=layout_state,
blank_state=blank_state,
stopped_state=stopped_state,
tick_count_state=tick_count_state,
play_active_state=play_active_state,
session_id_box=session_id_box,
)
# Resolve (or mint) the browser's session id once the page loads; updating the
# hidden box fires its change, which re-renders the Archive list for this user.
demo.load(None, None, [session_id_box], js=_SESSION_ID_JS)
return demo
# ── wiring (the integrator's core job) ──────────────────────────────────────────
def _h(handles: dict, *names):
"""First present handle among *names* (leaf units may name things variously)."""
for n in names:
if n in handles and handles[n] is not None:
return handles[n]
return None
def _wire(
*,
tabs: gr.Tabs,
lab_handles: dict,
show_handles: dict,
session_state: gr.State,
k_state: gr.State,
scenario_state: gr.State,
mind_reader_state: gr.State,
layout_state: gr.State,
blank_state: gr.State,
stopped_state: gr.State,
tick_count_state: gr.State,
play_active_state: gr.State,
session_id_box: gr.Textbox | None = None,
) -> None:
"""Connect Lab/Show component handles to session transport + HTML re-render.
Every handle lookup is defensive: if a leaf unit doesn't expose a widget yet, the
corresponding wiring is simply skipped so the shell still builds and runs."""
# Show output panes (rendered HTML).
stage_out = _h(show_handles, "stage", "stage_html", "constellation")
feed_out = _h(show_handles, "feed", "feed_html")
meters_out = _h(show_handles, "meters", "meters_html")
verdict_out = _h(show_handles, "verdict", "verdict_html")
show_outs = [c for c in (stage_out, feed_out, meters_out, verdict_out) if c is not None]
# The "FROM THE RAFTERS" gallery is driven on its own clock (a cards gr.State + a
# @gr.render in show.py), decoupled from show_outs so the autoplay re-render never
# resets the native audio player. Here we just poll the session into the cards state.
rafters_cards_state = _h(show_handles, "rafters_cards")
rafters_idx_state = _h(show_handles, "rafters_idx")
rafters_timer = _h(show_handles, "rafters_timer")
# Transport controls (looked up early so output lists can include the timer + the
# single Play/Pause hero button — its label travels in the tails so it reverts to
# "▶ Play" the instant the show stops on its own).
timer = _h(show_handles, "timer")
play_btn = _h(show_handles, "play", "play_btn")
_PLAY_LABEL = "▶ Play"
_PAUSE_LABEL = "❚❚ Pause"
# The tail appended to advancing handlers' outputs, in order:
# (timer?, stopped_state, play_active_state, play_btn?)
# ``_halt_tail``/``_run_tail`` keep the returned tuple aligned to whichever are wired.
_tail_outs = (
([timer] if timer is not None else [])
+ [stopped_state, play_active_state]
+ ([play_btn] if play_btn is not None else [])
)
def _halt_tail() -> tuple:
"""Tail values that STOP autoplay and reset the hero button to ▶ Play."""
return (
((gr.update(active=False),) if timer is not None else ())
+ (True, False)
+ ((gr.update(value=_PLAY_LABEL),) if play_btn is not None else ())
)
def _run_tail(play_active: bool = False, stopped: bool = False) -> tuple:
"""Tail values that leave autoplay running: keep the timer + button as-is."""
return (
((gr.update(),) if timer is not None else ())
+ (stopped, bool(play_active))
+ ((gr.update(),) if play_btn is not None else ())
)
# Outputs/values that (re)set the play state when a fresh run begins — shared by
# Summon (auto-start) and Restart. Order: (timer?, play_active_state, play_btn?).
_start_outs = (
([timer] if timer is not None else []) + [play_active_state] + ([play_btn] if play_btn is not None else [])
)
def _start_tail(active: bool) -> tuple:
"""Arm (or idle) the transport for a new run: timer, play_active, button label."""
label = _PAUSE_LABEL if active else _PLAY_LABEL
return (
((gr.update(active=active),) if timer is not None else ())
+ (active,)
+ ((gr.update(value=label),) if play_btn is not None else ())
)
def _stopped_panes(session, k: int, *, layout: str, mind_reader: bool, reason: str) -> tuple:
"""Render the Show at *k* but swap the verdict pane for the STOPPED banner.
Returns a tuple aligned to ``show_outs`` (verdict pane last when present), so
the run halts *visibly* without crashing the callback."""
panes = list(_render_at(session, k, layout=layout, mind_reader=mind_reader))
banner = _stopped_banner_html(reason)
if verdict_out is not None:
panes[3] = banner # the 4th pane (stage, feed, meters, verdict) is the verdict
elif feed_out is not None:
# No verdict pane wired — surface the halt in the feed so it is never silent.
panes[1] = banner
return _pad_values(tuple(panes), show_outs)
# Show transport controls. The scrubber/step/back handles no longer exist on the
# slimmed transport (one Play/Pause button); the lookups stay defensive so any wiring
# that references them is simply skipped.
scrubber = _h(show_handles, "scrubber", "slider", "scrub", "step_slider")
step_btn = _h(show_handles, "step", "step_btn", "next", "advance", "fwd_btn")
back_btn = _h(show_handles, "rewind", "back", "to_start", "first", "back_btn")
judge_btn = _h(show_handles, "judge_btn", "start_judging", "judge_now", "judge")
speed_radio = _h(show_handles, "speed", "speed_radio")
layout_radio = _h(show_handles, "layout", "layout_radio")
mind_toggle = _h(show_handles, "mind_reader", "read_minds", "minds")
restart_btn = _h(show_handles, "restart_btn", "restart", "reset_btn")
poke_send = _h(show_handles, "poke_send", "poke_btn", "poke")
poke_text = _h(show_handles, "poke_text", "poke_input")
poke_buttons = show_handles.get("poke_buttons") or show_handles.get("poke_btns") or []
# Lab controls.
summon_btn = _h(lab_handles, "summon", "summon_btn", "launch", "start")
scenario_in = _h(lab_handles, "scenario", "scenario_select", "scenario_dropdown")
seed_in = _h(lab_handles, "seed", "seed_in", "world_seed")
# Composer inputs — the per-cast Modal model picks (cast_models) and the run knobs;
# all looked up defensively so the shell still runs if the Lab omits a widget.
premise_in = _h(lab_handles, "premise")
cast_models_in = _h(lab_handles, "cast_models")
backend_in = _h(lab_handles, "inference_backend", "backend")
judge_model_in = _h(lab_handles, "judge_model")
judge_policy_in = _h(lab_handles, "judge_policy")
judge_strictness_in = _h(lab_handles, "judge_strictness")
tools_in = _h(lab_handles, "tools")
# New editable surface: per-agent edit States + the scenario roster/genesis/governor.
cast_tools_in = _h(lab_handles, "cast_tools")
cast_personas_in = _h(lab_handles, "cast_personas")
cast_schedules_in = _h(lab_handles, "cast_schedules")
cast_roster_in = _h(lab_handles, "cast_roster")
genesis_in = _h(lab_handles, "world")
max_turns_in = _h(lab_handles, "max_turns")
max_calls_in = _h(lab_handles, "max_calls_per_turn")
max_tokens_in = _h(lab_handles, "max_total_tokens")
hourly_budget_in = _h(lab_handles, "hourly_budget_usd")
def _scenario_title(value) -> str:
"""Resolve a Lab scenario selection (title or internal name) to a title key."""
if value in SCENARIOS:
return value
for title, name in SCENARIOS.items():
if value == name:
return title
return _DEFAULT_TITLE
def _compose_session(name, **knobs):
"""Build a session for a Lab-composed run: the selected Modal models drive the
cast (ADR-0022). The composed WorldConfig flows through ``Registry.from_world``
onto the exact same engine path as a config-file run. Any compose/validate error
degrades to the scenario's default cast so Summon always yields a runnable show
(and with no credentials the deterministic stub drives it, demo reproducible)."""
from src.core.registry import Registry
from src.ui.fishbowl.lab import collect_world_config
def _num(key):
v = knobs.get(key)
return v if isinstance(v, (int, float)) else None
def _dict(key):
v = knobs.get(key)
return v if isinstance(v, dict) else {}
try:
world = collect_world_config(
scenario=name,
premise=knobs.get("premise") or "",
seed=knobs.get("seed") or "",
cast_models=_dict("cast_models"),
judge_policy=knobs.get("judge_policy") or "Majority Vote",
judge_model=knobs.get("judge_model") or "",
judge_strictness=knobs.get("judge_strictness")
if isinstance(knobs.get("judge_strictness"), (int, float))
else 50,
tools=knobs.get("tools") if isinstance(knobs.get("tools"), list) else [],
tokens=_num("tokens"),
max_rounds=_num("max_rounds"),
backend=knobs.get("backend")
if isinstance(knobs.get("backend"), str) and knobs.get("backend")
else "modal",
cast_tools=_dict("cast_tools"),
cast_personas=_dict("cast_personas"),
cast_schedules=_dict("cast_schedules"),
cast_roster=knobs.get("cast_roster") if isinstance(knobs.get("cast_roster"), list) else None,
genesis=knobs.get("genesis") if isinstance(knobs.get("genesis"), str) else None,
max_turns=_num("max_turns"),
max_calls_per_turn=_num("max_calls_per_turn"),
max_total_tokens=_num("max_total_tokens"),
hourly_budget_usd=_num("hourly_budget_usd"),
)
return FishbowlSession(name, registry=Registry.from_world(world), tools=_tools)
except Exception:
return _new_session(name) # bad compose → default cast; Summon never breaks
# ── Summon: build a fresh session from the composed run, reset, jump to the Show ──
def on_summon(
scenario_value,
seed_value,
premise,
cast_models,
judge_model,
judge_policy,
judge_strictness,
tools,
backend,
cast_tools,
cast_personas,
cast_schedules,
cast_roster,
genesis,
max_turns,
max_calls_per_turn,
max_total_tokens,
hourly_budget_usd,
layout,
mind_reader,
session_id,
):
title = _scenario_title(scenario_value)
name = SCENARIOS.get(title, "")
session = (
_compose_session(
name,
premise=premise,
seed=seed_value,
cast_models=cast_models,
judge_model=judge_model,
judge_policy=judge_policy,
judge_strictness=judge_strictness,
tools=tools,
backend=backend,
cast_tools=cast_tools,
cast_personas=cast_personas,
cast_schedules=cast_schedules,
cast_roster=cast_roster,
genesis=genesis,
max_turns=max_turns,
max_calls_per_turn=max_calls_per_turn,
max_total_tokens=max_total_tokens,
hourly_budget_usd=hourly_budget_usd,
)
if name
else None
)
if session is not None:
session.reset((seed_value or "").strip(), session_id=(session_id or "").strip() or None)
k = session.head
else:
k = 0
out = _render_at(session, k, layout=layout, mind_reader=mind_reader)
# Summon starts a fresh run AND begins playing immediately: clear the stopped
# flag, reset the autoplay backstop, arm the timer, and flip the hero button to
# "❚❚ Pause" so the show comes alive the moment the visitor hits Summon.
return (
session,
k,
title,
gr.update(selected="show"),
*_pad_values(out, show_outs),
False,
0,
*_start_tail(True),
)
summon_evt = None # the Summon click event — chained to the kickoff so the show
# plays its first line immediately on Summon (set below when the button exists).
if summon_btn is not None:
summon_inputs = [
scenario_in if scenario_in is not None else scenario_state,
seed_in if seed_in is not None else blank_state, # empty seed when no widget
premise_in if premise_in is not None else blank_state,
cast_models_in if cast_models_in is not None else blank_state,
judge_model_in if judge_model_in is not None else blank_state,
judge_policy_in if judge_policy_in is not None else blank_state,
judge_strictness_in if judge_strictness_in is not None else blank_state,
tools_in if tools_in is not None else blank_state,
backend_in if backend_in is not None else blank_state,
cast_tools_in if cast_tools_in is not None else blank_state,
cast_personas_in if cast_personas_in is not None else blank_state,
cast_schedules_in if cast_schedules_in is not None else blank_state,
cast_roster_in if cast_roster_in is not None else blank_state,
genesis_in if genesis_in is not None else blank_state,
max_turns_in if max_turns_in is not None else blank_state,
max_calls_in if max_calls_in is not None else blank_state,
max_tokens_in if max_tokens_in is not None else blank_state,
hourly_budget_in if hourly_budget_in is not None else blank_state,
layout_state,
mind_reader_state,
session_id_box if session_id_box is not None else blank_state,
]
# The start tail (timer?, play_active, play_btn?) mirrors on_summon's return so the
# show auto-plays the instant it is summoned.
summon_evt = summon_btn.click(
on_summon,
inputs=summon_inputs,
outputs=[
session_state,
k_state,
scenario_state,
tabs,
*show_outs,
stopped_state,
tick_count_state,
*_start_outs,
],
)
# ── ↺ Restart → wipe the run and begin a *fresh context* on the same cast ────
# Unlike a scrub-to-start (pure replay), Restart opens a brand-new run: a fresh seed
# gives a genuinely different conversation (the deterministic stub is seed-driven), a
# new run_id isolates it in the ledger, and the play state re-arms so the new context
# starts playing at once. No tab switch (we're already on the Show) and no cast
# rebuild — it reuses the session the visitor composed. ``restart_evt`` is chained to
# the kickoff below so the first line lands immediately, just like Summon.
restart_evt = None
if restart_btn is not None and show_outs:
def on_restart(session, layout, mind_reader, session_id):
if session is None or getattr(session, "replay", False):
# Nothing live to restart (no Summon yet, or a read-only replay): leave
# the Show as-is and idle the transport rather than erroring.
out = _render_at(session, 0, layout=layout, mind_reader=mind_reader)
return (0, *_pad_values(out, show_outs), False, 0, *_start_tail(False))
session.reset(uuid.uuid4().hex[:8], session_id=(session_id or "").strip() or None)
k = session.head
out = _render_at(session, k, layout=layout, mind_reader=mind_reader)
return (k, *_pad_values(out, show_outs), False, 0, *_start_tail(True))
restart_evt = restart_btn.click(
on_restart,
inputs=[
session_state,
layout_state,
mind_reader_state,
session_id_box if session_id_box is not None else blank_state,
],
outputs=[k_state, *show_outs, stopped_state, tick_count_state, *_start_outs],
)
# ── Scenario picked → re-seed the dependent Lab fields from the registry ─────
# Without this, switching to a new world (e.g. the spy game) leaves the premise,
# seed, cast table, and narrator showing the previous scenario — and Summon would
# genesis with the wrong seed. Refreshing them makes "compose a spy game" one click.
# The cast picker re-seeds itself (it is a gr.render over the scenario + backend), as
# does the Judge model picker (the Lab owns both, since they depend on the chosen
# backend); we re-seed only the static, backend-independent fields here.
_scenario_fields = [
(_h(lab_handles, "premise"), "premise"),
(seed_in, "seed"),
(_h(lab_handles, "world"), "world"),
(_h(lab_handles, "narrator"), "narrator"),
]
_present_fields = [(handle, key) for handle, key in _scenario_fields if handle is not None]
if scenario_in is not None and _present_fields:
def on_scenario_change(scenario_value):
cfg = _registry.scenarios.get(SCENARIOS.get(_scenario_title(scenario_value), ""))
if cfg is None:
return tuple(gr.update() for _ in _present_fields)
# The seed is now an editable textbox (the preset dropdown that fills it is owned
# by the Lab and reseeds itself); refresh its value, not its choices.
updates = {
# Premise is a dropdown-that-supports-text: refresh its single preset
# (the new world's goal) and select it; a custom typed value still wins.
"premise": gr.update(choices=[cfg.goal] if cfg.goal else [], value=cfg.goal),
"seed": gr.update(value=cfg.default_seed),
"world": gr.update(value=cfg.genesis_text or ""),
"narrator": gr.update(value=scenario_voice(cfg.name)),
}
return tuple(updates[key] for _handle, key in _present_fields)
scenario_in.change(
on_scenario_change,
inputs=[scenario_in],
outputs=[handle for handle, _key in _present_fields],
)
# ── ⏭ / ▶ at head → step (generate) then render at the new head ─────────────
# Returns (k, *show_outs, *halt-tail) — a tripped governor stops the timer and
# paints the STOPPED banner instead of crashing the Gradio callback.
def step_at_head(session, layout, mind_reader):
if session is None:
out = _render_at(None, 0, layout=layout, mind_reader=mind_reader)
return (0, *_pad_values(out, show_outs), *_run_tail())
try:
session.step_one() # ⏭ advances one agent so each utterance shows on its own
except BudgetExceeded as exc:
reason = getattr(exc, "reason", None) or str(exc)
panes = _stopped_panes(session, session.head, layout=layout, mind_reader=mind_reader, reason=reason)
return (session.head, *panes, *_halt_tail())
k = session.head
out = _render_at(session, k, layout=layout, mind_reader=mind_reader)
return (k, *_pad_values(out, show_outs), *_run_tail())
if step_btn is not None and show_outs:
step_btn.click(
step_at_head,
inputs=[session_state, layout_state, mind_reader_state],
outputs=[k_state, *show_outs, *_tail_outs],
)
# ── ⚖ Start judging → silence the cast, the judge rules on the whole ledger ──
# The curtain call: stop autoplay (the halt-tail kills the timer), drain the cast's
# pending turns, and have the judge read every event of the run and land a verdict +
# winner. Renders the verdict pane on success; a cast with no judge halts visibly so
# the click is never silent. Returns (k, *show_outs, *halt-tail).
def start_judging(session, layout, mind_reader):
if session is None:
out = _render_at(None, 0, layout=layout, mind_reader=mind_reader)
return (0, *_pad_values(out, show_outs), *_halt_tail())
try:
ruled = session.force_verdict()
except BudgetExceeded as exc:
# A judge that itself trips a bound still halts visibly rather than crashing.
reason = getattr(exc, "reason", None) or str(exc)
panes = _stopped_panes(session, session.head, layout=layout, mind_reader=mind_reader, reason=reason)
return (session.head, *panes, *_halt_tail())
k = session.head
if ruled or session.has_verdict():
out = _render_at(session, k, layout=layout, mind_reader=mind_reader)
return (k, *_pad_values(out, show_outs), *_halt_tail())
# No judge in this cast — surface that instead of stopping silently.
panes = _stopped_panes(
session, k, layout=layout, mind_reader=mind_reader, reason="no judge in this cast to rule"
)
return (k, *panes, *_halt_tail())
if judge_btn is not None and show_outs:
judge_btn.click(
start_judging,
inputs=[session_state, layout_state, mind_reader_state],
outputs=[k_state, *show_outs, *_tail_outs],
)
# ── scrubber / ⏮ → pure prefix view (no stepping) ───────────────────────────
def scrub_to(session, k, layout, mind_reader):
kk = int(k or 0)
out = _render_at(session, kk, layout=layout, mind_reader=mind_reader)
return (kk, *_pad_values(out, show_outs))
if scrubber is not None and show_outs:
scrubber.change(
scrub_to,
inputs=[session_state, scrubber, layout_state, mind_reader_state],
outputs=[k_state, *show_outs],
)
def to_start(session, layout, mind_reader):
out = _render_at(session, 0, layout=layout, mind_reader=mind_reader)
return (0, *_pad_values(out, show_outs))
if back_btn is not None and show_outs:
back_btn.click(
to_start,
inputs=[session_state, layout_state, mind_reader_state],
outputs=[k_state, *show_outs],
)
# ── gr.Timer.tick → hybrid: advance k (replay) below head, else step ────────
# Loop-safe: stops autoplay on a tripped governor, on a verdict at the head (the
# show resolved), or after the governor-derived tick cap of consecutive generating
# ticks (the hard backstop). The cap tracks the scenario budget so a long show that
# ends on a late Judge verdict is never cut off early. Returns (k, *show_outs,
# *halt-tail, tick_count).
# A *generator*: when a tick is about to GENERATE a new line (vs. cheaply replay the
# existing prefix), it first yields a frame with a "who's thinking…" strip on the
# stage, then runs the (seconds-long) model call, then yields the result. So the
# theater shows who we're waiting on instead of freezing silently.
def on_tick(session, k, layout, mind_reader, tick_count, play_active):
kk = int(k or 0)
cap = session.autoplay_tick_cap if session is not None else _MAX_AUTO_TICKS
will_generate = (
session is not None
and not getattr(session, "replay", False)
and not session.has_verdict()
and kk >= session.head
and int(tick_count or 0) < cap
)
if will_generate and show_outs:
who = session.peek_next_actor_name()
phrase = f"{who} is thinking" if who else "the cast is thinking"
hint = _render_at(session, kk, layout=layout, mind_reader=mind_reader, thinking=phrase)
yield (kk, *_pad_values(hint, show_outs), *_run_tail(play_active), tick_count)
new_k, new_ticks, stop_reason = advance_one_tick(session, kk, tick_count, max_auto_ticks=cap)
if stop_reason is None:
out = _render_at(session, new_k, layout=layout, mind_reader=mind_reader)
yield (new_k, *_pad_values(out, show_outs), *_run_tail(play_active), new_ticks)
return
# The cast's run is over (budget/turn cap, a reached verdict, or replay end). If a
# judge can still rule and hasn't, bring it on — the limit is the curtain call, so
# the show ends on a verdict + winner rather than a silent STOPPED banner.
live = session is not None and not getattr(session, "replay", False)
if live and not session.has_verdict() and session.has_judge():
try:
session.force_verdict()
except BudgetExceeded:
pass # a judge that trips its own bound falls through to the banner below
new_k = session.head if session is not None else new_k
if session is not None and session.has_verdict():
# Render the ruling itself (the verdict pane), timer halted.
out = _render_at(session, new_k, layout=layout, mind_reader=mind_reader)
yield (new_k, *_pad_values(out, show_outs), *_halt_tail(), new_ticks)
return
# No judge to rule (or a replay reaching its end): halt visibly with the banner.
panes = _stopped_panes(session, new_k, layout=layout, mind_reader=mind_reader, reason=stop_reason)
yield (new_k, *panes, *_halt_tail(), new_ticks)
if timer is not None and show_outs:
timer.tick(
on_tick,
inputs=[session_state, k_state, layout_state, mind_reader_state, tick_count_state, play_active_state],
outputs=[k_state, *show_outs, *_tail_outs, tick_count_state],
)
# ── FROM THE RAFTERS gallery → poll the session into the cards state ──────────
# Decoupled from the show re-render so the native audio player isn't reset every
# autoplay tick (the cause of the "voice stuck loading"): this low-frequency clock
# rebuilds the critic's beat list and writes it to the cards gr.State ONLY when it
# actually changed (gr.skip() otherwise → @gr.render doesn't re-run → audio is
# left alone). When a NEW beat lands it jumps the view to it; navigating between
# cards (the @gr.render's ‹/› buttons) sets the index directly and is never clobbered.
if rafters_timer is not None and rafters_cards_state is not None and rafters_idx_state is not None:
def _refresh_rafters(session, old_cards, old_idx):
cards = _rafters_cards(session.snapshot(session.head)) if session is not None else []
old_cards = old_cards or []
if cards == old_cards:
return gr.skip(), gr.skip() # nothing new → don't touch the gallery
# A fresh beat arrives → reveal it; otherwise clamp the current view in range.
idx = len(cards) - 1 if len(cards) > len(old_cards) else min(int(old_idx or 0), max(0, len(cards) - 1))
return cards, idx
rafters_timer.tick(
_refresh_rafters,
inputs=[session_state, rafters_cards_state, rafters_idx_state],
outputs=[rafters_cards_state, rafters_idx_state],
show_progress="hidden",
)
# ── speed radio → timer interval; play/pause → timer.active ──────────────────
if speed_radio is not None and timer is not None:
def on_speed(speed):
return gr.update(value=SPEEDS.get(speed, SPEEDS["1×"]))
speed_radio.change(on_speed, inputs=[speed_radio], outputs=[timer])
if play_btn is not None and timer is not None:
# The single hero control: the same button plays and pauses, its label flipping
# between "▶ Play" and "❚❚ Pause". Each fresh ▶ resets the autoplay backstop so
# the next run gets a full tick budget rather than inheriting the old one.
def on_play(active):
new = not bool(active)
label = _PAUSE_LABEL if new else _PLAY_LABEL
return new, gr.update(active=new), 0, gr.update(value=label)
play_evt = play_btn.click(
on_play,
inputs=[play_active_state],
outputs=[play_active_state, timer, tick_count_state, play_btn],
)
if show_outs:
# Don't make the visitor wait a whole timer interval for the first line: the
# instant Play is pressed (play_active just flipped True), advance one step
# right away — reusing the tick path so the "who's thinking…" hint shows while
# the model runs. On pause (play_active now False) we simply re-render the
# current frame without stepping. Chained after the toggle so it reads the
# freshly-updated play_active/tick_count.
def kickoff(session, k, layout, mind_reader, tick_count, play_active):
if not play_active:
out = _render_at(session, int(k or 0), layout=layout, mind_reader=mind_reader)
yield (int(k or 0), *_pad_values(out, show_outs), *_run_tail(False), int(tick_count or 0))
return
yield from on_tick(session, k, layout, mind_reader, tick_count, play_active)
_kickoff_io = dict(
inputs=[session_state, k_state, layout_state, mind_reader_state, tick_count_state, play_active_state],
outputs=[k_state, *show_outs, *_tail_outs, tick_count_state],
)
play_evt.then(kickoff, **_kickoff_io)
# Summon and Restart both auto-start a fresh run; kick their first line off
# immediately too, instead of waiting a whole timer interval.
if summon_evt is not None:
summon_evt.then(kickoff, **_kickoff_io)
if restart_evt is not None:
restart_evt.then(kickoff, **_kickoff_io)
# ── layout radio + "Read their minds" → re-render the stage HTML ─────────────
if layout_radio is not None and show_outs:
def on_layout(value, session, k, mind_reader):
out = _render_at(session, int(k or 0), layout=value, mind_reader=mind_reader)
return (value, *_pad_values(out, show_outs))
layout_radio.change(
on_layout,
inputs=[layout_radio, session_state, k_state, mind_reader_state],
outputs=[layout_state, *show_outs],
)
if mind_toggle is not None and show_outs:
def on_minds(value, session, k, layout):
out = _render_at(session, int(k or 0), layout=layout, mind_reader=bool(value))
return (bool(value), *_pad_values(out, show_outs))
mind_toggle.change(
on_minds,
inputs=[mind_toggle, session_state, k_state, layout_state],
outputs=[mind_reader_state, *show_outs],
)
# ── poke buttons / poke_send → inject then render at the new head ────────────
# Injecting steps the conductor (a full turn of model calls), so a poke can trip the
# governor too — wrap it. A *generator*: yield a "the wood stirs…" hint first so the
# disturbance feels alive while the cast reacts, then the result (or the STOPPED card).
def _poke_after(session, text, label_value, layout, mind_reader, play_active):
if session is None:
out = _render_at(None, 0, layout=layout, mind_reader=mind_reader)
yield (0, *_pad_values(out, show_outs), *_run_tail(play_active))
return
if show_outs:
hint = _render_at(session, session.head, layout=layout, mind_reader=mind_reader, thinking="the wood stirs")
yield (session.head, *_pad_values(hint, show_outs), *_run_tail(play_active))
try:
session.inject((text or ""), label=label_value)
except BudgetExceeded as exc:
reason = getattr(exc, "reason", None) or str(exc)
panes = _stopped_panes(session, session.head, layout=layout, mind_reader=mind_reader, reason=reason)
yield (session.head, *panes, *_halt_tail())
return
k = session.head
out = _render_at(session, k, layout=layout, mind_reader=mind_reader)
yield (k, *_pad_values(out, show_outs), *_run_tail(play_active))
if poke_send is not None and poke_text is not None and show_outs:
# Free-text poke: the textbox supplies the text; a neutral label.
def on_poke_send(session, text, layout, mind_reader, play_active):
yield from _poke_after(session, text, None, layout, mind_reader, play_active)
poke_send.click(
on_poke_send,
inputs=[session_state, poke_text, layout_state, mind_reader_state, play_active_state],
outputs=[k_state, *show_outs, *_tail_outs],
)
for btn in poke_buttons:
if btn is None or not show_outs:
continue
# Preset poke buttons: the button's own label is both the disturbance text
# and its label; capture it per-button so each button injects its own.
label = str(getattr(btn, "value", None) or "DISTURBANCE")
def make_preset(label_value: str):
def _on_preset(session, layout, mind_reader, play_active):
yield from _poke_after(session, label_value, label_value, layout, mind_reader, play_active)
return _on_preset
btn.click(
make_preset(label),
inputs=[session_state, layout_state, mind_reader_state, play_active_state],
outputs=[k_state, *show_outs, *_tail_outs],
)
# ── small output helper (keep emitted values aligned to present panes) ──────────
def _pad_values(values, show_outs: list) -> tuple:
"""Trim a (stage, feed, meters, verdict) tuple to the present panes' count."""
return tuple(values[: len(show_outs)])
# ── Archive drawer ("my past sessions" → read-only replay) ──────────────────────
# Resolve (or mint) a per-browser session id from localStorage on page load. Kept in
# JS so the id is the user's own and survives reloads — Python only reads the carrier.
_SESSION_ID_JS = """
() => {
try {
let id = localStorage.getItem('fishbowl_session_id');
if (!id) {
id = (window.crypto && crypto.randomUUID)
? crypto.randomUUID()
: 'sess-' + Math.random().toString(36).slice(2) + Date.now().toString(36);
localStorage.setItem('fishbowl_session_id', id);
}
return id;
} catch (e) {
return 'sess-' + Math.random().toString(36).slice(2);
}
}
"""
def _title_for(value) -> str:
"""Resolve a scenario title *or* internal name to a display-title key."""
if value in SCENARIOS:
return value
for title, name in SCENARIOS.items():
if value == name:
return title
return _DEFAULT_TITLE
def _show_outs(show_handles: dict) -> list:
"""The present Show panes (stage, feed, meters, verdict), in render order."""
stage = _h(show_handles, "stage", "stage_html", "constellation")
feed = _h(show_handles, "feed", "feed_html")
meters = _h(show_handles, "meters", "meters_html")
verdict = _h(show_handles, "verdict", "verdict_html")
return [c for c in (stage, feed, meters, verdict) if c is not None]
def _archive_empty_html() -> str:
"""The drawer's empty state — no past runs for this world yet."""
return _fishbowl(
'
'
'
⟳ No past sessions
'
'
Summon the bowl, and your runs in this world '
"will gather here — yours alone, replayable any time.
"
"
",
role="fb-archive",
)
def _build_archive_drawer(*, scenario_handle, session_id_box, refs: dict, tabs, states: dict) -> None:
"""The Lab's "Past sessions" accordion: clickable phosphor cards → read-only replay.
A ``gr.render`` keyed on (scenario, session id) lists *this user's* runs for the
*current* world via :func:`list_runs`; clicking a card loads that run with
:func:`load_replay` and jumps to the Show. The list re-renders on world change,
when the session id resolves from localStorage, and on the manual refresh.
"""
if scenario_handle is None: # no scenario picker → nothing to scope a list to
return
with gr.Accordion("⟲ Past sessions · this world", open=False, elem_classes=["archive-drawer"]):
refresh = gr.Button("⟳ refresh", size="sm", elem_classes=["archive-refresh"], scale=0)
@gr.render(
inputs=[scenario_handle, session_id_box],
triggers=[scenario_handle.change, session_id_box.change, refresh.click],
)
def _render_archive(scenario_value, session_id):
name = SCENARIOS.get(_title_for(scenario_value), "")
runs = list_runs(name, session_id)
if not runs:
gr.HTML(_archive_empty_html())
return
show_outs = _show_outs(refs.get("show_handles") or {})
n_out = 6 + len(show_outs) # session, k, scenario, tabs, *panes, stopped, ticks
def _loader(run_id: str):
def _load(layout, mind_reader):
session = load_replay(run_id, registry=_registry, tools=_tools)
if session is None:
return tuple(gr.update() for _ in range(n_out))
k = session.head # land on the full discussion; scrub/▶ replays it
out = _render_at(session, k, layout=layout, mind_reader=mind_reader)
return (
session,
k,
_title_for(session.scenario_name),
gr.update(selected="show"),
*_pad_values(out, show_outs),
False,
0,
)
return _load
for summary in runs:
card = gr.Button(run_card_label(summary), elem_classes=["archive-card"])
card.click(
_loader(summary.run_id),
inputs=[states["layout"], states["mind"]],
outputs=[
states["session"],
states["k"],
states["scenario"],
tabs,
*show_outs,
states["stopped"],
states["ticks"],
],
)
# ── dev server port (ported from the original root app.py) ──────────────────────
def on_spaces() -> bool:
"""True when running inside a Hugging Face Space (incl. ZeroGPU).
HF injects ``SPACE_ID`` into every Space container; nothing else sets it."""
return bool(os.getenv("SPACE_ID"))
@spaces.GPU
def gpu_selftest() -> str:
"""ZeroGPU entrypoint — the one ``@spaces.GPU`` function the platform requires.
ZeroGPU aborts startup unless it detects a decorated GPU function (the decorator
registers itself at import, which is all the platform's startup check needs — this
need never be *called*, so it costs no GPU quota on page load). We still make it a
real probe rather than a no-op: a tiny on-device matmul that proves a GPU was
actually attached. The cast's inference runs on remote OpenAI-compatible endpoints,
so this is the engine's only direct touch of local silicon; torch is imported lazily
here so module import never initialises CUDA (which would trip ZeroGPU's fork guard)."""
import torch
if not torch.cuda.is_available(): # pragma: no cover — only reachable on real ZeroGPU
return "cuda unavailable"
device = torch.device("cuda")
probe = (torch.randn(256, 256, device=device) @ torch.randn(256, 256, device=device)).sum()
return f"{torch.cuda.get_device_name(device)} · checksum {probe.item():.3f}"
def dev_server_port() -> int:
# The platform's reverse proxy only reaches the app on the port it forwards to, so
# an explicit ``GRADIO_SERVER_PORT``/``PORT`` from the environment MUST win over the
# dev-range scanner below. On HF Spaces the proxy forwards to 7860 but only sets
# ``GRADIO_SERVER_NAME`` (not the port) — without this, the scanner picks 7960 and
# the Space builds "successfully" yet is permanently unreachable (the bug this fixes).
configured = os.getenv("GRADIO_SERVER_PORT") or os.getenv("PORT")
if configured:
return int(configured)
if on_spaces():
return 7860 # HF Spaces' fixed forward port; never scan a range it can't see.
for port in range(7960, 8060):
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
try:
sock.bind(("127.0.0.1", port))
return port
except OSError:
continue
raise RuntimeError("No free development port found in range 7960-8059.")
# ── module-level app + launch ────────────────────────────────────────────────────
demo = build_app()
def _launch_kwargs() -> dict:
"""Theme/css/head applied at launch() time (Gradio 6), when available."""
kwargs: dict = {"css": load_css()}
if FISHBOWL_HEAD:
kwargs["head"] = FISHBOWL_HEAD
theme = FishbowlTheme() if callable(FishbowlTheme) else FishbowlTheme
if theme is not None:
kwargs["theme"] = theme
# Serve the commentator's live media (the rafters-critic's generated images/audio) via
# Gradio's /file= path. Live artifacts are written here and referenced by path, so the
# exported ledger/trace stays lean; offline stub media is inlined as a data: URI and
# needs no allow-listing. The dir may not exist yet — created on first write.
from src.media import media_output_dir
media_dir = media_output_dir()
media_dir.mkdir(parents=True, exist_ok=True)
kwargs["allowed_paths"] = [str(media_dir)]
return kwargs
def launch(**overrides):
"""Launch the FISHBOWL theater with theme/css/head + a free dev port.
The single entry point the root shim and ``uv run app.py`` call; keeps the
no-API-key offline behaviour (the deterministic stub drives the cast)."""
kwargs = {"server_port": dev_server_port(), **_launch_kwargs()}
if on_spaces():
# Bind every interface so HF's proxy can reach us (defensive — HF also sets
# GRADIO_SERVER_NAME), and drop Gradio 6's SSR Node proxy: on ZeroGPU it spawns
# a second (Node) process and a +1 internal port purely to pre-render the first
# paint. This live, stream-driven theater gets ~nothing from SSR, so disabling
# it trims cold-start time and memory on the resource-capped Space.
kwargs.setdefault("server_name", "0.0.0.0")
kwargs.setdefault("ssr_mode", False)
kwargs.update(overrides)
return demo.launch(**kwargs)
if __name__ == "__main__":
launch()