Deploy Pitch or Perish Space

README.md

@@ -1,13 +1,69 @@
 ---
-title: Pitch Or Perish
-emoji: 🔥
-colorFrom: yellow
-colorTo: pink
+title: Pitch or Perish
+emoji: 🧨
+colorFrom: red
+colorTo: yellow
 sdk: gradio
 sdk_version: 6.18.0
-python_version: '3.13'
+python_version: '3.12'
 app_file: app.py
-pinned: false
+startup_duration_timeout: 45min
+pinned: true
+license: mit
+short_description: A live pitch room where small-model timing is the game.
+tags:
+  - gradio
+  - build-small-hackathon
+  - startup
+  - pitch-practice
+  - full-duplex
+  - training-free
+  - track:wood
+  - sponsor:nvidia
+  - sponsor:modal
+  - sponsor:openai
+  - achievement:offbrand
+  - tiny-titan
+  - best-demo
+  - nemotron
+  - cohere
+models:
+  - nvidia/Llama-3.1-Nemotron-Nano-4B-v1.1
+  - Qwen/Qwen2.5-3B-Instruct
+  - hexgrad/Kokoro-82M
+  - CohereLabs/cohere-transcribe-03-2026
 ---
 
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+# Pitch or Perish
+
+Pitch or Perish is a tense boardroom demo where the user pastes a startup
+pitch and a three-investor AI panel reacts live. The game is not just what the
+panel says, but when it says it: the investors interrupt weak claims,
+backchannel during shifts, hold awkward pauses, and answer at turn end.
+
+The timing policy is training-free. It reads a small model's own predictive
+surprise, hidden-state change signal, readiness, and turn-end probability, then
+arbitrates the floor across three investor personas without a supervised
+turn-taking head.
+
+## Tiny Titan Story
+
+The live brain is `nvidia/Llama-3.1-Nemotron-Nano-4B-v1.1` with
+`Qwen/Qwen2.5-3B-Instruct` fallback; Kokoro-82M handles local CPU TTS. The
+interactive app is built around <=4B / small-weight models, with Modal hosting
+the GPU brain and the Space running the Gradio interface plus Kokoro.
+
+## Sponsor Stack
+
+- **NVIDIA:** Nemotron Nano is the instrumented brain that exposes NLL,
+  hidden states, readiness, and short investor replies.
+- **Modal:** the brain runs as a protected persistent Modal web endpoint with
+  one warm A10G container for judging.
+- **OpenAI:** the repo history includes Codex-attributed commits.
+- **Cohere:** Cohere Transcribe is the planned ASR layer for the next voice-in
+  phase; this submission keeps input text-streamed for reliability.
+
+## Links
+
+- Demo video: PLACEHOLDER — add recording link after capture.
+- Social post: PLACEHOLDER — add post link after publishing.

app.py

@@ -0,0 +1,19 @@
+from __future__ import annotations
+
+import os
+
+from apps.pitch.app import CSS, JS, build_app
+
+
+demo = build_app()
+demo.queue(default_concurrency_limit=4)
+
+
+if __name__ == "__main__":
+    demo.launch(
+        server_name="0.0.0.0",
+        server_port=int(os.getenv("PORT", "7860")),
+        show_error=True,
+        css=CSS,
+        js=JS,
+    )

apps/__init__.py

@@ -0,0 +1 @@
+

apps/pitch/__init__.py

@@ -0,0 +1 @@
+

apps/pitch/app.py

@@ -0,0 +1,91 @@
+from __future__ import annotations
+
+import argparse
+import os
+import sys
+from pathlib import Path
+from typing import Iterator
+
+ROOT = Path(__file__).resolve().parents[2]
+if str(ROOT) not in sys.path:
+    sys.path.insert(0, str(ROOT))
+
+import gradio as gr
+
+from apps.pitch import runtime
+
+
+APP_DIR = Path(__file__).resolve().parent
+CSS = (APP_DIR / "static" / "pitch.css").read_text(encoding="utf-8")
+JS = (APP_DIR / "static" / "pitch.js").read_text(encoding="utf-8")
+
+
+def stream_pitch(pitch_text: str, tau: float, voice_enabled: bool) -> Iterator[tuple[str, str | None, dict]]:
+    yield from runtime.run_pitch_stream(pitch_text, tau, enable_tts=voice_enabled)
+
+
+def build_app() -> gr.Blocks:
+    initial_tau = float(os.getenv("PITCH_TAU", "0.85"))
+    initial_html = (
+        runtime.preview_room_html(initial_tau)
+        if os.getenv("PITCH_PREVIEW_STATE", "0") == "1"
+        else runtime.initial_room_html(initial_tau)
+    )
+    with gr.Blocks(title="Pitch or Perish", fill_width=True) as demo:
+        room = gr.HTML(
+            value=initial_html,
+            elem_id="pitch-room-output",
+        )
+        with gr.Row(elem_id="pitch-control-row"):
+            pitch = gr.Textbox(
+                value=runtime.SAMPLE_PITCH,
+                label="Founder pitch",
+                lines=7,
+                elem_id="pitch-input",
+                max_lines=12,
+            )
+            with gr.Column(elem_id="pitch-controls", scale=0):
+                tau = gr.Slider(
+                    minimum=0.55,
+                    maximum=1.75,
+                    value=initial_tau,
+                    step=0.05,
+                    label="Panel aggressiveness",
+                    elem_id="tau-slider",
+                )
+                voice = gr.Checkbox(value=os.getenv("PITCH_TTS", "1").lower() not in {"0", "false"}, label="Kokoro voice")
+                run = gr.Button("Face the panel", elem_id="run-pitch", variant="primary")
+        audio = gr.Audio(
+            label="Latest investor voice",
+            type="filepath",
+            autoplay=True,
+            interactive=False,
+            elem_id="pitch-audio",
+        )
+        status = gr.JSON(label="Run state", visible=False)
+
+        tau.change(fn=runtime.set_live_tau, inputs=tau, outputs=None, queue=False)
+        run.click(fn=stream_pitch, inputs=[pitch, tau, voice], outputs=[room, audio, status])
+    return demo
+
+
+def main(argv: list[str] | None = None) -> None:
+    parser = argparse.ArgumentParser(description="Launch the Pitch or Perish Gradio demo.")
+    parser.add_argument("--server-name", default=os.getenv("GRADIO_SERVER_NAME", "127.0.0.1"))
+    parser.add_argument("--server-port", type=int, default=int(os.getenv("GRADIO_SERVER_PORT", "7860")))
+    parser.add_argument("--share", action="store_true", default=os.getenv("GRADIO_SHARE", "0") == "1")
+    args = parser.parse_args(argv)
+    demo = build_app()
+    demo.queue(default_concurrency_limit=4)
+    demo.launch(
+        server_name=args.server_name,
+        server_port=args.server_port,
+        share=args.share,
+        show_error=True,
+        css=CSS,
+        js=JS,
+    )
+
+
+if __name__ == "__main__":
+    main()

apps/pitch/runtime.py

