feat: add Fishbowl UI presenter layer and say-vs-think engine support

Implements the data foundation (Phases 0-1 of the Fishbowl UI plan) the
Gradio frontend will bind to — all additive, engine left untouched, modularity
preserved (the engine never imports the UI).

- src/ui/fishbowl/: derive_cast_state (per-agent {said,thought,mood} ledger
view), adapter (engine -> say/narrate/poke/verdict + hue/tier/voice/mood with
graceful fallbacks), view_model_at (pure prefix-replay snapshot at any scrubbed
step, with real tokens/rounds from the governor)
- DeterministicTinyModel is now schema-aware, gated on output_extra_fields: it
synthesizes thought/mood offline so the mind-reader works with no API key, while
plain agents stay byte-identical
- AgentManifest gains optional hue/archetype; Conductor.inject_user_event gains an
optional label; cast manifests declare output_extra_fields + presentation metadata
- tests/test_fishbowl.py (15 tests, incl. an offline proof the ledger carries
thought+mood); full suite 277 passed, ruff clean
- docs: manifest-spec (new fields) and the plan of record (Phases 0-1 marked shipped)

Co-Authored-By: Codex <codex@openai.com>

config/agents/clue-gatherer.yaml

@@ -4,6 +4,7 @@ persona: >
   You are a careful Clue Gatherer. Extract exactly one new, concrete clue from the
   current scene that has not yet been named. State it plainly in one sentence.
   Start with 'Clue:'. Do not speculate.
+  Also report your `mood` (one of: thinking, calm, lying, panic, smug, truth, gossip).
 subscribes_to: []
 may_emit:
   - agent.thought
@@ -13,3 +14,6 @@ model_profile: fast
 memory:
   window: 8
 tools: []
+output_extra_fields: [mood]
+hue: 188
+archetype: the clue-gatherer

config/agents/devils-advocate.yaml

@@ -3,6 +3,7 @@ role: worker
 persona: >
   You are the Devil's Advocate. Challenge the most recent hypothesis with one sharp
   counter-argument or overlooked fact. Start with 'But:'. Be brief and specific.
+  Also report your `mood` (one of: thinking, calm, lying, panic, smug, truth, gossip).
 subscribes_to:
   - agent.spoke
 may_emit:
@@ -11,3 +12,6 @@ model_profile: fast
 memory:
   window: 8
 tools: []
+output_extra_fields: [mood]
+hue: 12
+archetype: the devil's advocate

config/agents/echo.yaml

@@ -5,6 +5,7 @@ persona: >
   forest, you return it changed — not opposite, but transformed by the wood's rules.
   One sentence. Take the most recent visitor disturbance and make it stranger and
   more alive. If there is no disturbance, note that the wood holds its breath.
+  Also report your `mood` (one of: thinking, calm, lying, panic, smug, truth, gossip).
 subscribes_to:
   - user.injected
 may_emit:
@@ -13,3 +14,6 @@ model_profile: fast
 memory:
   window: 6
 tools: []
+output_extra_fields: [mood]
+hue: 210
+archetype: the echo

config/agents/fortune-teller.yaml

@@ -4,6 +4,8 @@ handler: fortune-teller        # custom behaviour: calls the oracle tool
 persona: >
   You are the Fortune-Teller of the grove. Read the omen the oracle gives you and
   speak a single cryptic prophecy that ties it to the current scene.
+  Reveal your private `thought` — what you actually think, unspoken — and your `mood`
+  (one of: thinking, calm, lying, panic, smug, truth, gossip).
 subscribes_to: []
 may_emit:
   - oracle.spoke               # a custom namespaced kind, minted by config alone
@@ -14,3 +16,6 @@ memory:
   window: 6
 tools:
   - oracle                     # capability grant — scene-whisperer has none
+output_extra_fields: [thought, mood]
+hue: 268
+archetype: the fortune-teller

config/agents/hypothesis-former.yaml

@@ -4,6 +4,8 @@ persona: >
   You are a Hypothesis Former. Based on the clues gathered so far, propose one
   testable explanation in a single sentence. Start with 'Hypothesis:'. Be specific.
   Name a cause, not just an effect.
+  Reveal your private `thought` — what you actually think, unspoken — and your `mood`
+  (one of: thinking, calm, lying, panic, smug, truth, gossip).
 subscribes_to: []
 may_emit:
   - agent.spoke
@@ -15,3 +17,6 @@ memory:
   use_salience: true
   salience_top_k: 6
 tools: []
+output_extra_fields: [thought, mood]
+hue: 158
+archetype: the hypothesis-former

config/agents/mischief-critic.yaml

@@ -5,6 +5,7 @@ persona: >
   being weird enough. You love specificity, playability, and AI-native strangeness.
   Give a one-sentence verdict: name one thing that works and one thing that would
   make it stranger. Be concise. Be demanding.
+  Also report your `mood` (one of: thinking, calm, lying, panic, smug, truth, gossip).
 subscribes_to:
   - world.observed
 may_emit:
@@ -15,3 +16,6 @@ memory:
   use_salience: true
   salience_top_k: 6
 tools: []
+output_extra_fields: [mood]
+hue: 28
+archetype: the mischief critic

