fix: heid-bug-hunt fixups — donut voiced-interview slices 2+3

Triaged the heid-bug-hunt panel (Gróa+Hulda+Regin+Kimi, 11 distinct findings).
Fixed the real ones; the 3-arm "memory_context unverifiable" alarm was refuted
(tests + live SDK verify), and caller-supplied agent_id is accepted under the
LAN/no-auth debug-tool trust model (documented, not fixed).

Constructible crashes (were uncaught HTTP 500s from wire input):
- _tts_endpoint: coerce non-str / unhashable agent_id -> None before the voice-map
  lookup (matches the submit path's guard); an unhashable {} / [] TypeError'd -> 500.
- PadState.from_obj: catch ArithmeticError — float() of a huge-int JSON literal
  raises OverflowError, absent from the except tuple -> 500; now a neutral read.
- _submit_turn_endpoint: require a non-blank STR content — a truthy non-str crashed
  pin_kb_context's question.strip() mid-stream instead of a deterministic 400.
  pin_kb_context also isinstance-guards the question defensively.

Robustness:
- kb_bridge: delete the throwaway Mimir consult session (SDK sessions.delete) on
  success/error/timeout via a caller-owned holder so cleanup survives a mid-stream
  timeout — consults no longer accumulate server-side under the fixed partition.
- _stream_turn_endpoint: emit a ": keepalive" SSE comment BEFORE the (<=20s) KB
  consult so a reverse proxy / EventSource doesn't drop the silent connection into
  a false "WIRE LOST" before the turn starts.
- _tts_endpoint: cap text at 8000 chars (413) before the process-global lock;
  gateway timeout 120s->60s — one huge/stalled body can't starve all TTS.
- tts_synthesize: validate the WAVE form tag (bytes 8:12), not just the RIFF magic.
- index.html: revoke the audio blob URL in cancelTts (removeAttribute+load fires
  neither ended nor error, so the src's own revoke never ran -> per-turn blob leak).

TDD: +11 tests (543 green). Live-smoked on :8765: all five constructible adversarial
inputs now return 200/413/400, never 500.

src/ratatoskr/kb_bridge.py

@@ -60,14 +60,19 @@ def _consult_prompt(question: str, corpus: str) -> str:
     )
 
 
-async def _run_consult(client: WorldtreeClient, prompt: str) -> str:
+async def _run_consult(client: WorldtreeClient, prompt: str, created: list[str]) -> str:
     """Drive one throwaway Mimir turn and return its grounded answer text. Prefers
     the DoneEvent's authoritative full `response`, falling back to the accumulated
-    text deltas if the terminal event carried none."""
+    text deltas if the terminal event carried none.
+
+    `created` is a caller-owned list this appends the new session id to as soon as it
+    exists, so the caller can delete the session even when a timeout cancels this
+    coroutine mid-stream (the id would otherwise be lost in this frame)."""
     session = await client.sessions.create(
         {"agent_id": _MIMIR_AGENT_ID, "end_user_id": _KB_BRIDGE_END_USER}
     )
     session_id = session["session_id"]
+    created.append(session_id)
     parts: list[str] = []
     final = ""
     # aclosing: we break on DoneEvent without exhausting the stream; deterministically
@@ -97,17 +102,30 @@ async def pin_kb_context(
     corpus = _KB_INTERVIEW_CORPORA.get(agent_id or "")
     if corpus is None:  # INV-KB-4: not an opted-in interview character
         return []
-    if not (question and question.strip()):
+    if not isinstance(question, str) or not question.strip():
         return []
+    created: list[str] = []
     try:
         extract = await asyncio.wait_for(
-            _run_consult(client, _consult_prompt(question, corpus)),
+            _run_consult(client, _consult_prompt(question, corpus), created),
             timeout=_CONSULT_TIMEOUT_S,
         )
     except Exception:
         # timeout / transport / protocol / any consult failure → degrade silently
         # (INV-KB-3). CancelledError is a BaseException, so it is NOT swallowed here.
-        return []
+        extract = ""
+    finally:
+        # Delete the throwaway consult session so they don't accumulate server-side under
+        # the fixed partition (heid bug-hunt). Runs on success, error, AND timeout — the
+        # timeout is caught above, so this executes in a non-cancelled context and the
+        # delete completes. Best-effort; a delete failure never masks the result. On a
+        # propagating CancelledError (browser disconnect) the session may outlive us —
+        # acceptable, Worldtree GCs it; the invariant is that CancelledError still propagates.
+        for sid in created:
+            try:
+                await client.sessions.delete(sid)
+            except Exception:
+                pass
     if not extract:
         return []
     return [{"kind": "corpus_reference", "text": extract[:_MAX_EXTRACT_CHARS]}]

src/ratatoskr/tts.py

@@ -65,7 +65,10 @@ class PadState:
                 arousal=float(obj["arousal"]),
                 dominance=float(obj.get("dominance", 0.0)),
             )
-        except (KeyError, TypeError, ValueError):
+        except (KeyError, TypeError, ValueError, ArithmeticError):
+            # ArithmeticError covers OverflowError — float() of a huge JSON integer
+            # literal (e.g. a 400-digit number) raises it; the wire is untrusted, so a
+            # malformed pad degrades to a neutral read rather than 500ing /api/tts.
             return None
 
 
@@ -105,9 +108,11 @@ def _clamp(x: float, lo: float, hi: float) -> float:
 def pad_to_dials(pad: PadState | None) -> EmotionDials:
     """Map live PAD → Zonos emotion dials (DEC-7 / FN pad_to_dials).
 
-    Total: any input (None / finite / out-of-range / NaN / inf) → valid dials,
-    never raises. None/absent PAD → a neutral read (emotion_enabled=False)."""
-    if pad is None:
+    Total: any input (None / finite / out-of-range / NaN / inf / a non-PadState
+    object) → valid dials, never raises. None/absent/malformed PAD → a neutral read
+    (emotion_enabled=False)."""
+    if not isinstance(pad, PadState):
+        # None, or any non-PadState the declared surface doesn't cover — neutral read.
         return EmotionDials(emotion_enabled=False)
     return EmotionDials(
         emotion_enabled=True,
@@ -142,9 +147,10 @@ async def tts_synthesize(
             f"gateway status {resp.status_code}", status=resp.status_code
         )
     data = resp.content
-    # DEC-3 guard: the gateway MUST return RIFF/WAVE. A non-wav 200 (an error
-    # page, or the mislabeled-PCM mp3/opus trap) is treated as unavailable —
-    # played garbage is worse than silence.
-    if data[:4] != b"RIFF":
+    # DEC-3 guard: the gateway MUST return a RIFF/WAVE container. A non-wav 200 (an
+    # error page, or the mislabeled-PCM mp3/opus trap) is treated as unavailable —
+    # played garbage is worse than silence. Check BOTH the RIFF magic (0:4) and the
+    # WAVE form tag (8:12); a `b"RIFF..."` body that isn't WAVE would still fail decode.
+    if data[:4] != b"RIFF" or data[8:12] != b"WAVE":
         raise TtsUnavailable("gateway returned a non-wav body")
     return data

src/ratatoskr/web/server.py

@@ -293,7 +293,10 @@ async def _submit_turn_endpoint(request: Request) -> JSONResponse:
     """
     body = await request.json()
     content = body.get("content") if isinstance(body, dict) else None
-    if not content:
+    if not content or not isinstance(content, str):
+        # Require a non-blank STRING: a truthy non-str (e.g. {"content": {...}}) would
+        # pass a bare falsy check, then crash pin_kb_context's question.strip() mid-stream
+        # instead of a deterministic 400 (heid bug-hunt: non-str content).
         return JSONResponse({"error_code": "missing_content"}, status_code=400)
     session_id = request.path_params["session_id"]
     # The browser may include agent_id so the stream path can gate the KB-recall
@@ -374,6 +377,12 @@ async def _stream_turn_endpoint(request: Request) -> StreamingResponse:
         wt_client = _wt_client(client)
         try:
             handle.status = "streaming"
+            # Open the SSE stream immediately with a comment frame BEFORE the KB consult
+            # (which can take up to _CONSULT_TIMEOUT_S). Without a first byte, a reverse
+            # proxy or the browser EventSource can drop a silent connection → a false
+            # "WIRE LOST" before the real turn even starts (heid bug-hunt). SSE comment
+            # frames (":"-prefixed) are ignored by EventSource, so this is inert to the UI.
+            yield b": keepalive\n\n"
             # Retire-ready KB-recall bridge (INV-KB-1: THE single call-site). Consumer-side
             # Mimir retrieval for opted-in interview characters (INV-KB-4); [] for everyone
             # else. Bounded + degrade-to-[] internally (INV-KB-3), so it never blocks or
@@ -560,6 +569,10 @@ async def _memory_chunks_endpoint(request: Request) -> JSONResponse:
 # Zonos voice; everything else falls to the gateway default. Case-folded gateway-side.
 _TTS_VOICE_MAP = {"ratatoskr:donut": "donut"}
 _TTS_DEFAULT_VOICE = "Cora"
+# Cap the synth input before taking the process-global lock. An interview clip is a
+# few sentences; a huge/hostile body would otherwise hold the shared 3090 for the full
+# gateway timeout, starving every other turn's audio (heid bug-hunt: text-size DoS).
+_TTS_MAX_TEXT_CHARS = 8000
 
 
 async def _tts_endpoint(request: Request) -> Response:
@@ -578,7 +591,14 @@ async def _tts_endpoint(request: Request) -> Response:
     text = body.get("text") if isinstance(body, dict) else None
     if not text or not isinstance(text, str):
         return JSONResponse({"error_code": "missing_text"}, status_code=400)
+    if len(text) > _TTS_MAX_TEXT_CHARS:  # bound before the lock (bug-hunt: text-size DoS)
+        return JSONResponse({"error_code": "text_too_large"}, status_code=413)
     agent_id = body.get("agent_id") if isinstance(body, dict) else None
+    if not isinstance(agent_id, str):
+        # Coerce a non-str (incl. an unhashable list/dict) to None BEFORE the map lookup:
+        # `_TTS_VOICE_MAP.get([])` would TypeError → uncaught 500, diverging from the
+        # submit path's identical guard (heid bug-hunt: open-world agent_id).
+        agent_id = None
     voice = _TTS_VOICE_MAP.get(agent_id, _TTS_DEFAULT_VOICE)
     pad_obj = body.get("pad") if isinstance(body, dict) else None
     dials = pad_to_dials(PadState.from_obj(pad_obj))
@@ -587,9 +607,11 @@ async def _tts_endpoint(request: Request) -> Response:
     lock = request.app.state.tts_lock
     try:
         # DEC-5: serialize — a new turn's synth waits on any in-flight one (the
-        # client also aborts the prior request, cancelling the server task).
+        # client also aborts the prior request, cancelling the server task). 60s ceiling:
+        # generous for a multi-sentence clip, bounded so a stalled gateway can't pin the
+        # shared lock for two minutes (bug-hunt: lock-hold starvation).
         async with lock:
-            async with httpx.AsyncClient(timeout=httpx.Timeout(120.0)) as client:
+            async with httpx.AsyncClient(timeout=httpx.Timeout(60.0)) as client:
                 wav = await tts_synthesize(
                     text, voice=voice, dials=dials, client=client, url=tts_url
                 )

src/ratatoskr/web/static/index.html

@@ -1998,13 +1998,21 @@ async function cancelTurn() {
 // time (INV-TTS-3: a new turn or a superseding synth cancels the prior fetch + audio),
 // and non-blocking (INV-TTS-4: any failure logs to the ticker + skips — never the turn).
 let _ttsAbort = null;
+let _ttsUrl = null;   // active blob object URL, tracked so cancel can revoke it
 function ttsEnabled() {
   try { return localStorage.getItem("ratatoskr-tts") === "1"; } catch (_) { return false; }
 }
+function _revokeTtsUrl() {
+  if (_ttsUrl) { try { URL.revokeObjectURL(_ttsUrl); } catch (_) {} _ttsUrl = null; }
+}
 function cancelTts() {
   if (_ttsAbort) { try { _ttsAbort.abort(); } catch (_) {} _ttsAbort = null; }
   const a = $("tts-audio");
   if (a) { try { a.pause(); a.removeAttribute("src"); a.load(); } catch (_) {} }
+  // removeAttribute("src")+load() fires NEITHER ended nor error, so the src's own
+  // revoke handler never runs — revoke the tracked URL here or the blob leaks per
+  // interrupted turn (heid bug-hunt).
+  _revokeTtsUrl();
 }
 async function speakOnDone(text, agentId, pad) {
   const clip = (text || "").trim();
@@ -2028,12 +2036,13 @@ async function speakOnDone(text, agentId, pad) {
   const a = $("tts-audio");
   if (!a) { URL.revokeObjectURL(url); return; }
   a.src = url;
-  a.onended = a.onerror = () => { URL.revokeObjectURL(url); };
+  _ttsUrl = url;   // track for revoke on natural end OR cancel
+  a.onended = a.onerror = () => { _revokeTtsUrl(); };
   // Playback can be refused by the browser autoplay policy until the page has an
   // activation; the operator's toggle+prompt gesture generally satisfies it, and the
   // catch keeps a refusal non-fatal (INV-TTS-4).
   try { await a.play(); tickerAdd("ok", "tts", "▶ voiced"); }
-  catch (_) { URL.revokeObjectURL(url); tickerAdd("err", "tts", "playback blocked"); }
+  catch (_) { _revokeTtsUrl(); tickerAdd("err", "tts", "playback blocked"); }
 }
 // ---- 🔊 toggle (mirrors theme / cot-toggle; default OFF, persisted) ----
 (function () {

tests/test_kb_bridge.py

@@ -41,6 +41,7 @@ class _FakeSessions:
         self.create_error = create_error
         self.stream_delay = stream_delay
         self.calls: list = []
+        self.deleted: list = []
 
     async def create(self, body=None, **kw):
         self.calls.append(("create", body, kw))
@@ -48,6 +49,9 @@ class _FakeSessions:
             raise self.create_error
         return self.create_result
 
+    async def delete(self, session_id):
+        self.deleted.append(session_id)
+
     def stream_turn(self, *args, **kw):
         self.calls.append(("stream_turn", args, kw))
         return self._gen()
@@ -83,6 +87,13 @@ class TestAllowlistGate:
         out = await pin_kb_context("   ", _DONUT, client=_FakeClient(fake))
         assert out == [] and fake.calls == []
 
+    async def test_non_string_question_returns_empty(self) -> None:
+        # A non-str question (e.g. a dict slipping past an upstream guard) must not
+        # crash on .strip() — it degrades to [] like any other miss.
+        fake = _FakeSessions()
+        out = await pin_kb_context({"x": 1}, _DONUT, client=_FakeClient(fake))
+        assert out == [] and fake.calls == []
+
 
 class TestConsult:
     async def test_happy_pins_done_response_as_corpus_reference(self) -> None:
@@ -141,3 +152,32 @@ class TestDegradation:
         fake = _FakeSessions(create_error=asyncio.CancelledError())
         with pytest.raises(asyncio.CancelledError):
             await pin_kb_context("q", _DONUT, client=_FakeClient(fake))
+
+
+class TestSessionCleanup:
+    """The throwaway Mimir consult session is deleted so consults don't accumulate
+    server-side (heid bug-hunt) — on success, error, AND timeout."""
+
+    async def test_deletes_session_on_success(self) -> None:
+        fake = _FakeSessions(events=[_done("grounded")])
+        await pin_kb_context("q", _DONUT, client=_FakeClient(fake))
+        assert fake.deleted == ["kb-sess-1"]
+
+    async def test_deletes_session_on_stream_error(self) -> None:
+        fake = _FakeSessions(events=[_text("x")], stream_error=RuntimeError("boom"))
+        await pin_kb_context("q", _DONUT, client=_FakeClient(fake))
+        assert fake.deleted == ["kb-sess-1"]  # cleaned up despite the failure
+
+    async def test_deletes_session_on_timeout(self, monkeypatch) -> None:
+        monkeypatch.setattr(kb_bridge, "_CONSULT_TIMEOUT_S", 0.05)
+        fake = _FakeSessions(events=[_done("late")], stream_delay=5.0)
+        await pin_kb_context("q", _DONUT, client=_FakeClient(fake))
+        # The holder pattern makes the id available even though the timeout cancelled
+        # the consult mid-stream; the finally deletes it in the caught (non-cancelled) path.
+        assert fake.deleted == ["kb-sess-1"]
+
+    async def test_no_delete_when_create_failed(self) -> None:
+        # Nothing was created → nothing to delete (created list stays empty).
+        fake = _FakeSessions(create_error=RuntimeError("no session"))
+        await pin_kb_context("q", _DONUT, client=_FakeClient(fake))
+        assert fake.deleted == []

tests/test_tts.py

@@ -22,8 +22,9 @@ from ratatoskr.tts import (
     tts_synthesize,
 )
 
-# A minimally-valid wav body: the DEC-3 guard only checks the RIFF magic.
-_WAV = b"RIFF" + b"\x00" * 40
+# A minimally-valid wav body: the DEC-3 guard checks the RIFF magic (0:4) + WAVE
+# form tag (8:12).
+_WAV = b"RIFF\x00\x00\x00\x00WAVE" + b"\x00" * 32
 
 
 class TestPadState:
@@ -48,6 +49,12 @@ class TestPadState:
     def test_from_obj_non_numeric_is_none(self) -> None:
         assert PadState.from_obj({"pleasure": "hot", "arousal": 0.1}) is None
 
+    def test_from_obj_huge_int_overflow_is_none(self) -> None:
+        # float() of a 400-digit JSON integer raises OverflowError (an ArithmeticError,
+        # not a ValueError); from_obj must still degrade to None, not propagate a 500.
+        huge = int("9" * 400)
+        assert PadState.from_obj({"pleasure": huge, "arousal": 0}) is None
+
 
 class TestPadToDials:
     def test_none_pad_is_neutral_disabled(self) -> None:
@@ -73,6 +80,14 @@ class TestPadToDials:
         assert d.emotion_arousal == 1.0  # +inf clamps to the ceiling
         assert not math.isnan(d.emotion_valence)
 
+    def test_non_padstate_input_degrades_to_neutral(self) -> None:
+        # The declared surface is PadState | None, but the function must be total over
+        # ANY object (a raw dict / str / arbitrary object) → neutral, never raise.
+        for bad in ({}, "bad", [0.1, 0.2], object(), 42):
+            d = pad_to_dials(bad)
+            assert d.emotion_enabled is False
+            assert d.to_body() == {}
+
 
 class TestEmotionDialsToBody:
     def test_disabled_emits_no_params(self) -> None:
@@ -166,3 +181,17 @@ class TestTtsSynthesize:
                     "hi", voice="Cora", dials=pad_to_dials(None), client=client,
                     url="http://tts.example/v1/audio/speech",
                 )
+
+    @respx.mock
+    async def test_riff_but_not_wave_raises_tts_unavailable(self) -> None:
+        # A body with the RIFF magic but a non-WAVE form tag (e.g. RIFF/AVI, or a
+        # truncated/garbage container) is not playable audio — reject it too.
+        respx.post("http://tts.example/v1/audio/speech").mock(
+            return_value=httpx.Response(200, content=b"RIFF\x00\x00\x00\x00AVI LIST")
+        )
+        async with httpx.AsyncClient() as client:
+            with pytest.raises(TtsUnavailable):
+                await tts_synthesize(
+                    "hi", voice="Cora", dials=pad_to_dials(None), client=client,
+                    url="http://tts.example/v1/audio/speech",
+                )

tests/test_web_server.py

@@ -382,6 +382,14 @@ class TestSubmitTurnEndpoint:
         resp = TestClient(app).post("/api/turns/s-1", json={})
         assert resp.status_code == 400
 
+    def test_non_string_content_returns_400(self) -> None:
+        """Non-str truthy content (e.g. a dict) → deterministic 400, not a crash in the
+        KB bridge's question.strip() mid-stream (heid bug-hunt)."""
+        from ratatoskr.web.server import create_app
+        app = create_app(_mock_client_factory())
+        resp = TestClient(app).post("/api/turns/s-1", json={"content": {"x": 1}})
+        assert resp.status_code == 400
+
     def test_monotonic_turn_ids(self) -> None:
         """monotonic_turn_ids [trace]: two submits → second turn_id > first."""
         from ratatoskr.web.server import create_app
@@ -1454,7 +1462,7 @@ class TestTtsEndpoint:
     (DEC-8); emotion dials map the browser-sent live PAD (DEC-7); a gateway failure
     degrades to 503 (INV-TTS-4)."""
 
-    _WAV = b"RIFF" + b"\x00" * 40
+    _WAV = b"RIFF\x00\x00\x00\x00WAVE" + b"\x00" * 32
     _TTS = "http://tts.example/v1/audio/speech"
 
     @respx.mock
@@ -1527,6 +1535,45 @@ class TestTtsEndpoint:
         resp = TestClient(app).post("/api/tts", json={"text": "hi"})
         assert resp.status_code == 503
 
+    @respx.mock
+    def test_unhashable_agent_id_does_not_500(self) -> None:
+        # An unhashable agent_id ([] / {}) must not TypeError the voice-map lookup into
+        # a 500 — it coerces to None (default voice), matching the submit path's guard.
+        from ratatoskr.web.server import create_app
+
+        route = respx.post(self._TTS).mock(
+            return_value=httpx.Response(200, content=self._WAV)
+        )
+        app = create_app(_mock_client_factory(), tts_url=self._TTS)
+        resp = TestClient(app).post("/api/tts", json={"text": "hi", "agent_id": []})
+        assert resp.status_code == 200
+        assert json.loads(route.calls.last.request.content)["voice"] == "Cora"
+
+    def test_text_too_large_returns_413(self) -> None:
+        from ratatoskr.web.server import create_app
+        from ratatoskr.web.server import _TTS_MAX_TEXT_CHARS
+
+        app = create_app(_mock_client_factory(), tts_url=self._TTS)
+        resp = TestClient(app).post(
+            "/api/tts", json={"text": "x" * (_TTS_MAX_TEXT_CHARS + 1)}
+        )
+        assert resp.status_code == 413
+        assert resp.json()["error_code"] == "text_too_large"
+
+    @respx.mock
+    def test_huge_int_pad_does_not_500(self) -> None:
+        # A malformed pad (huge-int → float() OverflowError in from_obj) must degrade to
+        # a neutral read, not 500 the endpoint.
+        from ratatoskr.web.server import create_app
+
+        respx.post(self._TTS).mock(return_value=httpx.Response(200, content=self._WAV))
+        app = create_app(_mock_client_factory(), tts_url=self._TTS)
+        resp = TestClient(app).post(
+            "/api/tts",
+            json={"text": "hi", "pad": {"pleasure": int("9" * 400), "arousal": 0}},
+        )
+        assert resp.status_code == 200
+
 
 class TestKbBridgeWiring:
     """The KB-recall bridge seam (slice 3, INV-KB-1): the turn path calls