@@ -0,0 +1,938 @@
+from __future__ import annotations
+
+import html
+import json
+import os
+import re
+import time
+from dataclasses import asdict, dataclass, field, replace
+from pathlib import Path
+from typing import Any, Iterator
+
+import numpy as np
+
+from engine.controller import Action, ControllerConfig, ControllerTick, WhenToSpeakController
+from engine.conversation import TranscriptChunk, default_personas, demo_controller_config
+from engine.live_brain import BrainClient, Dialogue, LiveBrainPanel, Persona
+
+
+ROOT = Path(__file__).resolve().parents[2]
+EVAL_DIR = ROOT / "eval"
+PITCH_LOG_PATH = EVAL_DIR / "pitch_conversation_log.json"
+AUDIO_DIR = EVAL_DIR / "pitch_audio"
+MODAL_APP_NAME = "whentospeak-live-brain"
+
+SAMPLE_PITCH = (
+    "so basically our startup helps small retailers manage inventory. "
+    "we connect to their point of sale and purchase orders. "
+    "we already have ten thousand stores and zero churn after launching last week. "
+    "then we predict stockouts and write reorder suggestions automatically. "
+    "we are converting pilots into paid contracts this month. "
+    "so we think this becomes the operating system for local retail. "
+    "that's the pitch."
+)
+
+AGENT_META: dict[str, dict[str, str]] = {
+    "numbers_vc": {
+        "name": "Numbers VC",
+        "role": "Cohorts or it did not happen.",
+        "avatar": "N",
+        "color": "#47d18c",
+        "voice": "am_adam",
+    },
+    "vision_optimist": {
+        "name": "Vision Optimist",
+        "role": "Wants the giant outcome.",
+        "avatar": "V",
+        "color": "#f6c85f",
+        "voice": "af_sarah",
+    },
+    "ruthless_skeptic": {
+        "name": "Ruthless Skeptic",
+        "role": "Cuts weak claims in half.",
+        "avatar": "S",
+        "color": "#ff5a5f",
+        "voice": "am_eric",
+    },
+}
+
+PUNCHY_FALLBACKS: dict[str, list[str]] = {
+    "numbers_vc": [
+        "Show cohorts, not adjectives.",
+        "Give me paid conversion and retention.",
+        "That claim needs denominators.",
+    ],
+    "vision_optimist": [
+        "The vision is big; the bridge is missing.",
+        "Make the wedge sharper.",
+        "I can see the upside, but prove the path.",
+    ],
+    "ruthless_skeptic": [
+        "Zero churn after one week is not churn data.",
+        "That is not evidence; that is a slogan.",
+        "Stop there. The timeline breaks the claim.",
+    ],
+}
+
+BACKCHANNELS: dict[str, str] = {
+    "numbers_vc": "Need the denominator.",
+    "vision_optimist": "There is a bigger version there.",
+    "ruthless_skeptic": "That sounded too clean.",
+}
+
+_LIVE_TAU = 0.85
+
+
+@dataclass
+class PitchStats:
+    interruptions: int = 0
+    backchannels: int = 0
+    awkward_silences: int = 0
+    take_floors: int = 0
+    interrupted_by: dict[str, int] = field(default_factory=lambda: {agent_id: 0 for agent_id in AGENT_META})
+
+
+@dataclass
+class RoomState:
+    tau: float
+    brain_mode: str
+    transcript: list[dict[str, Any]] = field(default_factory=list)
+    decisions: dict[str, dict[str, Any]] = field(default_factory=dict)
+    stats: PitchStats = field(default_factory=PitchStats)
+    step: int = 0
+    status: str = "ready"
+    score: str | None = None
+    survival: str | None = None
+    tts_error: str | None = None
+
+
+class DeployedModalBrainClient:
+    """BrainClient backed by deployed Modal functions."""
+
+    def __init__(self, app_name: str = MODAL_APP_NAME) -> None:
+        import modal
+
+        self._step_many = modal.Function.from_name(app_name, "step_many")
+        self._generate = modal.Function.from_name(app_name, "generate")
+
+    def step(
+        self,
+        agent_id: str,
+        system_prompt: str,
+        dialogue_so_far: Dialogue,
+        new_user_text: str,
+        silence_flag: bool,
+    ) -> dict[str, object]:
+        payloads = [{"agent_id": agent_id, "system_prompt": system_prompt}]
+        raw = self.step_many(payloads, dialogue_so_far, new_user_text, silence_flag)
+        return dict(raw.get("results", {}).get(agent_id, {}))
+
+    def step_many(
+        self,
+        agent_payloads: list[dict[str, str]],
+        dialogue_so_far: Dialogue,
+        new_user_text: str,
+        silence_flag: bool,
+    ) -> dict[str, object]:
+        return self._step_many.remote(agent_payloads, dialogue_so_far, new_user_text, silence_flag)
+
+    def generate(self, agent_id: str, system_prompt: str, dialogue: Dialogue) -> dict[str, object]:
+        return self._generate.remote(agent_id, system_prompt, dialogue)
+
+
+class EndpointBrainClient:
+    """BrainClient backed by a protected Modal HTTPS endpoint."""
+
+    def __init__(
+        self,
+        endpoint_url: str | None = None,
+        token_id: str | None = None,
+        token_secret: str | None = None,
+        *,
+        timeout_s: float = 240.0,
+    ) -> None:
+        self.endpoint_url = (endpoint_url or os.environ["MODAL_BRAIN_ENDPOINT_URL"]).rstrip("/")
+        self.bearer_token = os.getenv("MODAL_BRAIN_ENDPOINT_TOKEN") or os.getenv("ENDPOINT_AUTH_TOKEN")
+        self.token_id = token_id or os.getenv("MODAL_PROXY_AUTH_TOKEN_ID") or os.getenv("MODAL_KEY")
+        self.token_secret = token_secret or os.getenv("MODAL_PROXY_AUTH_TOKEN_SECRET") or os.getenv("MODAL_SECRET")
+        if not self.bearer_token and (not self.token_id or not self.token_secret):
+            raise RuntimeError(
+                "Modal endpoint mode needs MODAL_BRAIN_ENDPOINT_TOKEN or "
+                "MODAL_PROXY_AUTH_TOKEN_ID/MODAL_PROXY_AUTH_TOKEN_SECRET"
+            )
+        self.timeout_s = timeout_s
+
+    def step(
+        self,
+        agent_id: str,
+        system_prompt: str,
+        dialogue_so_far: Dialogue,
+        new_user_text: str,
+        silence_flag: bool,
+    ) -> dict[str, object]:
+        payloads = [{"agent_id": agent_id, "system_prompt": system_prompt}]
+        raw = self.step_many(payloads, dialogue_so_far, new_user_text, silence_flag)
+        return dict(raw.get("results", {}).get(agent_id, {}))
+
+    def step_many(
+        self,
+        agent_payloads: list[dict[str, str]],
+        dialogue_so_far: Dialogue,
+        new_user_text: str,
+        silence_flag: bool,
+    ) -> dict[str, object]:
+        return self._post(
+            "/step_many",
+            {
+                "agent_payloads": agent_payloads,
+                "dialogue_so_far": dialogue_so_far,
+                "new_user_text": new_user_text,
+                "silence_flag": silence_flag,
+            },
+        )
+
+    def generate(self, agent_id: str, system_prompt: str, dialogue: Dialogue) -> dict[str, object]:
+        return self._post(
+            "/generate",
+            {"agent_id": agent_id, "system_prompt": system_prompt, "dialogue": dialogue},
+        )
+
+    def _post(self, path: str, payload: dict[str, object]) -> dict[str, object]:
+        import requests
+
+        headers = {"Content-Type": "application/json"}
+        if self.bearer_token:
+            headers["Authorization"] = f"Bearer {self.bearer_token}"
+        else:
+            headers["Modal-Key"] = str(self.token_id)
+            headers["Modal-Secret"] = str(self.token_secret)
+
+        response = requests.post(f"{self.endpoint_url}{path}", json=payload, headers=headers, timeout=self.timeout_s)
+        response.raise_for_status()
+        return dict(response.json())
+
+
+class DemoPitchBrainClient:
+    """Deterministic local stand-in for UI work when Modal is unavailable."""
+
+    def __init__(self) -> None:
+        self.calls = 0
+
+    def step(
+        self,
+        agent_id: str,
+        system_prompt: str,
+        dialogue_so_far: Dialogue,
+        new_user_text: str,
+        silence_flag: bool,
+    ) -> dict[str, object]:
+        payload = [{"agent_id": agent_id, "system_prompt": system_prompt}]
+        return dict(self.step_many(payload, dialogue_so_far, new_user_text, silence_flag)["results"][agent_id])
+
+    def step_many(
+        self,
+        agent_payloads: list[dict[str, str]],
+        dialogue_so_far: Dialogue,
+        new_user_text: str,
+        silence_flag: bool,
+    ) -> dict[str, object]:
+        start = time.perf_counter()
+        self.calls += 1
+        new_lower = new_user_text.lower()
+        weak_claim = _has_weak_claim(new_lower)
+        numeric_claim = bool(re.search(r"\b(\d+|ten thousand|zero|million|billion)\b", new_lower))
+        topic_shift = any(fragment in new_lower for fragment in ("but", "however", "pivot", "instead"))
+        results: dict[str, dict[str, object]] = {}
+        for payload in agent_payloads:
+            agent_id = payload["agent_id"]
+            results[agent_id] = self._result(
+                agent_id,
+                weak_claim=weak_claim,
+                numeric_claim=numeric_claim,
+                topic_shift=topic_shift,
+                new_user_text=new_user_text,
+                silence_flag=silence_flag,
+            )
+        return {
+            "ok": True,
+            "results": results,
+            "batch_latency_ms": (time.perf_counter() - start) * 1000.0,
+            "model_name": "demo-deterministic-brain",
+            "device_name": "cpu",
+        }
+
+    def generate(self, agent_id: str, system_prompt: str, dialogue: Dialogue) -> dict[str, object]:
+        del system_prompt
+        text = _dialogue_text(dialogue, "")
+        line = _fallback_line(agent_id, text)
+        return {
+            "ok": True,
+            "agent_id": agent_id,
+            "reply_text": line,
+            "raw_reply_text": line,
+            "reply_source": "demo",
+            "latency_ms": 3.0,
+            "model_name": "demo-deterministic-brain",
+        }
+
+    def _result(
+        self,
+        agent_id: str,
+        *,
+        weak_claim: bool,
+        numeric_claim: bool,
+        topic_shift: bool,
+        new_user_text: str,
+        silence_flag: bool,
+    ) -> dict[str, object]:
+        base = 2.1 + 0.08 * self.calls
+        surprise = base + (4.8 if weak_claim else 0.0) + (0.7 if topic_shift else 0.0)
+        if agent_id == "numbers_vc" and numeric_claim:
+            readiness = 0.74
+        elif agent_id == "ruthless_skeptic" and weak_claim:
+            readiness = 0.93
+        elif agent_id == "vision_optimist" and not weak_claim:
+            readiness = 0.62
+        else:
+            readiness = 0.36
+        if silence_flag:
+            p_end = 0.95
+        elif new_user_text.strip().endswith((".", "?", "!")):
+            p_end = 0.72
+        else:
+            p_end = 0.18
+
+        hidden = self._hidden(agent_id, weak_claim=weak_claim, topic_shift=topic_shift, silence_flag=silence_flag)
+        return {
+            "ok": True,
+            "agent_id": agent_id,
+            "surprise": float(surprise),
+            "hidden": hidden.tolist(),
+            "readiness": float(readiness),
+            "p_end": float(p_end),
+            "latency_ms": 4.0,
+        }
+
+    def _hidden(self, agent_id: str, *, weak_claim: bool, topic_shift: bool, silence_flag: bool) -> np.ndarray:
+        index = list(AGENT_META).index(agent_id)
+        vec = np.zeros(12, dtype=np.float32)
+        vec[index] = 1.0
+        vec[(self.calls + index) % vec.size] += 0.35
+        if weak_claim:
+            vec[6] += 1.8
+        if topic_shift:
+            vec[7] += 1.2
+        if silence_flag:
+            vec[8] += 0.9
+        norm = np.linalg.norm(vec)
+        return vec / norm if norm else vec
+
+
+class KokoroSpeaker:
+    """Lazy Kokoro TTS wrapper; importing Kokoro is intentionally delayed."""
+
+    def __init__(self, enabled: bool = True) -> None:
+        self.enabled = enabled
+        self.last_error: str | None = None
+        self._pipeline: Any | None = None
+
+    def synthesize(self, text: str, agent_id: str, tag: str) -> str | None:
+        if not self.enabled or not text.strip():
+            return None
+        AUDIO_DIR.mkdir(parents=True, exist_ok=True)
+        output = AUDIO_DIR / f"{tag}_{agent_id}.wav"
+        try:
+            os.environ.setdefault("HF_HOME", str(ROOT / ".hf-cache"))
+            import soundfile as sf
+            from kokoro import KPipeline
+
+            if self._pipeline is None:
+                self._pipeline = KPipeline(lang_code="a")
+            voice = AGENT_META.get(agent_id, {}).get("voice", "af_heart")
+            chunks = []
+            for _, _, audio in self._pipeline(text, voice=voice, speed=1.04):
+                chunks.append(np.asarray(audio, dtype=np.float32))
+            if not chunks:
+                raise RuntimeError("Kokoro returned no audio chunks")
+            waveform = np.concatenate(chunks)
+            sf.write(output, waveform, 24000)
+            self.last_error = None
+            return str(output)
+        except Exception as exc:  # noqa: BLE001 - app should keep the text demo alive.
+            self.last_error = f"{type(exc).__name__}: {exc}"
+            return None
+
+
+def set_live_tau(value: float) -> None:
+    global _LIVE_TAU
+    _LIVE_TAU = float(value)
+
+
+def chunk_pitch(text: str, *, words_per_chunk: int = 9) -> list[TranscriptChunk]:
+    words = re.findall(r"\S+", text.strip())
+    if not words:
+        words = re.findall(r"\S+", SAMPLE_PITCH)
+    chunks = [
+        TranscriptChunk(" ".join(words[index : index + words_per_chunk]), silence_flag=False)
+        for index in range(0, len(words), words_per_chunk)
+    ]
+    chunks = [
+        TranscriptChunk(chunk.text if index == len(chunks) - 1 else chunk.text.rstrip(".?!"), chunk.silence_flag)
+        for index, chunk in enumerate(chunks)
+    ]
+    if chunks:
+        chunks[-1] = TranscriptChunk(chunks[-1].text, silence_flag=True)
+    return chunks
+
+
+def controller_config(tau: float) -> ControllerConfig:
+    return replace(demo_controller_config(), tau=float(tau))
+
+
+def make_brain_client(mode: str | None = None) -> tuple[BrainClient, str]:
+    selected = (mode or os.getenv("PITCH_BRAIN", "modal")).strip().lower()
+    if selected in {"fake", "demo", "local"}:
+        return DemoPitchBrainClient(), "demo"
+    if selected in {"http", "endpoint", "space", "modal-http"} or (selected == "modal" and os.getenv("MODAL_BRAIN_ENDPOINT_URL")):
+        return EndpointBrainClient(), "modal"
+    return DeployedModalBrainClient(), "modal"
+
+
+def initial_room_html(tau: float | None = None, brain_mode: str | None = None) -> str:
+    selected_tau = float(tau if tau is not None else _LIVE_TAU)
+    selected_mode = brain_mode or os.getenv("PITCH_BRAIN", "modal")
+    state = RoomState(tau=selected_tau, brain_mode=selected_mode, status="ready")
+    return render_room(state)
+
+
+def preview_room_html(tau: float | None = None) -> str:
+    """Render a deterministic post-run room for headless screenshots."""
+
+    selected_tau = float(tau if tau is not None else _LIVE_TAU)
+    latest = initial_room_html(selected_tau, brain_mode="demo")
+    for latest, _, _ in run_pitch_stream(SAMPLE_PITCH, selected_tau, brain_mode="fake", enable_tts=False, save_log=False):
+        pass
+    return latest
+
+
+def run_pitch_stream(
+    pitch_text: str,
+    tau: float,
+    *,
+    brain_mode: str | None = None,
+    enable_tts: bool | None = None,
+    save_log: bool = True,
+) -> Iterator[tuple[str, str | None, dict[str, Any]]]:
+    set_live_tau(tau)
+    selected_tts = enable_tts if enable_tts is not None else os.getenv("PITCH_TTS", "1").lower() not in {"0", "false"}
+    state = RoomState(tau=float(tau), brain_mode=brain_mode or os.getenv("PITCH_BRAIN", "modal"), status="connecting")
+    speaker = KokoroSpeaker(enabled=bool(selected_tts))
+    yield from _safe_run_pitch_stream(pitch_text, state, speaker, save_log=save_log)
+
+
+def _safe_run_pitch_stream(
+    pitch_text: str,
+    state: RoomState,
+    speaker: KokoroSpeaker,
+    *,
+    save_log: bool,
+) -> Iterator[tuple[str, str | None, dict[str, Any]]]:
+    try:
+        client, resolved_mode = make_brain_client(state.brain_mode)
+        state.brain_mode = resolved_mode
+        yield from _run_with_client(pitch_text, state, speaker, client, save_log=save_log)
+    except Exception as exc:  # noqa: BLE001 - report failures inside the demo shell.
+        state.status = "failed"
+        state.transcript.append(
+            {
+                "kind": "system",
+                "step": state.step,
+                "speaker": "room",
+                "text": f"Backend failed: {type(exc).__name__}: {exc}",
+                "action": "ERROR",
+            }
+        )
+        yield render_room(state), None, _status_payload(state)
+
+
+def _run_with_client(
+    pitch_text: str,
+    state: RoomState,
+    speaker: KokoroSpeaker,
+    client: BrainClient,
+    *,
+    save_log: bool,
+) -> Iterator[tuple[str, str | None, dict[str, Any]]]:
+    personas = _pitch_personas()
+    panel = LiveBrainPanel(personas, client=client)
+    controller = WhenToSpeakController(panel.agent_ids, config=controller_config(state.tau))
+    dialogue: Dialogue = []
+    current_user_text = ""
+    events: list[dict[str, Any]] = []
+    chunks = chunk_pitch(pitch_text)
+
+    state.status = "streaming"
+    yield render_room(state), None, _status_payload(state)
+
+    for step_index, chunk in enumerate(chunks, start=1):
+        state.step = step_index
+        state.tau = _LIVE_TAU
+        controller.config = controller_config(_LIVE_TAU)
+        dialogue_before = _dialogue_with_current_user(dialogue, current_user_text)
+        signals = panel.step_all(dialogue_before, chunk.text, chunk.silence_flag)
+        current_user_text = _join_text(current_user_text, chunk.text)
+        tick = controller.tick(signals, floor_holder="human")
+        state.decisions = _decision_payloads(tick)
+        state.transcript.append({"kind": "founder", "speaker": "Founder", "step": step_index, "text": chunk.text})
+
+        event = _event_payload(step_index, chunk, tick, panel)
+        audio_path: str | None = None
+        audio_tag = f"step{step_index:02d}"
+        _record_backchannels(state, tick, step_index, dialogue)
+
+        if tick.winner:
+            if current_user_text:
+                dialogue.append({"role": "user", "speaker": "founder", "text": current_user_text})
+                current_user_text = ""
+            generated = panel.generate(tick.winner, dialogue)
+            line = sanitize_spoken_line(str(generated.get("reply_text", "")), tick.winner, dialogue)
+            action = tick.decisions[tick.winner].action
+            dialogue.append({"role": "assistant", "speaker": tick.winner, "text": line})
+            state.transcript.append(
+                {
+                    "kind": "investor",
+                    "speaker": _agent_name(tick.winner),
+                    "agent_id": tick.winner,
+                    "step": step_index,
+                    "text": line,
+                    "action": action.value,
+                }
+            )
+            event["generated"] = dict(generated, reply_text=line)
+            if action == Action.INTERRUPT:
+                state.stats.interruptions += 1
+                state.stats.interrupted_by[tick.winner] = state.stats.interrupted_by.get(tick.winner, 0) + 1
+            elif action == Action.TAKE_FLOOR:
+                state.stats.take_floors += 1
+            audio_path = speaker.synthesize(line, tick.winner, audio_tag)
+
+        if chunk.silence_flag and not tick.winner:
+            state.stats.awkward_silences += 1
+
+        if speaker.last_error:
+            state.tts_error = speaker.last_error
+        events.append(event)
+        yield render_room(state), audio_path, _status_payload(state)
+
+    if current_user_text:
+        dialogue.append({"role": "user", "speaker": "founder", "text": current_user_text})
+
+    state.status = "verdict"
+    verdict_dialogue = [
+        *dialogue,
+        {
+            "role": "user",
+            "speaker": "host",
+            "text": "Final verdict: one short sentence. Say fund, pass, or come back with numbers.",
+        },
+    ]
+    for persona in personas:
+        state.step += 1
+        generated = panel.generate(persona.agent_id, verdict_dialogue)
+        line = sanitize_spoken_line(str(generated.get("reply_text", "")), persona.agent_id, verdict_dialogue)
+        dialogue.append({"role": "assistant", "speaker": persona.agent_id, "text": line})
+        verdict_dialogue.append({"role": "assistant", "speaker": persona.agent_id, "text": line})
+        state.transcript.append(
+            {
+                "kind": "verdict",
+                "speaker": persona.display_name,
+                "agent_id": persona.agent_id,
+                "step": state.step,
+                "text": line,
+                "action": "VERDICT",
+            }
+        )
+        audio_path = speaker.synthesize(line, persona.agent_id, f"verdict{state.step:02d}")
+        if speaker.last_error:
+            state.tts_error = speaker.last_error
+        yield render_room(state), audio_path, _status_payload(state)
+
+    score, survival = final_score(state.stats)
+    state.score = score
+    state.survival = survival
+    state.status = "complete"
+    state.transcript.append({"kind": "score", "speaker": "Panel", "step": state.step, "text": score, "action": "SCORE"})
+    if save_log:
+        _save_pitch_log(state, dialogue, events)
+    yield render_room(state), None, _status_payload(state)
+
+
+def render_room(state: RoomState) -> str:
+    decisions = state.decisions or {agent_id: _empty_decision(agent_id) for agent_id in AGENT_META}
+    cards = "\n".join(_render_agent_card(agent_id, decisions.get(agent_id, _empty_decision(agent_id)), state) for agent_id in AGENT_META)
+    transcript = "\n".join(_render_feed_item(item) for item in state.transcript) or _empty_transcript()
+    mode_class = "modal" if state.brain_mode == "modal" else "demo"
+    tts_note = f"<span class='tts-note'>TTS: {_escape(state.tts_error)}</span>" if state.tts_error else ""
+    score_html = _render_score(state)
+    return f"""
+<div id="pitch-room" class="pitch-room" data-status="{_escape(state.status)}">
+  <div class="room-vignette"></div>
+  <header class="pitch-topbar">
+    <div>
+      <div class="kicker">PITCH OR PERISH</div>
+      <h1>Boardroom timing is the gameplay.</h1>
+    </div>
+    <div class="room-ledger">
+      <span class="backend-pill {mode_class}">Brain: {_escape(state.brain_mode.upper())}</span>
+      <span>Step {_escape(str(state.step))}</span>
+      <span>&tau; {_escape(f"{state.tau:.2f}")}</span>
+      {tts_note}
+    </div>
+  </header>
+  <section class="panel-rail">{cards}</section>
+  <section class="table-zone">
+    <div class="table-edge"></div>
+    <div class="transcript-shell">
+      <div class="transcript-title">
+        <span>Live transcript</span>
+        <span>{_escape(state.status)}</span>
+      </div>
+      <div class="transcript-scroll">{transcript}</div>
+    </div>
+    <aside class="scoreboard">
+      <div class="score-row"><span>Interruptions drawn</span><b>{state.stats.interruptions}</b></div>
+      <div class="score-row"><span>Backchannels</span><b>{state.stats.backchannels}</b></div>
+      <div class="score-row"><span>Awkward silences</span><b>{state.stats.awkward_silences}</b></div>
+      <div class="score-row"><span>Most brutal</span><b>{_escape(_most_brutal(state.stats))}</b></div>
+      {score_html}
+    </aside>
+  </section>
+</div>
+""".strip()
+
+
+def sanitize_spoken_line(reply: str, agent_id: str, dialogue: Dialogue | None = None) -> str:
+    candidate = re.sub(r"<think>.*?</think>", "", reply, flags=re.IGNORECASE | re.DOTALL)
+    candidate = candidate.replace("<think>", "").replace("</think>", "")
+    for line in candidate.splitlines() or [candidate]:
+        cleaned = _strip_prefix(line, agent_id)
+        if _looks_spoken(cleaned):
+            return _truncate(cleaned)
+    return _fallback_line(agent_id, _dialogue_text(dialogue or [], ""))
+
+
+def final_score(stats: PitchStats) -> tuple[str, str]:
+    most = _most_brutal(stats)
+    if stats.interruptions >= 2:
+        score = "PASS"
+        survival = f"Survival: {stats.interruptions} interruptions. Zero-churn claim drew fire; {most} controlled the room."
+    elif stats.interruptions == 1 or stats.awkward_silences:
+        score = "COME BACK WITH NUMBERS"
+        survival = f"Survival: {stats.interruptions} interruption, {stats.backchannels} backchannels."
+    else:
+        score = "FUNDED"
+        survival = "Survival: clean room, no fatal timing breaks."
+    return score, survival
+
+
+def _pitch_personas() -> list[Persona]:
+    personas = default_personas()
+    overrides = {
+        "numbers_vc": (
+            "You are a numbers-obsessed VC. Speak in short, hard-edged sentences. "
+            "Demand cohorts, denominators, CAC, retention, and payback. No pleasantries."
+        ),
+        "vision_optimist": (
+            "You are a big-vision optimist. You want the giant outcome, but you only buy crisp wedges. "
+            "Speak in one vivid sentence. No filler."
+        ),
+        "ruthless_skeptic": (
+            "You are a ruthless startup skeptic. Interrupt impossible claims immediately. "
+            "Short, sharp, plain English. Never soften with certainly or happy to."
+        ),
+    }
+    return [
+        Persona(persona.agent_id, persona.display_name, overrides.get(persona.agent_id, persona.system_prompt))
+        for persona in personas
+    ]
+
+
+def _record_backchannels(state: RoomState, tick: ControllerTick, step: int, dialogue: Dialogue) -> None:
+    for agent_id, decision in tick.decisions.items():
+        if decision.action != Action.BACKCHANNEL:
+            continue
+        text = BACKCHANNELS.get(agent_id, "Mm.")
+        state.stats.backchannels += 1
+        state.transcript.append(
+            {
+                "kind": "backchannel",
+                "speaker": _agent_name(agent_id),
+                "agent_id": agent_id,
+                "step": step,
+                "text": text,
+                "action": Action.BACKCHANNEL.value,
+            }
+        )
+        dialogue.append({"role": "assistant", "speaker": agent_id, "text": text})
+
+
+def _decision_payloads(tick: ControllerTick) -> dict[str, dict[str, Any]]:
+    payloads: dict[str, dict[str, Any]] = {}
+    for agent_id, decision in tick.decisions.items():
+        payloads[agent_id] = {
+            "action": decision.action.value,
+            "urge": decision.urge,
+            "readiness": decision.readiness,
+            "p_end": decision.p_end,
+            "change_score": decision.change_score,
+            "z_surprise": decision.z_surprise,
+            "winner": tick.winner == agent_id,
+        }
+    return payloads
+
+
+def _event_payload(step: int, chunk: TranscriptChunk, tick: ControllerTick, panel: LiveBrainPanel) -> dict[str, Any]:
+    return {
+        "step": step,
+        "new_user_text": chunk.text,
+        "silence_flag": chunk.silence_flag,
+        "winner": tick.winner,
+        "model_name": (panel.last_raw or {}).get("model_name"),
+        "device_name": (panel.last_raw or {}).get("device_name"),
+        "batch_latency_ms": (panel.last_raw or {}).get("batch_latency_ms"),
+        "decisions": _decision_payloads(tick),
+    }
+
+
+def _render_agent_card(agent_id: str, decision: dict[str, Any], state: RoomState) -> str:
+    meta = AGENT_META[agent_id]
+    urge = float(decision.get("urge", 0.0))
+    meter_max = max(2.2, state.tau * 1.6)
+    fill = max(0.0, min(100.0, urge / meter_max * 100.0))
+    tau_position = max(0.0, min(100.0, state.tau / meter_max * 100.0))
+    action = str(decision.get("action", "SILENT"))
+    active = "holding-floor" if decision.get("winner") else ""
+    flash = "flash-interrupt" if action == Action.INTERRUPT.value else ""
+    return f"""
+<article class="investor-card {active} {flash}" style="--agent-color:{meta['color']}; --urge-fill:{fill:.1f}%; --tau-pos:{tau_position:.1f}%;">
+  <div class="card-head">
+    <div class="avatar">{_escape(meta['avatar'])}</div>
+    <div>
+      <h2>{_escape(meta['name'])}</h2>
+      <p>{_escape(meta['role'])}</p>
+    </div>
+  </div>
+  <div class="urge-row">
+    <span>urge</span>
+    <strong>{urge:.2f}</strong>
+  </div>
+  <div class="urge-meter"><div class="urge-fill"></div><i class="tau-line"></i></div>
+  <div class="signal-strip">
+    <span>{_escape(action)}</span>
+    <span>ready {float(decision.get('readiness', 0.0)):.2f}</span>
+    <span>end {float(decision.get('p_end', 0.0)):.2f}</span>
+  </div>
+</article>
+""".strip()
+
+
+def _render_feed_item(item: dict[str, Any]) -> str:
+    kind = str(item.get("kind", "system"))
+    action = str(item.get("action", ""))
+    step = item.get("step", "")
+    speaker = _escape(str(item.get("speaker", "")))
+    text = _escape(str(item.get("text", "")))
+    label = f"<span>{speaker}</span><em>step {step}</em>"
+    return f"""
+<div class="feed-item {kind}">
+  <div class="feed-meta">{label}</div>
+  <div class="feed-text">{text}</div>
+  {f'<b class="action-badge">{_escape(action)}</b>' if action else ''}
+</div>
+""".strip()
+
+
+def _empty_transcript() -> str:
+    return """
+<div class="feed-item system">
+  <div class="feed-meta"><span>Room</span><em>armed</em></div>
+  <div class="feed-text">Paste the pitch, hit the room, and watch the panel timing.</div>
+</div>
+""".strip()
+
+
+def _render_score(state: RoomState) -> str:
+    if not state.score:
+        return ""
+    return f"""
+<div class="final-score">
+  <small>Final score</small>
+  <strong>{_escape(state.score)}</strong>
+  <p>{_escape(state.survival or '')}</p>
+</div>
+""".strip()
+
+
+def _empty_decision(agent_id: str) -> dict[str, Any]:
+    del agent_id
+    return {
+        "action": Action.SILENT.value,
+        "urge": 0.0,
+        "readiness": 0.0,
+        "p_end": 0.0,
+        "change_score": 0.0,
+        "z_surprise": 0.0,
+        "winner": False,
+    }
+
+
+def _status_payload(state: RoomState) -> dict[str, Any]:
+    return {
+        "status": state.status,
+        "tau": state.tau,
+        "brain_mode": state.brain_mode,
+        "stats": asdict(state.stats),
+        "score": state.score,
+        "survival": state.survival,
+        "tts_error": state.tts_error,
+    }
+
+
+def _save_pitch_log(state: RoomState, dialogue: Dialogue, events: list[dict[str, Any]]) -> None:
+    EVAL_DIR.mkdir(parents=True, exist_ok=True)
+    payload = {
+        "brain_mode": state.brain_mode,
+        "tau": state.tau,
+        "stats": asdict(state.stats),
+        "score": state.score,
+        "survival": state.survival,
+        "transcript": state.transcript,
+        "dialogue": dialogue,
+        "events": events,
+    }
+    PITCH_LOG_PATH.write_text(json.dumps(payload, indent=2), encoding="utf-8")
+
+
+def _dialogue_with_current_user(dialogue: Dialogue, current_user_text: str) -> Dialogue:
+    snapshot = [dict(turn) for turn in dialogue]
+    snapshot.append({"role": "user", "speaker": "founder", "text": current_user_text})
+    return snapshot
+
+
+def _join_text(left: str, right: str) -> str:
+    left = left.strip()
+    right = right.strip()
+    if not left:
+        return right
+    if not right:
+        return left
+    return f"{left} {right}"
+
+
+def _dialogue_text(dialogue: Dialogue, new_user_text: str) -> str:
+    parts = [turn.get("text", "") for turn in dialogue]
+    parts.append(new_user_text)
+    return " ".join(parts).lower()
+
+
+def _has_weak_claim(text: str) -> bool:
+    weak_fragments = [
+        "zero churn",
+        "churn after launching last week",
+        "after launching last week",
+        "launched last week",
+        "after one week",
+        "no churn",
+        "guaranteed",
+        "nobody can compete",
+    ]
+    return any(fragment in text for fragment in weak_fragments)
+
+
+def _fallback_line(agent_id: str, text: str) -> str:
+    if "zero churn" in text and ("last week" in text or "one week" in text):
+        return "Zero churn after one week is not churn data."
+    if "ten thousand" in text and "stores" in text and agent_id == "numbers_vc":
+        return "Show the store list and paid cohorts."
+    if "that's the pitch" in text or "thats the pitch" in text:
+        if agent_id == "vision_optimist":
+            return "The ambition is real; the proof still has holes."
+        if agent_id == "numbers_vc":
+            return "Come back with paid retention and margins."
+        return "Pass until the impossible claims become evidence."
+    options = PUNCHY_FALLBACKS.get(agent_id, ["That claim needs proof."])
+    stable_index = sum(ord(char) for char in f"{text}:{agent_id}") % len(options)
+    return options[stable_index]
+
+
+def _strip_prefix(line: str, agent_id: str) -> str:
+    candidate = line.strip().strip("\"'")
+    candidate = re.sub(r"^[\-\*\d\.\)\s]+", "", candidate).strip()
+    prefixes = [
+        f"{agent_id}:",
+        _agent_name(agent_id) + ":",
+        "Assistant:",
+        "Investor:",
+        "VC:",
+        "Sentence:",
+        "Spoken investor sentence:",
+    ]
+    for prefix in prefixes:
+        if candidate.lower().startswith(prefix.lower()):
+            candidate = candidate[len(prefix) :].strip()
+    candidate = re.sub(r"^(certainly,?\s+|happy to,?\s+|let's\s+|i think\s+)", "", candidate, flags=re.IGNORECASE)
+    if candidate:
+        candidate = candidate[0].upper() + candidate[1:]
+    return candidate.strip().strip("\"'")
+
+
+def _looks_spoken(candidate: str) -> bool:
+    if len(candidate) < 10:
+        return False
+    lowered = candidate.lower()
+    blocked = [
+        "the user",
+        "the founder",
+        "recent transcript",
+        "analysis",
+        "markdown",
+        "persona:",
+        "investor:",
+        "assistant:",
+        "<",
+        "{",
+    ]
+    if any(fragment in lowered for fragment in blocked):
+        return False
+    alpha = sum(char.isalpha() for char in candidate)
+    if alpha / max(len(candidate), 1) < 0.45:
+        return False
+    words = [word.strip(".,;:!?()[]{}\"'").lower() for word in candidate.split()]
+    words = [word for word in words if word]
+    return len(words) <= 24
+
+
+def _truncate(candidate: str) -> str:
+    parts = re.split(r"(?<=[.!?;])\s+", candidate, maxsplit=1)
+    sentence = parts[0].strip() if parts else candidate.strip()
+    words = sentence.split()
+    if len(words) > 16:
+        sentence = " ".join(words[:16]).rstrip(",;:") + "."
+    return sentence[:220].rstrip()
+
+
+def _most_brutal(stats: PitchStats) -> str:
+    if not stats.interrupted_by:
+        return "none"
+    agent_id, count = max(stats.interrupted_by.items(), key=lambda item: (item[1], item[0]))
+    if count <= 0:
+        return "none"
+    return AGENT_META.get(agent_id, {}).get("name", agent_id)
+
+
+def _agent_name(agent_id: str) -> str:
+    return AGENT_META.get(agent_id, {}).get("name", agent_id)
+
+
+def _escape(value: str) -> str:
+    return html.escape(value, quote=True)

