feat(tts): config-driven voices + two-voice dialogue/narration split (DEC-11)

Voice assignment moves from the hardcoded server map to ~/.config/ratatoskr/
voices.json (per-agent voice + optional narration_voice). An agent with a
narration_voice gets a two-voice split: quoted speech in `voice`, narration in
`narration_voice`, synthesized per-span and stitched under one WAV header.

- new src/ratatoskr/voices.py: load_voice_config (degrade-not-crash),
  segment_dialogue (quote-based, straight + curly), resolve_voice_spans
- tts.py: tts_stream_stitched replaces tts_stream — serial per-span synth,
  span 0 verbatim, spans 1..N header-stripped -> one gapless 48kHz stream;
  a single-span list is a byte-identical passthrough (no single-voice regression)
- server.py: _tts_endpoint resolves spans from app.state.voice_config; the
  hardcoded _TTS_VOICE_MAP is retired; create_app gains a voice_config param
- entrypoint.py: loads voices.json at startup
- contract DEC-11 + INV-TTS-5/6/7; initial config donut->donut,
  sindra->miranda (dialogue) / emmie (narration)

Live-verified on :8765: Sindra mixed turn -> 2 dots calls (emmie+miranda)
stitched into one 48kHz WAV with a single RIFF header; Donut single-voice
unchanged. 545 tests green (incl. new test_voices.py).

docs/contracts/donut_voiced_interview.contract.md

@@ -236,9 +236,38 @@ each independently shippable. Slice order is chosen for fastest visible result.
   becomes POST (DEC-10a) so the full text rides the body, not a length-capped URL; the
   outer text cap rises 2000→8000 (a shared-3090 hold bound, not a URL bound).
 
+- **DEC-11 — config-driven voices + two-voice dialogue/narration split (2026-08-11, operator-directed).**
+  Voice assignment moves from the hardcoded `_TTS_VOICE_MAP` (DEC-8) to a rata-side config file
+  `~/.config/ratatoskr/voices.json` (beside local_agents.json + provider.env; NOT folded into the
+  agent-index schema — isolated from its v1→v2 silent-drop foot-gun). Supersedes DEC-8's hardcoded map.
+  - **Schema:** `{"default": "<voice>", "agents": {"<agent_id>": {"voice": "<voice>", "narration_voice": "<voice>"?}}}`.
+    `voice` = the agent's dialogue/primary voice; an unmapped agent (or one with no `voice`) falls to
+    `default`. Voice names are GATEWAY-validated (GET /v1/voices), not client-asserted.
+  - **`narration_voice` is OPTIONAL and its PRESENCE is the two-voice switch** (no separate flag).
+    Absent → the whole turn is one span in `voice` (byte-identical to the prior single-call passthrough —
+    this is why dialogue-only Donut needs NO special-casing). Present → the turn is SEGMENTED into
+    dialogue vs narration spans; dialogue → `voice`, narration → `narration_voice`.
+  - **Segmentation (FN segment_dialogue):** QUOTED text (straight `"` OR curly `“ ”`) = dialogue;
+    text OUTSIDE quotes = narration. Order preserved; empty/whitespace spans dropped. An unbalanced
+    trailing open-quote → its run-to-end is dialogue (best-effort, never raises). dots' server-side
+    curly→ASCII fold is pronunciation-only and does NOT affect boundary detection (we match both styles).
+  - **Synthesis (FN tts_stream_stitched):** spans synth SERIALLY (dots single-consumer) into ONE
+    continuous stream — span 0 streamed as wav VERBATIM (header + PCM), spans 1..N streamed as wav with
+    the leading WAV header STRIPPED (accumulate-until-`data`, emit after `data`+8) so the browser decodes
+    one gapless 48kHz mono s16le stream after a single leading header. The single-span case is EXACTLY the
+    prior tts_stream passthrough (INV-TTS-6, no regression). The `yielded_any` degrade pivot spans the whole
+    sequence: a pre-first-byte failure on ANY span before span 0 has committed → TtsUnavailable → 503; a
+    failure after ≥1 byte committed → degrade (drop the tail, keep what played), never raise into the 200.
+  - **Config load (FN load_voice_config):** entrypoint reads voices.json at startup → `create_app(voice_config)`
+    → `app.state.voice_config`; `_tts_endpoint` resolves spans per turn (FN resolve_voice_spans). An
+    absent/malformed file → the built-in DEFAULT_VOICE_CONFIG (donut→donut; sindra→miranda dialogue + emmie
+    narration; glados default), logged — degrade-not-crash (INV-TTS-5).
+  - **Initial voices.json:** `donut → {voice: donut}` (dialogue-only, single voice); `sindra →
+    {voice: miranda, narration_voice: emmie}`.
+
 ## Invariants
 
-- **INV-TTS-1 [hard]** — the TTS gateway host/URL (chatterbox-fast :8197) never
+- **INV-TTS-1 [hard]** — the TTS gateway host/URL (dots-tts :8198) never
   reaches the browser; all synthesis goes through `/api/tts`.
 - **INV-TTS-2 [hard]** — TTS is opt-in: a 🔊 toggle (default OFF), persisted to
   localStorage (mirrors the theme/cot-toggle pattern). No speech without it.
@@ -249,6 +278,16 @@ each independently shippable. Slice order is chosen for fastest visible result.
   scoped to GENUINE failure: a committed-200 mid-stream/later-chunk degrade writes a
   `tts_degrade` stderr line (server) or a `no WAV header` ticker (browser); a browser-side
   ABORT/cancel (INV-TTS-3 new-turn) is deliberately SILENT — cancellation is not a failure.
