Deploy Decompress MiniCPM demo

README.md

@@ -1,13 +1,76 @@
 ---
 title: Decompress
-emoji: 📈
-colorFrom: purple
-colorTo: red
+emoji: 🌿
+colorFrom: green
+colorTo: blue
 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 MiniCPM voice companion that knows when not to talk.
+tags:
+  - gradio
+  - build-small-hackathon
+  - stress-management
+  - wellness
+  - full-duplex
+  - training-free
+  - rag
+  - track:backyard
+  - sponsor:openbmb
+  - sponsor:cohere
+  - sponsor:modal
+  - sponsor:openai
+  - achievement:offbrand
+  - tiny-titan
+  - minicpm
+  - cohere
+models:
+  - openbmb/MiniCPM3-4B
+  - CohereLabs/cohere-transcribe-03-2026
+  - hexgrad/Kokoro-82M
+  - sentence-transformers/all-MiniLM-L6-v2
 ---
 
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+# Decompress
+
+Decompress is a calm, bounded voice/text check-in companion. The point is not
+just what it says; it is when it chooses not to say anything. The same
+training-free when-to-speak controller from Pitch or Perish drives the timing:
+MiniCPM exposes surprise, hidden-state deltas, readiness, and turn-end
+probability, then the controller waits, backchannels, or responds at the pause.
+
+## Backyard Use
+
+This is the practical track version of the method: a small companion for a
+five-minute decompression ritual after work, before sleep, or between stressful
+tasks. It gives one grounded next step and cites the source used by the RAG
+retriever. It is wellness support only, not medical advice, diagnosis, therapy,
+or crisis care.
+
+## Tiny Titan / Sponsor Stack
+
+- **OpenBMB:** `openbmb/MiniCPM3-4B` is the load-bearing brain for NLL,
+  hidden states, readiness, and generated companion lines.
+- **Cohere:** `CohereLabs/cohere-transcribe-03-2026` powers push-to-talk voice
+  input before the exact same text-streamed controller runs.
+- **Modal:** MiniCPM and Cohere ASR run as protected A10G endpoints with one
+  warm container during judging.
+- **OpenAI:** the repo history includes Codex-attributed implementation commits.
+- **Small weights:** the app uses <=4B LLM/ASR models, Kokoro-82M for CPU TTS,
+  and all-MiniLM-L6-v2 for local RAG retrieval.
+
+## Sources
+
+The shipped corpus is intentionally small and conservative. It paraphrases
+open-access stress-management, CBT, breathwork, and grounding sources from WHO,
+NHS, NIH NCCIH, University of Rochester Medical Center, and Scientific Reports,
+with citations displayed in the UI.
+
+## 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.decompress.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/decompress/__init__.py

@@ -0,0 +1 @@
+"""Decompress Gradio app package."""

apps/decompress/app.py

@@ -0,0 +1,123 @@
+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.decompress import runtime
+
+
+APP_DIR = Path(__file__).resolve().parent
+CSS = (APP_DIR / "static" / "decompress.css").read_text(encoding="utf-8")
+JS = (APP_DIR / "static" / "decompress.js").read_text(encoding="utf-8")
+
+
+def stream_checkin(checkin_text: str, tau: float, voice_enabled: bool) -> Iterator[tuple[str, str | None, dict]]:
+    yield from runtime.run_checkin_stream(checkin_text, tau, enable_tts=voice_enabled)
+
+
+def stream_voice_checkin(audio_clip: str | None, tau: float, voice_enabled: bool) -> Iterator[tuple[str, str | None, dict, str]]:
+    yield from runtime.run_voice_checkin_stream(audio_clip, tau, enable_tts=voice_enabled)
+
+
+def build_app() -> gr.Blocks:
+    initial_tau = float(os.getenv("DECOMPRESS_TAU", "0.92"))
+    initial_html = (
+        runtime.preview_room_html(initial_tau)
+        if os.getenv("DECOMPRESS_PREVIEW_STATE", "0") == "1"
+        else runtime.initial_room_html(initial_tau)
+    )
+    with gr.Blocks(title="Decompress", fill_width=True) as demo:
+        room = gr.HTML(value=initial_html, elem_id="decompress-room-output")
+        with gr.Row(elem_id="decompress-control-row"):
+            checkin = gr.Textbox(
+                value=runtime.SAMPLE_CHECKIN,
+                label="What is on your mind?",
+                lines=6,
+                max_lines=10,
+                elem_id="decompress-input",
+            )
+            with gr.Column(elem_id="decompress-controls", scale=0):
+                tau = gr.Slider(
+                    minimum=0.65,
+                    maximum=1.45,
+                    value=initial_tau,
+                    step=0.05,
+                    label="Companion timing",
+                    elem_id="decompress-tau",
+                )
+                voice = gr.Checkbox(
+                    value=os.getenv("DECOMPRESS_TTS", "1").lower() not in {"0", "false"},
+                    label="Kokoro voice",
+                    elem_id="decompress-tts-toggle",
+                )
+                run = gr.Button("Start check-in", elem_id="run-decompress", variant="primary")
+                mic = gr.Audio(
+                    sources=["microphone", "upload"],
+                    type="filepath",
+                    format="wav",
+                    label="Speak your check-in",
+                    elem_id="decompress-mic",
+                )
+                voice_run = gr.Button("Use voice check-in", elem_id="run-decompress-voice", variant="secondary")
+                asr_text = gr.Textbox(
+                    value="",
+                    label="Cohere transcript",
+                    lines=3,
+                    max_lines=5,
+                    interactive=False,
+                    elem_id="decompress-transcript",
+                )
+        audio = gr.Audio(
+            label="Latest companion voice",
+            type="filepath",
+            autoplay=True,
+            interactive=False,
+            elem_id="decompress-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_checkin,
+            inputs=[checkin, tau, voice],
+            outputs=[room, audio, status],
+            api_name="stream_checkin",
+        )
+        voice_run.click(
+            fn=stream_voice_checkin,
+            inputs=[mic, tau, voice],
+            outputs=[room, audio, status, asr_text],
+            api_name="stream_voice_checkin",
+        )
+    return demo
+
+
+def main(argv: list[str] | None = None) -> None:
+    parser = argparse.ArgumentParser(description="Launch the Decompress 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", "7861")))
+    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/decompress/corpus/stress_sources.json