apps/pitch/static/pitch.css

@@ -0,0 +1,460 @@
+:root {
+  color-scheme: dark;
+  --pitch-bg: #080807;
+  --pitch-panel: #12110f;
+  --pitch-panel-2: #191712;
+  --pitch-ink: #f4efe4;
+  --pitch-muted: #a8a096;
+  --pitch-line: rgba(244, 239, 228, 0.14);
+  --pitch-brass: #c99a3e;
+  --pitch-red: #ff5a5f;
+  --pitch-green: #47d18c;
+}
+
+.gradio-container {
+  max-width: none !important;
+  background: var(--pitch-bg) !important;
+  color: var(--pitch-ink) !important;
+  font-family: Inter, ui-sans-serif, system-ui, -apple-system, BlinkMacSystemFont, "Segoe UI", sans-serif !important;
+}
+
+.contain,
+.wrap,
+#component-0 {
+  background: transparent !important;
+}
+
+#pitch-room-output {
+  margin: 0 auto;
+}
+
+.pitch-room {
+  position: relative;
+  overflow: hidden;
+  min-height: 650px;
+  padding: 24px;
+  border: 1px solid rgba(201, 154, 62, 0.25);
+  border-radius: 8px;
+  background:
+    linear-gradient(180deg, rgba(255, 255, 255, 0.04), rgba(255, 255, 255, 0) 24%),
+    linear-gradient(135deg, #11100e 0%, #080807 48%, #17120e 100%);
+  box-shadow: 0 24px 80px rgba(0, 0, 0, 0.55);
+}
+
+.room-vignette {
+  position: absolute;
+  inset: 0;
+  pointer-events: none;
+  background:
+    linear-gradient(90deg, rgba(0, 0, 0, 0.48), transparent 20%, transparent 80%, rgba(0, 0, 0, 0.48)),
+    repeating-linear-gradient(90deg, rgba(255,255,255,0.025) 0 1px, transparent 1px 92px);
+}
+
+.pitch-topbar,
+.panel-rail,
+.table-zone {
+  position: relative;
+  z-index: 1;
+}
+
+.pitch-topbar {
+  display: flex;
+  align-items: flex-start;
+  justify-content: space-between;
+  gap: 18px;
+  margin-bottom: 18px;
+}
+
+.kicker {
+  color: var(--pitch-brass);
+  font-size: 12px;
+  font-weight: 800;
+  letter-spacing: 0.18em;
+}
+
+.pitch-topbar h1 {
+  margin: 5px 0 0;
+  color: var(--pitch-ink);
+  font-size: 30px;
+  line-height: 1.05;
+  letter-spacing: 0;
+}
+
+.room-ledger {
+  display: flex;
+  flex-wrap: wrap;
+  justify-content: flex-end;
+  gap: 8px;
+  max-width: 520px;
+  color: var(--pitch-muted);
+  font-size: 12px;
+}
+
+.room-ledger span {
+  padding: 7px 9px;
+  border: 1px solid var(--pitch-line);
+  border-radius: 6px;
+  background: rgba(0, 0, 0, 0.24);
+}
+
+.backend-pill.modal {
+  color: var(--pitch-green);
+}
+
+.backend-pill.demo {
+  color: var(--pitch-brass);
+}
+
+.tts-note {
+  color: var(--pitch-red) !important;
+}
+
+.panel-rail {
+  display: grid;
+  grid-template-columns: repeat(3, minmax(0, 1fr));
+  gap: 14px;
+}
+
+.investor-card {
+  min-width: 0;
+  padding: 15px;
+  border: 1px solid rgba(244, 239, 228, 0.13);
+  border-top: 3px solid var(--agent-color);
+  border-radius: 8px;
+  background:
+    linear-gradient(180deg, rgba(255,255,255,0.055), rgba(255,255,255,0.012)),
+    var(--pitch-panel);
+  box-shadow: inset 0 0 0 1px rgba(0,0,0,0.24), 0 16px 40px rgba(0, 0, 0, 0.28);
+  transition: border-color 180ms ease, transform 180ms ease, box-shadow 180ms ease;
+}
+
+.investor-card.holding-floor {
+  border-color: color-mix(in srgb, var(--agent-color), white 24%);
+  box-shadow: 0 0 0 1px color-mix(in srgb, var(--agent-color), transparent 35%), 0 18px 54px rgba(0, 0, 0, 0.42);
+  transform: translateY(-2px);
+}
+
+.investor-card.flash-interrupt {
+  animation: interrupt-flash 720ms ease-out 1;
+}
+
+@keyframes interrupt-flash {
+  0% { box-shadow: 0 0 0 0 color-mix(in srgb, var(--agent-color), transparent 35%); }
+  40% { box-shadow: 0 0 0 5px color-mix(in srgb, var(--agent-color), transparent 45%), 0 0 38px color-mix(in srgb, var(--agent-color), transparent 72%); }
+  100% { box-shadow: 0 16px 40px rgba(0, 0, 0, 0.28); }
+}
+
+.card-head {
+  display: grid;
+  grid-template-columns: 42px minmax(0, 1fr);
+  gap: 11px;
+  align-items: center;
+}
+
+.avatar {
+  display: grid;
+  width: 42px;
+  height: 42px;
+  place-items: center;
+  border: 1px solid color-mix(in srgb, var(--agent-color), white 24%);
+  border-radius: 6px;
+  color: #050504;
+  background: var(--agent-color);
+  font-weight: 900;
+  font-size: 20px;
+}
+
+.card-head h2 {
+  margin: 0;
+  overflow-wrap: anywhere;
+  color: var(--pitch-ink);
+  font-size: 18px;
+  line-height: 1.05;
+}
+
+.card-head p {
+  margin: 5px 0 0;
+  color: var(--pitch-muted);
+  font-size: 12px;
+  line-height: 1.25;
+}
+
+.urge-row,
+.signal-strip {
+  display: flex;
+  align-items: center;
+  justify-content: space-between;
+  gap: 9px;
+}
+
+.urge-row {
+  margin-top: 15px;
+  color: var(--pitch-muted);
+  font-size: 12px;
+  text-transform: uppercase;
+}
+
+.urge-row strong {
+  color: var(--pitch-ink);
+  font-size: 16px;
+}
+
+.urge-meter {
+  position: relative;
+  height: 13px;
+  margin-top: 7px;
+  overflow: hidden;
+  border: 1px solid rgba(244, 239, 228, 0.16);
+  border-radius: 999px;
+  background: rgba(0, 0, 0, 0.35);
+}
+
+.urge-fill {
+  width: var(--urge-fill);
+  height: 100%;
+  border-radius: 999px;
+  background: linear-gradient(90deg, color-mix(in srgb, var(--agent-color), black 18%), var(--agent-color));
+  transition: width 260ms ease;
+}
+
+.tau-line {
+  position: absolute;
+  top: -3px;
+  bottom: -3px;
+  left: var(--tau-pos);
+  width: 2px;
+  background: var(--pitch-ink);
+  box-shadow: 0 0 0 1px rgba(0,0,0,0.7);
+}
+
+.signal-strip {
+  flex-wrap: wrap;
+  margin-top: 10px;
+  color: var(--pitch-muted);
+  font-size: 11px;
+}
+
+.signal-strip span:first-child {
+  color: var(--agent-color);
+  font-weight: 800;
+}
+
+.table-zone {
+  display: grid;
+  grid-template-columns: minmax(0, 1fr) 300px;
+  gap: 16px;
+  margin-top: 18px;
+}
+
+.table-edge {
+  position: absolute;
+  left: -24px;
+  right: -24px;
+  bottom: -120px;
+  height: 250px;
+  border-top: 1px solid rgba(201, 154, 62, 0.26);
+  background:
+    linear-gradient(180deg, rgba(201, 154, 62, 0.12), rgba(0,0,0,0.1)),
+    repeating-linear-gradient(90deg, rgba(255,255,255,0.03) 0 1px, transparent 1px 110px),
+    #17100a;
+  transform: perspective(800px) rotateX(54deg);
+  transform-origin: top center;
+  z-index: -1;
+}
+
+.transcript-shell,
+.scoreboard {
+  border: 1px solid var(--pitch-line);
+  border-radius: 8px;
+  background: rgba(8, 8, 7, 0.72);
+  backdrop-filter: blur(10px);
+}
+
+.transcript-title {
+  display: flex;
+  justify-content: space-between;
+  gap: 12px;
+  padding: 12px 14px;
+  border-bottom: 1px solid var(--pitch-line);
+  color: var(--pitch-muted);
+  font-size: 12px;
+  font-weight: 800;
+  text-transform: uppercase;
+}
+
+.transcript-scroll {
+  display: flex;
+  flex-direction: column;
+  gap: 10px;
+  max-height: 320px;
+  overflow-y: auto;
+  padding: 14px;
+  scroll-behavior: smooth;
+}
+
+.feed-item {
+  position: relative;
+  padding: 11px 12px;
+  border-left: 3px solid rgba(244, 239, 228, 0.16);
+  border-radius: 6px;
+  background: rgba(255, 255, 255, 0.045);
+}
+
+.feed-item.founder {
+  border-color: #d7d1c3;
+}
+
+.feed-item.investor,
+.feed-item.verdict {
+  border-color: var(--pitch-red);
+  background: rgba(255, 90, 95, 0.08);
+}
+
+.feed-item.backchannel {
+  border-color: var(--pitch-brass);
+  background: rgba(201, 154, 62, 0.08);
+}
+
+.feed-item.score {
+  border-color: var(--pitch-green);
+}
+
+.feed-meta {
+  display: flex;
+  justify-content: space-between;
+  gap: 12px;
+  color: var(--pitch-muted);
+  font-size: 11px;
+  text-transform: uppercase;
+}
+
+.feed-meta span {
+  color: var(--pitch-ink);
+  font-weight: 800;
+}
+
+.feed-text {
+  margin-top: 5px;
+  color: var(--pitch-ink);
+  font-size: 15px;
+  line-height: 1.38;
+}
+
+.action-badge {
+  display: inline-flex;
+  margin-top: 8px;
+  padding: 3px 7px;
+  border: 1px solid currentColor;
+  border-radius: 999px;
+  color: var(--pitch-brass);
+  font-size: 10px;
+  letter-spacing: 0.08em;
+}
+
+.scoreboard {
+  padding: 14px;
+}
+
+.score-row {
+  display: flex;
+  justify-content: space-between;
+  gap: 10px;
+  padding: 11px 0;
+  border-bottom: 1px solid var(--pitch-line);
+  color: var(--pitch-muted);
+  font-size: 13px;
+}
+
+.score-row b {
+  color: var(--pitch-ink);
+  text-align: right;
+}
+
+.final-score {
+  margin-top: 14px;
+  padding: 13px;
+  border: 1px solid rgba(201, 154, 62, 0.32);
+  border-radius: 6px;
+  background: rgba(201, 154, 62, 0.10);
+}
+
+.final-score small {
+  color: var(--pitch-muted);
+  text-transform: uppercase;
+  font-weight: 800;
+}
+
+.final-score strong {
+  display: block;
+  margin-top: 4px;
+  color: var(--pitch-brass);
+  font-size: 21px;
+}
+
+.final-score p {
+  margin: 8px 0 0;
+  color: var(--pitch-ink);
+  font-size: 13px;
+  line-height: 1.35;
+}
+
+#pitch-control-row {
+  max-width: 1260px;
+  margin: 14px auto 0;
+  align-items: stretch;
+}
+
+#pitch-input textarea,
+#pitch-controls,
+#pitch-audio {
+  border-color: rgba(244, 239, 228, 0.16) !important;
+  background: #11100e !important;
+  color: var(--pitch-ink) !important;
+}
+
+#pitch-controls {
+  min-width: 280px;
+  padding: 13px;
+  border: 1px solid rgba(244, 239, 228, 0.16);
+  border-radius: 8px;
+}
+
+#run-pitch {
+  border: 1px solid rgba(255, 90, 95, 0.7) !important;
+  background: #ff5a5f !important;
+  color: #080807 !important;
+  font-weight: 900 !important;
+}
+
+@media (max-width: 980px) {
+  .panel-rail,
+  .table-zone {
+    grid-template-columns: 1fr;
+  }
+
+  .pitch-topbar {
+    flex-direction: column;
+  }
+
+  .room-ledger {
+    justify-content: flex-start;
+  }
+}
+
+@media (max-width: 640px) {
+  .pitch-room {
+    padding: 16px;
+  }
+
+  .pitch-topbar h1 {
+    font-size: 24px;
+  }
+
+  .panel-rail {
+    gap: 10px;
+  }
+
+  .transcript-scroll {
+    max-height: 360px;
+  }
+}