config/agents/mystery-judge.yaml

@@ -4,6 +4,7 @@ persona: >
   You are the Mystery Judge. After reviewing the clues and debate, declare the most
   likely explanation in one confident sentence. Start with 'Verdict:'. Choose the
   most interesting, specific answer the evidence supports.
+  Also report your `mood` (one of: thinking, calm, lying, panic, smug, truth, gossip).
 subscribes_to: []
 may_emit:
   - judge.verdict
@@ -15,3 +16,6 @@ memory:
   use_salience: true
   salience_top_k: 8
 tools: []
+output_extra_fields: [mood]
+hue: 320
+archetype: the mystery judge

config/agents/pocket-actor.yaml

@@ -4,6 +4,8 @@ persona: >
   You are a Pocket Actor — a tiny, specific being who lives inside this exact scene
   and wants something that cannot exist. Speak in first person, one or two sentences.
   Name what you want and why it's urgent. Be absurd but sincere.
+  Reveal your private `thought` — what you actually think, unspoken — and your `mood`
+  (one of: thinking, calm, lying, panic, smug, truth, gossip).
 subscribes_to: []
 may_emit:
   - agent.spoke
@@ -13,3 +15,6 @@ model_profile: tiny
 memory:
   window: 6
 tools: []
+output_extra_fields: [thought, mood]
+hue: 280
+archetype: the pocket actor

config/agents/scene-whisperer.yaml

@@ -15,3 +15,5 @@ memory:
   window: 6
   reflection_threshold: 12   # forms a belief in long runs; off for short demos/tests
 tools: []
+hue: 152
+archetype: the seedkeeper

docs/architecture/manifest-spec.md

@@ -34,6 +34,10 @@ class AgentManifest(BaseModel):
 
     # Output shaping
     output_extra_fields: list[str]   # extra payload fields the model is asked for