@@ -0,0 +1,44 @@
+[
+  {
+    "id": "who-doing-what-matters",
+    "title": "Doing What Matters in Times of Stress",
+    "publisher": "World Health Organization",
+    "url": "https://www.who.int/publications/i/item/9789240003927",
+    "text": "A practical stress check-in can start by grounding attention in the present, noticing what is happening in the body and room, and then choosing one small action that fits the person's values. The guide frames these as short self-help skills for coping with adversity, not as medical treatment."
+  },
+  {
+    "id": "nhs-self-help-cbt",
+    "title": "Self-help CBT techniques",
+    "publisher": "NHS Every Mind Matters",
+    "url": "https://www.nhs.uk/every-mind-matters/mental-wellbeing-tips/self-help-cbt-techniques/",
+    "text": "CBT-inspired self-help treats thoughts, feelings, and behavior as linked. For everyday stress, a useful move is to notice an unhelpful thought pattern, step back from it, and break a problem into one practical next action."
+  },
+  {
+    "id": "nhs-thought-record",
+    "title": "Thought record",
+    "publisher": "NHS Every Mind Matters",
+    "url": "https://www.nhs.uk/every-mind-matters/mental-wellbeing-tips/self-help-cbt-techniques/thought-record/",
+    "text": "A thought record is a CBT exercise for writing down a situation, feelings, the unhelpful thought, evidence for and against it, and a more balanced alternative. In a brief companion flow, this can be simplified into: name the thought, name one fact, and choose a kinder neutral sentence."
+  },
+  {
+    "id": "nccih-relaxation",
+    "title": "Mind and Body Approaches for Stress and Anxiety: What the Science Says",
+    "publisher": "NIH NCCIH",
+    "url": "https://www.nccih.nih.gov/health/providers/digest/mind-and-body-approaches-for-stress-science",
+    "text": "Relaxation techniques may be helpful for stress-related symptoms and are generally considered safe for healthy people, while some people can have negative experiences. Keep suggestions gentle, optional, and brief; do not present them as treatment."
+  },
+  {
+    "id": "urmc-54321",
+    "title": "5-4-3-2-1 Coping Technique for Anxiety",
+    "publisher": "University of Rochester Medical Center",
+    "url": "https://www.urmc.rochester.edu/behavioral-health-partners/bhp-blog/april-2018/5-4-3-2-1-coping-technique-for-anxiety",
+    "text": "The 5-4-3-2-1 grounding exercise redirects attention to the present by naming things the person can see, feel, hear, smell, and taste. It is useful when thoughts are bouncing around and the person needs a concrete anchor."
+  },
+  {
+    "id": "breathwork-meta",
+    "title": "Effect of breathwork on stress and mental health",
+    "publisher": "Scientific Reports",
+    "url": "https://www.nature.com/articles/s41598-022-27247-y",
+    "text": "A meta-analysis found breathwork may help stress and mental health outcomes, with cautions about heterogeneity. In this app, breathing prompts should stay simple and optional, such as a longer exhale or one slow breath."
+  }
+]

apps/decompress/runtime.py