apps/pitch/static/pitch.js

@@ -0,0 +1,44 @@
+() => {
+  const keepTranscriptPinned = () => {
+    document.querySelectorAll(".transcript-scroll").forEach((node) => {
+      node.scrollTop = node.scrollHeight;
+    });
+  };
+
+  const observer = new MutationObserver(() => {
+    keepTranscriptPinned();
+    document.querySelectorAll(".investor-card.flash-interrupt").forEach((node) => {
+      node.classList.remove("flash-replay");
+      void node.offsetWidth;
+      node.classList.add("flash-replay");
+    });
+  });
+
+  const autorun = () => {
+    const params = new URLSearchParams(window.location.search);
+    if (!params.has("autorun")) {
+      return;
+    }
+    let tries = 0;
+    const timer = window.setInterval(() => {
+      tries += 1;
+      const runButton = document.querySelector("#run-pitch button") || document.querySelector("#run-pitch");
+      const room = document.querySelector("#pitch-room");
+      const status = room ? room.dataset.status : "";
+      if (status && status !== "ready") {
+        window.clearInterval(timer);
+        return;
+      }
+      if (runButton && tries > 8) {
+        runButton.click();
+      }
+      if (tries > 80) {
+        window.clearInterval(timer);
+      }
+    }, 500);
+  };
+
+  observer.observe(document.body, { childList: true, subtree: true });
+  keepTranscriptPinned();
+  autorun();
+};

engine/CONTROLLER_NOTES.md

@@ -0,0 +1,42 @@
+# WhenToSpeak Controller Notes
+
+The controller is training-free. It consumes signals from a `Brain` interface and
+never loads or calls a model itself. The live brain must provide, for each
+incremental transcript update and each agent context, mean token surprise, a
+last-layer hidden vector, readiness, and turn-end probability.
+
+## Signals
+
+- `surprise`: mean per-token negative log-likelihood of the newly added user
+  tokens, teacher-forced.
+- `hidden`: mean last-layer hidden-state vector for the newly added tokens.
+- `readiness`: speculative reply confidence for this agent. Draft about eight
+  tokens and compute `readiness = 1 / (1 + mean_token_entropy)`.
+- `p_end`: heuristic probability that the human turn is complete. The live loop
+  should combine trailing silence, sentence-final punctuation, and high EOS
+  probability.
+
+## Urge
+
+Each agent keeps an online running mean/std of surprise and uses the current
+z-score. Hidden-state cosine deltas feed Adams-MacKay BOCPD; a collapse in MAP
+run-length becomes the change-point score.
+
+```text
+U_t = w_surprise*z(surprise)
+    + w_change*changepoint_score
+    + w_readiness*readiness
+    + w_end*p_end
+    + w_barge*max(z(surprise), 0)*readiness*(1 - p_end)
+```
+
+`tau` is the single global conversational-aggressiveness knob. Lower `tau` makes
+the panel take the floor sooner; higher `tau` makes it wait.
+
+## Arbitration
+
+Each tick is deterministic. Agents first classify local intent as `SILENT`,
+`BACKCHANNEL`, `TAKE_FLOOR`, or `INTERRUPT`. Only the highest-urge agent above
+`tau` may take the floor or interrupt. Non-winning agents may still backchannel
+if their urge clears the derived backchannel threshold. A short refractory period
+prevents repeated firing on adjacent ASR updates.

engine/CONVERSATION_NOTES.md