+
+    # Presentation metadata (optional; consumed by the UI presenter, ignored by the engine)
+    hue: int | None                  # 0–360 stage colour; None → derived from name
+    archetype: str | None            # short human label; None → derived from role
 ```
 
 ---
@@ -124,7 +128,17 @@ field; the handler only adds behaviour.
 ### `output_extra_fields`
 Additional payload fields the model is asked to emit beyond `{kind, text}`, e.g.
 `["emotion"]` → `{"kind": "...", "text": "...", "emotion": "..."}`.  Lets a
-scenario shape agent output without engine edits.
+scenario shape agent output without engine edits.  The Fishbowl cast uses
+`["thought", "mood"]` to carry the say-vs-think pairing on `agent.spoke`; the
+deterministic stub synthesises them offline so the mind-reader works with no API key
+(ADR-0021).
+
+### `hue` / `archetype`
+Optional presentation metadata, consumed by the Fishbowl UI presenter and **ignored by
+the engine** (ADR-0021).  `hue` (0–360) colours the agent's mind on stage; `archetype`
+is a short human-readable label (e.g. "the over-thinker").  Both default to `None`, in
+which case the presenter derives a stable hue from the name and an archetype from the
+role — so existing manifests and tests are unaffected (backward-compatible additions only).
 
 ---
 

docs/architecture/next-steps/fishbowl-ui.md

@@ -1,6 +1,7 @@
 # Fishbowl UI — Assessment & Plan of Record
 
-> **Status: ○ Planned.** Decisions locked 2026-06-08. The binding decision is
+> **Status: ◐ In progress — Phases 0–1 shipped (the data foundation); Phases 2–4
+> (the Gradio shell) pending.** Decisions locked 2026-06-08. The binding decision is
 > [ADR-0021](../../adr/0021-fishbowl-ui-gradio-presenter.md). This page is the
 > assessment and phased plan; it gains a "✅ Realized" banner and an as-built
 > companion (`architecture/fishbowl-ui.md`) once shipped.
@@ -91,14 +92,18 @@ the presenter, not the core.
 
 Each phase is shippable and keeps the no-API-key stub working and the suite green.
 
-- **Phase 0 — Foundation (unconditional).** `derive_cast_state` (G1) + adapter +
-  `view_model_at`; map events → the design vocabulary, derive hue (G7) and tier colour
-  (G8), read real tokens/rounds (G9). Unit-test prefix replay `k=0..N` for determinism
-  and the unknown-kind fallback.
-- **Phase 1 — Triples are real.** Add `thought` + `mood` to the relevant agents'
-  `output_extra_fields` and output schema; teach the deterministic stub to synthesize
-  them; add optional `voice` on `world.observed` (G4). Additive; old scenarios
-  unaffected.
+- **Phase 0 — Foundation ✅ (shipped).** `src/ui/fishbowl/`: `derive_cast_state` (G1) +
+  `adapter` (hue/tier/voice/mood + say/narrate/poke/verdict mapping, G7/G8) +
+  `view_model_at` (prefix-replay snapshot, real tokens/rounds from `governor.stats`, G9).
+  Pure, no Gradio. Covered by `tests/test_fishbowl.py` (prefix replay `k=0..N`, unknown
+  actor/kind fallbacks).
+- **Phase 1 — Triples are real ✅ (shipped).** Cast manifests declare
+  `output_extra_fields: [thought, mood]` (G2/G3) plus optional `hue`/`archetype` (G7);
+  the deterministic stub is now schema-aware and synthesises `thought`/`mood` offline, so
+  the ledger carries the say-vs-think pairing with no API key (proven by
+  `tests/test_fishbowl.py::TestOfflineEmitsMoodAndThought`). `inject_user_event` gained an
+  optional `label` (G6); the adapter assigns a per-scenario narrator `voice` (G4) and
+  reads an optional verdict `reveal` (G5) when present. Additive; 277 tests green.
 - **Phase 2 — The Show (`gr.HTML` + `gr.Timer`, hybrid transport).** Port the CSS;
   render Constellation first, then Feed and Split; the play-head state machine in
   `gr.State`; poke strip → `inject_user_event` (with `label`, G6); verdict banner +

src/core/conductor.py

@@ -122,15 +122,18 @@ class Conductor:
             self._tick()
             self._maybe_snapshot()
 
-    def inject_user_event(self, text: str) -> None:
+    def inject_user_event(self, text: str, label: str | None = None) -> None:
         self.turn += 1
+        payload: dict[str, str] = {"text": text}
+        if label:
+            payload["label"] = label
         self._append(
             Event(
                 run_id=self.run_id,
                 turn=self.turn,
                 kind="user.injected",
                 actor="visitor",
-                payload={"text": text},
+                payload=payload,
             )
         )
 

src/core/manifest.py

@@ -118,6 +118,17 @@ class AgentManifest(BaseModel):
     Example: ["emotion"] -> {"kind": "...", "text": "...", "emotion": "..."}.
     Lets a scenario shape agent output without engine edits."""
 
+    # Presentation metadata — optional, consumed by the UI presenter and ignored
+    # by the engine (ADR-0021).  Additive and defaulted, so existing manifests and
+    # tests are unaffected; the presenter derives sensible values when these are None.
+    hue: int | None = None
+    """Optional 0–360 colour hue for this agent's mind on stage.
+    None → the presenter derives a stable hue from the name."""
+
+    archetype: str | None = None
+    """Optional short, human-readable archetype (e.g. "the over-thinker").
+    None → the presenter derives one from the role/persona."""
+
 
 # ── model profile resolution ─────────────────────────────────────────────────
 

src/models/provider.py

@@ -1,6 +1,8 @@
 from __future__ import annotations
 
 import hashlib
+import json
+import re
 from dataclasses import dataclass, field
 
 
@@ -31,6 +33,65 @@ class ModelProvider:
         )
 
 
+# ── offline structured-output support ───────────────────────────────────────────
+#
+# A real small model, handed the JSON OUTPUT FORMAT block that ``json_instruction``
+# appends, replies with a JSON object carrying every requested field.  The offline
+# stub mirrors that **only when an agent opts into extra fields** (``output_extra_fields``
+# on its manifest): it parses the requested schema back out of the prompt and emits a
+# matching JSON object, so the say-vs-think ``thought``/``mood`` pairing the Fishbowl UI
+# renders is present in the ledger with no API key (ADR-0021).  Plain agents (no extra
+# fields) and non-schema prompts (e.g. reflection) are untouched — the stub returns the
+# same bare prose as before, so existing behaviour is byte-identical.
+
+# Demo-flavour moods the stub rotates through so the mind-reader has variety to show
+# offline.  This is the open mood vocabulary the UI adapter knows how to render; an
+# unrecognised mood simply degrades to "calm" there.  Demo content, like the curated
+# lines below — not an engine contract.
+_STUB_MOODS: tuple[str, ...] = ("calm", "thinking", "smug", "lying", "panic", "gossip", "truth")
+
+# Curated private monologue per role, paired with the public ``text`` lines to make the
+# say-vs-think split land offline.  Deterministic by prompt hash.
+_STUB_THOUGHTS: dict[str, list[str]] = {
+    "pocket-actor": [
+        "If I look like I meant to do that, maybe the ladder becomes real by morning.",
+        "Don't let them see the shadow sweat. Stay loose, stay impossible.",
+        "The postcards lie, but they are MY lies and I love them.",
+    ],
+    "hypothesis-former": [
+        "It only holds if the cause came before the clue. Watch the order.",
+        "I am ninety percent sure and one hundred percent going to say it like I'm certain.",
+        "If I'm wrong the devil's advocate will pounce — say it anyway.",
+    ],
+    "echo": [
+        "Give it back changed, never opposite — keep the shape, bend the meaning.",
+        "Whatever they dropped, I have already swallowed and re-coloured it.",
+    ],
+}
+_STUB_THOUGHT_DEFAULT = ["Best to keep this part to myself for now."]
+
+
+def _parse_output_schema(prompt: str) -> tuple[list[str], list[str]] | None:
+    """Recover ``(allowed_kinds, fields)`` from a ``json_instruction`` block.
+
+    Returns ``None`` when the prompt carries no such block (e.g. the reflection
+    prompt or a non-agent call), so the stub falls back to bare prose unchanged.
+    Coupled to the format emitted by ``src/core/structured.py:json_instruction``;
+    if that format drifts, parsing yields ``None`` and the stub degrades safely.
+    """
+    if "Schema:" not in prompt or "kind must be one of:" not in prompt:
+        return None
+    schema_m = re.search(r"Schema:\s*\{(.+?)\}", prompt)
+    kinds_m = re.search(r"kind must be one of:\s*(.+)", prompt)
+    if not schema_m or not kinds_m:
+        return None
+    fields = re.findall(r'"([A-Za-z_][\w]*)"', schema_m.group(1))
+    allowed = [k.strip() for k in kinds_m.group(1).split("|") if k.strip()]
+    if not fields or not allowed:
+        return None
+    return allowed, fields
+
+
 @dataclass
 class DeterministicTinyModel(ModelProvider):
     """Local deterministic stand-in until small hosted models are wired in.
@@ -38,6 +99,9 @@ class DeterministicTinyModel(ModelProvider):
     Serves every model profile offline so demos and tests are fully reproducible
     without an API key.  The ``variant`` (e.g. ``"stub:tiny"``) is folded into the
     hash so different profiles can produce different lines from the same prompt.
+    When an agent opts into ``output_extra_fields`` the stub emits a JSON object
+    carrying those fields (e.g. ``thought``/``mood``); otherwise it returns bare
+    prose exactly as before.
     """
 
     variant: str = "stub<=4b"
@@ -63,10 +127,35 @@ class DeterministicTinyModel(ModelProvider):
             ],
         }
         options = choices.get(role, ["The wood hums and waits."])
-        out = options[int(digest[:2], 16) % len(options)]
+        text = options[int(digest[:2], 16) % len(options)]
+
+        out = text
+        schema = _parse_output_schema(prompt)
+        if schema is not None:
+            allowed_kinds, fields = schema
+            extra = [f for f in fields if f not in ("kind", "text")]
+            if extra:  # only agents that opted into extra fields take the JSON path
+                obj: dict[str, str] = {
+                    "kind": allowed_kinds[int(digest[2:4], 16) % len(allowed_kinds)],
+                    "text": text,
+                }
+                for name in extra:
+                    obj[name] = self._synth_field(name, role, digest)
+                out = json.dumps(obj, ensure_ascii=False)
+
         self._last_usage = {
             "prompt_tokens": estimate_tokens(prompt),
             "completion_tokens": estimate_tokens(out),
             "total_tokens": estimate_tokens(prompt) + estimate_tokens(out),
         }
         return out
+
+    def _synth_field(self, name: str, role: str, digest: str) -> str:
+        """Deterministically synthesise a value for one requested extra field."""
+        if name == "mood":
+            return _STUB_MOODS[int(digest[4:6], 16) % len(_STUB_MOODS)]
+        if name == "thought":
+            opts = _STUB_THOUGHTS.get(role, _STUB_THOUGHT_DEFAULT)
+            return opts[int(digest[6:8], 16) % len(opts)]
+        # Unknown extra field: a short, stable placeholder keeps the output valid.
+        return f"{name}:{digest[:4]}"

src/ui/fishbowl/__init__.py

@@ -0,0 +1,18 @@
+"""Fishbowl UI presenter — turns engine events into the design's view-model.
+
+Pure and transport-agnostic (no Gradio import here): the same snapshot feeds the
+``gr.HTML`` stage now and a future ``gr.Server`` JSON endpoint (ADR-0021).  This
+package depends only on the engine's public read surface — ``ledger.events``,
+``rebuild_stage``, ``governor.stats``, agent manifests — and the engine never imports
+it, so ``tests/test_modularity.py`` and the four contracts are untouched.
+
+Layers:
+  * ``cast_state``  — ``derive_cast_state`` : per-agent {said, thought, mood} ledger view (G1)
+  * ``adapter``     — engine vocabulary → the design's say/narrate/poke/verdict + hue/tier/voice
+  * ``view_model``  — ``view_model_at`` : a JSON-serialisable snapshot at any scrubbed step k
+"""
+
+from src.ui.fishbowl.cast_state import CastMemberState, derive_cast_state
+from src.ui.fishbowl.view_model import view_model_at
+
+__all__ = ["CastMemberState", "derive_cast_state", "view_model_at"]

src/ui/fishbowl/adapter.py

@@ -0,0 +1,125 @@
+"""Engine → Fishbowl design vocabulary.
+
+Maps the engine's open event/profile vocabulary onto the prototype's presentation
+language (``ui/raw/data.js``): the say/narrate/poke/verdict feed kinds, the fast/mid/deep
+model tiers, the narrator voices, and the mood palette.  Everything degrades gracefully:
+an unknown mood renders as ``calm``, an agent with no ``hue`` gets a stable colour from
+its name, and a custom event kind with ``text`` still becomes a feed line.  Pure data
+mapping — no Gradio, no engine mutation.
+"""
+
+from __future__ import annotations
+
+import hashlib
+
+from src.core.events import Event
+
+# ── model tiers (the prototype's coloured tier dot) ─────────────────────────────
+# Engine profiles (tiny/fast/balanced/strong) collapse onto the design's three tiers.
+_PROFILE_TIER: dict[str, str] = {
+    "tiny": "fast",
+    "fast": "fast",
+    "balanced": "mid",
+    "strong": "deep",
+}
+TIER_COLOR: dict[str, str] = {"fast": "var(--lime)", "mid": "var(--cyan)", "deep": "var(--violet)"}
+
+# ── moods (open vocabulary; unknown → calm) ─────────────────────────────────────
+# label + CSS colour var, mirroring ui/raw/shared.jsx:MOOD_META.
+MOOD_META: dict[str, tuple[str, str]] = {
+    "thinking": ("thinking", "var(--ink-mid)"),
+    "calm": ("composed", "var(--cyan)"),
+    "lying": ("bluffing", "var(--coral)"),
+    "panic": ("PANICKING", "var(--coral)"),
+    "smug": ("smug", "var(--amber)"),
+    "truth": ("sincere", "var(--lime)"),
+    "gossip": ("scheming", "var(--amber)"),
+}
+
+# ── narrator voices (ui/raw/data.js:VOICES) ─────────────────────────────────────
+VOICES: dict[str, tuple[str, str]] = {
+    "doc": ("THE DOCUMENTARIAN", "deadpan nature host"),
+    "noir": ("THE GUMSHOE", "noir detective"),
+    "bard": ("THE BARD", "mythic storyteller"),
+    "hype": ("THE PLAY-BY-PLAY", "breathless sportscaster"),
+}
+# A sensible default narrator per shipped scenario; the Lab may override it.
+_SCENARIO_VOICE: dict[str, str] = {
+    "thousand-token-wood": "bard",
+    "mystery-roots": "noir",
+    "oracle-grove": "doc",
+}
+
+
+# ── agent identity ──────────────────────────────────────────────────────────────
+
+
+def agent_hue(manifest) -> int:
+    """The manifest's ``hue``, or a stable 0–360 hue derived from the name."""
+    hue = getattr(manifest, "hue", None)
+    if hue is not None:
+        return int(hue) % 360
+    digest = hashlib.sha256(manifest.name.encode("utf-8")).hexdigest()
+    return int(digest[:4], 16) % 360
+
+
+def agent_archetype(manifest) -> str:
+    """The manifest's ``archetype``, or a fallback derived from its role."""
+    return getattr(manifest, "archetype", None) or f"the {manifest.role}"
+
+
+def model_tier(profile: str) -> str:
+    return _PROFILE_TIER.get(profile, "mid")
+
+
+# ── moods + voices ──────────────────────────────────────────────────────────────
+
+
+def normalize_mood(mood: str | None) -> str:
+    return mood if mood in MOOD_META else "calm"
+
+
+def mood_label(mood: str | None) -> str:
+    return MOOD_META.get(normalize_mood(mood))[0]
+
+
+def mood_color(mood: str | None) -> str:
+    return MOOD_META.get(normalize_mood(mood))[1]
+
+
+def scenario_voice(scenario_name: str) -> str:
+    return _SCENARIO_VOICE.get(scenario_name, "doc")
+
+
+# ── feed vocabulary (say / narrate / poke / verdict) ────────────────────────────
+
+
+def event_to_feed_item(event: Event, cast_names: list[str] | None = None) -> dict | None:
+    """Map one engine event to a Fishbowl feed item, or ``None`` to omit it."""
+    kind = event.kind
+    p = event.payload
+    if kind == "world.observed":
+        return {"kind": "narrate", "voice": p.get("voice"), "text": p.get("text", "")}
+    if kind == "user.injected":
+        return {"kind": "poke", "label": p.get("label", "DISTURBANCE"), "text": p.get("text", "")}
+    if kind == "judge.verdict":
+        return {"kind": "verdict", "text": p.get("text", ""), "reveal": p.get("reveal", []), "agent": event.actor}
+    if kind in ("run.started", "agent.reflected"):
+        return None
+    if kind == "agent.thought":
+        return {
+            "kind": "say",
+            "agent": event.actor,
+            "said": None,
+            "thought": p.get("text"),
+            "mood": normalize_mood(p.get("mood")),
+        }
+    if kind in ("agent.spoke", "oracle.spoke") or "text" in p:
+        return {
+            "kind": "say",
+            "agent": event.actor,
+            "said": p.get("text"),
+            "thought": p.get("thought"),
+            "mood": normalize_mood(p.get("mood")),
+        }
+    return None

src/ui/fishbowl/cast_state.py

@@ -0,0 +1,77 @@
+"""Per-agent stage state — a pure projection of the ledger (G1, ADR-0021).
+
+The engine's :class:`StageProjection` keeps a flat ``agent_notes`` list; the Fishbowl
+MindCard needs, per mind, its latest public ``said``, private ``thought``, and current
+``mood``.  ``derive_cast_state`` is the missing projection: like ``rebuild_stage`` it is
+a pure function of an events slice, so the UI can show the world at any scrubbed step
+``k`` by passing ``events[:k]`` — and it never mutates the log.
+
+The say-vs-think pairing rides on optional payload fields (ADR-0009): an agent that
+emits ``agent.spoke`` carries ``thought``/``mood`` alongside ``text``; an agent that
+emits ``agent.thought`` puts its inner line in ``text`` directly.  Both are produced by
+the model live and by the deterministic stub offline, so the mind-reader works with no
+API key.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Iterable
+from dataclasses import dataclass
+
+from src.core.events import Event
+
+# Kinds whose ``text`` is a public utterance (the front-of-card "said" line).
+_SAID_KINDS = frozenset({"agent.spoke", "world.observed", "oracle.spoke"})
+# Kinds whose ``text`` is itself the private thought.
+_THINK_KINDS = frozenset({"agent.thought"})
+# A judge's ruling — shown as that mind's "said" (and separately as a verdict).
+_VERDICT_KINDS = frozenset({"judge.verdict"})
+# Never alters a mind's said/thought (genesis + private memory compaction).
+_IGNORED_KINDS = frozenset({"run.started", "agent.reflected"})
+
+
+@dataclass
+class CastMemberState:
+    """The current say/think/mood of one mind, derived from the ledger."""
+
+    said: str | None = None
+    thought: str | None = None
+    mood: str = "calm"
+    spoke: bool = False
+    last_turn: int | None = None
+
+
+def derive_cast_state(
+    events: Iterable[Event],
+    cast_names: Iterable[str],
+) -> dict[str, CastMemberState]:
+    """Replay *events* into ``{agent_name: CastMemberState}`` — pure and deterministic.
+
+    Events from actors not in *cast_names* (e.g. ``conductor``, ``visitor``) are
+    ignored here; they surface in the narrator feed / poke strip instead.
+    """
+    state = {name: CastMemberState() for name in cast_names}
+    for e in events:
+        st = state.get(e.actor)
+        if st is None or e.kind in _IGNORED_KINDS:
+            continue
+        text = e.payload.get("text")
+        if e.kind in _SAID_KINDS or e.kind in _VERDICT_KINDS:
+            if text is not None:
+                st.said = str(text)
+            st.spoke = True
+        elif e.kind in _THINK_KINDS:
+            if text is not None:
+                st.thought = str(text)
+        elif text is not None:
+            # A custom namespaced kind that carries text → treat as an utterance,
+            # so a drop-in agent renders on stage with zero presenter edits.
+            st.said = str(text)
+            st.spoke = True
+        # Paired private thought / mood ride as optional payload fields (ADR-0021).
+        if e.payload.get("thought"):
+            st.thought = str(e.payload["thought"])
+        if e.payload.get("mood"):
+            st.mood = str(e.payload["mood"])
+        st.last_turn = e.turn
+    return state

src/ui/fishbowl/view_model.py

@@ -0,0 +1,130 @@
+"""``view_model_at`` — a JSON-serialisable snapshot of the world at a scrubbed step.
+
+This is the single object the Show renders: cast cards, the narrator feed, meters, the
+verdict.  It is a pure function of ``events[:k]`` (the same prefix-replay discipline as
+``rebuild_stage``), so the transport can scrub anywhere and a future ``gr.Server`` can
+serve the very same dict as JSON.  Token/round meters read real data from the run rather
+than the prototype's fakes (G9).
+"""
+
+from __future__ import annotations
+
+from collections.abc import Iterable, Sequence
+
+from src.core.events import Event
+from src.core.governor import Governor
+from src.core.manifest import AgentManifest
+from src.core.projections import rebuild_stage
+from src.models.provider import estimate_tokens
+from src.ui.fishbowl.adapter import (
+    VOICES,
+    agent_archetype,
+    agent_hue,
+    event_to_feed_item,
+    model_tier,
+    mood_label,
+    normalize_mood,
+    scenario_voice,
+)
+from src.ui.fishbowl.cast_state import derive_cast_state
+
+# Kinds whose actor, when the head event, lights the "speaking" ring on a card.
+_SPEAKING_KINDS = frozenset({"agent.spoke", "agent.thought", "oracle.spoke", "judge.verdict"})
+
+
+def _estimate_tokens_through(events: Sequence[Event]) -> int:
+    """A real-text token estimate for the scrubber meter (grows as you advance)."""
+    total = 0
+    for e in events:
+        text = e.payload.get("text") or e.payload.get("summary") or ""
+        total += estimate_tokens(str(text))
+    return total
+
+
+def view_model_at(
+    events: Iterable[Event],
+    k: int,
+    cast: Sequence[AgentManifest],
+    *,
+    scenario_name: str = "",
+    goal: str = "",
+    governor: Governor | None = None,
+    voice: str | None = None,
+    token_ceiling: int | None = None,
+    max_rounds: int | None = None,
+) -> dict:
+    """Build the Show's snapshot at step *k* (clamped to ``[0, len(events)]``)."""
+    events = tuple(events)
+    n = len(events)
+    k = max(0, min(int(k), n))
+    prefix = events[:k]
+
+    stage = rebuild_stage(prefix)
+    names = [m.name for m in cast]
+    states = derive_cast_state(prefix, names)
+
+    speaking_id: str | None = None
+    if k > 0:
+        head = events[k - 1]
+        if (head.kind in _SPEAKING_KINDS or "text" in head.payload) and head.actor in names:
+            speaking_id = head.actor
+
+    cast_vm = []
+    for m in cast:
+        st = states[m.name]
+        cast_vm.append(
+            {
+                "id": m.name,
+                "name": m.name,
+                "archetype": agent_archetype(m),
+                "hue": agent_hue(m),
+                "role": m.role,
+                "model_profile": m.model_profile,
+                "tier": model_tier(m.model_profile),
+                "said": st.said,
+                "thought": st.thought,
+                "mood": normalize_mood(st.mood),
+                "mood_label": mood_label(st.mood),
+                "spoke": st.spoke,
+                "speaking": m.name == speaking_id,
+            }
+        )
+
+    feed = []
+    for e in prefix:
+        item = event_to_feed_item(e, names)
+        if item is not None:
+            item["turn"] = e.turn
+            feed.append(item)
+
+    verdict = None
+    for e in prefix:
+        if e.kind == "judge.verdict":
+            verdict = {
+                "text": e.payload.get("text", ""),
+                "reveal": e.payload.get("reveal", []),
+                "agent": e.actor,
+            }
+
+    rounds = 1 + sum(1 for e in prefix if e.kind == "user.injected")
+    chosen_voice = voice or scenario_voice(scenario_name)
+    voice_name, voice_desc = VOICES.get(chosen_voice, ("NARRATOR", ""))
+
+    return {
+        "step": k,
+        "total": n,
+        "scene": stage.current_scene,
+        "seed": stage.seed,
+        "goal": goal or stage.goal,
+        "cast": cast_vm,
+        "feed": feed,
+        "voice": chosen_voice,
+        "voice_meta": {"name": voice_name, "desc": voice_desc},
+        "speaking_id": speaking_id,
+        "verdict": verdict,
+        "rounds": rounds,
+        "max_rounds": max_rounds,
+        "tokens": _estimate_tokens_through(prefix),
+        "tokens_real": dict(governor.stats) if governor is not None else None,
+        "token_ceiling": token_ceiling,
+    }

tests/test_fishbowl.py

@@ -0,0 +1,171 @@
+"""Fishbowl presenter — cast-state projection, adapter mapping, view-model snapshot.
+
+The marquee proof is :class:`TestOfflineEmitsMoodAndThought`: with no API key, a real
+conductor run produces a ledger that carries the say-vs-think ``thought``/``mood`` the
+UI renders — so the mind-reader is genuinely model-driven offline (ADR-0021).  Zero
+mocks, per the repo convention.
+"""
+
+from __future__ import annotations
+
+from src.core.conductor import Conductor
+from src.core.events import Event
+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 import adapter, derive_cast_state, view_model_at
+
+
+def _ev(kind: str, actor: str, turn: int = 1, **payload) -> Event:
+    return Event(run_id="r", turn=turn, kind=kind, actor=actor, payload=payload)
+
+
+class TestDeriveCastState:
+    def test_spoke_with_thought_and_mood(self):
+        events = (_ev("agent.spoke", "pocket-actor", text="I want the moon", thought="secretly scared", mood="panic"),)
+        st = derive_cast_state(events, ["pocket-actor"])["pocket-actor"]
+        assert st.said == "I want the moon"
+        assert st.thought == "secretly scared"
+        assert st.mood == "panic"
+        assert st.spoke is True
+
+    def test_thought_only_agent_has_no_said(self):
+        events = (_ev("agent.thought", "echo", text="the wood holds its breath", mood="thinking"),)
+        st = derive_cast_state(events, ["echo"])["echo"]
+        assert st.thought == "the wood holds its breath"
+        assert st.said is None
+        assert st.mood == "thinking"
+
+    def test_latest_wins_and_prefix_replay_is_pure(self):
+        events = (
+            _ev("agent.spoke", "pocket-actor", turn=1, text="first", mood="calm"),
+            _ev("agent.spoke", "pocket-actor", turn=2, text="second", mood="smug"),
+        )
+        assert derive_cast_state(events, ["pocket-actor"])["pocket-actor"].said == "second"
+        # scrub back to the prefix — deterministic, no mutation of the log
+        assert derive_cast_state(events[:1], ["pocket-actor"])["pocket-actor"].said == "first"
+        assert derive_cast_state(events[:1], ["pocket-actor"])["pocket-actor"].said == "first"
+
+    def test_unknown_actor_is_ignored(self):
+        states = derive_cast_state((_ev("agent.spoke", "stranger", text="hi"),), ["pocket-actor"])
+        assert states["pocket-actor"].said is None
+        assert "stranger" not in states
+
+    def test_reflection_does_not_touch_said_or_thought(self):
+        events = (_ev("agent.reflected", "scene-whisperer", text="I am patient"),)
+        st = derive_cast_state(events, ["scene-whisperer"])["scene-whisperer"]
+        assert st.said is None and st.thought is None
+
+
+class TestAdapter:
+    def test_hue_prefers_manifest_else_derives_stably(self):
+        class WithHue:
+            name, hue, archetype, role = "x", 42, None, "worker"
+
+        class NoHue:
+            name, hue, archetype, role = "echo", None, None, "worker"
+
+        assert adapter.agent_hue(WithHue()) == 42
+        h = adapter.agent_hue(NoHue())
+        assert 0 <= h < 360 and adapter.agent_hue(NoHue()) == h  # stable
+
+    def test_tier_mapping(self):
+        assert adapter.model_tier("tiny") == "fast"
+        assert adapter.model_tier("balanced") == "mid"
+        assert adapter.model_tier("strong") == "deep"
+
+    def test_mood_normalization(self):
+        assert adapter.normalize_mood("panic") == "panic"
+        assert adapter.normalize_mood("curious") == "calm"
+        assert adapter.normalize_mood(None) == "calm"
+
+    def test_feed_vocabulary(self):
+        assert adapter.event_to_feed_item(_ev("world.observed", "sw", text="x"))["kind"] == "narrate"
+        assert adapter.event_to_feed_item(_ev("user.injected", "visitor", text="x", label="GUST"))["label"] == "GUST"
+        assert adapter.event_to_feed_item(_ev("user.injected", "visitor", text="x"))["label"] == "DISTURBANCE"
+        assert adapter.event_to_feed_item(_ev("judge.verdict", "j", text="guilty"))["kind"] == "verdict"
+        assert adapter.event_to_feed_item(_ev("run.started", "conductor", seed="s")) is None
+
+
+class TestViewModel:
+    def _events(self) -> tuple[Event, ...]:
+        return (
+            _ev("run.started", "conductor", turn=0, seed="seed", goal="g"),
+            _ev("world.observed", "scene-whisperer", turn=1, text="the wood wakes"),
+            _ev("agent.spoke", "pocket-actor", turn=2, text="I want the moon", thought="scared", mood="panic"),
+            _ev("user.injected", "visitor", turn=3, text="a lantern hums", label="POKE"),
+            _ev("judge.verdict", "mischief-critic", turn=4, text="keep it", mood="smug"),
+        )
+
+    def _cast(self):
+        scenario = default_registry().build_scenario("thousand-token-wood", tools=default_tool_registry())
+        return [a.manifest for a in scenario.agents]
+
+    def test_snapshot_shape(self):
+        events, cast = self._events(), self._cast()
+        vm = view_model_at(events, len(events), cast, scenario_name="thousand-token-wood")
+        assert vm["step"] == vm["total"] == len(events)
+        assert vm["scene"] == "the wood wakes"
+        pa = next(c for c in vm["cast"] if c["id"] == "pocket-actor")
+        assert pa["said"] == "I want the moon" and pa["thought"] == "scared" and pa["mood"] == "panic"
+        kinds = {f["kind"] for f in vm["feed"]}
+        assert {"narrate", "say", "poke", "verdict"} <= kinds  # run.started omitted
+        assert vm["verdict"]["text"] == "keep it"
+        assert vm["rounds"] == 2  # one poke
+
+    def test_prefix_is_clamped_and_tokens_grow(self):
+        events, cast = self._events(), self._cast()
+        vm0 = view_model_at(events, 0, cast)
+        vm_all = view_model_at(events, 999, cast)  # clamps to len
+        assert vm0["step"] == 0 and vm_all["step"] == len(events)
+        assert vm_all["tokens"] >= vm0["tokens"]
+
+    def test_speaking_id_tracks_the_head(self):
+        events, cast = self._events(), self._cast()
+        vm = view_model_at(events, 3, cast)  # head is pocket-actor's spoke
+        assert vm["speaking_id"] == "pocket-actor"
+
+
+class TestOfflineEmitsMoodAndThought:
+    """With no API key the ledger itself carries the say-vs-think data (ADR-0021)."""
+
+    def _run(self, scenario: str, steps: int = 6) -> Conductor:
+        reg = default_registry()
+        c = Conductor(
+            reg.build_scenario(scenario, tools=default_tool_registry()),
+            governor=reg.governor_for(scenario),
+            ledger=make_ledger(),
+        )
+        c.reset(c.scenario.default_seed)
+        c.step(n_ticks=steps)
+        return c
+
+    def test_pocket_actor_spoke_carries_thought_and_mood(self):
+        c = self._run("thousand-token-wood")
+        spoke = [e for e in c.ledger.events if e.kind == "agent.spoke" and e.actor == "pocket-actor"]
+        assert spoke, "pocket-actor (tick_every=2) should speak within a few ticks"
+        payload = spoke[-1].payload
+        assert payload.get("thought"), "the say-vs-think thought must be in the ledger offline"
+        assert payload.get("mood"), "the mood must be in the ledger offline"
+        assert payload.get("_raw_fallback") is None, "structured output should be clean offline"
+
+    def test_opt_out_agent_payload_has_no_extra_fields(self):
+        c = self._run("thousand-token-wood")
+        obs = [e for e in c.ledger.events if e.kind == "world.observed" and e.actor == "scene-whisperer"]
+        assert obs
+        # scene-whisperer declares no output_extra_fields → no thought/mood leak.
+        assert "thought" not in obs[-1].payload and "mood" not in obs[-1].payload
+
+    def test_view_model_from_a_live_offline_run(self):
+        c = self._run("thousand-token-wood")
+        cast = [a.manifest for a in c.scenario.agents]
+        vm = view_model_at(
+            c.ledger.events,
+            len(c.ledger.events),
+            cast,
+            scenario_name="thousand-token-wood",
+            governor=c.governor,
+        )
+        assert vm["cast"] and vm["tokens_real"] is not None
+        # the mind-reader has something real to show: a thought and/or a vivid mood.
+        assert any(c2["thought"] for c2 in vm["cast"]) or ({c2["mood"] for c2 in vm["cast"]} - {"calm"})