+- **INV-TTS-5 [hard]** — voice config is degrade-not-crash: an absent, unreadable, or malformed
+  `voices.json` (bad JSON, wrong types, missing keys) falls back to the built-in DEFAULT_VOICE_CONFIG
+  and logs; it NEVER crashes `create_app`/the server. Per-agent malformed entries fall to `default`.
+- **INV-TTS-6 [hard]** — the single-span path is a byte-identical passthrough: an agent with no
+  `narration_voice` (e.g. Donut), or any turn that segments to one span, produces the exact stream the
+  prior single `tts_stream` call did (one leading WAV header + PCM, verbatim). No regression for the
+  dialogue-only / single-voice case.
+- **INV-TTS-7 [hard]** — a stitched multi-span stream carries EXACTLY ONE WAV header (span 0's); spans
+  1..N are header-stripped before their PCM is emitted, so the browser decodes one continuous s16le
+  stream (never a RIFF header buried mid-stream).
 - **INV-KB-1 [hard]** — the KB bridge is import-isolated behind a single seam:
   `server.py`'s turn path calls exactly one function `pin_kb_context(question,
   agent_id) -> list[memory_context] | []`. Retiring the bridge = delete

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "ratatoskr"
-version = "0.22.2"
+version = "0.22.3"
 description = "Worldtree Conversation API debug console (web + headless CLI) — multi-pane observability"
 readme = "README.md"
 requires-python = ">=3.12"

src/ratatoskr/tts.py

@@ -3,10 +3,13 @@
 Migrated off chatterbox-fast 2026-08-10 (operator-directed, after an A/B win —
 "very good"). dots-tts (rednote-hilab `dots.tts-soar`, irv-ml1 :8198) is an
 OpenAI-shaped `/v1/audio/speech` gateway — closer to the Zonos-era client than
-chatterbox's bespoke `/tts`. It carries over the two Zonos-era subsystem drops:
+chatterbox's bespoke `/tts`. It carries over one Zonos-era drop and reintroduces one primitive:
   - NO affect/emotion knob (dots, like chatterbox Turbo, exposes no
     valence/arousal/emotion dial), so voice stays flat (DEC-7 stays retired); and
-  - NO client-side chunk-and-concatenate — dots streams a whole turn from one call.
+  - a SINGLE-VOICE turn streams from one dots call (no chunking), but the two-voice
+    dialogue/narration split (DEC-11) synthesizes per-span and STITCHES the PCM under one
+    leading WAV header (tts_stream_stitched). A single-span list is a verbatim passthrough,
+    so the single-voice case is unchanged.
 This module is the SINGLE swap seam for voice synthesis: the `/api/tts` route in
 web/server.py is its only caller.
 
@@ -61,48 +64,78 @@ def gateway_body(text: str, voice: str) -> dict:
     return {"input": text, "voice": voice, "response_format": "wav", "stream": True}
 
 
-async def tts_stream(
-    text: str,
+def _log_degrade(event: str, detail: str) -> None:
+    """Write the INV-TTS-4 `tts_degrade` stderr line for a committed-200 mid-stream failure."""
+    sys.stderr.write(f'{{"kind":"tts_degrade","event":"{event}","detail":"{detail}"}}\n')
+
+
+def _strip_to_pcm(acc: bytes) -> bytes | None:
+    """Given accumulated header bytes of a span past the first, return the PCM after the WAV
+    `data` chunk id+size, or None if `data` hasn't arrived yet (keep accumulating). Robust to
+    the header spanning chunks (INV-TTS-7)."""
+    di = acc.find(b"data")
+    if di < 0 or di + 8 > len(acc):
+        return None
+    return acc[di + 8:]
+
+
+async def tts_stream_stitched(
+    spans: list[tuple[str, str]],
     *,
-    voice: str,
     client: httpx.AsyncClient,
     url: str = DOTS_TTS_URL,
 ) -> AsyncIterator[bytes]:
-    """Open the gateway's CHUNKED stream and yield WAV bytes as they synthesize.
+    """Synthesize an ordered list of (voice, text) spans SERIALLY into ONE continuous stream.
 
-    response_format:"wav"/stream:true emits a streaming int16 WAV (RIFF/data sizes =
-    placeholders) over `transfer-encoding: chunked`, streaming ahead of real-time
-    (RTF ~0.22, infra-ops) — designed to be played progressively by the browser Web Audio
-    path. So we PROXY THE CHUNKS STRAIGHT THROUGH: never buffer, never rewrite the header.
-    dots streams a whole turn from this SINGLE call — there is no client-side
-    chunk-and-concatenate wrapper.
+    dots is single-consumer, so spans render in order. The FIRST emitting span is passed
+    through VERBATIM (its WAV header + PCM); every later span is streamed as wav too but with
+    its leading WAV header STRIPPED (accumulate until `data`, emit after data+8), so the
+    browser decodes one gapless 48kHz mono s16le stream after a single leading header
+    (INV-TTS-7). A single-span list is therefore byte-identical to a plain gateway proxy
+    (INV-TTS-6): the whole-turn single-voice / dialogue-only case is a no-regression passthrough.
 
-    The error policy (the `yielded_any` pivot):
-      - a non-200 OPEN, or a connect/transport failure BEFORE the first byte, raises
-        TtsUnavailable so the endpoint peek can still return a 503 (INV-TTS-4) — nothing
-        committed yet.
-      - a transport drop AFTER >= 1 byte has already streamed (the 200 is committed) DEGRADES:
-        log a `tts_degrade` line, end the generator, keep what played. It NEVER raises into the
-        committed StreamingResponse (which would corrupt it with an ASGI trace).
+    The `yielded_any` degrade pivot spans the WHOLE sequence (INV-TTS-4):
+      - a non-200 OPEN or a transport failure BEFORE any byte has been committed → TtsUnavailable
+        (the endpoint peek turns it into a 503; nothing committed yet).
+      - any failure AFTER >= 1 byte has streamed (the 200 is committed) → DEGRADE: log a
+        `tts_degrade` line, drop the tail, keep what played. NEVER raise into the committed 200.
     """
-    assert text, "tts_stream: text must be non-empty (the endpoint guards this)"
+    assert spans, "tts_stream_stitched: spans must be non-empty (the endpoint guards this)"
     yielded_any = False
-    try:
-        async with client.stream("POST", url, json=gateway_body(text, voice)) as resp:
-            if resp.status_code != 200:
-                raise TtsUnavailable(
-                    f"gateway status {resp.status_code}", status=resp.status_code
-                )
-            async for chunk in resp.aiter_bytes():
-                yielded_any = True
-                yield chunk
-    except httpx.RequestError as exc:
-        if yielded_any:
-            # committed-200 mid-stream drop → degrade + log (INV-TTS-4 "logs + skips").
-            sys.stderr.write(
-                f'{{"kind":"tts_degrade","event":"stream_dropped",'
-                f'"exc":"{type(exc).__name__}"}}\n'
-            )
-            return
-        # open failure, pre-commit → the endpoint peek turns this into a 503.
-        raise TtsUnavailable(f"gateway transport failure: {exc}") from exc
+    header_emitted = False  # has the single leading WAV header been passed through yet?
+    for voice, text in spans:
+        if not text.strip():
+            continue
+        strip_header = header_emitted
+        acc = b""
+        found_pcm = not strip_header  # first emitting span passes through immediately
+        try:
+            async with client.stream("POST", url, json=gateway_body(text, voice)) as resp:
+                if resp.status_code != 200:
+                    if yielded_any:
+                        _log_degrade("span_status", str(resp.status_code))
+                        return
+                    raise TtsUnavailable(
+                        f"gateway status {resp.status_code}", status=resp.status_code
+                    )
+                async for chunk in resp.aiter_bytes():
+                    if not found_pcm:
+                        acc += chunk
+                        pcm = _strip_to_pcm(acc)
+                        if pcm is None:
+                            if len(acc) > 65536:  # no data header in a sane window → malformed
+                                if yielded_any:
+                                    _log_degrade("no_data_header", "")
+                                    return
+                                raise TtsUnavailable("stitched span: no WAV data header")
+                            continue
+                        chunk, found_pcm, acc = pcm, True, b""
+                    if chunk:
+                        yielded_any = True
+                        header_emitted = True
+                        yield chunk
+        except httpx.RequestError as exc:
+            if yielded_any:
+                _log_degrade("stream_dropped", type(exc).__name__)
+                return
+            raise TtsUnavailable(f"gateway transport failure: {exc}") from exc

src/ratatoskr/voices.py

@@ -0,0 +1,125 @@
+"""Rata-side voice configuration + dialogue/narration segmentation (DEC-11).
+
+Voice assignment is config-driven, not hardcoded: `~/.config/ratatoskr/voices.json`
+maps each agent to a dots voice, with an OPTIONAL `narration_voice` whose presence turns
+on the two-voice split (dialogue in `voice`, narration in `narration_voice`).
+
+  schema: {"default": "<voice>",
+           "agents": {"<agent_id>": {"voice": "<v>", "narration_voice": "<v>"?}}}
+
+The split rule (segment_dialogue): QUOTED text (straight " or curly “ ”) = dialogue; text
+outside quotes = narration. An agent with NO narration_voice always voices the whole turn in
+`voice` (single span) — which is why dialogue-only Donut needs no special-casing.
+
+Config load is degrade-not-crash (INV-TTS-5): an absent file is normal (→ built-in default);
+a malformed one logs + falls back. Pure + self-contained (stdlib only).
+"""
+
+from __future__ import annotations
+
+import copy
+import json
+import os
+import sys
+
+# Built-in fallback (INV-TTS-5) and the initial shipped config. donut is dialogue-only
+# (no narration_voice → single voice); sindra splits miranda (spoken) / emmie (narration).
+DEFAULT_VOICE_CONFIG: dict = {
+    "default": "glados",
+    "agents": {
+        "ratatoskr:donut": {"voice": "donut"},
+        "ratatoskr:sindra": {"voice": "miranda", "narration_voice": "emmie"},
+    },
+}
+
+_DEFAULT_PATH = "~/.config/ratatoskr/voices.json"
+
+# Double-quote boundary chars. Straight " TOGGLES; curly “ opens, ” closes (directional).
+# Single quotes / apostrophes (' ’) are NOT boundaries — they stay inside spans so  # noqa: RUF003
+# possessives/contractions ("Donut's") never split a word.
+_QUOTE_OPEN = "“"   # “
+_QUOTE_CLOSE = "”"  # ”
+_QUOTE_STRAIGHT = '"'
+
+
+def load_voice_config(path: str | None = None) -> dict:
+    """Load voices.json → config dict. Absent file → built-in default (normal, not an error);
+    malformed/unreadable → built-in default + a logged `voice_config_error` (INV-TTS-5)."""
+    p = os.path.expanduser(path or _DEFAULT_PATH)
+    try:
+        with open(p, encoding="utf-8") as f:
+            data = json.load(f)
+    except FileNotFoundError:
+        return copy.deepcopy(DEFAULT_VOICE_CONFIG)
+    except (OSError, json.JSONDecodeError) as exc:
+        sys.stderr.write(
+            f'{{"kind":"voice_config_error","path":{json.dumps(p)},"exc":"{type(exc).__name__}"}}\n'
+        )
+        return copy.deepcopy(DEFAULT_VOICE_CONFIG)
+    if not isinstance(data, dict) or not isinstance(data.get("agents"), dict):
+        sys.stderr.write(
+            f'{{"kind":"voice_config_error","path":{json.dumps(p)},'
+            f'"exc":"missing agents mapping"}}\n'
+        )
+        return copy.deepcopy(DEFAULT_VOICE_CONFIG)
+    data.setdefault("default", DEFAULT_VOICE_CONFIG["default"])
+    return data
+
+
+def segment_dialogue(text: str) -> list[tuple[str, str]]:
+    """Split text into ordered (kind, span) pairs, kind ∈ {"dialogue","narration"}.
+
+    Quoted runs are dialogue, the rest narration. Straight " toggles quote-state; curly “/”
+    are directional. Quote chars are delimiters (dropped from spans). Empty/whitespace spans
+    are dropped. Unbalanced (a trailing open quote) → its run-to-end is dialogue (best-effort,
+    never raises). Word order is preserved.
+    """
+    spans: list[tuple[str, str]] = []
+    buf: list[str] = []
+    in_quote = False
+
+    def flush() -> None:
+        s = "".join(buf)
+        if s.strip():
+            spans.append(("dialogue" if in_quote else "narration", s))
+        buf.clear()
+
+    for ch in text:
+        if ch == _QUOTE_STRAIGHT:
+            flush()
+            in_quote = not in_quote
+        elif ch == _QUOTE_OPEN:
+            flush()
+            in_quote = True
+        elif ch == _QUOTE_CLOSE:
+            flush()
+            in_quote = False
+        else:
+            buf.append(ch)
+    flush()
+    return spans
+
+
+def resolve_voice_spans(config: dict, agent_id: object, text: str) -> list[tuple[str, str]]:
+    """Resolve a turn to an ordered list of (voice, text) synthesis spans.
+
+    Unmapped agent (or non-str agent_id) → one span in `default`. Mapped agent with no
+    `narration_voice` → one span in `voice`. Mapped agent WITH `narration_voice` → segment
+    into dialogue/narration spans (dialogue→voice, narration→narration_voice). Always returns
+    at least one span for non-empty text (segmentation that empties falls back to a single span).
+    """
+    default = config.get("default", DEFAULT_VOICE_CONFIG["default"])
+    agents = config.get("agents", {})
+    entry = agents.get(agent_id) if isinstance(agent_id, str) and isinstance(agents, dict) else None
+    if not isinstance(entry, dict):
+        return [(default, text)]
+    voice = entry.get("voice") or default
+    if not isinstance(voice, str):
+        voice = default
+    narr = entry.get("narration_voice")
+    if not narr or not isinstance(narr, str):
+        return [(voice, text)]
+    spans = segment_dialogue(text)
+    if not spans:
+        return [(voice, text)]
+    return [(voice if kind == "dialogue" else narr, span) for kind, span in spans]

src/ratatoskr/web/entrypoint.py

@@ -107,8 +107,13 @@ def main(argv: list[str] | None = None) -> int:
     import httpx
 
     from ratatoskr.cli import USER_AGENT
+    from ratatoskr.voices import load_voice_config
     from ratatoskr.web.server import create_app
 
+    # Config-driven voices (DEC-11): ~/.config/ratatoskr/voices.json (per-agent voice +
+    # optional narration_voice). Absent/malformed → the built-in default (INV-TTS-5).
+    voice_config = load_voice_config()
+
     def client_factory() -> httpx.AsyncClient:
         return httpx.AsyncClient(
             base_url=server_url,
@@ -128,6 +133,7 @@ def main(argv: list[str] | None = None) -> int:
         memory_read_url=memory_read_url,
         admin_key=admin_key,
         tts_url=tts_url,
+        voice_config=voice_config,
     )
 
     # Boot banner to stderr (so stdout stays clean for piping).

src/ratatoskr/web/server.py

@@ -9,6 +9,7 @@ passes a factory that bakes in WORLDTREE_API_URL + WORLDTREE_API_KEY.
 from __future__ import annotations
 
 import asyncio
+import copy
 import itertools
 import json
 from collections.abc import AsyncIterator, Callable, Mapping
@@ -68,8 +69,9 @@ from ratatoskr.sse_client import (
 from ratatoskr.tts import (
     DOTS_TTS_URL,
     TtsUnavailable,
-    tts_stream,
+    tts_stream_stitched,
 )
+from ratatoskr.voices import DEFAULT_VOICE_CONFIG, resolve_voice_spans
 
 
 def _wt_client(
@@ -542,11 +544,9 @@ async def _memory_chunks_endpoint(request: Request) -> JSONResponse:
     return JSONResponse(r.json(), status_code=r.status_code)
 
 
-# Per-character voice map (DEC-8): interview characters resolve to their registered
-# dots voice name (GET /v1/voices lists them); everything else falls to the gateway default.
-# NOTE the case: dots wants lowercase "donut" (Zonos used "Donut").
-_TTS_VOICE_MAP = {"ratatoskr:donut": "donut", "ratatoskr:sindra": "miranda"}
-_TTS_DEFAULT_VOICE = "glados"  # a dots voice (chatterbox "glados_25s" / Zonos "Cora" do not exist here)
+# Voice assignment is config-driven (DEC-11): app.state.voice_config (from voices.json) maps
+# each agent to a dots voice + optional narration_voice; resolve_voice_spans turns a turn into
+# ordered (voice, text) synthesis spans. The old hardcoded _TTS_VOICE_MAP is retired.
 # The text rides the POST body (DEC-10a), so URL length is not the bound — this is a safety
 # ceiling on the shared-GPU hold. dots streams a whole turn from one call (no client concat),
 # so a single call voices the whole turn; ~8000 chars still covers any real interview turn
@@ -567,9 +567,10 @@ def _truncate_at_boundary(text: str, limit: int) -> str:
 async def _tts_endpoint(request: Request) -> Response:
     """POST /api/tts {text, agent_id?} → audio/wav, STREAMED chunked from the dots-tts
     gateway (FN tts_endpoint). POST (not GET) so an arbitrarily long turn rides the body, not a
-    length-capped URL. dots streams a whole turn from one call, so a single tts_stream call
-    proxies it — bytes straight through (one leading WAV header + s16le PCM @ 48kHz), and the
-    browser decodes one gapless stream.
+    length-capped URL. resolve_voice_spans (DEC-11) turns the turn into ordered (voice, text)
+    spans — one for a single-voice agent, or dialogue/narration spans when the agent has a
+    narration_voice — and tts_stream_stitched synthesizes them into one gapless 48kHz stream
+    (one leading WAV header + s16le PCM).
 
     Server-side proxy (DEC-4 / INV-TTS-1: the gateway host never reaches the browser).
     Voice per-character (DEC-8). No affect modulation — DEC-7 retired with the Zonos migration.
@@ -591,14 +592,12 @@ async def _tts_endpoint(request: Request) -> Response:
     if not text.strip():
         return JSONResponse({"error_code": "missing_text"}, status_code=400)
     text = _truncate_at_boundary(text, _TTS_MAX_TEXT_CHARS)
-    # agent_id is an UNTRUSTED open-world body field: a non-str (unhashable list/dict) would
-    # TypeError on the voice-map .get(), so guard the type and degrade to the default voice
-    # rather than 500 (INV-TTS-4). (The Zonos-era p/a PAD fields are gone — DEC-7 retired.)
+    # agent_id is an UNTRUSTED open-world body field; resolve_voice_spans guards a non-str
+    # (→ default voice, DEC-11) rather than 500 (INV-TTS-4). It returns >=1 (voice, text) span:
+    # one for a single-voice/dialogue-only agent, or dialogue/narration spans when the agent
+    # has a narration_voice. (The Zonos-era p/a PAD fields are gone — DEC-7 retired.)
     agent_id = body.get("agent_id")
-    voice = (
-        _TTS_VOICE_MAP.get(agent_id, _TTS_DEFAULT_VOICE)
-        if isinstance(agent_id, str) else _TTS_DEFAULT_VOICE
-    )
+    spans = resolve_voice_spans(request.app.state.voice_config, agent_id, text)
 
     tts_url = request.app.state.tts_url
     lock = request.app.state.tts_lock
@@ -610,10 +609,10 @@ async def _tts_endpoint(request: Request) -> Response:
     client = httpx.AsyncClient(
         timeout=httpx.Timeout(connect=10.0, read=120.0, write=10.0, pool=10.0)
     )
-    gen = tts_stream(text, voice=voice, client=client, url=tts_url)
+    gen = tts_stream_stitched(spans, client=client, url=tts_url)
 
     async def _release() -> None:
-        await gen.aclose()      # unwinds tts_stream's `async with` → closes the gateway resp
+        await gen.aclose()      # unwinds the stitched gen's `async with` → closes the gateway resp
         await client.aclose()
         lock.release()
 
@@ -808,6 +807,7 @@ def create_app(
     memory_read_url: str | None = None,
     admin_key: str | None = None,
     tts_url: str | None = None,
+    voice_config: dict | None = None,
 ) -> Starlette:
     """Construct the Starlette app — wire routes + state per FN create_app.
 
@@ -905,6 +905,11 @@ def create_app(
     # don't contend the shared GPU (DEC-5).
     app.state.tts_url = tts_url or DOTS_TTS_URL
     app.state.tts_lock = asyncio.Lock()
+    # Config-driven voices (DEC-11): the per-agent voice + optional narration_voice map.
+    # entrypoint loads voices.json; tests may inject; None → the built-in default (INV-TTS-5).
+    app.state.voice_config = voice_config if voice_config is not None else copy.deepcopy(
+        DEFAULT_VOICE_CONFIG
+    )
     # INV-002: turn registry is in-process memory, keyed (session_id, turn_id)
     app.state.turn_registry = {}
     return app

tests/test_tts.py

@@ -1,10 +1,10 @@
 """Tests for ratatoskr.tts — the STREAMING dots-tts gateway client.
 
-tts_stream proxies the gateway's chunked response verbatim (no buffering, no header
-rewrite — the placeholder-size streaming WAV is meant to be played progressively). It
-is the sole synthesis primitive: dots streams a whole turn from one call, so there is
-no client-side chunk-and-concatenate, and no affect dials (dots has no emotion knob).
-The mid-stream degrade policy is folded in.
+tts_stream_stitched is the sole synthesis primitive: it synthesizes an ordered list of
+(voice, text) spans serially into one continuous stream — the first span verbatim, later
+spans header-stripped (DEC-11 two-voice split). A single-span list is a verbatim passthrough
+(no buffering, no header rewrite), so the single-voice / dialogue-only case is unchanged. No
+affect dials (dots has no emotion knob). The mid-stream degrade policy spans the sequence.
 """
 
 import httpx
@@ -15,7 +15,7 @@ from ratatoskr.tts import (
     DOTS_TTS_URL,
     TtsUnavailable,
     gateway_body,
-    tts_stream,
+    tts_stream_stitched,
 )
 
 _URL = "http://tts.example/v1/audio/speech"
@@ -25,6 +25,12 @@ _WAV = (
     b"RIFF\xff\xff\xff\xffWAVEfmt \x10\x00\x00\x00" + b"\x00" * 20
     + b"data\xff\xff\xff\xff" + b"\x11\x22" * 64
 )
+# A second span's WAV with distinct PCM — its header is stripped when stitched after span 0.
+_WAV2 = (
+    b"RIFF\xff\xff\xff\xffWAVEfmt \x10\x00\x00\x00" + b"\x00" * 20
+    + b"data\xff\xff\xff\xff" + b"\x33\x44" * 32
+)
+_PCM2 = b"\x33\x44" * 32  # the part of _WAV2 after `data`+size (what stitching keeps)
 
 
 async def _drain(gen) -> bytes:
@@ -34,6 +40,12 @@ async def _drain(gen) -> bytes:
     return out
 
 
+def _json_voice(route, i: int) -> str:
+    import json as _json
+
+    return _json.loads(route.calls[i].request.content)["voice"]
+
+
 class _RaisingByteStream(httpx.AsyncByteStream):
     """A 200-body stream that yields `head` then drops mid-stream (an httpx.ReadError, a
     RequestError subclass) — models a gateway connection drop AFTER the response committed."""
@@ -73,12 +85,15 @@ class TestGatewayBody:
             assert dead not in b
 
 
-class TestTtsStream:
+class TestTtsStreamStitched:
     @respx.mock
-    async def test_streams_chunks_and_posts_openai_body(self) -> None:
+    async def test_single_span_verbatim_and_posts_openai_body(self) -> None:
+        # A single-span list is a verbatim passthrough (INV-TTS-6) with the OpenAI body.
         route = respx.post(_URL).mock(return_value=httpx.Response(200, content=_WAV))
         async with httpx.AsyncClient() as client:
-            out = await _drain(tts_stream("hello there", voice="donut", client=client, url=_URL))
+            out = await _drain(
+                tts_stream_stitched([("donut", "hello there")], client=client, url=_URL)
+            )
         assert out == _WAV  # passed through verbatim — no header rewrite
         import json as _json
 
@@ -89,12 +104,39 @@ class TestTtsStream:
         assert body["stream"] is True
 
     @respx.mock
-    async def test_default_url_is_dots(self) -> None:
-        route = respx.post(DOTS_TTS_URL).mock(
-            return_value=httpx.Response(200, content=_WAV)
+    async def test_two_spans_stitched_one_header(self) -> None:
+        # Span 0 verbatim (its WAV header + PCM), span 1 header-STRIPPED → one continuous
+        # stream with exactly one leading header (INV-TTS-7). Distinct voices per span.
+        route = respx.post(_URL).mock(
+            side_effect=[httpx.Response(200, content=_WAV), httpx.Response(200, content=_WAV2)]
         )
         async with httpx.AsyncClient() as client:
-            await _drain(tts_stream("hi", voice="donut", client=client))
+            out = await _drain(
+                tts_stream_stitched(
+                    [("miranda", "spoken bit"), ("emmie", "narrated bit")],
+                    client=client, url=_URL,
+                )
+            )
+        assert out == _WAV + _PCM2                      # span1's header dropped, PCM kept
+        assert out.count(b"RIFF") == 1                  # exactly one WAV header
+        assert _json_voice(route, 0) == "miranda" and _json_voice(route, 1) == "emmie"
+
+    @respx.mock
+    async def test_empty_span_skipped(self) -> None:
+        route = respx.post(_URL).mock(return_value=httpx.Response(200, content=_WAV))
+        async with httpx.AsyncClient() as client:
+            out = await _drain(
+                tts_stream_stitched(
+                    [("miranda", "   "), ("donut", "real")], client=client, url=_URL
+                )
+            )
+        assert out == _WAV and len(route.calls) == 1     # blank span never synthesized
+
+    @respx.mock
+    async def test_default_url_is_dots(self) -> None:
+        route = respx.post(DOTS_TTS_URL).mock(return_value=httpx.Response(200, content=_WAV))
+        async with httpx.AsyncClient() as client:
+            await _drain(tts_stream_stitched([("donut", "hi")], client=client))
         assert route.called  # the module default points at the dots-tts gateway
 
     @respx.mock
@@ -102,7 +144,7 @@ class TestTtsStream:
         respx.post(_URL).mock(return_value=httpx.Response(500, content=b"boom"))
         async with httpx.AsyncClient() as client:
             with pytest.raises(TtsUnavailable) as exc:
-                await _drain(tts_stream("hi", voice="Cora", client=client, url=_URL))
+                await _drain(tts_stream_stitched([("glados", "hi")], client=client, url=_URL))
         assert exc.value.status == 500
 
     @respx.mock
@@ -110,16 +152,27 @@ class TestTtsStream:
         respx.post(_URL).mock(side_effect=httpx.ConnectError("refused"))
         async with httpx.AsyncClient() as client:
             with pytest.raises(TtsUnavailable):
-                await _drain(tts_stream("hi", voice="Cora", client=client, url=_URL))
+                await _drain(tts_stream_stitched([("glados", "hi")], client=client, url=_URL))
 
     @respx.mock
     async def test_mid_stream_drop_after_first_byte_degrades_not_raises(self) -> None:
         # The 200 is committed once bytes flow; a later transport drop must DEGRADE
-        # (return what streamed), never raise — the pivot is yielded_any, folded in from
-        # the retired tts_stream_long. Keeps a committed StreamingResponse from an ASGI trace.
-        respx.post(_URL).mock(
-            return_value=httpx.Response(200, stream=_RaisingByteStream(_WAV))
+        # (return what streamed), never raise — the pivot is yielded_any. Keeps a committed
+        # StreamingResponse from an ASGI trace.
+        respx.post(_URL).mock(return_value=httpx.Response(200, stream=_RaisingByteStream(_WAV)))
+        async with httpx.AsyncClient() as client:
+            out = await _drain(tts_stream_stitched([("donut", "hi")], client=client, url=_URL))
+        assert out == _WAV  # head kept, no raise
+
+    @respx.mock
+    async def test_later_span_open_fail_after_commit_degrades(self) -> None:
+        # Span 0 commits a 200 + bytes; span 1's OPEN then 500s. Because the stream is already
+        # committed, this DEGRADES (keep span 0), never raises into the 200 (INV-TTS-4).
+        route = respx.post(_URL).mock(
+            side_effect=[httpx.Response(200, content=_WAV), httpx.Response(500, content=b"boom")]
         )
         async with httpx.AsyncClient() as client:
-            out = await _drain(tts_stream("hi", voice="donut", client=client, url=_URL))
-        assert out == _WAV  # head kept, no raise
+            out = await _drain(
+                tts_stream_stitched([("miranda", "a"), ("emmie", "b")], client=client, url=_URL)
+            )
+        assert out == _WAV and len(route.calls) == 2  # span0 kept, span1 attempted then dropped

tests/test_voices.py

@@ -0,0 +1,116 @@
+"""Tests for ratatoskr.voices — config load + dialogue/narration segmentation (DEC-11)."""
+
+import json
+from typing import ClassVar
+
+from ratatoskr.voices import (
+    DEFAULT_VOICE_CONFIG,
+    load_voice_config,
+    resolve_voice_spans,
+    segment_dialogue,
+)
+
+
+class TestSegmentDialogue:
+    def test_narration_only_no_quotes(self) -> None:
+        assert segment_dialogue("She tilts her head.") == [("narration", "She tilts her head.")]
+
+    def test_straight_quotes_dialogue(self) -> None:
+        assert segment_dialogue('"Hi there."') == [("dialogue", "Hi there.")]
+
+    def test_curly_quotes_dialogue(self) -> None:
+        assert segment_dialogue("“Hi there.”") == [("dialogue", "Hi there.")]
+
+    def test_mixed_narration_and_dialogue_order_preserved(self) -> None:
+        text = "She smiles. “Hello.” She left."
+        assert segment_dialogue(text) == [
+            ("narration", "She smiles. "),
+            ("dialogue", "Hello."),
+            ("narration", " She left."),
+        ]
+
+    def test_empty_and_whitespace_spans_dropped(self) -> None:
+        # The between-quotes gap here is a single space → dropped; leading empty narration too.
+        assert segment_dialogue('"A" "B"') == [("dialogue", "A"), ("dialogue", "B")]
+
+    def test_apostrophe_is_not_a_boundary(self) -> None:
+        # Curly and straight apostrophes stay inside the span (no word split).
+        assert segment_dialogue("Donut’s crown and Carl's cat") == [  # noqa: RUF001
+            ("narration", "Donut’s crown and Carl's cat")  # noqa: RUF001
+        ]
+
+    def test_unbalanced_trailing_open_quote_is_dialogue(self) -> None:
+        assert segment_dialogue('She said "hello') == [
+            ("narration", "She said "),
+            ("dialogue", "hello"),
+        ]
+
+    def test_empty_text(self) -> None:
+        assert segment_dialogue("") == []
+
+
+class TestResolveVoiceSpans:
+    CFG: ClassVar[dict] = {
+        "default": "glados",
+        "agents": {
+            "ratatoskr:donut": {"voice": "donut"},
+            "ratatoskr:sindra": {"voice": "miranda", "narration_voice": "emmie"},
+        },
+    }
+
+    def test_unmapped_agent_single_default_voice(self) -> None:
+        assert resolve_voice_spans(self.CFG, "mimir", "hello") == [("glados", "hello")]
+
+    def test_non_str_agent_id_single_default(self) -> None:
+        assert resolve_voice_spans(self.CFG, ["donut"], "hi") == [("glados", "hi")]
+
+    def test_mapped_no_narration_single_voice(self) -> None:
+        # Donut is dialogue-only (no narration_voice) → whole turn in one voice, no split,
+        # even though the text has no quotes.
+        assert resolve_voice_spans(self.CFG, "ratatoskr:donut", "The crown is mine.") == [
+            ("donut", "The crown is mine.")
+        ]
+
+    def test_mapped_with_narration_splits(self) -> None:
+        spans = resolve_voice_spans(
+            self.CFG, "ratatoskr:sindra", "She smiles. “Hello.”"
+        )
+        assert spans == [("emmie", "She smiles. "), ("miranda", "Hello.")]
+
+    def test_mapped_with_narration_all_dialogue(self) -> None:
+        assert resolve_voice_spans(self.CFG, "ratatoskr:sindra", '"Only speech here."') == [
+            ("miranda", "Only speech here.")
+        ]
+
+    def test_segmentation_empty_falls_back_to_single_span(self) -> None:
+        # Text that segments to nothing (only quote chars) still yields one span, never [].
+        out = resolve_voice_spans(self.CFG, "ratatoskr:sindra", '""')
+        assert len(out) == 1 and out[0][0] == "miranda"
+
+
+class TestLoadVoiceConfig:
+    def test_absent_file_returns_builtin_default(self, tmp_path) -> None:
+        cfg = load_voice_config(str(tmp_path / "nope.json"))
+        assert cfg == DEFAULT_VOICE_CONFIG
+        assert cfg is not DEFAULT_VOICE_CONFIG  # a copy, not the shared constant
+
+    def test_valid_file_loaded(self, tmp_path) -> None:
+        p = tmp_path / "voices.json"
+        p.write_text(json.dumps({"default": "glados", "agents": {"x:y": {"voice": "emmie"}}}))
+        cfg = load_voice_config(str(p))
+        assert cfg["agents"]["x:y"]["voice"] == "emmie"
+
+    def test_malformed_json_falls_back(self, tmp_path) -> None:
+        p = tmp_path / "voices.json"
+        p.write_text("{not json")
+        assert load_voice_config(str(p)) == DEFAULT_VOICE_CONFIG
+
+    def test_missing_agents_mapping_falls_back(self, tmp_path) -> None:
+        p = tmp_path / "voices.json"
+        p.write_text(json.dumps({"default": "glados"}))
+        assert load_voice_config(str(p)) == DEFAULT_VOICE_CONFIG
+
+    def test_default_key_backfilled(self, tmp_path) -> None:
+        p = tmp_path / "voices.json"
+        p.write_text(json.dumps({"agents": {"x:y": {"voice": "emmie"}}}))
+        assert load_voice_config(str(p))["default"] == "glados"

tests/test_web_server.py

@@ -1506,19 +1506,47 @@ class TestTtsEndpoint:
         assert body["voice"] == "glados"    # dots default voice (DEC-8)
 
     @respx.mock
-    def test_sindra_resolves_miranda_voice(self) -> None:
+    def test_sindra_dialogue_resolves_miranda_voice(self) -> None:
+        # Sindra has a narration_voice, so QUOTED text is dialogue → miranda (DEC-11).
         from ratatoskr.web.server import create_app
 
-        route = respx.post(self._TTS).mock(
-            return_value=httpx.Response(200, content=self._WAV)
-        )
+        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": "Hello there.", "agent_id": "ratatoskr:sindra"}
+            "/api/tts", json={"text": '"Hello there."', "agent_id": "ratatoskr:sindra"}
         )
         assert resp.status_code == 200
-        body = json.loads(route.calls.last.request.content)
-        assert body["voice"] == "miranda"   # per-character map (DEC-8)
+        assert len(route.calls) == 1
+        assert json.loads(route.calls.last.request.content)["voice"] == "miranda"
+
+    @respx.mock
+    def test_sindra_narration_resolves_emmie_voice(self) -> None:
+        # Unquoted text is narration → the narration_voice (emmie) (DEC-11).
+        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": "She tilts her head.", "agent_id": "ratatoskr:sindra"}
+        )
+        assert resp.status_code == 200
+        assert json.loads(route.calls.last.request.content)["voice"] == "emmie"
+
+    @respx.mock
+    def test_sindra_two_voice_stitches_one_header(self) -> None:
+        # A mixed turn → narration span (emmie) + dialogue span (miranda), stitched into one
+        # stream with a single leading WAV header (INV-TTS-7); span 1's header is stripped.
+        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": 'She smiles. "Hello."', "agent_id": "ratatoskr:sindra"}
+        )
+        assert resp.status_code == 200
+        assert len(route.calls) == 2
+        assert [json.loads(c.request.content)["voice"] for c in route.calls] == ["emmie", "miranda"]
+        assert resp.content.count(b"RIFF") == 1   # one header for the whole stitched stream
 
     def test_missing_text_returns_400(self) -> None:
         from ratatoskr.web.server import create_app
@@ -1557,7 +1585,7 @@ class TestTtsEndpoint:
 
     @respx.mock
     def test_non_str_agent_id_degrades_no_500(self) -> None:
-        # An unhashable agent_id (list/dict) would TypeError on _TTS_VOICE_MAP.get().
+        # An unhashable agent_id (list/dict) is guarded by resolve_voice_spans → default voice.
         from ratatoskr.web.server import create_app
 
         route = respx.post(self._TTS).mock(
@@ -1588,7 +1616,7 @@ class TestTtsEndpoint:
             raise RuntimeError("boom during peek")
             yield b""  # unreachable — marks this an async generator
 
-        monkeypatch.setattr(_server, "tts_stream", _boom)
+        monkeypatch.setattr(_server, "tts_stream_stitched", _boom)
         app = create_app(_mock_client_factory(), tts_url=self._TTS)
         with pytest.raises(RuntimeError):
             TestClient(app).post("/api/tts", json={"text": "hi"})

uv.lock

@@ -463,7 +463,7 @@ wheels = [
 
 [[package]]
 name = "ratatoskr"
-version = "0.22.2"
+version = "0.22.3"
 source = { editable = "." }
 dependencies = [
     { name = "httpx" },