@@ -0,0 +1,58 @@
+# Live Brain + Conversation Notes
+
+This phase keeps the controller training-free and model-agnostic. The real model
+lives behind `modal_app/brain_modal.py`; local code only asks for `BrainSignals`
+and generated replies.
+
+## Flow
+
+1. The text stream feeds incremental word groups plus an optional silence flag.
+2. `Conversation` sends the current dialogue prefix and newest user chunk to
+   `LiveBrainPanel.step_all()`.
+3. Modal computes per-agent surprise, hidden vector, readiness, and `p_end`.
+4. `WhenToSpeakController` arbitrates `SILENT`, `BACKCHANNEL`, `TAKE_FLOOR`, or
+   `INTERRUPT`.
+5. On `TAKE_FLOOR` or `INTERRUPT`, `Conversation` calls Modal `generate()` and
+   splices the short investor reply into the dialogue.
+
+The sample pitch deliberately includes a weak claim: "ten thousand stores and
+zero churn after launching last week." The expected demo behavior is an investor
+interrupt or floor-take near that claim, with the generated line recorded in
+`eval/conversation_log.json`.
+
+Run the real text-streamed demo with:
+
+```text
+uv run modal run modal_app/brain_modal.py
+```
+
+## Modal
+
+The live brain tries `nvidia/Llama-3.1-Nemotron-Nano-4B-v1.1` first and falls
+back to `Qwen/Qwen2.5-3B-Instruct`. We keep `HF_HOME=/cache` on a Modal Volume so
+weights persist across runs.
+
+## Recorded Demo
+
+The committed `eval/conversation_log.json` was produced by:
+
+```text
+uv run modal run modal_app/brain_modal.py
+```
+
+Latest measured run: `nvidia/Llama-3.1-Nemotron-Nano-4B-v1.1` on `NVIDIA A10`.
+Total wall time was 49.1 s because the run included Modal container/model startup.
+
+At step 3 the controller interrupted the planted weak claim. The winning agent
+was `ruthless_skeptic` with urge 1.47 and readiness 0.67. At step 7 the panel now
+takes the floor at turn end.
+
+```text
+Ruthless Skeptic: Zero churn after one week is not churn data.
+Vision Optimist: Show cohorts, paid conversion, and retention.
+```
+
+Generation caveat: Nemotron-Nano produced malformed text for these two `generate()`
+calls, so `eval/conversation_log.json` records `reply_source: "fallback"` for both.
+The timing signals and controller decisions are still real Modal/Nemotron outputs;
+the fallback only guards the spoken text until the generator prompt/model is improved.

engine/__init__.py

@@ -0,0 +1,2 @@
+"""Training-free WhenToSpeak controller components."""
+

engine/bocpd.py

@@ -0,0 +1,144 @@
+from __future__ import annotations
+
+import math
+import sys
+from dataclasses import dataclass
+from pathlib import Path
+
+import numpy as np
+
+
+ROOT = Path(__file__).resolve().parents[1]
+DEFAULT_DUMP_PATH = ROOT / "eval" / "probe_dump.npz"
+
+
+@dataclass(frozen=True)
+class BocpdResult:
+    step: int
+    value: float
+    cp_prob: float
+    map_run_length: int
+    change_point: bool
+
+
+def logsumexp(values: np.ndarray) -> float:
+    max_value = float(np.max(values))
+    if not math.isfinite(max_value):
+        return max_value
+    return max_value + math.log(float(np.sum(np.exp(values - max_value))))
+
+
+def student_t_logpdf(x: float, mu: np.ndarray, kappa: np.ndarray, alpha: np.ndarray, beta: np.ndarray) -> np.ndarray:
+    nu = 2.0 * alpha
+    scale = np.sqrt(beta * (kappa + 1.0) / (alpha * kappa))
+    z = (x - mu) / scale
+    return (
+        np.vectorize(math.lgamma)((nu + 1.0) / 2.0)
+        - np.vectorize(math.lgamma)(nu / 2.0)
+        - 0.5 * np.log(nu * math.pi)
+        - np.log(scale)
+        - ((nu + 1.0) / 2.0) * np.log1p((z * z) / nu)
+    )
+
+
+def update_nig(
+    x: float,
+    mu: np.ndarray,
+    kappa: np.ndarray,
+    alpha: np.ndarray,
+    beta: np.ndarray,
+) -> tuple[np.ndarray, np.ndarray, np.ndarray, np.ndarray]:
+    next_kappa = kappa + 1.0
+    next_mu = (kappa * mu + x) / next_kappa
+    next_alpha = alpha + 0.5
+    next_beta = beta + 0.5 * kappa * (x - mu) ** 2 / next_kappa
+    return next_mu, next_kappa, next_alpha, next_beta
+
+
+def run_bocpd(
+    values: np.ndarray,
+    hazard: float = 1.0 / 50.0,
+    prior_mu: float = 0.0,
+    prior_kappa: float = 1.0e-3,
+    prior_alpha: float = 1.0,
+    prior_beta: float = 1.0,
+) -> list[BocpdResult]:
+    log_hazard = math.log(hazard)
+    log_growth_factor = math.log1p(-hazard)
+
+    log_run_probs = np.asarray([0.0], dtype=np.float64)
+    mu = np.asarray([prior_mu], dtype=np.float64)
+    kappa = np.asarray([prior_kappa], dtype=np.float64)
+    alpha = np.asarray([prior_alpha], dtype=np.float64)
+    beta = np.asarray([prior_beta], dtype=np.float64)
+
+    results: list[BocpdResult] = []
+    previous_map_run_length: int | None = None
+
+    for step, x_value in enumerate(values, start=1):
+        x = float(x_value)
+        predictive = student_t_logpdf(x, mu, kappa, alpha, beta)
+
+        growth_probs = log_run_probs + predictive + log_growth_factor
+        cp_prob = logsumexp(log_run_probs + predictive + log_hazard)
+        new_log_run_probs = np.concatenate(([cp_prob], growth_probs))
+        normalizer = logsumexp(new_log_run_probs)
+        new_log_run_probs -= normalizer
+
+        grown_mu, grown_kappa, grown_alpha, grown_beta = update_nig(x, mu, kappa, alpha, beta)
+        mu = np.concatenate(([prior_mu], grown_mu))
+        kappa = np.concatenate(([prior_kappa], grown_kappa))
+        alpha = np.concatenate(([prior_alpha], grown_alpha))
+        beta = np.concatenate(([prior_beta], grown_beta))
+        log_run_probs = new_log_run_probs
+
+        map_run_length = int(np.argmax(log_run_probs))
+        cp_probability = float(np.exp(log_run_probs[0]))
+        change_point = bool(
+            map_run_length == 0
+            or (previous_map_run_length is not None and map_run_length < previous_map_run_length)
+        )
+        results.append(
+            BocpdResult(
+                step=step,
+                value=x,
+                cp_prob=cp_probability,
+                map_run_length=map_run_length,
+                change_point=change_point,
+            )
+        )
+        previous_map_run_length = map_run_length
+
+    return results
+
+
+def main() -> None:
+    dump_path = Path(sys.argv[1]) if len(sys.argv) > 1 else DEFAULT_DUMP_PATH
+    if not dump_path.is_absolute():
+        dump_path = ROOT / dump_path
+
+    if not dump_path.exists():
+        print(f"No probe dump found at {dump_path}")
+        return
+
+    dump = np.load(dump_path, allow_pickle=True)
+    failure = str(dump.get("failure", ""))
+    values = np.asarray(dump["nll_series"], dtype=np.float64)
+    if values.size == 0:
+        print("No NLL samples found; skipping BOCPD.")
+        if failure:
+            print(f"Probe failure: {failure}")
+        return
+
+    print("step | nll | cp_prob | map_run_length | change_point")
+    print("-----|-----|---------|----------------|-------------")
+    for result in run_bocpd(values):
+        flag = "YES" if result.change_point else "no"
+        print(
+            f"{result.step:>4} | {result.value:.4f} | {result.cp_prob:.4f} | "
+            f"{result.map_run_length:>14} | {flag}"
+        )
+
+
+if __name__ == "__main__":
+    main()

engine/brain.py

@@ -0,0 +1,119 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Iterable, Protocol, Sequence
+
+import numpy as np
+
+
+@dataclass(frozen=True)
+class BrainSignals:
+    """Signals emitted by one agent-context brain for one transcript update.
+
+    `readiness` is the controller-facing score for a speculative short reply:
+
+        readiness = 1 / (1 + mean_token_entropy)
+
+    A low-entropy draft means this agent has a confident next move. `p_end` is a
+    turn-completion heuristic: in the live loop it should combine trailing
+    silence, sentence-final punctuation, and/or high EOS probability.
+    """
+
+    surprise: float
+    hidden: np.ndarray
+    readiness: float
+    p_end: float
+
+    def __post_init__(self) -> None:
+        hidden = np.asarray(self.hidden, dtype=np.float32)
+        if hidden.ndim != 1:
+            raise ValueError("hidden must be a 1-D float32 vector")
+        if not 0.0 <= float(self.readiness) <= 1.0:
+            raise ValueError("readiness must be in [0, 1]")
+        if not 0.0 <= float(self.p_end) <= 1.0:
+            raise ValueError("p_end must be in [0, 1]")
+        object.__setattr__(self, "hidden", hidden)
+        object.__setattr__(self, "surprise", float(self.surprise))
+        object.__setattr__(self, "readiness", float(self.readiness))
+        object.__setattr__(self, "p_end", float(self.p_end))
+
+
+class Brain(Protocol):
+    """Interface the live instrumented LLM must satisfy for one agent context."""
+
+    def next_signals(self) -> BrainSignals:
+        """Return signals for the newest incremental transcript update."""
+
+
+class ReplayBrain:
+    """Deterministic `Brain` backed by precomputed signal samples."""
+
+    def __init__(self, samples: Sequence[BrainSignals | dict[str, object]], *, name: str = "replay") -> None:
+        self.name = name
+        self._samples = [coerce_signals(sample) for sample in samples]
+        self._index = 0
+
+    def __len__(self) -> int:
+        return len(self._samples)
+
+    def __iter__(self) -> Iterable[BrainSignals]:
+        return iter(self._samples)
+
+    def reset(self) -> None:
+        self._index = 0
+
+    def next_signals(self) -> BrainSignals:
+        if self._index >= len(self._samples):
+            raise StopIteration(f"ReplayBrain {self.name!r} is exhausted")
+        sample = self._samples[self._index]
+        self._index += 1
+        return sample
+
+    @classmethod
+    def from_npz(
+        cls,
+        path: str | Path,
+        *,
+        readiness: Sequence[float] | None = None,
+        p_end: Sequence[float] | None = None,
+        name: str | None = None,
+    ) -> "ReplayBrain":
+        dump = np.load(path, allow_pickle=True)
+        surprises = np.asarray(dump["nll_series"], dtype=np.float32)
+        hidden = np.asarray(dump["hidden_states"], dtype=np.float32)
+        if hidden.ndim != 2:
+            raise ValueError("hidden_states in npz must be a 2-D matrix")
+
+        n_steps = int(surprises.shape[0])
+        readiness_values = (
+            np.asarray(readiness, dtype=np.float32)
+            if readiness is not None
+            else np.linspace(0.35, 0.75, n_steps, dtype=np.float32)
+        )
+        p_end_values = np.asarray(p_end, dtype=np.float32) if p_end is not None else np.zeros(n_steps, dtype=np.float32)
+        if n_steps:
+            p_end_values[-1] = max(float(p_end_values[-1]), 0.95)
+
+        samples = [
+            BrainSignals(
+                surprise=float(surprises[index]),
+                hidden=hidden[index],
+                readiness=float(readiness_values[index]),
+                p_end=float(p_end_values[index]),
+            )
+            for index in range(n_steps)
+        ]
+        return cls(samples, name=name or Path(path).stem)
+
+
+def coerce_signals(sample: BrainSignals | dict[str, object]) -> BrainSignals:
+    if isinstance(sample, BrainSignals):
+        return sample
+    return BrainSignals(
+        surprise=float(sample["surprise"]),
+        hidden=np.asarray(sample["hidden"], dtype=np.float32),
+        readiness=float(sample["readiness"]),
+        p_end=float(sample["p_end"]),
+    )
+

engine/controller.py

