"""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_agent_model_falls_back_to_profile_without_endpoint(self): class NoEndpoint: name, role, model_profile, model_endpoint = "echo", "worker", "balanced", None assert adapter.agent_model(NoEndpoint()) == "balanced" assert adapter.agent_tier(NoEndpoint()) == "mid" def test_agent_model_honours_endpoint_override(self): # An unknown/offline catalogue key degrades to the raw key (never blank), and the # tier still falls back to the declared profile — the Show stays honest offline. class WithEndpoint: name, role, model_profile, model_endpoint = "pocket", "worker", "fast", "no-such-key" assert adapter.agent_model(WithEndpoint()) == "no-such-key" assert adapter.agent_tier(WithEndpoint()) == "fast" 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" # A cast member's world.observed is credited to that agent (a say line), while a # non-cast actor (genesis / the scenario itself) stays the anonymous narrator. cast = ["scene-whisperer"] whisper = adapter.event_to_feed_item(_ev("world.observed", "scene-whisperer", text="the wood wakes"), cast) assert whisper["kind"] == "say" and whisper["agent"] == "scene-whisperer" genesis = adapter.event_to_feed_item(_ev("world.observed", "thousand-token-wood", text="stirs awake"), cast) assert genesis["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"), # Genesis: emitted by the scenario itself (not a cast member) → narrator voice. _ev("world.observed", "thousand-token-wood", turn=0, text="The wood stirs awake around 'seed'."), _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" # Every card carries the model it is running (profile name when it routes by # profile) so the Show can badge each agent with its model. assert all(c["model"] for c in vm["cast"]) assert pa["model"] == pa["model_profile"] # no endpoint override → profile name 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"] == 4 # the sim-turn at the play-head (head event is turn 4) 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, 4, 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"})