@@ -0,0 +1,1021 @@
+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.brain import BrainSignals
+from engine.controller import Action, ControllerConfig, ControllerTick, WhenToSpeakController
+from engine.conversation import TranscriptChunk
+from engine.live_brain import BrainClient, Dialogue, signals_from_raw
+
+
+ROOT = Path(__file__).resolve().parents[2]
+APP_DIR = Path(__file__).resolve().parent
+EVAL_DIR = ROOT / "eval"
+DECOMPRESS_LOG_PATH = EVAL_DIR / "decompress_conversation_log.json"
+AUDIO_DIR = EVAL_DIR / "decompress_audio"
+CORPUS_PATH = APP_DIR / "corpus" / "stress_sources.json"
+
+AGENT_ID = "decompress"
+DISPLAY_NAME = "Decompress"
+DEFAULT_EMBEDDER = "sentence-transformers/all-MiniLM-L6-v2"
+SAMPLE_CHECKIN = (
+    "I had a tense day at work and I keep replaying a meeting in my head. "
+    "My shoulders are tight, I feel behind on everything, and I just want five minutes "
+    "to come down before I try to sleep."
+)
+
+CALM_BACKCHANNELS = [
+    "Mm. Keep going.",
+    "I'm here with you.",
+    "That sounds heavy.",
+    "Take your time.",
+]
+
+FALLBACK_LINES = [
+    "Let's slow it down: one easy breath, then name one thing you can feel.",
+    "Make it smaller: unclench your jaw and choose one next tiny step.",
+    "Park one sentence on paper, then let the next exhale be longer.",
+]
+
+_LIVE_TAU = 0.92
+
+
+@dataclass(frozen=True)
+class RetrievedSource:
+    id: str
+    title: str
+    publisher: str
+    url: str
+    text: str
+    score: float
+
+
+@dataclass
+class DecompressStats:
+    backchannels: int = 0
+    take_floors: int = 0
+    interrupts: int = 0
+    holds: int = 0
+
+
+@dataclass
+class DecompressState:
+    tau: float
+    brain_mode: str
+    status: str = "ready"
+    step: int = 0
+    transcript: list[dict[str, Any]] = field(default_factory=list)
+    decision: dict[str, Any] = field(default_factory=dict)
+    citations: list[dict[str, Any]] = field(default_factory=list)
+    stats: DecompressStats = field(default_factory=DecompressStats)
+    tts_error: str | None = None
+    asr_text: str | None = None
+    asr_latency_ms: float | None = None
+    asr_model_name: str | None = None
+    embedding_model: str = "not-loaded"
+    response_latency_ms: float | None = None
+    model_name: str | None = None
+    device_name: str | None = None
+
+
+class EndpointDecompressBrainClient:
+    """BrainClient backed by the protected MiniCPM Modal HTTPS endpoint."""
+
+    def __init__(self, endpoint_url: str | None = None, *, timeout_s: float = 240.0) -> None:
+        self.endpoint_url = (endpoint_url or os.environ["DECOMPRESS_BRAIN_ENDPOINT_URL"]).rstrip("/")
+        self.bearer_token = os.getenv("DECOMPRESS_BRAIN_ENDPOINT_TOKEN") or os.getenv("ENDPOINT_AUTH_TOKEN")
+        self.transcribe_endpoint_url = (
+            os.getenv("DECOMPRESS_TRANSCRIBE_ENDPOINT_URL") or self.endpoint_url
+        ).rstrip("/")
+        self.transcribe_bearer_token = os.getenv("DECOMPRESS_TRANSCRIBE_ENDPOINT_TOKEN") or self.bearer_token
+        if not self.bearer_token:
+            raise RuntimeError("DECOMPRESS_BRAIN_ENDPOINT_TOKEN is required for endpoint mode")
+        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]:
+        raw = self.step_many(
+            [{"agent_id": agent_id, "system_prompt": system_prompt}],
+            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 transcribe(self, audio_path: str, language: str = "en") -> dict[str, object]:
+        import requests
+
+        headers = self._auth_headers(self.transcribe_bearer_token)
+        with Path(audio_path).open("rb") as audio_file:
+            files = {"file": (Path(audio_path).name, audio_file, "audio/wav")}
+            data = {"language": language}
+            response = requests.post(
+                f"{self.transcribe_endpoint_url}/transcribe",
+                files=files,
+                data=data,
+                headers=headers,
+                timeout=max(self.timeout_s, 600.0),
+            )
+        response.raise_for_status()
+        return dict(response.json())
+
+    def _post(self, path: str, payload: dict[str, object]) -> dict[str, object]:
+        import requests
+
+        response = requests.post(
+            f"{self.endpoint_url}{path}",
+            json=payload,
+            headers={"Content-Type": "application/json", **self._auth_headers(self.bearer_token)},
+            timeout=self.timeout_s,
+        )
+        response.raise_for_status()
+        return dict(response.json())
+
+    @staticmethod
+    def _auth_headers(bearer_token: str | None) -> dict[str, str]:
+        if not bearer_token:
+            return {}
+        return {"Authorization": f"Bearer {bearer_token}"}
+
+
+class DemoDecompressBrainClient:
+    """Deterministic local stand-in for tests and UI work."""
+
+    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]:
+        raw = self.step_many(
+            [{"agent_id": agent_id, "system_prompt": system_prompt}],
+            dialogue_so_far,
+            new_user_text,
+            silence_flag,
+        )
+        return dict(raw["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]:
+        del dialogue_so_far
+        start = time.perf_counter()
+        self.calls += 1
+        text = new_user_text.lower()
+        stress_words = sum(word in text for word in ["tense", "replaying", "tight", "behind", "sleep", "work"])
+        shift = any(word in text for word in ["but", "then", "before", "after"])
+        results = {}
+        for payload in agent_payloads:
+            hidden = np.zeros(10, dtype=np.float32)
+            hidden[0] = 1.0
+            hidden[(self.calls % 5) + 1] = 0.35
+            if shift:
+                hidden[7] = 1.2
+            if silence_flag:
+                hidden[8] = 1.0
+            hidden /= max(float(np.linalg.norm(hidden)), 1.0)
+            p_end = 0.94 if silence_flag else (0.68 if new_user_text.strip().endswith((".", "?", "!")) else 0.18)
+            results[payload["agent_id"]] = {
+                "ok": True,
+                "agent_id": payload["agent_id"],
+                "surprise": float(2.0 + stress_words * 0.7 + self.calls * 0.03),
+                "hidden": hidden.tolist(),
+                "readiness": 0.72 if silence_flag else 0.48,
+                "p_end": p_end,
+                "latency_ms": 3.0,
+                "model_name": "demo-minicpm-brain",
+                "device_name": "cpu-test",
+            }
+        return {
+            "ok": True,
+            "results": results,
+            "batch_latency_ms": (time.perf_counter() - start) * 1000.0,
+            "model_name": "demo-minicpm-brain",
+            "device_name": "cpu-test",
+        }
+
+    def generate(self, agent_id: str, system_prompt: str, dialogue: Dialogue) -> dict[str, object]:
+        del agent_id, system_prompt
+        text = _dialogue_text(dialogue, "")
+        line = fallback_line(text)
+        return {
+            "ok": True,
+            "agent_id": AGENT_ID,
+            "reply_text": line,
+            "raw_reply_text": line,
+            "reply_source": "demo",
+            "latency_ms": 4.0,
+            "model_name": "demo-minicpm-brain",
+        }
+
+    def transcribe(self, audio_path: str, language: str = "en") -> dict[str, object]:
+        del audio_path, language
+        return {
+            "ok": True,
+            "text": SAMPLE_CHECKIN,
+            "language": "en",
+            "latency_ms": 5.0,
+            "duration_s": 4.0,
+            "model_name": "demo-cohere-transcribe",
+        }
+
+
+class KokoroCalmSpeaker:
+    """Lazy Kokoro TTS wrapper; text still works if TTS fails."""
+
+    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, 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}_decompress.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")
+            chunks = []
+            for _, _, audio in self._pipeline(text, voice=os.getenv("DECOMPRESS_VOICE", "af_heart"), speed=0.92):
+                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 - keep demo alive on CPU/TTS failures.
+            self.last_error = f"{type(exc).__name__}: {exc}"
+            return None
+
+
+class DecompressRetriever:
+    def __init__(self, corpus_path: Path = CORPUS_PATH, model_name: str = DEFAULT_EMBEDDER) -> None:
+        self.corpus_path = corpus_path
+        self.model_name = model_name
+        self.docs = json.loads(corpus_path.read_text(encoding="utf-8"))
+        self.embedding_model = "lexical-fallback"
+        self._model: Any | None = None
+        self._embeddings: np.ndarray | None = None
+        self._load_error: str | None = None
+
+    def retrieve(self, query: str, *, k: int = 2) -> list[RetrievedSource]:
+        query = " ".join(query.strip().split()) or "stress breathing grounding"
+        if self._ensure_sentence_embeddings():
+            assert self._model is not None and self._embeddings is not None
+            query_vec = np.asarray(self._model.encode([query], normalize_embeddings=True), dtype=np.float32)[0]
+            scores = np.dot(self._embeddings, query_vec)
+        else:
+            scores = np.asarray([_lexical_score(query, doc["text"]) for doc in self.docs], dtype=np.float32)
+
+        ranked = sorted(enumerate(scores.tolist()), key=lambda item: item[1], reverse=True)[:k]
+        return [
+            RetrievedSource(
+                id=str(self.docs[index]["id"]),
+                title=str(self.docs[index]["title"]),
+                publisher=str(self.docs[index]["publisher"]),
+                url=str(self.docs[index]["url"]),
+                text=str(self.docs[index]["text"]),
+                score=float(score),
+            )
+            for index, score in ranked
+        ]
+
+    def _ensure_sentence_embeddings(self) -> bool:
+        if self._embeddings is not None:
+            return True
+        if self._load_error is not None:
+            return False
+        try:
+            from sentence_transformers import SentenceTransformer
+
+            os.environ.setdefault("HF_HOME", str(ROOT / ".hf-cache"))
+            self._model = SentenceTransformer(self.model_name)
+            texts = [f"{doc['title']}. {doc['text']}" for doc in self.docs]
+            self._embeddings = np.asarray(self._model.encode(texts, normalize_embeddings=True), dtype=np.float32)
+            self.embedding_model = self.model_name
+            return True
+        except Exception as exc:  # noqa: BLE001 - local tests should not require the embedder.
+            self._load_error = f"{type(exc).__name__}: {exc}"
+            self.embedding_model = f"lexical-fallback ({self._load_error})"
+            return False
+
+
+_RETRIEVER: DecompressRetriever | None = None
+
+
+def get_retriever() -> DecompressRetriever:
+    global _RETRIEVER
+    if _RETRIEVER is None:
+        _RETRIEVER = DecompressRetriever(model_name=os.getenv("DECOMPRESS_EMBEDDER", DEFAULT_EMBEDDER))
+    return _RETRIEVER
+
+
+def set_live_tau(value: float) -> None:
+    global _LIVE_TAU
+    _LIVE_TAU = float(value)
+
+
+def chunk_checkin(text: str, *, words_per_chunk: int = 11) -> list[TranscriptChunk]:
+    words = re.findall(r"\S+", text.strip())
+    if not words:
+        words = re.findall(r"\S+", SAMPLE_CHECKIN)
+    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 gentle_controller_config(tau: float) -> ControllerConfig:
+    return ControllerConfig(
+        tau=float(tau),
+        w_surprise=0.25,
+        w_change=0.20,
+        w_readiness=0.65,
+        w_end=1.35,
+        w_barge=0.0,
+        min_readiness=0.08,
+        take_floor_p_end=0.68,
+        interrupt_p_end_max=0.05,
+        backchannel_p_end_max=0.40,
+        backchannel_tau_fraction=0.75,
+        barge_tau_fraction=10.0,
+        turn_end_tau_discount=0.55,
+        refractory_steps=2,
+    )
+
+
+def make_brain_client(mode: str | None = None) -> tuple[BrainClient, str]:
+    selected = (mode or os.getenv("DECOMPRESS_BRAIN", "modal")).strip().lower()
+    if selected in {"fake", "demo", "local"}:
+        return DemoDecompressBrainClient(), "demo"
+    return EndpointDecompressBrainClient(), "modal"
+
+
+def initial_room_html(tau: float | None = None, brain_mode: str | None = None) -> str:
+    state = DecompressState(
+        tau=float(tau if tau is not None else _LIVE_TAU),
+        brain_mode=brain_mode or os.getenv("DECOMPRESS_BRAIN", "modal"),
+        status="ready",
+    )
+    state.citations = [asdict(source) for source in get_retriever().retrieve(SAMPLE_CHECKIN, k=2)]
+    state.embedding_model = get_retriever().embedding_model
+    return render_room(state)
+
+
+def preview_room_html(tau: float | None = None) -> str:
+    latest = initial_room_html(tau, brain_mode="demo")
+    for latest, _, _ in run_checkin_stream(SAMPLE_CHECKIN, tau or _LIVE_TAU, brain_mode="fake", enable_tts=False, save_log=False):
+        pass
+    return latest
+
+
+def run_checkin_stream(
+    checkin_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("DECOMPRESS_TTS", "1").lower() not in {"0", "false"}
+    state = DecompressState(tau=float(tau), brain_mode=brain_mode or os.getenv("DECOMPRESS_BRAIN", "modal"), status="connecting")
+    speaker = KokoroCalmSpeaker(enabled=bool(selected_tts))
+    yield from _safe_run_checkin_stream(checkin_text, state, speaker, save_log=save_log)
+
+
+def run_voice_checkin_stream(
+    audio_clip: Any,
+    tau: float,
+    *,
+    brain_mode: str | None = None,
+    enable_tts: bool | None = None,
+    save_log: bool = True,
+) -> Iterator[tuple[str, str | None, dict[str, Any], str]]:
+    set_live_tau(tau)
+    selected_tts = enable_tts if enable_tts is not None else os.getenv("DECOMPRESS_TTS", "1").lower() not in {"0", "false"}
+    state = DecompressState(tau=float(tau), brain_mode=brain_mode or os.getenv("DECOMPRESS_BRAIN", "modal"), status="transcribing")
+    speaker = KokoroCalmSpeaker(enabled=bool(selected_tts))
+    state.transcript.append(
+        {
+            "kind": "system",
+            "speaker": "Cohere Transcribe",
+            "step": 0,
+            "text": "Listening to the recorded check-in.",
+            "action": "ASR",
+        }
+    )
+    yield render_room(state), None, _status_payload(state), ""
+
+    try:
+        audio_path = _audio_clip_path(audio_clip)
+        client, resolved_mode = make_brain_client(state.brain_mode)
+        state.brain_mode = resolved_mode
+        if not hasattr(client, "transcribe"):
+            raise RuntimeError("selected brain client does not support transcription")
+        raw_asr = client.transcribe(audio_path, language="en")  # type: ignore[attr-defined]
+        if not raw_asr.get("ok", False):
+            raise RuntimeError(str(raw_asr.get("failure", "transcription failed")))
+        transcript = " ".join(str(raw_asr.get("text", "")).strip().split())
+        if not transcript:
+            raise RuntimeError("Cohere Transcribe returned an empty transcript")
+        state.asr_text = transcript
+        state.asr_latency_ms = float(raw_asr.get("latency_ms", 0.0) or 0.0)
+        state.asr_model_name = str(raw_asr.get("model_name", ""))
+        state.transcript.append(
+            {
+                "kind": "system",
+                "speaker": "Cohere transcript",
+                "step": 0,
+                "text": transcript,
+                "action": "ASR",
+            }
+        )
+        yield render_room(state), None, _status_payload(state), transcript
+
+        for html_frame, audio_path_out, status in _run_with_client(
+            transcript,
+            state,
+            speaker,
+            client,
+            save_log=save_log,
+        ):
+            yield html_frame, audio_path_out, status, transcript
+    except Exception as exc:  # noqa: BLE001 - surface failures in the app shell.
+        state.status = "failed"
+        state.transcript.append(
+            {
+                "kind": "system",
+                "speaker": "room",
+                "step": state.step,
+                "text": f"Voice path failed: {type(exc).__name__}: {exc}",
+                "action": "ERROR",
+            }
+        )
+        yield render_room(state), None, _status_payload(state), state.asr_text or ""
+
+
+def _safe_run_checkin_stream(
+    checkin_text: str,
+    state: DecompressState,
+    speaker: KokoroCalmSpeaker,
+    *,
+    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(checkin_text, state, speaker, client, save_log=save_log)
+    except Exception as exc:  # noqa: BLE001 - keep the visible shell alive.
+        state.status = "failed"
+        state.transcript.append(
+            {
+                "kind": "system",
+                "speaker": "room",
+                "step": state.step,
+                "text": f"Backend failed: {type(exc).__name__}: {exc}",
+                "action": "ERROR",
+            }
+        )
+        yield render_room(state), None, _status_payload(state)
+
+
+def _run_with_client(
+    checkin_text: str,
+    state: DecompressState,
+    speaker: KokoroCalmSpeaker,
+    client: BrainClient,
+    *,
+    save_log: bool,
+) -> Iterator[tuple[str, str | None, dict[str, Any]]]:
+    retriever = get_retriever()
+    controller = WhenToSpeakController([AGENT_ID], config=gentle_controller_config(state.tau))
+    dialogue: Dialogue = []
+    current_user_text = ""
+    events: list[dict[str, Any]] = []
+    chunks = chunk_checkin(checkin_text)
+    state.status = "listening"
+    state.embedding_model = retriever.embedding_model
+    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 = gentle_controller_config(_LIVE_TAU)
+        query_text = _join_text(current_user_text, chunk.text)
+        retrieved = retriever.retrieve(query_text, k=2)
+        state.citations = [asdict(source) for source in retrieved]
+        state.embedding_model = retriever.embedding_model
+        system_prompt = companion_system_prompt(retrieved)
+
+        dialogue_before = _dialogue_with_current_user(dialogue, current_user_text)
+        raw = client.step_many(
+            [{"agent_id": AGENT_ID, "system_prompt": system_prompt}],
+            dialogue_before,
+            chunk.text,
+            chunk.silence_flag,
+        )
+        raw_result = dict(raw.get("results", {}).get(AGENT_ID, {}))
+        signals = signals_from_raw(raw_result)
+        current_user_text = _join_text(current_user_text, chunk.text)
+        tick = controller.tick({AGENT_ID: signals}, floor_holder="human")
+        state.decision = _decision_payload(tick)
+        state.model_name = str(raw.get("model_name", "") or raw_result.get("model_name", "") or "")
+        state.device_name = str(raw.get("device_name", "") or raw_result.get("device_name", "") or "")
+        state.transcript.append({"kind": "user", "speaker": "You", "step": step_index, "text": chunk.text})
+        event = _event_payload(step_index, chunk, tick, raw, raw_result)
+
+        audio_path: str | None = None
+        decision = tick.decisions[AGENT_ID]
+        if decision.action == Action.BACKCHANNEL:
+            line = backchannel_line(step_index, current_user_text)
+            state.stats.backchannels += 1
+            dialogue.append({"role": "assistant", "speaker": AGENT_ID, "text": line})
+            state.transcript.append(
+                {
+                    "kind": "companion backchannel",
+                    "speaker": DISPLAY_NAME,
+                    "step": step_index,
+                    "text": line,
+                    "action": Action.BACKCHANNEL.value,
+                }
+            )
+            audio_path = speaker.synthesize(line, f"step{step_index:02d}_backchannel")
+        elif tick.winner:
+            if current_user_text:
+                dialogue.append({"role": "user", "speaker": "you", "text": current_user_text})
+                current_user_text = ""
+            generated = client.generate(AGENT_ID, system_prompt, dialogue)
+            line = sanitize_spoken_line(str(generated.get("reply_text", "")), dialogue)
+            dialogue.append({"role": "assistant", "speaker": AGENT_ID, "text": line})
+            state.response_latency_ms = float(generated.get("latency_ms", 0.0) or 0.0)
+            state.transcript.append(
+                {
+                    "kind": "companion",
+                    "speaker": DISPLAY_NAME,
+                    "step": step_index,
+                    "text": line,
+                    "action": decision.action.value,
+                }
+            )
+            event["generated"] = dict(generated, reply_text=line)
+            if decision.action == Action.INTERRUPT:
+                state.stats.interrupts += 1
+            else:
+                state.stats.take_floors += 1
+            audio_path = speaker.synthesize(line, f"step{step_index:02d}_reply")
+        elif chunk.silence_flag:
+            state.stats.holds += 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": "you", "text": current_user_text})
+
+    if state.stats.take_floors == 0:
+        retrieved = retriever.retrieve(_dialogue_text(dialogue, ""), k=2)
+        state.citations = [asdict(source) for source in retrieved]
+        system_prompt = companion_system_prompt(retrieved)
+        generated = client.generate(AGENT_ID, system_prompt, dialogue)
+        line = sanitize_spoken_line(str(generated.get("reply_text", "")), dialogue)
+        dialogue.append({"role": "assistant", "speaker": AGENT_ID, "text": line})
+        state.stats.take_floors += 1
+        state.response_latency_ms = float(generated.get("latency_ms", 0.0) or 0.0)
+        state.transcript.append(
+            {
+                "kind": "companion",
+                "speaker": DISPLAY_NAME,
+                "step": state.step,
+                "text": line,
+                "action": Action.TAKE_FLOOR.value,
+            }
+        )
+        audio_path = speaker.synthesize(line, f"step{state.step:02d}_final")
+        yield render_room(state), audio_path, _status_payload(state)
+
+    state.status = "complete"
+    if save_log:
+        _save_decompress_log(state, dialogue, events)
+    yield render_room(state), None, _status_payload(state)
+
+
+def companion_system_prompt(sources: list[RetrievedSource]) -> str:
+    source_lines = [
+        f"[{index}] {source.publisher}, {source.title}: {source.text}"
+        for index, source in enumerate(sources, start=1)
+    ]
+    return "\n".join(
+        [
+            "You are Decompress, a calm five-minute de-stress companion.",
+            "Use the grounding notes below for practical, non-medical suggestions.",
+            "Listen first. Use short backchannel-like wording if the user is still talking.",
+            "When the user pauses, give one gentle next step in one sentence.",
+            "Do not diagnose, do not prescribe, and do not imply therapy or medical care.",
+            "If the user may be in immediate danger, tell them to contact emergency support or a trusted person now.",
+            "Grounding notes:",
+            *source_lines,
+        ]
+    )
+
+
+def render_room(state: DecompressState) -> str:
+    decision = state.decision or _empty_decision()
+    transcript = "\n".join(_render_feed_item(item) for item in state.transcript) or _empty_transcript()
+    citations = "\n".join(_render_citation(item) for item in state.citations) or _empty_citations()
+    score = min(100.0, max(0.0, float(decision.get("urge", 0.0)) / max(state.tau * 1.45, 1.0) * 100.0))
+    tau_pos = min(100.0, max(0.0, state.tau / max(state.tau * 1.45, 1.0) * 100.0))
+    action = str(decision.get("action", "SILENT"))
+    tts_note = f"<span class='soft-note'>TTS: {_escape(state.tts_error)}</span>" if state.tts_error else ""
+    return f"""
+<main id="decompress-room" class="decompress-room" data-status="{_escape(state.status)}">
+  <section class="calm-hero">
+    <div class="breath-dial" aria-hidden="true"><span></span></div>
+    <div class="hero-copy">
+      <div class="kicker">DECOMPRESS</div>
+      <h1>A companion that waits for the right beat.</h1>
+      <p>MiniCPM listens through the when-to-speak controller, then answers with one grounded next step.</p>
+    </div>
+    <div class="session-chip">
+      <span>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>
+  </section>
+  <section class="companion-grid">
+    <article class="companion-card {action.lower()}">
+      <div class="card-top">
+        <div class="moon-mark">D</div>
+        <div>
+          <h2>Decompress</h2>
+          <p>Calm timing, not assistant chatter.</p>
+        </div>
+      </div>
+      <div class="urge-readout">
+        <span>urge</span>
+        <strong>{float(decision.get("urge", 0.0)):.2f}</strong>
+      </div>
+      <div class="soft-meter" style="--fill:{score:.1f}%; --tau:{tau_pos:.1f}%;">
+        <div class="soft-fill"></div><i></i>
+      </div>
+      <div class="signal-row">
+        <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>
+      <div class="stats-line">
+        <span>{state.stats.backchannels} backchannels</span>
+        <span>{state.stats.take_floors} replies</span>
+        <span>{state.stats.holds} holds</span>
+      </div>
+    </article>
+    <article class="transcript-panel">
+      <div class="panel-title"><span>Check-in transcript</span><em>{_escape(state.status)}</em></div>
+      <div class="decompress-scroll">{transcript}</div>
+    </article>
+    <aside class="grounding-panel">
+      <div class="panel-title"><span>Grounded by</span><em>{_escape(state.embedding_model)}</em></div>
+      <div class="citation-list">{citations}</div>
+      <p class="disclaimer">Wellness support only. Not medical advice, diagnosis, or crisis care.</p>
+    </aside>
+  </section>
+</main>
+""".strip()
+
+
+def sanitize_spoken_line(reply: 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)
+        if _looks_spoken(cleaned):
+            return _truncate(cleaned)
+    return fallback_line(_dialogue_text(dialogue or [], ""))
+
+
+def fallback_line(text: str) -> str:
+    lowered = text.lower()
+    if any(fragment in lowered for fragment in ("hurt myself", "kill myself", "suicide", "not safe")):
+        return "If you might be unsafe, contact emergency support or a trusted person now."
+    if "sleep" in lowered or "late" in lowered:
+        return "Park one sentence on paper, then let the next exhale be longer."
+    if "work" in lowered or "meeting" in lowered or "behind" in lowered:
+        return "Make work smaller: unclench your jaw and choose one next tiny step."
+    stable_index = sum(ord(char) for char in lowered) % len(FALLBACK_LINES)
+    return FALLBACK_LINES[stable_index]
+
+
+def backchannel_line(step: int, text: str) -> str:
+    seed = step + len(text)
+    return CALM_BACKCHANNELS[seed % len(CALM_BACKCHANNELS)]
+
+
+def _decision_payload(tick: ControllerTick) -> dict[str, Any]:
+    decision = tick.decisions[AGENT_ID]
+    return {
+        "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,
+        "winner": tick.winner == AGENT_ID,
+    }
+
+
+def _event_payload(
+    step: int,
+    chunk: TranscriptChunk,
+    tick: ControllerTick,
+    raw: dict[str, object],
+    raw_result: dict[str, object],
+) -> dict[str, Any]:
+    return {
+        "step": step,
+        "new_user_text": chunk.text,
+        "silence_flag": chunk.silence_flag,
+        "winner": tick.winner,
+        "model_name": raw.get("model_name") or raw_result.get("model_name"),
+        "device_name": raw.get("device_name") or raw_result.get("device_name"),
+        "batch_latency_ms": raw.get("batch_latency_ms"),
+        "decision": _decision_payload(tick),
+        "brain_latency_ms": raw_result.get("latency_ms"),
+        "surprise": raw_result.get("surprise"),
+    }
+
+
+def _empty_decision() -> dict[str, Any]:
+    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: DecompressState) -> dict[str, Any]:
+    return {
+        "status": state.status,
+        "tau": state.tau,
+        "brain_mode": state.brain_mode,
+        "stats": asdict(state.stats),
+        "citations": state.citations,
+        "tts_error": state.tts_error,
+        "asr_text": state.asr_text,
+        "asr_latency_ms": state.asr_latency_ms,
+        "asr_model_name": state.asr_model_name,
+        "embedding_model": state.embedding_model,
+        "response_latency_ms": state.response_latency_ms,
+        "model_name": state.model_name,
+        "device_name": state.device_name,
+    }
+
+
+def _save_decompress_log(state: DecompressState, 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,
+        "status": state.status,
+        "stats": asdict(state.stats),
+        "citations": state.citations,
+        "asr_text": state.asr_text,
+        "asr_latency_ms": state.asr_latency_ms,
+        "asr_model_name": state.asr_model_name,
+        "embedding_model": state.embedding_model,
+        "response_latency_ms": state.response_latency_ms,
+        "model_name": state.model_name,
+        "device_name": state.device_name,
+        "transcript": state.transcript,
+        "dialogue": dialogue,
+        "events": events,
+    }
+    DECOMPRESS_LOG_PATH.write_text(json.dumps(payload, indent=2), encoding="utf-8")
+
+
+def _audio_clip_path(audio_clip: Any) -> str:
+    if audio_clip is None:
+        raise ValueError("record a check-in clip before starting")
+    if isinstance(audio_clip, (str, os.PathLike)):
+        path = Path(audio_clip)
+    elif isinstance(audio_clip, dict):
+        path = Path(str(audio_clip.get("path") or audio_clip.get("name") or ""))
+    elif isinstance(audio_clip, (tuple, list)) and audio_clip:
+        path = Path(str(audio_clip[0]))
+    else:
+        raise TypeError(f"unsupported audio clip value: {type(audio_clip).__name__}")
+    if not path.exists():
+        raise FileNotFoundError(str(path))
+    return str(path)
+
+
+def _dialogue_with_current_user(dialogue: Dialogue, current_user_text: str) -> Dialogue:
+    snapshot = [dict(turn) for turn in dialogue]
+    snapshot.append({"role": "user", "speaker": "you", "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 _strip_prefix(line: str) -> str:
+    candidate = line.strip().strip("\"'")
+    candidate = re.sub(r"^[\-\*\d\.\)\s]+", "", candidate).strip()
+    for prefix in ["Decompress:", "Assistant:", "Companion:", "Sentence:", "Spoken line:"]:
+        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) < 8:
+        return False
+    lowered = candidate.lower()
+    blocked = [
+        "the user",
+        "recent transcript",
+        "analysis",
+        "markdown",
+        "medical advice",
+        "diagnosis",
+        "as a therapist",
+        "decompress:",
+        "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
+    return len(candidate.split()) <= 28
+
+
+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) > 24:
+        sentence = " ".join(words[:24]).rstrip(",;:") + "."
+    return sentence[:220].rstrip()
+
+
+def _render_feed_item(item: dict[str, Any]) -> str:
+    kind = str(item.get("kind", "system")).replace(" ", "-")
+    action = str(item.get("action", ""))
+    step = item.get("step", "")
+    speaker = _escape(str(item.get("speaker", "")))
+    text = _escape(str(item.get("text", "")))
+    return f"""
+<div class="soft-feed {kind}">
+  <div class="feed-meta"><span>{speaker}</span><em>step {step}</em></div>
+  <div class="feed-text">{text}</div>
+  {f'<b>{_escape(action)}</b>' if action else ''}
+</div>
+""".strip()
+
+
+def _empty_transcript() -> str:
+    return """
+<div class="soft-feed system">
+  <div class="feed-meta"><span>Room</span><em>ready</em></div>
+  <div class="feed-text">Type or record what is on your mind. The companion will wait for the pause.</div>
+</div>
+""".strip()
+
+
+def _render_citation(item: dict[str, Any]) -> str:
+    title = _escape(str(item.get("title", "")))
+    publisher = _escape(str(item.get("publisher", "")))
+    url = _escape(str(item.get("url", "")))
+    score = float(item.get("score", 0.0) or 0.0)
+    return f"""
+<a class="citation-card" href="{url}" target="_blank" rel="noreferrer">
+  <strong>{title}</strong>
+  <span>{publisher}</span>
+  <em>match {score:.2f}</em>
+</a>
+""".strip()
+
+
+def _empty_citations() -> str:
+    return "<div class='citation-empty'>Sources appear when the check-in starts.</div>"
+
+
+def _lexical_score(query: str, text: str) -> float:
+    query_tokens = _tokens(query)
+    text_tokens = _tokens(text)
+    if not query_tokens or not text_tokens:
+        return 0.0
+    overlap = query_tokens & text_tokens
+    return float(len(overlap) / max(len(query_tokens), 1))
+
+
+def _tokens(text: str) -> set[str]:
+    stop = {
+        "the",
+        "and",
+        "for",
+        "that",
+        "with",
+        "you",
+        "your",
+        "this",
+        "are",
+        "was",
+        "have",
+        "has",
+        "but",
+        "not",
+        "one",
+    }
+    return {token for token in re.findall(r"[a-z]{3,}", text.lower()) if token not in stop}
+
+
+def _escape(value: str) -> str:
+    return html.escape(value, quote=True)

apps/decompress/static/decompress.css

@@ -0,0 +1,432 @@
+:root {
+  --ink: #1f2a2c;
+  --paper: #f4f7f5;
+  --mist: #dce8e3;
+  --sage: #6e9b8f;
+  --moss: #355c55;
+  --blue: #8fb7c7;
+  --rose: #d88d84;
+  --plum: #59465d;
+  --line: rgba(31, 42, 44, 0.18);
+}
+
+body,
+.gradio-container {
+  background:
+    linear-gradient(135deg, rgba(31, 42, 44, 0.035) 25%, transparent 25%) 0 0 / 34px 34px,
+    linear-gradient(145deg, #eef4f1 0%, #d7e5e1 46%, #edf2f6 100%) !important;
+  color: var(--ink) !important;
+  font-family: "Aptos", "Segoe UI", sans-serif !important;
+}
+
+.gradio-container {
+  max-width: none !important;
+}
+
+#decompress-control-row {
+  width: min(1180px, calc(100vw - 36px));
+  margin: 18px auto 28px;
+  align-items: stretch;
+}
+
+#decompress-input textarea,
+#decompress-transcript textarea {
+  border: 1px solid rgba(31, 42, 44, 0.18) !important;
+  background: rgba(255, 255, 255, 0.68) !important;
+  color: var(--ink) !important;
+  border-radius: 6px !important;
+  box-shadow: inset 0 1px 0 rgba(255, 255, 255, 0.75) !important;
+}
+
+#decompress-controls {
+  min-width: 280px;
+}
+
+#run-decompress,
+#run-decompress-voice {
+  border-radius: 6px !important;
+  border: 0 !important;
+  min-height: 44px !important;
+}
+
+#run-decompress {
+  background: var(--moss) !important;
+}
+
+#run-decompress-voice {
+  background: rgba(89, 70, 93, 0.12) !important;
+  color: var(--plum) !important;
+}
+
+.decompress-room {
+  position: relative;
+  width: min(1180px, calc(100vw - 36px));
+  margin: 24px auto 0;
+  padding: 0;
+  color: var(--ink);
+}
+
+.calm-hero {
+  min-height: 270px;
+  display: grid;
+  grid-template-columns: minmax(170px, 230px) 1fr auto;
+  gap: 26px;
+  align-items: center;
+  border-bottom: 1px solid var(--line);
+}
+
+.breath-dial {
+  width: min(19vw, 210px);
+  min-width: 150px;
+  aspect-ratio: 1;
+  border-radius: 50%;
+  border: 1px solid rgba(53, 92, 85, 0.26);
+  display: grid;
+  place-items: center;
+  background:
+    linear-gradient(135deg, rgba(255, 255, 255, 0.72), rgba(220, 232, 227, 0.48)),
+    repeating-conic-gradient(from 0deg, rgba(53, 92, 85, 0.18) 0deg 3deg, transparent 3deg 12deg);
+  box-shadow: 0 20px 50px rgba(31, 42, 44, 0.12);
+}
+
+.breath-dial span {
+  width: 54%;
+  aspect-ratio: 1;
+  border-radius: 50%;
+  border: 1px solid rgba(216, 141, 132, 0.55);
+  animation: breathe 5.8s ease-in-out infinite;
+  background: rgba(255, 255, 255, 0.36);
+}
+
+@keyframes breathe {
+  0%,
+  100% {
+    transform: scale(0.82);
+    opacity: 0.62;
+  }
+  50% {
+    transform: scale(1.08);
+    opacity: 1;
+  }
+}
+
+.hero-copy .kicker {
+  color: var(--moss);
+  font-size: 0.78rem;
+  letter-spacing: 0.22em;
+  font-weight: 800;
+}
+
+.hero-copy h1 {
+  margin: 10px 0 12px;
+  font-size: clamp(2.2rem, 5vw, 5.8rem);
+  line-height: 0.95;
+  letter-spacing: 0;
+  max-width: 760px;
+  color: var(--ink);
+  font-family: Georgia, "Times New Roman", serif;
+  font-weight: 500;
+}
+
+.hero-copy p {
+  max-width: 660px;
+  font-size: 1.04rem;
+  line-height: 1.6;
+  color: rgba(31, 42, 44, 0.75);
+}
+
+.session-chip {
+  display: grid;
+  gap: 8px;
+  align-self: start;
+  justify-items: end;
+  padding-top: 20px;
+  font-size: 0.78rem;
+  color: rgba(31, 42, 44, 0.7);
+}
+
+.session-chip span {
+  border: 1px solid var(--line);
+  border-radius: 999px;
+  padding: 6px 10px;
+  background: rgba(255, 255, 255, 0.45);
+}
+
+.soft-note {
+  max-width: 260px;
+  border-color: rgba(216, 141, 132, 0.35) !important;
+}
+
+.companion-grid {
+  display: grid;
+  grid-template-columns: 320px minmax(360px, 1fr) 300px;
+  gap: 18px;
+  margin-top: 18px;
+  align-items: stretch;
+}
+
+.companion-card,
+.transcript-panel,
+.grounding-panel {
+  border: 1px solid var(--line);
+  background: rgba(255, 255, 255, 0.48);
+  box-shadow: 0 18px 45px rgba(31, 42, 44, 0.08);
+  backdrop-filter: blur(16px);
+  border-radius: 8px;
+}
+
+.companion-card {
+  padding: 18px;
+  min-height: 330px;
+  display: flex;
+  flex-direction: column;
+  justify-content: space-between;
+}
+
+.companion-card.take_floor,
+.companion-card.backchannel {
+  border-color: rgba(53, 92, 85, 0.5);
+  box-shadow: 0 18px 55px rgba(53, 92, 85, 0.2);
+}
+
+.card-top {
+  display: flex;
+  gap: 13px;
+  align-items: center;
+}
+
+.moon-mark {
+  width: 48px;
+  aspect-ratio: 1;
+  border-radius: 50%;
+  display: grid;
+  place-items: center;
+  background: var(--ink);
+  color: var(--paper);
+  font-family: Georgia, "Times New Roman", serif;
+  font-size: 1.35rem;
+}
+
+.card-top h2 {
+  margin: 0;
+  font-family: Georgia, "Times New Roman", serif;
+  font-weight: 500;
+  font-size: 1.7rem;
+}
+
+.card-top p {
+  margin: 2px 0 0;
+  color: rgba(31, 42, 44, 0.66);
+}
+
+.urge-readout {
+  display: flex;
+  justify-content: space-between;
+  align-items: end;
+  margin-top: 32px;
+  color: rgba(31, 42, 44, 0.68);
+}
+
+.urge-readout strong {
+  font-size: 2.35rem;
+  font-family: Georgia, "Times New Roman", serif;
+  color: var(--moss);
+  font-weight: 500;
+}
+
+.soft-meter {
+  height: 18px;
+  border: 1px solid rgba(31, 42, 44, 0.18);
+  border-radius: 999px;
+  position: relative;
+  overflow: hidden;
+  background: rgba(255, 255, 255, 0.55);
+}
+
+.soft-fill {
+  width: var(--fill);
+  height: 100%;
+  background: linear-gradient(90deg, var(--blue), var(--sage), var(--rose));
+  transition: width 360ms ease;
+}
+
+.soft-meter i {
+  position: absolute;
+  left: var(--tau);
+  top: -5px;
+  bottom: -5px;
+  width: 2px;
+  background: var(--plum);
+}
+
+.signal-row,
+.stats-line {
+  display: flex;
+  flex-wrap: wrap;
+  gap: 7px;
+}
+
+.signal-row span,
+.stats-line span {
+  padding: 6px 8px;
+  border-radius: 999px;
+  background: rgba(31, 42, 44, 0.06);
+  color: rgba(31, 42, 44, 0.68);
+  font-size: 0.78rem;
+}
+
+.transcript-panel,
+.grounding-panel {
+  min-height: 330px;
+  display: flex;
+  flex-direction: column;
+}
+
+.panel-title {
+  height: 44px;
+  display: flex;
+  align-items: center;
+  justify-content: space-between;
+  padding: 0 14px;
+  border-bottom: 1px solid var(--line);
+  color: rgba(31, 42, 44, 0.68);
+  text-transform: uppercase;
+  font-size: 0.72rem;
+  letter-spacing: 0.1em;
+}
+
+.panel-title em {
+  font-style: normal;
+  text-transform: none;
+  letter-spacing: 0;
+  max-width: 180px;
+  overflow: hidden;
+  text-overflow: ellipsis;
+  white-space: nowrap;
+}
+
+.decompress-scroll {
+  flex: 1;
+  max-height: 390px;
+  overflow-y: auto;
+  padding: 14px;
+  display: flex;
+  flex-direction: column;
+  gap: 10px;
+}
+
+.soft-feed {
+  border-left: 3px solid rgba(31, 42, 44, 0.16);
+  padding: 10px 12px;
+  background: rgba(255, 255, 255, 0.42);
+  border-radius: 0 6px 6px 0;
+}
+
+.soft-feed.companion,
+.soft-feed.companion-backchannel {
+  border-left-color: var(--moss);
+  background: rgba(110, 155, 143, 0.13);
+}
+
+.soft-feed.system {
+  border-left-color: var(--blue);
+}
+
+.feed-meta {
+  display: flex;
+  justify-content: space-between;
+  color: rgba(31, 42, 44, 0.58);
+  font-size: 0.74rem;
+  margin-bottom: 5px;
+}
+
+.feed-meta em {
+  font-style: normal;
+}
+
+.feed-text {
+  line-height: 1.48;
+}
+
+.soft-feed b {
+  display: inline-block;
+  margin-top: 7px;
+  font-size: 0.7rem;
+  color: var(--moss);
+  letter-spacing: 0.08em;
+}
+
+.citation-list {
+  padding: 14px;
+  display: grid;
+  gap: 10px;
+}
+
+.citation-card,
+.citation-empty {
+  display: block;
+  text-decoration: none;
+  color: var(--ink);
+  padding: 12px;
+  border-radius: 6px;
+  border: 1px solid rgba(31, 42, 44, 0.13);
+  background: rgba(255, 255, 255, 0.38);
+}
+
+.citation-card strong {
+  display: block;
+  font-family: Georgia, "Times New Roman", serif;
+  font-size: 1.02rem;
+  font-weight: 500;
+}
+
+.citation-card span,
+.citation-card em,
+.citation-empty {
+  display: block;
+  margin-top: 5px;
+  font-size: 0.78rem;
+  color: rgba(31, 42, 44, 0.62);
+  font-style: normal;
+}
+
+.disclaimer {
+  margin: auto 14px 14px;
+  padding-top: 14px;
+  border-top: 1px solid var(--line);
+  color: rgba(31, 42, 44, 0.64);
+  font-size: 0.82rem;
+  line-height: 1.45;
+}
+
+@media (max-width: 980px) {
+  .calm-hero,
+  .companion-grid {
+    grid-template-columns: 1fr;
+  }
+
+  .session-chip {
+    justify-items: start;
+    grid-template-columns: repeat(3, max-content);
+    padding-top: 0;
+  }
+
+  .breath-dial {
+    width: 150px;
+  }
+}
+
+@media (max-width: 640px) {
+  .decompress-room,
+  #decompress-control-row {
+    width: min(100vw - 20px, 1180px);
+  }
+
+  .hero-copy h1 {
+    font-size: 2.4rem;
+  }
+
+  .session-chip {
+    grid-template-columns: 1fr;
+  }
+}

apps/decompress/static/decompress.js

@@ -0,0 +1,20 @@
+(() => {
+  const scrollTranscript = () => {
+    const pane = document.querySelector("#decompress-room .decompress-scroll");
+    if (pane) pane.scrollTop = pane.scrollHeight;
+  };
+
+  const observer = new MutationObserver(scrollTranscript);
+  const start = () => {
+    const root = document.querySelector("#decompress-room-output");
+    if (!root) return;
+    observer.observe(root, { childList: true, subtree: true });
+    scrollTranscript();
+  };
+
+  if (document.readyState === "loading") {
+    document.addEventListener("DOMContentLoaded", start);
+  } else {
+    start();
+  }
+})();

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,8 @@
+--extra-index-url https://download.pytorch.org/whl/cpu
+gradio==6.18.0
+torch==2.12.0+cpu
+kokoro==0.9.4
+numpy==2.4.6
+requests==2.32.5
+soundfile==0.14.0
+sentence-transformers>=5.0.0,<6.0.0