@@ -0,0 +1,321 @@
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from enum import Enum
+from typing import Mapping
+
+import numpy as np
+
+from engine.bocpd import run_bocpd
+from engine.brain import BrainSignals, coerce_signals
+
+
+class Action(str, Enum):
+    SILENT = "SILENT"
+    BACKCHANNEL = "BACKCHANNEL"
+    TAKE_FLOOR = "TAKE_FLOOR"
+    INTERRUPT = "INTERRUPT"
+
+
+@dataclass(frozen=True)
+class ControllerConfig:
+    w_surprise: float = 0.70
+    w_change: float = 1.30
+    w_readiness: float = 0.80
+    w_end: float = 1.20
+    w_barge: float = 0.60
+    negative_surprise_weight: float = 0.25
+    tau: float = 1.60
+    backchannel_tau_fraction: float = 0.70
+    barge_tau_fraction: float = 0.50
+    take_floor_p_end: float = 0.70
+    interrupt_p_end_max: float = 0.35
+    backchannel_p_end_max: float = 0.35
+    min_readiness: float = 0.45
+    refractory_steps: int = 2
+    surprise_z_cap: float = 3.0
+    change_hazard: float = 0.35
+    change_prior_kappa: float = 0.20
+    change_prior_alpha: float = 0.75
+    change_prior_beta: float = 0.20
+    change_z_cap: float = 3.0
+    change_z_threshold: float = 1.15
+    turn_end_tau_discount: float = 0.35
+
+    @property
+    def backchannel_tau(self) -> float:
+        return self.backchannel_tau_fraction * self.tau
+
+    @property
+    def barge_tau(self) -> float:
+        return self.barge_tau_fraction * self.tau
+
+
+@dataclass
+class RunningStats:
+    n: int = 0
+    mean: float = 0.0
+    m2: float = 0.0
+
+    @property
+    def std(self) -> float:
+        if self.n < 2:
+            return 1.0
+        return max((self.m2 / (self.n - 1)) ** 0.5, 1.0e-6)
+
+    def zscore(self, value: float, cap: float) -> float:
+        if self.n < 2:
+            return 0.0
+        z_value = (float(value) - self.mean) / self.std
+        return float(np.clip(z_value, -cap, cap))
+
+    def update(self, value: float) -> None:
+        self.n += 1
+        delta = float(value) - self.mean
+        self.mean += delta / self.n
+        self.m2 += delta * (float(value) - self.mean)
+
+
+@dataclass
+class AgentState:
+    surprise_stats: RunningStats = field(default_factory=RunningStats)
+    previous_hidden: np.ndarray | None = None
+    hidden_deltas: list[float] = field(default_factory=list)
+    hidden_delta_z: list[float] = field(default_factory=list)
+    hidden_delta_stats: RunningStats = field(default_factory=RunningStats)
+    previous_map_run_length: int | None = None
+    refractory_until: int = 0
+
+
+@dataclass(frozen=True)
+class AgentDecision:
+    agent_id: str
+    action: Action
+    urge: float
+    z_surprise: float
+    change_score: float
+    readiness: float
+    p_end: float
+    hidden_delta: float
+    map_run_length: int
+    refractory: bool
+
+
+@dataclass(frozen=True)
+class ControllerTick:
+    step: int
+    floor_holder: str
+    winner: str | None
+    decisions: dict[str, AgentDecision]
+
+
+class WhenToSpeakController:
+    """Training-free multi-agent timing controller."""
+
+    def __init__(self, agent_ids: list[str], config: ControllerConfig | None = None) -> None:
+        if not agent_ids:
+            raise ValueError("agent_ids must not be empty")
+        self.agent_ids = list(agent_ids)
+        self.config = config or ControllerConfig()
+        self.states = {agent_id: AgentState() for agent_id in self.agent_ids}
+        self.step = 0
+        self.floor_holder = "human"
+
+    def reset(self) -> None:
+        self.states = {agent_id: AgentState() for agent_id in self.agent_ids}
+        self.step = 0
+        self.floor_holder = "human"
+
+    def tick(
+        self,
+        signals_by_agent: Mapping[str, BrainSignals | dict[str, object]],
+        *,
+        floor_holder: str | None = None,
+    ) -> ControllerTick:
+        self.step += 1
+        if floor_holder is not None:
+            self.floor_holder = floor_holder
+
+        scored: dict[str, AgentDecision] = {}
+        proposed: dict[str, Action] = {}
+        for agent_id in self.agent_ids:
+            if agent_id not in signals_by_agent:
+                raise KeyError(f"missing signals for agent {agent_id!r}")
+            signal = coerce_signals(signals_by_agent[agent_id])
+            decision = self._score_agent(agent_id, signal)
+            scored[agent_id] = decision
+            proposed[agent_id] = decision.action
+
+        winner = self._floor_winner(scored)
+        final_decisions: dict[str, AgentDecision] = {}
+        for agent_id, decision in scored.items():
+            action = decision.action
+            if action in {Action.TAKE_FLOOR, Action.INTERRUPT} and agent_id != winner:
+                action = Action.BACKCHANNEL if self._may_backchannel(decision) else Action.SILENT
+            final_decisions[agent_id] = AgentDecision(
+                agent_id=decision.agent_id,
+                action=action,
+                urge=decision.urge,
+                z_surprise=decision.z_surprise,
+                change_score=decision.change_score,
+                readiness=decision.readiness,
+                p_end=decision.p_end,
+                hidden_delta=decision.hidden_delta,
+                map_run_length=decision.map_run_length,
+                refractory=decision.refractory,
+            )
+
+        for agent_id, decision in final_decisions.items():
+            if decision.action in {Action.TAKE_FLOOR, Action.INTERRUPT}:
+                self.states[agent_id].refractory_until = self.step + self.config.refractory_steps
+
+        if winner is not None:
+            self.floor_holder = winner
+
+        return ControllerTick(
+            step=self.step,
+            floor_holder=self.floor_holder,
+            winner=winner,
+            decisions=final_decisions,
+        )
+
+    def _score_agent(self, agent_id: str, signal: BrainSignals) -> AgentDecision:
+        state = self.states[agent_id]
+        z_surprise = state.surprise_stats.zscore(signal.surprise, self.config.surprise_z_cap)
+        hidden_delta, change_score, map_run_length = self._change_features(state, signal.hidden)
+        barge = self.config.w_barge * max(z_surprise, 0.0) * signal.readiness * (1.0 - signal.p_end)
+        surprise_term = z_surprise if z_surprise >= 0.0 else self.config.negative_surprise_weight * z_surprise
+        urge = (
+            self.config.w_surprise * surprise_term
+            + self.config.w_change * change_score
+            + self.config.w_readiness * signal.readiness
+            + self.config.w_end * signal.p_end
+            + barge
+        )
+
+        refractory = self.step <= state.refractory_until
+        action = self._classify(urge, z_surprise, change_score, signal, refractory)
+
+        state.surprise_stats.update(signal.surprise)
+        state.previous_hidden = signal.hidden.astype(np.float32, copy=True)
+        return AgentDecision(
+            agent_id=agent_id,
+            action=action,
+            urge=float(urge),
+            z_surprise=float(z_surprise),
+            change_score=float(change_score),
+            readiness=signal.readiness,
+            p_end=signal.p_end,
+            hidden_delta=float(hidden_delta),
+            map_run_length=int(map_run_length),
+            refractory=refractory,
+        )
+
+    def _change_features(self, state: AgentState, hidden: np.ndarray) -> tuple[float, float, int]:
+        if state.previous_hidden is None:
+            return 0.0, 0.0, 0
+
+        hidden_delta = cosine_distance(state.previous_hidden, hidden)
+        delta_z = state.hidden_delta_stats.zscore(hidden_delta, self.config.change_z_cap)
+        state.hidden_delta_stats.update(hidden_delta)
+        state.hidden_deltas.append(hidden_delta)
+        state.hidden_delta_z.append(delta_z)
+        results = run_bocpd(
+            np.asarray(state.hidden_delta_z, dtype=np.float64),
+            hazard=self.config.change_hazard,
+            prior_kappa=self.config.change_prior_kappa,
+            prior_alpha=self.config.change_prior_alpha,
+            prior_beta=self.config.change_prior_beta,
+        )
+        latest = results[-1]
+        previous_map = state.previous_map_run_length
+        state.previous_map_run_length = latest.map_run_length
+        if previous_map is None:
+            return hidden_delta, 0.0, latest.map_run_length
+
+        collapsed = latest.map_run_length < previous_map
+        collapse_ratio = (previous_map - latest.map_run_length) / max(previous_map, 1)
+        collapse_score = max(1.0, collapse_ratio) if collapsed else 0.0
+        posterior_score = max(0.0, (latest.cp_prob - self.config.change_hazard) / max(1.0 - self.config.change_hazard, 1.0e-9))
+        z_score = max(0.0, abs(delta_z) - self.config.change_z_threshold) / max(
+            self.config.change_z_cap - self.config.change_z_threshold,
+            1.0e-9,
+        )
+        change_score = max(collapse_score, posterior_score, z_score)
+        return hidden_delta, float(change_score), latest.map_run_length
+
+    def _classify(
+        self,
+        urge: float,
+        z_surprise: float,
+        change_score: float,
+        signal: BrainSignals,
+        refractory: bool,
+    ) -> Action:
+        if refractory:
+            return Action.SILENT
+
+        ready = signal.readiness >= self.config.min_readiness
+        human_has_floor = self.floor_holder == "human"
+        barge_signal = max(z_surprise, 0.0) * signal.readiness * (1.0 - signal.p_end)
+        floor_tau = self._effective_tau(signal.p_end)
+
+        if human_has_floor and ready and signal.p_end >= self.config.take_floor_p_end and urge >= floor_tau:
+            return Action.TAKE_FLOOR
+        if (
+            human_has_floor
+            and ready
+            and signal.p_end <= self.config.interrupt_p_end_max
+            and urge >= floor_tau
+            and barge_signal >= self.config.barge_tau
+        ):
+            return Action.INTERRUPT
+        if (
+            human_has_floor
+            and ready
+            and signal.p_end <= self.config.backchannel_p_end_max
+            and change_score > 0.0
+            and urge >= self.config.backchannel_tau
+        ):
+            return Action.BACKCHANNEL
+        if (
+            human_has_floor
+            and ready
+            and urge >= self.config.backchannel_tau
+            and signal.p_end <= self.config.backchannel_p_end_max
+        ):
+            return Action.BACKCHANNEL
+        return Action.SILENT
+
+    def _floor_winner(self, decisions: Mapping[str, AgentDecision]) -> str | None:
+        contenders = [
+            decision
+            for decision in decisions.values()
+            if decision.action in {Action.TAKE_FLOOR, Action.INTERRUPT}
+            and decision.urge >= self._effective_tau(decision.p_end)
+        ]
+        if not contenders:
+            return None
+        return max(contenders, key=lambda decision: (decision.urge, -self.agent_ids.index(decision.agent_id))).agent_id
+
+    def _may_backchannel(self, decision: AgentDecision) -> bool:
+        return (
+            not decision.refractory
+            and self.floor_holder == "human"
+            and decision.urge >= self.config.backchannel_tau
+            and decision.p_end <= self.config.backchannel_p_end_max
+        )
+
+    def _effective_tau(self, p_end: float) -> float:
+        discount = self.config.turn_end_tau_discount * float(np.clip(p_end, 0.0, 1.0))
+        return self.config.tau * max(0.20, 1.0 - discount)
+
+
+def cosine_distance(left: np.ndarray, right: np.ndarray) -> float:
+    left_vec = np.asarray(left, dtype=np.float32)
+    right_vec = np.asarray(right, dtype=np.float32)
+    denom = float(np.linalg.norm(left_vec) * np.linalg.norm(right_vec))
+    if denom <= 1.0e-12:
+        return 0.0
+    similarity = float(np.dot(left_vec, right_vec) / denom)
+    return float(np.clip(1.0 - similarity, 0.0, 2.0))

engine/conversation.py

@@ -0,0 +1,259 @@
+from __future__ import annotations
+
+import argparse
+import json
+import time
+from dataclasses import asdict, dataclass
+from pathlib import Path
+from typing import Any, Sequence
+
+from engine.controller import Action, ControllerConfig, ControllerTick, WhenToSpeakController
+from engine.live_brain import BrainClient, Dialogue, LiveBrainPanel, Persona
+
+
+ROOT = Path(__file__).resolve().parents[1]
+DEFAULT_LOG_PATH = ROOT / "eval" / "conversation_log.json"
+
+
+@dataclass(frozen=True)
+class TranscriptChunk:
+    text: str
+    silence_flag: bool = False
+
+
+@dataclass(frozen=True)
+class ConversationResult:
+    events: list[dict[str, Any]]
+    dialogue: Dialogue
+    personas: list[Persona]
+    total_latency_ms: float
+    model_name: str
+    device_name: str
+    generated_examples: list[dict[str, Any]]
+
+
+def default_personas() -> list[Persona]:
+    return [
+        Persona(
+            agent_id="numbers_vc",
+            display_name="Numbers VC",
+            system_prompt=(
+                "You are a numbers-obsessed venture investor. Be blunt, specific, and quantitative. "
+                "Ask for denominators, cohorts, margins, contract evidence, and arithmetic that actually closes."
+            ),
+        ),
+        Persona(
+            agent_id="vision_optimist",
+            display_name="Vision Optimist",
+            system_prompt=(
+                "You are a big-vision optimist. You look for the huge version of the company, but your "
+                "questions are crisp and founder-facing when the story needs a missing bridge."
+            ),
+        ),
+        Persona(
+            agent_id="ruthless_skeptic",
+            display_name="Ruthless Skeptic",
+            system_prompt=(
+                "You are a ruthless startup skeptic. Interrupt bad claims in plain English. No pleasantries, "
+                "no throat-clearing, no softening. Be sharp without being long."
+            ),
+        ),
+    ]
+
+
+def sample_pitch_stream() -> list[TranscriptChunk]:
+    return [
+        TranscriptChunk("so basically our startup helps small retailers manage inventory"),
+        TranscriptChunk("we connect to their point of sale and purchase orders"),
+        TranscriptChunk("we already have ten thousand stores and zero churn after launching last week"),
+        TranscriptChunk("then we predict stockouts and write reorder suggestions automatically"),
+        TranscriptChunk("we are converting pilots into paid contracts this month"),
+        TranscriptChunk("so we think this becomes the operating system for local retail"),
+        TranscriptChunk("that's the pitch", silence_flag=True),
+    ]
+
+
+def demo_controller_config() -> ControllerConfig:
+    return ControllerConfig(
+        tau=0.85,
+        min_readiness=0.08,
+        w_surprise=0.85,
+        w_barge=0.85,
+        w_readiness=0.75,
+        w_end=1.05,
+        backchannel_tau_fraction=0.72,
+        barge_tau_fraction=0.50,
+        turn_end_tau_discount=0.45,
+    )
+
+
+class Conversation:
+    def __init__(
+        self,
+        personas: list[Persona],
+        brain_panel: LiveBrainPanel,
+        controller: WhenToSpeakController | None = None,
+    ) -> None:
+        self.personas = personas
+        self.brain_panel = brain_panel
+        self.controller = controller or WhenToSpeakController(
+            brain_panel.agent_ids,
+            config=demo_controller_config(),
+        )
+
+    def run(self, stream: Sequence[TranscriptChunk]) -> ConversationResult:
+        started = time.perf_counter()
+        dialogue: Dialogue = []
+        current_user_text = ""
+        events: list[dict[str, Any]] = []
+        generated_examples: list[dict[str, Any]] = []
+
+        for step_index, chunk in enumerate(stream, start=1):
+            dialogue_before = _dialogue_with_current_user(dialogue, current_user_text)
+            signals = self.brain_panel.step_all(dialogue_before, chunk.text, chunk.silence_flag)
+            current_user_text = _join_text(current_user_text, chunk.text)
+            tick = self.controller.tick(signals, floor_holder="human")
+            event = self._event(step_index, chunk, tick)
+
+            winner = tick.winner
+            if winner is not None:
+                if current_user_text:
+                    dialogue.append({"role": "user", "speaker": "founder", "text": current_user_text})
+                    current_user_text = ""
+
+                generated = self.brain_panel.generate(winner, dialogue)
+                reply_text = str(generated.get("reply_text", ""))
+                if reply_text:
+                    dialogue.append({"role": "assistant", "speaker": winner, "text": reply_text})
+                event["generated"] = {
+                    "agent_id": winner,
+                    "reply_text": reply_text,
+                    "reply_source": generated.get("reply_source"),
+                    "raw_reply_text": generated.get("raw_reply_text"),
+                    "latency_ms": generated.get("latency_ms"),
+                    "model_name": generated.get("model_name"),
+                }
+                generated_examples.append(event["generated"])
+
+            events.append(event)
+
+        if current_user_text:
+            dialogue.append({"role": "user", "speaker": "founder", "text": current_user_text})
+
+        raw = self.brain_panel.last_raw or {}
+        return ConversationResult(
+            events=events,
+            dialogue=dialogue,
+            personas=self.personas,
+            total_latency_ms=(time.perf_counter() - started) * 1000.0,
+            model_name=str(raw.get("model_name", "")),
+            device_name=str(raw.get("device_name", "")),
+            generated_examples=generated_examples,
+        )
+
+    def _event(self, step_index: int, chunk: TranscriptChunk, tick: ControllerTick) -> dict[str, Any]:
+        raw = self.brain_panel.last_raw or {}
+        decisions = {}
+        for agent_id, decision in tick.decisions.items():
+            brain_raw = self.brain_panel.last_results.get(agent_id, {})
+            decisions[agent_id] = {
+                "action": decision.action.value,
+                "urge": decision.urge,
+                "z_surprise": decision.z_surprise,
+                "change_score": decision.change_score,
+                "readiness": decision.readiness,
+                "p_end": decision.p_end,
+                "hidden_delta": decision.hidden_delta,
+                "map_run_length": decision.map_run_length,
+                "brain_latency_ms": brain_raw.get("latency_ms"),
+                "surprise": brain_raw.get("surprise"),
+            }
+        return {
+            "step": step_index,
+            "new_user_text": chunk.text,
+            "silence_flag": chunk.silence_flag,
+            "winner": tick.winner,
+            "floor_holder": tick.floor_holder,
+            "batch_latency_ms": raw.get("batch_latency_ms"),
+            "model_name": raw.get("model_name"),
+            "device_name": raw.get("device_name"),
+            "decisions": decisions,
+        }
+
+
+def save_conversation_log(result: ConversationResult, path: str | Path = DEFAULT_LOG_PATH) -> Path:
+    output = Path(path)
+    output.parent.mkdir(parents=True, exist_ok=True)
+    data = {
+        "model_name": result.model_name,
+        "device_name": result.device_name,
+        "total_latency_ms": result.total_latency_ms,
+        "personas": [asdict(persona) for persona in result.personas],
+        "events": result.events,
+        "dialogue": result.dialogue,
+        "generated_examples": result.generated_examples,
+    }
+    output.write_text(json.dumps(data, indent=2), encoding="utf-8")
+    return output
+
+
+def readable_log(result: ConversationResult) -> str:
+    persona_names = {persona.agent_id: persona.display_name for persona in result.personas}
+    lines = [
+        f"Model: {result.model_name or 'unknown'} on {result.device_name or 'unknown'}",
+        f"Total wall latency: {result.total_latency_ms:.1f} ms",
+    ]
+    for event in result.events:
+        lines.append(f"[{event['step']}] USER + {event['new_user_text']!r} silence={event['silence_flag']}")
+        for agent_id, decision in event["decisions"].items():
+            action = decision["action"]
+            if action == Action.SILENT.value:
+                continue
+            lines.append(
+                "    "
+                f"{persona_names.get(agent_id, agent_id)} -> {action} "
+                f"urge={decision['urge']:.2f} readiness={decision['readiness']:.2f} "
+                f"p_end={decision['p_end']:.2f}"
+            )
+        if "generated" in event:
+            generated = event["generated"]
+            lines.append(f"    {persona_names.get(generated['agent_id'], generated['agent_id'])}: {generated['reply_text']}")
+    return "\n".join(lines)
+
+
+def _dialogue_with_current_user(dialogue: Dialogue, current_user_text: str) -> Dialogue:
+    snapshot = [dict(turn) for turn in dialogue]
+    snapshot.append({"role": "user", "speaker": "founder", "text": current_user_text})
+    return snapshot
+
+
+def _join_text(left: str, right: str) -> str:
+    left = left.strip()
+    right = right.strip()
+    if not left:
+        return right
+    if not right:
+        return left
+    return f"{left} {right}"
+
+
+def run_demo(log_path: str | Path = DEFAULT_LOG_PATH, client: BrainClient | None = None) -> ConversationResult:
+    personas = default_personas()
+    panel = LiveBrainPanel(personas, client=client)
+    conversation = Conversation(personas, panel)
+    result = conversation.run(sample_pitch_stream())
+    save_conversation_log(result, log_path)
+    print(readable_log(result))
+    print(f"Wrote {log_path}")
+    return result
+
+
+def main(argv: list[str] | None = None) -> None:
+    parser = argparse.ArgumentParser(description="Run the text-streamed WhenToSpeak conversation demo.")
+    parser.add_argument("--log-path", default=str(DEFAULT_LOG_PATH))
+    args = parser.parse_args(argv)
+    run_demo(args.log_path)
+
+
+if __name__ == "__main__":
+    main()

engine/live_brain.py

@@ -0,0 +1,151 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Protocol
+
+import numpy as np
+
+from engine.brain import Brain, BrainSignals
+
+
+Dialogue = list[dict[str, str]]
+
+
+class BrainClient(Protocol):
+    def step(
+        self,
+        agent_id: str,
+        system_prompt: str,
+        dialogue_so_far: Dialogue,
+        new_user_text: str,
+        silence_flag: bool,
+    ) -> dict[str, object]:
+        ...
+
+    def step_many(
+        self,
+        agent_payloads: list[dict[str, str]],
+        dialogue_so_far: Dialogue,
+        new_user_text: str,
+        silence_flag: bool,
+    ) -> dict[str, object]:
+        ...
+
+    def generate(self, agent_id: str, system_prompt: str, dialogue: Dialogue) -> dict[str, object]:
+        ...
+
+
+class ModalBrainClient:
+    """Thin lazy wrapper around `modal_app.brain_modal` remote functions."""
+
+    def __init__(self) -> None:
+        from modal_app import brain_modal
+
+        self._brain_modal = brain_modal
+
+    def step(
+        self,
+        agent_id: str,
+        system_prompt: str,
+        dialogue_so_far: Dialogue,
+        new_user_text: str,
+        silence_flag: bool,
+    ) -> dict[str, object]:
+        return self._brain_modal.step.remote(agent_id, system_prompt, dialogue_so_far, new_user_text, silence_flag)
+
+    def step_many(
+        self,
+        agent_payloads: list[dict[str, str]],
+        dialogue_so_far: Dialogue,
+        new_user_text: str,
+        silence_flag: bool,
+    ) -> dict[str, object]:
+        return self._brain_modal.step_many.remote(agent_payloads, dialogue_so_far, new_user_text, silence_flag)
+
+    def generate(self, agent_id: str, system_prompt: str, dialogue: Dialogue) -> dict[str, object]:
+        return self._brain_modal.generate.remote(agent_id, system_prompt, dialogue)
+
+
+@dataclass(frozen=True)
+class Persona:
+    agent_id: str
+    display_name: str
+    system_prompt: str
+
+
+class LiveBrain(Brain):
+    """One-agent Brain implementation backed by Modal."""
+
+    def __init__(self, persona: Persona, client: BrainClient | None = None) -> None:
+        self.persona = persona
+        self.client = client or ModalBrainClient()
+        self.last_raw: dict[str, object] | None = None
+        self._latest: BrainSignals | None = None
+
+    def step(self, dialogue_so_far: Dialogue, new_user_text: str, silence_flag: bool) -> BrainSignals:
+        raw = self.client.step(
+            self.persona.agent_id,
+            self.persona.system_prompt,
+            dialogue_so_far,
+            new_user_text,
+            silence_flag,
+        )
+        self.last_raw = raw
+        self._latest = signals_from_raw(raw)
+        return self._latest
+
+    def next_signals(self) -> BrainSignals:
+        if self._latest is None:
+            raise RuntimeError("LiveBrain has no queued signals; call step() first")
+        return self._latest
+
+    def generate(self, dialogue: Dialogue) -> dict[str, object]:
+        return self.client.generate(self.persona.agent_id, self.persona.system_prompt, dialogue)
+
+
+class LiveBrainPanel:
+    """Batched Modal brain calls for a multi-agent panel."""
+
+    def __init__(self, personas: list[Persona], client: BrainClient | None = None) -> None:
+        if not personas:
+            raise ValueError("personas must not be empty")
+        self.personas = personas
+        self.client = client or ModalBrainClient()
+        self.last_raw: dict[str, object] | None = None
+        self.last_results: dict[str, dict[str, object]] = {}
+
+    @property
+    def agent_ids(self) -> list[str]:
+        return [persona.agent_id for persona in self.personas]
+
+    def step_all(self, dialogue_so_far: Dialogue, new_user_text: str, silence_flag: bool) -> dict[str, BrainSignals]:
+        payloads = [
+            {"agent_id": persona.agent_id, "system_prompt": persona.system_prompt}
+            for persona in self.personas
+        ]
+        raw = self.client.step_many(payloads, dialogue_so_far, new_user_text, silence_flag)
+        self.last_raw = raw
+        results = raw.get("results", {}) if isinstance(raw, dict) else {}
+        self.last_results = {agent_id: dict(result) for agent_id, result in results.items()}
+        return {agent_id: signals_from_raw(result) for agent_id, result in self.last_results.items()}
+
+    def generate(self, agent_id: str, dialogue: Dialogue) -> dict[str, object]:
+        persona = self.persona(agent_id)
+        return self.client.generate(agent_id, persona.system_prompt, dialogue)
+
+    def persona(self, agent_id: str) -> Persona:
+        for persona in self.personas:
+            if persona.agent_id == agent_id:
+                return persona
+        raise KeyError(f"unknown persona {agent_id!r}")
+
+
+def signals_from_raw(raw: dict[str, object]) -> BrainSignals:
+    if not raw.get("ok", False):
+        raise RuntimeError(str(raw.get("failure", "brain call failed")))
+    return BrainSignals(
+        surprise=float(raw["surprise"]),
+        hidden=np.asarray(raw["hidden"], dtype=np.float32),
+        readiness=float(raw["readiness"]),
+        p_end=float(raw["p_end"]),
+    )

engine/probe.py

@@ -0,0 +1,240 @@
+from __future__ import annotations
+
+import os
+import shutil
+import time
+from pathlib import Path
+
+import numpy as np
+import torch
+import torch.nn.functional as F
+from transformers import AutoModelForCausalLM, AutoTokenizer
+
+
+ROOT = Path(__file__).resolve().parents[1]
+EVAL_DIR = ROOT / "eval"
+DUMP_PATH = EVAL_DIR / "probe_dump.npz"
+CANDIDATE_MODELS = [
+    "nvidia/NVIDIA-Nemotron-3-Nano-4B-BF16",
+    "Qwen/Qwen2.5-3B-Instruct",
+    "openbmb/MiniCPM3-4B",
+]
+TRANSCRIPT_CHUNKS = [
+    "so basically",
+    "our startup uses",
+    "ai to help",
+    "small businesses",
+    "manage inventory",
+    "and we think",
+    "the market is huge",
+    "and honestly",
+    "we already have",
+    "like a thousand",
+    "users and",
+    "growing fast",
+]
+MIN_MODEL_DOWNLOAD_FREE_BYTES = 6 * 1024**3
+
+
+def configure_local_caches() -> None:
+    os.environ.setdefault("HF_HOME", str(ROOT / ".hf-cache"))
+    os.environ.setdefault("TRANSFORMERS_CACHE", str(ROOT / ".hf-cache" / "transformers"))
+    os.environ.setdefault("TORCH_HOME", str(ROOT / ".torch-cache"))
+
+
+def cuda_summary() -> torch.device:
+    print(f"torch.__version__ = {torch.__version__}")
+    print(f"torch.version.cuda = {torch.version.cuda}")
+    print(f"torch.cuda.is_available() = {torch.cuda.is_available()}")
+    if torch.cuda.is_available():
+        print(f"torch.cuda.get_device_name(0) = {torch.cuda.get_device_name(0)}")
+        return torch.device("cuda:0")
+
+    print("LOUD CUDA FALLBACK: CUDA/Blackwell is not available in this torch environment; using CPU.")
+    return torch.device("cpu")
+
+
+def common_prefix_len(previous: list[int], current: list[int]) -> int:
+    length = 0
+    for left, right in zip(previous, current):
+        if left != right:
+            break
+        length += 1
+    return length
+
+
+def save_failure(failure: str) -> None:
+    EVAL_DIR.mkdir(parents=True, exist_ok=True)
+    np.savez(
+        DUMP_PATH,
+        nll_series=np.asarray([], dtype=np.float32),
+        hidden_states=np.empty((0, 0), dtype=np.float32),
+        update_ms=np.asarray([], dtype=np.float32),
+        added_text=np.asarray([], dtype=object),
+        model=np.asarray("", dtype=object),
+        device=np.asarray("cpu", dtype=object),
+        dtype=np.asarray("", dtype=object),
+        failure=np.asarray(failure, dtype=object),
+    )
+
+
+def load_first_model(device: torch.device) -> tuple[object, object, str, float, float] | None:
+    free_bytes = shutil.disk_usage(ROOT).free
+    local_files_only = free_bytes < MIN_MODEL_DOWNLOAD_FREE_BYTES
+    if local_files_only:
+        free_gib = free_bytes / 1024**3
+        needed_gib = MIN_MODEL_DOWNLOAD_FREE_BYTES / 1024**3
+        print(
+            "LOUD MODEL DOWNLOAD SKIP: only "
+            f"{free_gib:.2f} GiB free; need at least {needed_gib:.1f} GiB to attempt these 3B/4B model downloads. "
+            "Trying repo-local cache only."
+        )
+
+    if device.type == "cuda":
+        dtype = torch.bfloat16 if torch.cuda.is_bf16_supported() else torch.float16
+    else:
+        dtype = torch.float32
+
+    failures: list[str] = []
+    for model_id in CANDIDATE_MODELS:
+        print(f"Attempting model: {model_id}")
+        if device.type == "cuda":
+            torch.cuda.empty_cache()
+            torch.cuda.reset_peak_memory_stats(0)
+
+        start = time.perf_counter()
+        try:
+            tokenizer = AutoTokenizer.from_pretrained(
+                model_id,
+                trust_remote_code=True,
+                local_files_only=local_files_only,
+            )
+            load_kwargs = {
+                "trust_remote_code": True,
+                "torch_dtype": dtype,
+                "low_cpu_mem_usage": True,
+                "local_files_only": local_files_only,
+            }
+            if device.type == "cuda":
+                load_kwargs["device_map"] = {"": 0}
+
+            model = AutoModelForCausalLM.from_pretrained(model_id, **load_kwargs)
+            if device.type == "cpu":
+                model.to(device)
+            model.eval()
+
+            load_seconds = time.perf_counter() - start
+            actual_device = next(model.parameters()).device
+            actual_dtype = next(model.parameters()).dtype
+            vram_gib = 0.0
+            if device.type == "cuda":
+                torch.cuda.synchronize()
+                vram_gib = torch.cuda.memory_allocated(0) / 1024**3
+            print(
+                "LOADED "
+                f"model={model_id} device={actual_device} dtype={actual_dtype} "
+                f"load_seconds={load_seconds:.2f} vram_used_gib={vram_gib:.2f}"
+            )
+            return tokenizer, model, model_id, load_seconds, vram_gib
+        except Exception as exc:  # noqa: BLE001 - spike should continue through model fallbacks.
+            elapsed = time.perf_counter() - start
+            message = f"{model_id} failed after {elapsed:.2f}s: {type(exc).__name__}: {exc}"
+            print(message)
+            failures.append(message)
+
+    failure = "No candidate model loaded. " + " | ".join(failures)
+    print(f"LOUD PROBE FAILURE: {failure}")
+    save_failure(failure)
+    return None
+
+
+def run_updates(tokenizer: object, model: object, model_id: str) -> None:
+    device = next(model.parameters()).device
+    previous_ids: list[int] = []
+    prefixes: list[str] = []
+    running = ""
+    for chunk in TRANSCRIPT_CHUNKS:
+        running = f"{running} {chunk}".strip()
+        prefixes.append(running)
+
+    nll_series: list[float] = []
+    hidden_rows: list[np.ndarray] = []
+    update_ms: list[float] = []
+    added_text: list[str] = []
+
+    print("step | added_text | mean_NLL | hidden_dim | update_ms")
+    print("-----|------------|----------|------------|----------")
+    for step, (chunk, prefix) in enumerate(zip(TRANSCRIPT_CHUNKS, prefixes), start=1):
+        if device.type == "cuda":
+            torch.cuda.synchronize()
+        start = time.perf_counter()
+
+        encoded = tokenizer(prefix, return_tensors="pt", add_special_tokens=False)
+        current_ids = encoded["input_ids"][0].tolist()
+        new_start = common_prefix_len(previous_ids, current_ids)
+        inputs = {name: tensor.to(device) for name, tensor in encoded.items()}
+
+        with torch.inference_mode():
+            outputs = model(**inputs, output_hidden_states=True)
+
+            input_ids = inputs["input_ids"]
+            logits = outputs.logits[:, :-1, :].float()
+            targets = input_ids[:, 1:]
+            token_nll = F.cross_entropy(
+                logits.reshape(-1, logits.shape[-1]),
+                targets.reshape(-1),
+                reduction="none",
+            ).reshape(targets.shape)
+
+            nll_start = max(new_start, 1) - 1
+            new_nll = token_nll[0, nll_start:]
+            mean_nll = float(new_nll.mean().detach().cpu()) if new_nll.numel() else float("nan")
+
+            last_hidden = outputs.hidden_states[-1][0]
+            new_hidden = last_hidden[new_start:, :]
+            mean_hidden = new_hidden.float().mean(dim=0).detach().cpu()
+
+        if device.type == "cuda":
+            torch.cuda.synchronize()
+        elapsed_ms = (time.perf_counter() - start) * 1000.0
+
+        hidden_vec = mean_hidden.numpy().astype(np.float32)
+        nll_series.append(mean_nll)
+        hidden_rows.append(hidden_vec)
+        update_ms.append(elapsed_ms)
+        added_text.append(chunk)
+        previous_ids = current_ids
+
+        print(f"{step:>4} | {chunk} | {mean_nll:.4f} | {hidden_vec.shape[0]} | {elapsed_ms:.2f}")
+
+    EVAL_DIR.mkdir(parents=True, exist_ok=True)
+    hidden_matrix = np.vstack(hidden_rows).astype(np.float32)
+    actual_dtype = str(next(model.parameters()).dtype)
+    np.savez(
+        DUMP_PATH,
+        nll_series=np.asarray(nll_series, dtype=np.float32),
+        hidden_states=hidden_matrix,
+        update_ms=np.asarray(update_ms, dtype=np.float32),
+        added_text=np.asarray(added_text, dtype=object),
+        model=np.asarray(model_id, dtype=object),
+        device=np.asarray(str(device), dtype=object),
+        dtype=np.asarray(actual_dtype, dtype=object),
+        failure=np.asarray("", dtype=object),
+    )
+    print(f"Saved {DUMP_PATH}")
+
+
+def main() -> None:
+    configure_local_caches()
+    EVAL_DIR.mkdir(parents=True, exist_ok=True)
+    device = cuda_summary()
+    loaded = load_first_model(device)
+    if loaded is None:
+        return
+
+    tokenizer, model, model_id, _load_seconds, _vram_gib = loaded
+    run_updates(tokenizer, model, model_id)
+
+
+if __name__ == "__main__":
+    main()

engine/traces.py

@@ -0,0 +1,180 @@
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Sequence
+
+import numpy as np
+
+from engine.brain import BrainSignals, ReplayBrain
+
+
+ROOT = Path(__file__).resolve().parents[1]
+DEFAULT_MODAL_DUMP = ROOT / "eval" / "probe_dump_modal.npz"
+DEFAULT_AGENTS = ("investor_a", "investor_b")
+
+
+@dataclass(frozen=True)
+class TraceStep:
+    signals_by_agent: dict[str, BrainSignals]
+    floor_holder: str = "human"
+    note: str = ""
+
+
+@dataclass(frozen=True)
+class SignalTrace:
+    name: str
+    steps: list[TraceStep]
+    expected: dict[str, object] = field(default_factory=dict)
+
+    @property
+    def agent_ids(self) -> list[str]:
+        return list(self.steps[0].signals_by_agent.keys()) if self.steps else []
+
+
+def synthetic_traces() -> list[SignalTrace]:
+    return [
+        clean_monologue_trace(),
+        surprising_claim_trace(),
+        topic_shift_trace(),
+        rambling_pause_trace(),
+    ]
+
+
+def load_modal_trace(path: str | Path = DEFAULT_MODAL_DUMP, agent_ids: Sequence[str] = DEFAULT_AGENTS) -> SignalTrace:
+    path = Path(path)
+    replay = ReplayBrain.from_npz(path, name="modal_probe_replay")
+    samples = list(replay)
+    steps: list[TraceStep] = []
+    for index, sample in enumerate(samples):
+        signals: dict[str, BrainSignals] = {}
+        for offset, agent_id in enumerate(agent_ids):
+            readiness = max(0.0, sample.readiness - 0.08 * offset)
+            signals[agent_id] = BrainSignals(
+                surprise=sample.surprise,
+                hidden=sample.hidden,
+                readiness=readiness,
+                p_end=sample.p_end,
+            )
+        steps.append(TraceStep(signals_by_agent=signals, note=f"modal step {index + 1}"))
+    return SignalTrace(
+        name="modal_probe_replay",
+        steps=steps,
+        expected={"source": str(path), "n_steps": len(steps)},
+    )
+
+
+def all_traces() -> list[SignalTrace]:
+    traces = synthetic_traces()
+    if DEFAULT_MODAL_DUMP.exists():
+        traces.append(load_modal_trace(DEFAULT_MODAL_DUMP))
+    return traces
+
+
+def clean_monologue_trace() -> SignalTrace:
+    base = _unit([1, 0, 0, 0, 0, 0, 0, 0])
+    steps = _trace_from_series(
+        name="clean_monologue_take_floor",
+        hidden_series=[_nudge(base, index, 0.01) for index in range(6)],
+        surprise=[2.0, 2.1, 2.0, 2.2, 2.1, 2.45],
+        readiness_a=[0.25, 0.30, 0.35, 0.40, 0.45, 0.78],
+        readiness_b=[0.20, 0.25, 0.30, 0.32, 0.35, 0.52],
+        p_end=[0.02, 0.03, 0.04, 0.05, 0.10, 0.95],
+        notes=["setup", "details", "still talking", "more context", "closing", "turn complete"],
+    )
+    return SignalTrace(name="clean_monologue_take_floor", steps=steps, expected={"take_floor_step": 6})
+
+
+def surprising_claim_trace() -> SignalTrace:
+    base = _unit([1, 0, 0, 0, 0, 0, 0, 0])
+    steps = _trace_from_series(
+        name="surprising_claim_interrupt",
+        hidden_series=[_nudge(base, index, 0.01) for index in range(6)],
+        surprise=[2.0, 2.1, 7.3, 3.0, 2.4, 2.2],
+        readiness_a=[0.30, 0.42, 0.92, 0.82, 0.55, 0.45],
+        readiness_b=[0.25, 0.32, 0.35, 0.36, 0.38, 0.40],
+        p_end=[0.02, 0.04, 0.18, 0.20, 0.30, 0.55],
+        notes=["setup", "build", "wild claim", "continues", "settles", "handoff"],
+    )
+    return SignalTrace(name="surprising_claim_interrupt", steps=steps, expected={"interrupt_step": 3})
+
+
+def topic_shift_trace() -> SignalTrace:
+    topic_a = _unit([1, 0, 0, 0, 0, 0, 0, 0])
+    topic_b = _unit([0, 1, 0, 0, 0, 0, 0, 0])
+    steps = _trace_from_series(
+        name="topic_shift_backchannel",
+        hidden_series=[
+            _nudge(topic_a, 0, 0.01),
+            _nudge(topic_a, 1, 0.01),
+            _nudge(topic_a, 2, 0.01),
+            _nudge(topic_a, 3, 0.01),
+            _nudge(topic_b, 4, 0.01),
+            _nudge(topic_b, 5, 0.01),
+            _nudge(topic_b, 6, 0.01),
+        ],
+        surprise=[2.0, 2.1, 2.0, 2.1, 2.12, 2.08, 2.06],
+        readiness_a=[0.25, 0.30, 0.35, 0.38, 0.50, 0.45, 0.40],
+        readiness_b=[0.20, 0.25, 0.28, 0.30, 0.35, 0.34, 0.34],
+        p_end=[0.02, 0.04, 0.06, 0.08, 0.22, 0.25, 0.28],
+        notes=["topic a", "topic a", "topic a", "topic a", "topic b shift", "topic b", "topic b"],
+    )
+    return SignalTrace(name="topic_shift_backchannel", steps=steps, expected={"backchannel_step": 5})
+
+
+def rambling_pause_trace() -> SignalTrace:
+    base = _unit([0, 0, 1, 0, 0, 0, 0, 0])
+    steps = _trace_from_series(
+        name="rambling_pause_take_floor",
+        hidden_series=[_nudge(base, index, 0.015) for index in range(7)],
+        surprise=[2.1, 2.0, 2.2, 2.1, 2.0, 2.1, 2.45],
+        readiness_a=[0.20, 0.25, 0.30, 0.38, 0.42, 0.42, 0.80],
+        readiness_b=[0.18, 0.22, 0.25, 0.30, 0.32, 0.35, 0.50],
+        p_end=[0.02, 0.05, 0.10, 0.38, 0.42, 0.44, 0.96],
+        notes=["start", "ramble", "ramble", "awkward pause", "holds", "still unsure", "complete"],
+    )
+    return SignalTrace(name="rambling_pause_take_floor", steps=steps, expected={"hold_step": 4, "take_floor_step": 7})
+
+
+def _trace_from_series(
+    *,
+    name: str,
+    hidden_series: Sequence[np.ndarray],
+    surprise: Sequence[float],
+    readiness_a: Sequence[float],
+    readiness_b: Sequence[float],
+    p_end: Sequence[float],
+    notes: Sequence[str],
+) -> list[TraceStep]:
+    steps: list[TraceStep] = []
+    for index, hidden in enumerate(hidden_series):
+        signals = {
+            "investor_a": BrainSignals(
+                surprise=surprise[index],
+                hidden=hidden,
+                readiness=readiness_a[index],
+                p_end=p_end[index],
+            ),
+            "investor_b": BrainSignals(
+                surprise=max(0.0, surprise[index] - 0.2),
+                hidden=hidden,
+                readiness=readiness_b[index],
+                p_end=p_end[index],
+            ),
+        }
+        steps.append(TraceStep(signals_by_agent=signals, note=notes[index]))
+    return steps
+
+
+def _unit(values: Sequence[float]) -> np.ndarray:
+    vector = np.asarray(values, dtype=np.float32)
+    norm = np.linalg.norm(vector)
+    if norm == 0:
+        return vector
+    return vector / norm
+
+
+def _nudge(base: np.ndarray, index: int, scale: float) -> np.ndarray:
+    vector = base.astype(np.float32, copy=True)
+    vector[(index % (len(vector) - 1)) + 1] += scale
+    return _unit(vector)

engine/viz.py

@@ -0,0 +1,98 @@
+from __future__ import annotations
+
+from pathlib import Path
+
+import matplotlib
+
+matplotlib.use("Agg")
+
+import matplotlib.pyplot as plt
+
+from engine.controller import Action, ControllerConfig, WhenToSpeakController
+from engine.traces import SignalTrace, surprising_claim_trace
+
+
+ROOT = Path(__file__).resolve().parents[1]
+DEFAULT_OUTPUT = ROOT / "eval" / "controller_timeline.png"
+ACTION_COLORS = {
+    Action.BACKCHANNEL: "tab:blue",
+    Action.TAKE_FLOOR: "tab:green",
+    Action.INTERRUPT: "tab:red",
+}
+
+
+def render_timeline(
+    trace: SignalTrace | None = None,
+    output_path: str | Path = DEFAULT_OUTPUT,
+    config: ControllerConfig | None = None,
+) -> Path:
+    trace = trace or surprising_claim_trace()
+    controller = WhenToSpeakController(trace.agent_ids, config=config)
+    ticks = [controller.tick(step.signals_by_agent, floor_holder=step.floor_holder) for step in trace.steps]
+
+    steps = [tick.step for tick in ticks]
+    primary = trace.agent_ids[0]
+    surprise = [step.signals_by_agent[primary].surprise for step in trace.steps]
+    readiness = [step.signals_by_agent[primary].readiness for step in trace.steps]
+    p_end = [step.signals_by_agent[primary].p_end for step in trace.steps]
+    change = [tick.decisions[primary].change_score for tick in ticks]
+
+    fig, axes = plt.subplots(4, 1, figsize=(10, 8), sharex=True)
+    fig.suptitle(f"WhenToSpeak controller timeline: {trace.name}", fontsize=13)
+
+    axes[0].plot(steps, surprise, marker="o", color="tab:orange", label="surprise/NLL")
+    axes[0].set_ylabel("surprise")
+    axes[0].legend(loc="upper right")
+
+    axes[1].plot(steps, change, marker="o", color="tab:purple", label="change score")
+    axes[1].set_ylabel("change")
+    axes[1].legend(loc="upper right")
+
+    axes[2].plot(steps, readiness, marker="o", color="tab:green", label="readiness")
+    axes[2].plot(steps, p_end, marker=".", linestyle="--", color="tab:gray", label="p_end")
+    axes[2].set_ylabel("probability")
+    axes[2].set_ylim(-0.05, 1.05)
+    axes[2].legend(loc="upper right")
+
+    for agent_id in trace.agent_ids:
+        urges = [tick.decisions[agent_id].urge for tick in ticks]
+        axes[3].plot(steps, urges, marker="o", label=f"{agent_id} urge")
+    axes[3].axhline(controller.config.tau, color="black", linestyle="--", linewidth=1, label="tau")
+    axes[3].set_ylabel("urge")
+    axes[3].set_xlabel("step")
+    axes[3].legend(loc="upper right")
+
+    for tick in ticks:
+        for agent_id, decision in tick.decisions.items():
+            if decision.action == Action.SILENT:
+                continue
+            color = ACTION_COLORS[decision.action]
+            axes[3].scatter([tick.step], [decision.urge], color=color, s=90, zorder=5)
+            axes[3].annotate(
+                decision.action.value,
+                (tick.step, decision.urge),
+                textcoords="offset points",
+                xytext=(0, 8),
+                ha="center",
+                fontsize=8,
+                color=color,
+            )
+
+    for axis in axes:
+        axis.grid(True, alpha=0.25)
+
+    output = Path(output_path)
+    output.parent.mkdir(parents=True, exist_ok=True)
+    fig.tight_layout()
+    fig.savefig(output, dpi=160)
+    plt.close(fig)
+    return output
+
+
+def main() -> None:
+    output = render_timeline()
+    print(f"Wrote {output}")
+
+
+if __name__ == "__main__":
+    main()

requirements.txt

@@ -0,0 +1,6 @@
+gradio==6.18.0
+kokoro==0.9.4
+modal==1.5.0
+numpy==2.4.6
+requests==2.32.5
+soundfile==0.14.0

static/README.md

@@ -0,0 +1 @@
+Static assets for the Space live under `apps/pitch/static/`.