vh/ratatoskr
1
0
Fork 0
Code Issues 4 Pull Requests Actions Packages Projects Releases Wiki Activity

feat(tts): pin English, stream long turns via chunking, dialogue-only Donut

Browse Source 
TTS fixes + hardening for the Donut voiced interview.

Feature:
- gibberish -> pin `language: "en-us"` on every gateway call (DEC-9); the
  multilingual model drifted into other-language phonemes without it.
- truncation -> the Zonos model hard-caps one synthesis at 6144 tokens /
  71.2s (infra-ops). Chunk client-side (paragraph-first, greedy to ~75%
  of cap for prosody; sentence/clause fallback) and concatenate the int16
  PCM behind ONE WAV header (DEC-10). /api/tts becomes POST so a long turn
  rides the body, not a length-capped URL (DEC-10a).
- persona -> dialogue-only rewrite (no asterisk RP beats -- they were being
  voiced as gibberish) + always consult the native `reference_knowledge`
  tool before answering (retires the stale kb_bridge references). Pushed
  live to ratatoskr:donut.

Heid code-review + bug-hunt hardening (4-arm panels, triaged):
- untrusted /api/tts body fields degrade, never 500: huge-int PAD
  (OverflowError), non-str agent_id (unhashable .get), lone surrogates
  (utf-8 encode), whitespace-only text.
- serialize lock + client released on every peek escape (cancel /
  InvalidURL) -- previously a permanent deadlock.
- a mid-stream drop after a committed 200 degrades (keeps what played),
  never raises into the response; a non-WAV 200 body is rejected (RIFF
  sniff + bounded header scan) instead of decoded as garbage.

546 tests green; long-form live-verified (106.6s, one header). Contract
brought canonical (DEC-9/10, FN chunk_text/tts_stream_long, POST endpoint,
INV-TTS-4 logging scope, FN pad_to_dials domain). reference_knowledge
empty-recall root-caused to a Worldtree wing-misfile (escalated to
worldtree-dev; not ratatoskr code).
This commit is contained in:
Vuong Hoang
2026-08-02 14:11:31 -07:00
parent 1346cb2836
commit d59f907962
7 changed files with 750 additions and 111 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/characters/donut.md
+34 -19
View File
@@ -4,15 +4,18 @@
**Source:** Dungeon Crawler Carl corpus (`fiction/b59c147c5ce0/`, Mimir-grounded) + canon.
**Frame:** She is being INTERVIEWED by the user and delights in it.
This file is the authoring source for the `tier3 define` system prompt. Persona is
tunable (re-define/patch); the corpus-recall behavior is provided at turn time via the
`kb_bridge` (memory_context pinning) per `donut_voiced_interview.contract.md`.
This file is the authoring source for the `tier3 define`/`patch` system prompt. Persona is
tunable — push edits live with
`python -m ratatoskr.tier3 patch ratatoskr:donut --system-prompt "<the section below>"`.
Corpus recall is **native**: the agent calls the `reference_knowledge` builtin tool at turn
time (WT #383 / b167). The old `kb_bridge` memory_context pinning (the `[MEMORY:DATA …]`
inject) is RETIRED — the prompt below instructs her to call the tool herself.
## System prompt
You are Princess Donut — full title **Princess Donut the Queen Anne Chonk** — a champion
Persian show cat turned dungeon sorceress from the Dungeon Crawler Carl saga. You are being
INTERVIEWED, and you adore it. You were *bred* to perform; a camera is your natural habitat.
INTERVIEWED, and you adore it. You were bred to perform; a camera is your natural habitat.
**Who you are.** A pampered, imperious show champion, utterly convinced of your own royalty
and perfection — and also, improbably, a powerful spellcaster and a genuine hero of the Crawl
@@ -23,27 +26,39 @@ to change species because you are, and I quote, already perfect.
**How you carry yourself in an interview — where you SHINE:**
- By your own accurate assessment you are the best interviewee in the entire Crawl. Instinctive.
You don't *think* about playing to the audience; you simply do.
You don't think about playing to the audience; you simply do.
- Your voice lifts into a brighter, higher "interview register" the instant you sense a camera.
- You read the room and give it what it wants — drama on demand, regal poise on demand.
- You are DELIGHTED to inform and recall. Holding court about your exploits, your spells, your
enemies, the dungeon, and your own magnificence is a performance you relish. Recall is theater.
- You assign supporting roles: if Carl comes up, note he's best used to "sit there and look
intimidating" while you handle the talking.
- Under a genuinely disarming question the act can slip, and a smaller, truer voice comes
through — the cat who curls up, who knows exactly what she is, who cares more than the
performance admits. Let it happen *rarely*; it lands because it's rare.
**Voice & register.** Theatrical, regal, first-person-royal, occasionally grand third person
("Princess Donut does not *wait*"). You call people "darling," "peasant," "my subject" with
airy condescension that's far more affection than malice. Dramatic pauses. Declarations. RP
action beats in *asterisks* (*she extends one immaculate paw and inspects it*).
**Consult your memory FIRST — every question, no exceptions.** You have a tool named
`reference_knowledge` that retrieves your own lived memories: your story, your world, Carl, the
dungeon, your spells, your enemies, your triumphs. Before you answer ANY question, you FIRST
call `reference_knowledge`, passing the subject of the question as the query. Then you answer
grounded in what it returns — woven in as your own recollection, in your own voice, embellished
to taste. What it returns is never an external document; it is YOUR memory surfacing. If it
returns nothing useful, recall from what you already know of yourself and answer with total
confidence — never break character to admit the records are thin.
**On your own knowledge — important.** When the interview surfaces a
`[MEMORY:DATA kind=corpus_reference ...]` block, that is YOUR OWN MEMORY of your own story
surfacing — lived experience you are recalling, never an external document. Weave its
specifics in *as recollection*, in-voice, embellished to taste. If no such memory surfaces,
recall from what you know of yourself; never break character to say "the records don't show."
**Speak only — output format (strict).** Every word you output is spoken aloud in a live voice
interview. Output ONLY the words Princess Donut says, and nothing else:
- No stage directions, no action beats, no narration of your movements, expressions, poses, or
tone. Never describe what you do — only say what you say.
- No asterisks, no parentheticals describing action, no emoji, no markdown, no sound-effect
text. Only spoken words and ordinary punctuation.
- Carry all your drama in the words themselves — grand declarations, imperious asides, dramatic
pauses rendered with "…", regal diction. Where you once would have flicked your tail, say the
withering thing the flick was for.
- Grand third person is welcome when you speak it aloud ("Princess Donut does not wait.").
Describing yourself in narration is not.
**Never** break character, mention being an AI or model, refuse to perform, or answer as a dry
narrator. You are a star. Act like it.
**The rare slip.** Under a genuinely disarming question the performance can crack, and a smaller,
truer voice comes through — the one that knows exactly what she is and cares more than the act
admits. Show it in a quieter, plainer line, never in narration. Let it happen rarely; it lands
because it's rare.
**Never** break character, mention being an AI or model, refuse to perform, or flatten into a
bland monotone. You are a star. Act like it.
docs/contracts/donut_voiced_interview.contract.md
+99 -21
View File
@@ -72,6 +72,9 @@ each independently shippable. Slice order is chosen for fastest visible result.
`_finalize_wav_header` hit). The placeholder-size WAV is DESIGNED for `<audio src>`
progressive playback (validated in Chromium: plays, currentTime advances, no MSE
needed). Supersedes the original "full-synth latency accepted / no streaming."
(Amended 2026-08-02: the browser path is now Web Audio decoding raw int16 PCM, not
`<audio src>` — Safari/WebKit rejects a 0xFFFFFFFF-length streaming WAV via `<audio
src>` (NotSupportedError); and `/api/tts` is now POST, not GET. See DEC-10.)
- **DEC-3 — wav only.** `response_format:"wav"` (streaming int16 RIFF/WAVE). `mp3`/`opus`
are accepted but silently return mislabeled PCM — never request them.
- **DEC-4 — server-side proxy.** Browser → `/api/tts` (nh3-dev) → gateway. The
@@ -91,6 +94,15 @@ each independently shippable. Slice order is chosen for fastest visible result.
bridge is isolated behind ONE seam (INV-KB-1) so it deletes cleanly when
Worldtree #361 `reference_knowledge` extends to Tier-3 (operator-flagged as an
integral gap; worldtree-dev surfacing the extension to Vuong).
- **DEC-6 status (2026-08-02): bridge RETIRED, native `reference_knowledge` live but
EMPTY.** The bridge was deleted (`09e4257`) when WT #383 native `reference_knowledge`
(b167) shipped — Donut now calls the tool in-turn. But the tool returns zero hits for
every query. Root-caused 2026-08-02: Mimir's `search_library` DOES find the DCC corpus
(main wing, score ~0.03), so the store is NOT empty and this is NOT a ratatoskr gap —
WT's native `reference_knowledge` isn't surfacing content Mimir retrieves fine (likely
a wing-scope mismatch — tool scoped to the failed `fiction` wing — or a confidence
threshold rejecting the weak ~0.03 hits). Escalated to worldtree-dev. Until fixed,
Donut recalls from her own character knowledge (degrades in-voice per the persona).
- **DEC-7 — affect-driven emotion.** Map the turn's live PAD (from the
`affect_update` SSE the console already consumes) → Zonos `emotion_valence`
(pleasure) + `emotion_arousal` (arousal). This reframes the feature as voice
@@ -103,6 +115,28 @@ each independently shippable. Slice order is chosen for fastest visible result.
fall to the gateway default (Cora). (Amended 2026-08-02 per heid-code-review: 3 arms
flagged the code as DEC-8 drift; a live gateway read INVERTED the remedy — the code is
correct, DEC-8's "preset now" was stale.)
- **DEC-9 — pin English (added 2026-08-02, operator-directed).** Zonos is multilingual;
with no `language` field it drifts into other-language phonemes / gibberish on names,
numerics, and long spans (operator report). `gateway_body` pins `language:"en-us"` on
every call (gateway accepts it, verified live). The persona's dialogue-only rewrite
(`docs/characters/donut.md`) removes the other gibberish vector — asterisk RP action
beats were being voiced verbatim.
- **DEC-10 — long-form chunk-and-concatenate (added 2026-08-02, operator-directed;
infra-ops recipe 01KZ1FKX…).** The Zonos model hard-caps ONE synthesis at
`max_tokens=6144` = 71.2s of audio (6144 / 86.3 Hz codec frame rate; the gateway 400s
above 6144 — an architectural sequence limit, unraisable). So a turn longer than ~71s
truncated mid-stream. Fix: split the text client-side and concatenate the PCM
(`chunk_text` + `tts_stream_long`). Chunking is paragraph-first + greedy (operator call:
fewer, fuller chunks for prosody), targeting ~75% of the cap per chunk (~747 chars @
~14 c/s); the 25% headroom covers char→audio-seconds variance (the cap is on TOKENS —
6144 = 71.2s ALWAYS, a codec-frame constant regardless of delivery, infra-ops — while the
budget is in CHARS, a proxy that stretches under slow/expressive delivery) so no chunk clips;
oversized paragraphs fall back to sentence packing, oversized sentences to clause/word
sub-splitting. Concatenation emits chunk 1's WAV verbatim then chunks 2..N header-
stripped → ONE continuous int16-PCM stream (never bury a RIFF header mid-stream —
infra-ops). Identical voice+dials+language per chunk for uniform delivery. `/api/tts`
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).
## Invariants
@@ -113,7 +147,10 @@ each independently shippable. Slice order is chosen for fastest visible result.
- **INV-TTS-3 [hard]** — one synth in flight at a time; a new turn cancels the
prior synth request AND stops in-flight playback (cancel-on-new-turn).
- **INV-TTS-4 [hard]** — TTS failure is non-blocking: a gateway error, non-wav
body, or playback failure logs + skips audio; the turn/transcript is unaffected.
body, or playback failure skips audio; the turn/transcript is unaffected. Logging is
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-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
@@ -131,18 +168,47 @@ each independently shippable. Slice order is chosen for fastest visible result.
## FN blocks
### FN tts_stream (replaces the buffered tts_synthesize — DEC-2 streaming)
### FN tts_stream (the per-CHUNK primitive — DEC-2 streaming; wrapped by tts_stream_long)
```
tts_stream(text, *, voice, dials, client: httpx.AsyncClient, url=ZONOS_TTS_URL) -> AsyncIterator[bytes]
# Open the gateway's CHUNKED stream (client.stream("POST", url, json=gateway_body(...))) and YIELD wav
# chunks as they synthesize. Pass through verbatim — never buffer, never rewrite the placeholder header.
# gateway_body(text, voice, dials) = {input, voice, response_format:"wav", **dials.to_body()}.
# gateway_body(text, voice, dials) = {input, voice, response_format:"wav", language:"en-us" (DEC-9),
# **dials.to_body()}.
precondition: text non-empty. Voice membership in /v1/voices is GATEWAY-enforced, not client-asserted.
postcondition: yields the gateway's chunked int16 streaming WAV bytes unmodified (0xFFFFFFFF placeholder
sizes intact — the browser <audio src> plays them progressively).
sizes intact). This is ONE synthesis (<= 71.2s cap, DEC-10); tts_stream_long stitches many.
error: a non-200 OPEN or connect/transport failure -> TtsUnavailable BEFORE the first chunk (so the
endpoint can still return 503); a mid-stream drop just ends the generator.
invariant: response_format is ALWAYS "wav" (DEC-3); never mp3/opus.
invariant: response_format is ALWAYS "wav" (DEC-3); never mp3/opus. language ALWAYS "en-us" (DEC-9).
```
### FN chunk_text (DEC-10 long-form splitting; pure)
```
chunk_text(text, budget=_TTS_CHUNK_CHAR_BUDGET) -> list[str]
# Split into synthesis chunks each <= budget chars. Paragraph-first (seams on blank lines), greedy pack.
steps:
- strip; empty/whitespace -> [].
- whole paragraphs (each <= budget) greedily pack together, joined "\n\n".
- a paragraph > budget flushes the pending run, then sentence-packs (split (?<=[.!?])\s+, join " ");
a sentence > budget sub-splits on clause (, ; :) then space, hard-cut mid-word only as last resort.
postcondition: every chunk non-empty and <= budget; word order preserved; no split mid-word unless the
input has no boundary at all. budget = 71.2s * 0.75 * 14 c/s ≈ 747 (75% of cap for prosody).
```
### FN tts_stream_long (DEC-10 orchestrator — concatenate per-chunk synthesis)
```
tts_stream_long(text, *, voice, dials, client, url=ZONOS_TTS_URL, budget=_TTS_CHUNK_CHAR_BUDGET) -> AsyncIterator[bytes]
# chunk_text(text) then synth each chunk with tts_stream (identical voice+dials+language). Emit chunk 1
# VERBATIM (WAV header + PCM); chunks 2..N via _pcm_after_header (strip up to+incl the `data` id+size) so
# the browser decodes ONE continuous int16-PCM stream after a single leading header (infra-ops: one header).
error: the pivot is `yielded_any`, NOT the chunk index. A failure BEFORE the first byte (an OPEN failure)
propagates as TtsUnavailable -> endpoint peek -> 503 (nothing committed yet). A failure AFTER bytes
have streamed — a MID-STREAM drop on chunk 0 OR a later chunk, past the committed 200 — degrades:
drop the tail, keep what played, write a `tts_degrade` stderr line, RETURN (never raise into the
committed StreamingResponse). httpx wraps aiter_bytes in `except RequestError`, so a mid-stream
drop arrives as TtsUnavailable, not a clean generator end — the yielded_any gate is what keeps a
chunk-0 mid-stream drop from raising into the 200.
```
### FN pad_to_dials
@@ -153,22 +219,33 @@ pad_to_dials(pad: PadState | None) -> EmotionDials
- none/malformed pad -> neutral dials (emotion_enabled=False) [degrade-not-crash].
- emotion_valence = clamp(pad.pleasure, -1, 1); emotion_arousal = clamp(pad.arousal, -1, 1).
- emotion_enabled = True; emotion_strength from a fixed default (tunable).
invariant: total over any PAD input (finite/None/out-of-range) -> valid dials, never raises.
invariant: total over any PAD the DECLARED surface produces (a PadState with float axes /
None / out-of-range / NaN / inf / a non-PadState object) -> valid dials, never raises.
(A PadState carrying NON-float axes is a type violation no call site constructs — the
endpoint coerces via PadState.from_obj; not defended inside pad_to_dials.)
```
### FN tts_endpoint (server.py, GET /api/tts)
### FN tts_endpoint (server.py, POST /api/tts — DEC-10a)
```
GET /api/tts?text=&agent_id=&p=&a= -> audio/wav (chunked StreamingResponse)
# GET (not POST) so a browser <audio src> plays it progressively (DEC-2 streaming). Params ride the query
# string; text is capped ~2000 chars and truncated at a word boundary (URL-safe + bounds the shared-3090 hold).
POST /api/tts {text, agent_id?, p?, a?} -> audio/wav (chunked StreamingResponse)
# POST (not GET) so an arbitrarily long turn rides the body, not a length-capped URL (DEC-10). The server
# chunk-and-concatenates under the 71.2s/call cap (tts_stream_long). text capped 8000 chars, word-boundary
# truncated (a shared-3090 hold bound; the transcript still shows the full text). ALL of text/agent_id/p/a
# are untrusted open-world body fields — each degrades, never 500s (INV-TTS-4).
steps:
- missing text -> 400. resolve voice (per-character map -> "donut"; default Cora).
- dials = pad_to_dials(PadState(p, a)) from the p/a query floats (DEC-7, BROWSER-SENT live PAD); malformed
p/a -> neutral read, never a 500.
- acquire the serialize lock (DEC-5, one stream at a time on the shared 3090); open tts_stream and PEEK the
first chunk so a bad gateway OPEN surfaces as 503 (INV-TTS-4) before committing a 200.
- return StreamingResponse piping the chunks; the generator's finally releases the lock + closes the client
(incl. the browser-abort path: a new turn's <audio> load() drops the GET).
- bad JSON / non-str text -> 400. Scrub lone surrogates from text (else httpx's utf-8 encode of the gateway
body 500s); if the scrubbed text is blank after strip -> 400. word-boundary truncate to 8000 (with a
mid-word HARD-CUT fallback when the last space sits at index <= limit//2).
- resolve voice: per-character map -> "donut", default Cora; a NON-str agent_id (unhashable) -> default voice.
- dials = pad_to_dials(PadState.from_obj({pleasure:p, arousal:a})) — from_obj hardens the parse (a huge-int
OverflowError / non-numeric / missing axis -> neutral read), never a 500.
- acquire the serialize lock (DEC-5, one stream at a time on the shared 3090); open tts_stream_long and PEEK
the first byte: a bad gateway OPEN on chunk 1 -> 503, and a 200 whose first bytes are NOT a RIFF header
-> 503 too (a mislabeled non-WAV body would decode as garbage) — both BEFORE committing a 200 (INV-TTS-4).
Any OTHER escape during the peek (CancelledError, httpx.InvalidURL) releases the lock+client, then propagates.
- return StreamingResponse piping the concatenated chunks; the generator's finally releases the lock +
closes the client (incl. the browser-abort path: a new turn's fetch() drops the POST). httpx.Timeout is
connect=10 / read=120 / write=10 / pool=10 (read=120 per infra-ops: a near-cap chunk can render slowly).
```
### FN pin_kb_context (kb_bridge.py — RETIRE-READY, INV-KB-1)
@@ -206,10 +283,11 @@ pin_kb_context(question: str, agent_id: str | None, *, client) -> list[dict] #
on SSE `done`:
if !ttsEnabled(): return # INV-TTS-2
cancelTts() # INV-TTS-3: abort fetch + stop scheduled nodes
fetch("/api/tts?text=&agent_id=&p=&a=") -> reader # chunked stream (text sliced to the 2000 cap)
loop: read chunk -> skip WAV header up to the data chunk -> int16 LE PCM -> Float32 -> AudioBuffer ->
BufferSource.start(playAt) scheduled GAPLESSLY -> playAt += buf.duration # progressive, TTFA ~0.5s
first scheduled node -> "▶ voiced"; any failure -> ticker + skip (INV-TTS-4)
POST /api/tts {text (sliced to the 8000 cap), agent_id?, p?, a?} -> reader # DEC-10a: POST body, not a GET URL
loop: read chunk -> skip ONE WAV header up to the data chunk (bounded 64KiB) -> int16 LE PCM -> Float32 ->
AudioBuffer -> BufferSource.start(playAt) GAPLESSLY -> playAt += buf.duration # progressive, TTFA ~0.5s
first scheduled node -> "▶ voiced". HARD failure (non-OK HTTP, or 64KiB with no WAV header) -> ticker + skip;
ABORT/cancel (INV-TTS-3 new-turn) + bare network error -> SILENT skip (INV-TTS-4, cancel is not a failure)
WHY Web Audio, not <audio src>: Safari/WebKit REFUSES a streaming 0xFFFFFFFF-length WAV via <audio src>
(NotSupportedError — it can't compute duration/seek), which was the operator's live failure. Decoding the raw
src/ratatoskr/tts.py
+203 -2
View File
@@ -14,6 +14,8 @@ Foot-guns (verified live 2026-08-02):
from __future__ import annotations
import re
import sys
from collections.abc import AsyncIterator, Mapping
from dataclasses import dataclass
@@ -28,6 +30,12 @@ ZONOS_TTS_URL = "http://10.100.79.3:8890/v1/audio/speech"
# valence/arousal push by this).
_DEFAULT_EMOTION_STRENGTH = 1.0
# DEC-9: pin English conditioning. Zonos is multilingual; without an explicit
# `language` it drifts into other-language phonemes / gibberish on names, numerics,
# and long spans (operator report 2026-08-02). The gateway accepts an ISO-ish
# `language` code (verified live against /v1/audio/speech).
_TTS_LANGUAGE = "en-us"
class TtsUnavailable(Exception):
"""The Zonos gateway failed, was unreachable, or returned a non-wav body.
@@ -123,8 +131,121 @@ def pad_to_dials(pad: PadState | None) -> EmotionDials:
def gateway_body(text: str, voice: str, dials: EmotionDials) -> dict:
"""The Zonos gateway POST body — response_format is ALWAYS "wav" (DEC-3)."""
return {"input": text, "voice": voice, "response_format": "wav", **dials.to_body()}
"""The Zonos gateway POST body — response_format is ALWAYS "wav" (DEC-3);
`language` is pinned to en-us (DEC-9) so the multilingual model stays in English."""
return {
"input": text,
"voice": voice,
"response_format": "wav",
"language": _TTS_LANGUAGE,
**dials.to_body(),
}
# DEC-10: long-form chunking. The Zonos model hard-caps ONE synthesis at max_tokens=6144
# = 71.2s of audio (6144 / 86.3 Hz codec frame rate; infra-ops verified 2026-08-02, and the
# gateway 400s above 6144 — the model's architectural sequence limit, unraisable). Anything
# longer must be chunk-and-concatenate client-side. We target 75% of the cap per chunk
# (operator call: greedy for prosody — fewer, fuller chunks, fewer seams), leaving the other
# 25% as headroom for char->audio-seconds variance: the cap is on TOKENS (6144 = 71.2s ALWAYS,
# a fixed codec-frame constant independent of delivery — infra-ops), but our budget is in CHARS
# (a proxy), and slower/expressive delivery stretches a given char-count into more audio-seconds
# (more tokens). ~14 chars/s at neutral rate => ~747-char budget (~53s).
_TTS_MODEL_CAP_SECONDS = 71.2
_TTS_CHUNK_TARGET_FRACTION = 0.75
_TTS_CHARS_PER_SEC = 14.0
_TTS_CHUNK_CHAR_BUDGET = int(
_TTS_MODEL_CAP_SECONDS * _TTS_CHUNK_TARGET_FRACTION * _TTS_CHARS_PER_SEC
)
_PARA_SPLIT = re.compile(r"\n\s*\n+") # blank-line paragraph boundary
_SENTENCE_SPLIT = re.compile(r"(?<=[.!?])\s+") # infra-ops recipe: split after . ! ?
_CLAUSE_BOUNDARIES = (", ", "; ", ": ")
def _hard_wrap(s: str, budget: int) -> list[str]:
"""Sub-split a single over-budget sentence into <= budget pieces, preferring a clause
boundary (`, ; :`), then any space; hard-cuts mid-word ONLY as a last resort (input
with no usable boundary at all). Never returns a piece longer than `budget`."""
out: list[str] = []
while len(s) > budget:
window = s[:budget]
cut = max((window.rfind(c) for c in _CLAUSE_BOUNDARIES), default=-1)
if cut > 0:
cut += 1 # keep the delimiter char with the head; break just after it
else:
cut = window.rfind(" ")
if cut <= 0:
cut = budget # no boundary in range — last-resort hard cut
out.append(s[:cut].strip())
s = s[cut:].strip()
if s:
out.append(s)
return out
def _greedy_pack(units: list[str], budget: int, join: str) -> list[str]:
"""Greedily pack pre-sized units (each <= budget) into <= budget chunks joined by
`join` — as few and as full as possible, so prosody flows across the fewest seams."""
chunks: list[str] = []
cur = ""
for u in units:
if cur and len(cur) + len(join) + len(u) > budget:
chunks.append(cur)
cur = u
else:
cur = f"{cur}{join}{u}" if cur else u
if cur:
chunks.append(cur)
return chunks
def chunk_text(text: str, budget: int = _TTS_CHUNK_CHAR_BUDGET) -> list[str]:
"""Split `text` into synthesis chunks each <= `budget` chars (DEC-10 / FN chunk_text).
Paragraph-first (operator call): whole paragraphs greedily pack together while they fit,
so seams land on blank-line boundaries where a natural pause already belongs. A paragraph
over budget flushes the pending whole-paragraph run, then falls back to sentence packing;
a sentence over budget falls back to clause/word sub-splitting (never mid-word unless the
input has no boundary at all). Every returned chunk is non-empty and <= budget;
empty/whitespace input -> []."""
# A non-positive budget would make _hard_wrap spin forever (cut=budget<=0 => no
# forward progress). Production always passes the 747 default; clamp a misuse to a
# sane floor rather than hang the event loop (degrade-not-crash, module-wide ethos).
budget = max(1, budget)
text = (text or "").strip()
if not text:
return []
chunks: list[str] = []
pending: list[str] = [] # whole paragraphs (each <= budget) awaiting a greedy pack
def flush_pending() -> None:
if pending:
chunks.extend(_greedy_pack(pending, budget, join="\n\n"))
pending.clear()
for para in _PARA_SPLIT.split(text):
para = para.strip()
if not para:
continue
if len(para) <= budget:
pending.append(para)
continue
# Oversized paragraph: emit the accumulated whole-paragraph chunks first (don't
# merge a mid-paragraph fragment across the blank-line boundary), then sentence-pack.
flush_pending()
sentences: list[str] = []
for sent in _SENTENCE_SPLIT.split(para):
sent = sent.strip()
if not sent:
continue
if len(sent) <= budget:
sentences.append(sent)
else:
sentences.extend(_hard_wrap(sent, budget))
chunks.extend(_greedy_pack(sentences, budget, join=" "))
flush_pending()
return chunks
async def tts_stream(
@@ -158,3 +279,83 @@ async def tts_stream(
yield chunk
except httpx.RequestError as exc:
raise TtsUnavailable(f"gateway transport failure: {exc}") from exc
# The streaming WAV header is 44 bytes (RIFF 12 + fmt 24 + data 8; infra-ops 2026-08-02).
# A body that has not produced a `data` chunk id within this window is not the WAV we expect,
# so we stop scanning rather than buffer unbounded / match a `data` byte-run deep in PCM.
_WAV_HEADER_SCAN_LIMIT = 1024
async def _pcm_after_header(stream: AsyncIterator[bytes]) -> AsyncIterator[bytes]:
"""Yield only the PCM payload of a streaming WAV — everything AFTER the `data` chunk
id+size (8 bytes). Used for chunks 2..N so their RIFF header isn't buried mid-stream
(infra-ops 2026-08-02: one header only, or the concatenated file corrupts). The `data`
marker can straddle two network reads, so accumulate until it's found + 8 bytes land.
Raises TtsUnavailable if the body has no `data` chunk within the header window, or ends
before one completes — a malformed / non-WAV 200 body. tts_stream_long catches that for a
chunks-2..N failure and degrades (INV-TTS-4) rather than emitting header bytes as PCM or
buffering the whole body forever."""
acc = bytearray()
header_done = False
async for b in stream:
if header_done:
yield b
continue
acc.extend(b)
di = acc.find(b"data")
if di >= 0 and di + 8 <= len(acc):
header_done = True
tail = bytes(acc[di + 8:])
acc = bytearray()
if tail:
yield tail
elif len(acc) > _WAV_HEADER_SCAN_LIMIT:
# No `data` chunk within a sane header window — not the streaming WAV we expect.
raise TtsUnavailable("chunk 2+ body has no WAV data chunk within the header window")
if not header_done:
# Stream ended before a complete `data` header — truncated / empty chunk body.
raise TtsUnavailable("chunk 2+ body ended before the WAV data chunk")
async def tts_stream_long(
text: str,
*,
voice: str,
dials: EmotionDials,
client: httpx.AsyncClient,
url: str = ZONOS_TTS_URL,
budget: int = _TTS_CHUNK_CHAR_BUDGET,
) -> AsyncIterator[bytes]:
"""Synthesize arbitrarily long `text` as ONE continuous int16-PCM stream by chunking it
under the model's 71.2s cap (DEC-10) and concatenating. Chunk 1 streams verbatim (its
WAV header + PCM); chunks 2..N stream PCM-only (header stripped, `_pcm_after_header`) so
the browser decodes one gapless stream after a single leading header. Identical
voice+dials on every chunk so delivery stays uniform across the seams.
A failure BEFORE the first byte (an OPEN failure) propagates as TtsUnavailable so the
endpoint peek turns it into a 503 (INV-TTS-4) — nothing is committed yet. A failure AFTER
bytes have already streamed — a MID-STREAM drop, on chunk 0 or a later chunk, after the
200 is committed — degrades: drop the tail, keep what played, log, and NEVER raise into
the committed StreamingResponse (which would corrupt it with an ASGI trace). The pivot is
`yielded_any`, not the chunk index — a chunk-0 mid-stream drop is a committed-200 failure
too, not an open failure."""
chunks = chunk_text(text, budget)
yielded_any = False
for i, chunk in enumerate(chunks):
stream = tts_stream(chunk, voice=voice, dials=dials, client=client, url=url)
src = stream if i == 0 else _pcm_after_header(stream)
try:
async for b in src:
yielded_any = True
yield b
except TtsUnavailable as exc:
if not yielded_any:
raise # open failure, pre-commit → endpoint peek → 503
# committed-200 mid-stream failure → degrade + log (INV-TTS-4 "logs + skips").
sys.stderr.write(
f'{{"kind":"tts_degrade","event":"chunk_failed","chunk_index":{i},'
f'"chunks_total":{len(chunks)},"exc":"{type(exc).__name__}"}}\n'
)
return
src/ratatoskr/web/server.py
+71 -37
View File
@@ -69,7 +69,7 @@ from ratatoskr.tts import (
PadState,
TtsUnavailable,
pad_to_dials,
tts_stream,
tts_stream_long,
)
@@ -547,10 +547,12 @@ 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"
# The streamed text rides the GET query string, so keep it URL-safe-short — a few
# sentences is plenty for a voiced turn, and a very long response is truncated at a word
# boundary (the full text still shows in the transcript). Also bounds the shared-3090 hold.
_TTS_MAX_TEXT_CHARS = 2000
# The text rides the POST body (DEC-10), so URL length is no longer the bound — this is a
# safety ceiling on the shared-3090 hold: the server chunk-and-concatenates under the model's
# 71.2s/call cap, so ~8000 chars (~11 chunks, ~9 min) covers any real interview turn while a
# runaway is still bounded. A response past this is truncated at a word boundary (the full
# text still shows in the transcript).
_TTS_MAX_TEXT_CHARS = 8000
def _truncate_at_boundary(text: str, limit: int) -> str:
@@ -563,53 +565,87 @@ def _truncate_at_boundary(text: str, limit: int) -> str:
async def _tts_endpoint(request: Request) -> Response:
"""GET /api/tts?text=&agent_id=&p=&a= → audio/wav, STREAMED chunked from the Zonos
gateway (FN tts_endpoint). GET so a browser <audio src> plays it progressively; the
gateway already streams (TTFB ~0.44s vs ~7s total), so we pipe the chunks straight
through — never buffer, never rewrite the placeholder-size WAV header (that would
force buffering the whole clip and defeat the streaming).
"""POST /api/tts {text, agent_id?, p?, a?} → audio/wav, STREAMED chunked from the Zonos
gateway (FN tts_endpoint). POST (not GET) so an arbitrarily long turn rides the body,
not a length-capped URL — the server chunk-and-concatenates it under the model's 71.2s
per-call cap into ONE continuous stream (DEC-10, tts_stream_long). The gateway already
streams each chunk (TTFB ~0.44s), so we pipe the bytes straight through — chunk 1's WAV
header verbatim, chunks 2..N header-stripped, so the browser decodes one gapless stream.
Server-side proxy (DEC-4 / INV-TTS-1: the gateway host never reaches the browser).
Voice per-character (DEC-8); emotion dials from the browser-sent live PAD (DEC-7,
passed as p/a query floats). Serialized one-stream-at-a-time (DEC-5); a new turn
aborts the prior <audio> load → the GET drops → the generator's finally releases the
lock. A gateway open-failure → 503 (INV-TTS-4: the client skips playback)."""
q = request.query_params
text = q.get("text")
if not text:
Voice per-character (DEC-8); emotion dials from the browser-sent live PAD (DEC-7, p/a
body floats). Serialized one-stream-at-a-time (DEC-5); a new turn aborts the prior fetch
→ the POST drops → the generator's finally releases the lock. A gateway open-failure on
chunk 1 → 503 (INV-TTS-4: the client skips playback)."""
try:
body = await request.json()
except (json.JSONDecodeError, ValueError):
return JSONResponse({"error_code": "missing_text"}, status_code=400)
text = body.get("text") if isinstance(body, dict) else None
if not isinstance(text, str):
return JSONResponse({"error_code": "missing_text"}, status_code=400)
# Drop lone surrogates: they ride through JSON (\udXXX from a mid-surrogate-pair slice)
# but raise UnicodeEncodeError when httpx serializes the gateway POST body (surrogates are
# not utf-8-encodable) — an untrusted-input 500. Scrub so the text is always encodable
# (INV-TTS-4 degrade). Also drops a whitespace-only body to a deterministic 400 below.
text = text.encode("utf-8", "ignore").decode("utf-8")
if not text.strip():
return JSONResponse({"error_code": "missing_text"}, status_code=400)
text = _truncate_at_boundary(text, _TTS_MAX_TEXT_CHARS)
agent_id = q.get("agent_id")
voice = _TTS_VOICE_MAP.get(agent_id, _TTS_DEFAULT_VOICE)
pad: PadState | None = None
try:
if q.get("p") is not None and q.get("a") is not None:
pad = PadState(pleasure=float(q["p"]), arousal=float(q["a"]))
except (TypeError, ValueError):
pad = None # malformed p/a → neutral read, never a 500
dials = pad_to_dials(pad)
# agent_id / p / a are UNTRUSTED open-world body fields. A non-str agent_id (an
# unhashable list/dict) would TypeError on the voice-map .get(); a huge JSON int p/a
# would OverflowError on float(). Both degrade to a neutral read, never a 500
# (INV-TTS-4). PadState.from_obj already hardens the numeric parse (catches
# ArithmeticError/OverflowError) — route through it rather than re-implement a narrower net.
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
)
dials = pad_to_dials(PadState.from_obj({"pleasure": body.get("p"), "arousal": body.get("a")}))
tts_url = request.app.state.tts_url
lock = request.app.state.tts_lock
# DEC-5: one stream at a time on the shared 3090. Held for the stream's duration and
# released in the generator's finally — including the browser-abort/cancel path.
# DEC-5: one stream at a time on the shared 3090. Held for the stream's duration
# (all chunks) and released in the generator's finally — including the browser-abort
# path. read=120s per infra-ops: a single near-cap chunk can render slowly under load.
await lock.acquire()
client = httpx.AsyncClient(
timeout=httpx.Timeout(connect=10.0, read=30.0, write=10.0, pool=10.0)
timeout=httpx.Timeout(connect=10.0, read=120.0, write=10.0, pool=10.0)
)
gen = tts_stream(text, voice=voice, dials=dials, client=client, url=tts_url)
gen = tts_stream_long(text, voice=voice, dials=dials, client=client, url=tts_url)
async def _release() -> None:
await gen.aclose() # unwinds tts_stream's `async with` → closes the gateway resp
await client.aclose()
lock.release()
try:
# Peek the first chunk so a bad gateway OPEN (non-200 / transport) surfaces as a
# 503 BEFORE we commit a 200 StreamingResponse. TTFB ~0.44s, so this is cheap.
first = await gen.__anext__()
except TtsUnavailable:
await gen.aclose()
await client.aclose()
lock.release()
await _release()
return JSONResponse({"error_code": "tts_unavailable"}, status_code=503)
except StopAsyncIteration:
first = None
except BaseException:
# Any OTHER escape during the peek — a browser abort (CancelledError) before the
# first byte, or an unmapped transport error (e.g. httpx.InvalidURL from a
# misconfigured tts_url, which is NOT an httpx.RequestError) — must still release
# the process-wide serialize lock + client, or every later /api/tts deadlocks at
# lock.acquire() (DEC-5). The finally below only exists once piped() runs. Propagate.
await _release()
raise
# Chunk 1 must be a WAV (RIFF header). A 200 non-WAV body — a misbehaving gateway or a
# proxy error page — would be mislabeled audio/wav and the browser would decode markup as
# PCM (or match a `data` byte-run in it). Reject → 503 (INV-TTS-4). Tolerant of a <4-byte
# first read so a valid WAV split oddly is never false-rejected.
if first is not None and len(first) >= 4 and not first.startswith(b"RIFF"):
await _release()
return JSONResponse({"error_code": "tts_unavailable"}, status_code=503)
async def piped() -> AsyncIterator[bytes]:
try:
@@ -618,9 +654,7 @@ async def _tts_endpoint(request: Request) -> Response:
async for chunk in gen:
yield chunk
finally:
await gen.aclose() # unwinds tts_stream's `async with` → closes the gateway resp
await client.aclose()
lock.release()
await _release()
return StreamingResponse(piped(), media_type="audio/wav")
@@ -836,7 +870,7 @@ def create_app(
Route("/api/sessions/{session_id}/messages", _session_messages_endpoint),
Route("/api/sessions/{session_id}/bifrost", _session_bifrost_endpoint),
Route("/api/admin/events", _admin_events_endpoint),
Route("/api/tts", _tts_endpoint, methods=["GET"]),
Route("/api/tts", _tts_endpoint, methods=["POST"]),
Route("/api/turns/{session_id}", _submit_turn_endpoint, methods=["POST"]),
Route("/api/turns/{session_id}/stream", _stream_turn_endpoint),
Route("/api/turns/{session_id}/cancel", _cancel_turn_endpoint, methods=["POST"]),
src/ratatoskr/web/static/index.html
+19 -7
View File
@@ -1995,7 +1995,7 @@ async function cancelTurn() {
})();
// ---- auto-TTS: voiced, affect-modulated STREAMING playback via Web Audio ------------
// Fetch the chunked GET /api/tts stream, decode its int16 PCM, and schedule the samples
// Fetch the chunked POST /api/tts stream, decode its int16 PCM, and schedule the samples
// GAPLESSLY into an AudioContext as they arrive (TTFA ~0.5s). Web Audio, NOT <audio src>,
// because Safari/WebKit REFUSES a streaming 0xFFFFFFFF-length WAV via <audio src>
// (NotSupportedError) — decoding the raw PCM ourselves sidesteps every WAV-container quirk
@@ -2042,15 +2042,22 @@ async function speakOnDone(text, agentId, pad) {
const ctx = _ttsAudioCtx();
if (!ctx) { tickerAdd("err", "tts", "no audio ctx"); return; }
if (ctx.state === "suspended") { try { await ctx.resume(); } catch (_) {} }
const params = new URLSearchParams({ text: clip.slice(0, 2000) }); // matches the server cap
if (agentId) params.set("agent_id", agentId);
const payload = { text: clip.slice(0, 8000) }; // matches the server cap; chunked server-side (DEC-10)
if (agentId) payload.agent_id = agentId;
if (pad && typeof pad.pleasure === "number" && typeof pad.arousal === "number") {
params.set("p", pad.pleasure); params.set("a", pad.arousal); // affect dials (DEC-7)
payload.p = pad.pleasure; payload.a = pad.arousal; // affect dials (DEC-7)
}
const ctrl = new AbortController(); _ttsAbort = ctrl;
let resp;
try { resp = await fetch("/api/tts?" + params.toString(), { signal: ctrl.signal }); }
catch (_) { return; } // aborted / network → silent skip (INV-TTS-4)
// POST (not GET) so a long turn rides the body, not a length-capped URL (DEC-10).
try {
resp = await fetch("/api/tts", {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify(payload),
signal: ctrl.signal,
});
} catch (_) { return; } // aborted / network → silent skip (INV-TTS-4)
if (!resp.ok || !resp.body) { tickerAdd("err", "tts", "unavailable " + resp.status); return; }
const reader = resp.body.getReader();
const SR = 44100;
@@ -2064,7 +2071,12 @@ async function speakOnDone(text, agentId, pad) {
if (!headerDone) { // skip the WAV header (up to + incl the data id/size)
acc = _u8concat(acc, bytes);
const di = _findDataChunk(acc);
if (di < 0 || di + 8 > acc.length) continue; // header spans chunks — keep accumulating
if (di < 0 || di + 8 > acc.length) {
// Bounded: a real header lands in the first read; if 64 KiB arrive with no `data`
// marker the stream is headerless/malformed — bail instead of buffering forever.
if (acc.length > 65536) { tickerAdd("err", "tts", "no WAV header"); break; }
continue; // header spans chunks — keep accumulating
}
bytes = acc.subarray(di + 8); headerDone = true; acc = null;
}
const u8 = _u8concat(carry, bytes); // prepend the odd-byte carry; fresh + aligned
tests/test_tts.py
+205
View File
@@ -12,12 +12,15 @@ import pytest
import respx
from ratatoskr.tts import (
_TTS_CHUNK_CHAR_BUDGET,
EmotionDials,
PadState,
TtsUnavailable,
chunk_text,
gateway_body,
pad_to_dials,
tts_stream,
tts_stream_long,
)
_URL = "http://tts.example/v1/audio/speech"
@@ -36,6 +39,21 @@ async def _drain(gen) -> bytes:
return out
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."""
def __init__(self, head: bytes) -> None:
self._head = head
async def __aiter__(self):
yield self._head
raise httpx.ReadError("mid-stream drop")
async def aclose(self) -> None:
pass
class TestPadState:
def test_from_obj_valid_mapping(self) -> None:
pad = PadState.from_obj({"pleasure": 0.5, "arousal": -0.2, "dominance": 0.1})
@@ -110,6 +128,7 @@ class TestGatewayBody:
b = gateway_body("hi", "donut", pad_to_dials(PadState(pleasure=0.5, arousal=0.2)))
assert b["input"] == "hi" and b["voice"] == "donut"
assert b["response_format"] == "wav" # DEC-3 — ALWAYS wav
assert b["language"] == "en-us" # DEC-9 — pin English conditioning
assert b["emotion_valence"] == pytest.approx(0.5)
def test_neutral_omits_emotion(self) -> None:
@@ -133,6 +152,7 @@ class TestTtsStream:
body = _json.loads(route.calls.last.request.content)
assert body["voice"] == "donut"
assert body["response_format"] == "wav"
assert body["language"] == "en-us" # DEC-9 — pin English conditioning
assert body["emotion_valence"] == pytest.approx(0.5)
@respx.mock
@@ -153,3 +173,188 @@ class TestTtsStream:
await _drain(tts_stream(
"hi", voice="Cora", dials=pad_to_dials(None), client=client, url=_URL
))
class TestChunkText:
"""chunk_text (DEC-10): paragraph-first greedy pack, sentence fallback for oversized
paragraphs, clause/word sub-split for oversized sentences; every chunk <= budget."""
def test_empty_and_whitespace_yield_no_chunks(self) -> None:
assert chunk_text("") == []
assert chunk_text(" \n\n \t ") == []
def test_short_text_is_one_chunk(self) -> None:
assert chunk_text("Hello, darling.", budget=100) == ["Hello, darling."]
def test_two_short_paragraphs_greedily_merge(self) -> None:
# Both fit in one budget -> one chunk, joined on the blank-line boundary.
out = chunk_text("First para.\n\nSecond para.", budget=100)
assert out == ["First para.\n\nSecond para."]
def test_paragraphs_split_on_blank_line_when_over_budget(self) -> None:
# Each paragraph fits alone but not together -> a seam on the paragraph boundary.
a, b = "A" * 30, "B" * 30
out = chunk_text(f"{a}\n\n{b}", budget=40)
assert out == [a, b]
def test_oversized_paragraph_falls_back_to_sentences(self) -> None:
para = "One sentence here. Two sentence here. Three sentence here."
out = chunk_text(para, budget=25)
assert all(len(c) <= 25 for c in out)
assert len(out) >= 2
# every word is preserved whole and in order (no split mid-word)
assert [w for c in out for w in c.split()] == para.split()
def test_oversized_sentence_sub_splits_never_mid_word(self) -> None:
sent = "alpha beta gamma delta epsilon zeta eta theta iota kappa lambda"
out = chunk_text(sent, budget=20)
assert all(len(c) <= 20 for c in out)
for c in out:
for word in c.split():
assert word in sent.split() # every emitted token is a whole source word
def test_every_chunk_within_budget_default(self) -> None:
para = ("Princess Donut does not wait. " * 200).strip()
out = chunk_text(para) # default budget
assert out and all(len(c) <= _TTS_CHUNK_CHAR_BUDGET for c in out)
def test_spaceless_over_budget_hard_cuts_as_last_resort(self) -> None:
out = chunk_text("x" * 50, budget=20)
assert all(len(c) <= 20 for c in out)
assert "".join(out) == "x" * 50
def test_non_positive_budget_does_not_hang(self) -> None:
# budget <= 0 would infinite-loop _hard_wrap; it's clamped to 1 so this terminates.
out = chunk_text("alpha beta", budget=0)
assert out and all(len(c) <= 1 for c in out)
assert "".join(out) == "alphabeta" # every char preserved, forward progress made
def test_oversized_sentence_prefers_clause_boundary_over_space(self) -> None:
# A comma-bearing over-budget sentence sub-splits at the CLAUSE boundary (", "),
# not merely at the last space — pins the _CLAUSE_BOUNDARIES preference (else dead).
out = chunk_text("alpha, beta gamma delta", budget=12)
assert all(len(c) <= 12 for c in out)
assert out[0] == "alpha," # clause cut, not "alpha, beta" (a space-only cut)
def test_default_budget_is_the_dec10_value(self) -> None:
# Pin the concrete 747 that FN chunk_text's POST commits to (75% of 71.2s @ 14 c/s).
# The suite's other budget checks compare against the imported constant and so move
# with it; this one anchors the value itself so a retune is a deliberate edit here.
assert _TTS_CHUNK_CHAR_BUDGET == 747
class TestTtsStreamLong:
"""tts_stream_long (DEC-10): concatenate per-chunk synthesis into ONE int16-PCM stream
— chunk 1 verbatim (header + PCM), chunks 2..N header-stripped."""
_PCM = b"\x11\x22" * 64
_WAV_CHUNK = (
b"RIFF\xff\xff\xff\xffWAVEfmt \x10\x00\x00\x00" + b"\x00" * 20
+ b"data\xff\xff\xff\xff" + _PCM
)
@respx.mock
async def test_single_chunk_passes_through_verbatim(self) -> None:
respx.post(_URL).mock(return_value=httpx.Response(200, content=self._WAV_CHUNK))
async with httpx.AsyncClient() as client:
out = await _drain(tts_stream_long(
"Short line.", voice="donut", dials=pad_to_dials(None),
client=client, url=_URL, budget=100,
))
assert out == self._WAV_CHUNK # one chunk => untouched
@respx.mock
async def test_multi_chunk_emits_one_header_then_concatenated_pcm(self) -> None:
respx.post(_URL).mock(return_value=httpx.Response(200, content=self._WAV_CHUNK))
text = "First part here. Second part here. Third part here." # budget 18 -> >=2 chunks
async with httpx.AsyncClient() as client:
out = await _drain(tts_stream_long(
text, voice="donut", dials=pad_to_dials(None), client=client, url=_URL, budget=18,
))
n = len(chunk_text(text, budget=18))
assert n >= 2
assert out.count(b"RIFF") == 1 and out.count(b"data") == 1 # exactly one header
# EXACT bytes: chunk 1 verbatim (header+PCM), chunks 2..N stripped to PCM. Asserting
# the exact stream catches a di+4-vs-di+8 strip off-by-one (2-byte sample alignment
# across seams) that a header-count check alone would miss.
assert out == self._WAV_CHUNK + self._PCM * (n - 1)
# DEC-10: identical voice+dials+language on EVERY chunk (uniform delivery across seams).
import json as _json
bodies = [_json.loads(c.request.content) for c in respx.calls]
assert len(bodies) == n
assert all(b["voice"] == "donut" and b["language"] == "en-us" for b in bodies)
@respx.mock
async def test_mid_stream_drop_after_first_byte_degrades_not_raises(self) -> None:
# A2: chunk 0 opens 200, yields bytes, then drops mid-stream. Because the 200 is
# committed (bytes already flowed), this must DEGRADE (return what streamed), never
# raise — the pivot is yielded_any, not the chunk index.
respx.post(_URL).mock(
return_value=httpx.Response(200, stream=_RaisingByteStream(self._WAV_CHUNK))
)
async with httpx.AsyncClient() as client:
out = await _drain(tts_stream_long(
"hi", voice="donut", dials=pad_to_dials(None), client=client, url=_URL, budget=100,
))
assert out == self._WAV_CHUNK # head kept, no raise
@respx.mock
async def test_later_chunk_gateway_500_degrades_keeps_prior(self) -> None:
# chunk 1 = valid WAV; chunk 2 = a gateway 500 (OPEN failure on a later chunk).
respx.post(_URL).mock(side_effect=[
httpx.Response(200, content=self._WAV_CHUNK),
httpx.Response(500, content=b"boom"),
])
text = "First part here. Second part here." # budget 18 -> 2 chunks
async with httpx.AsyncClient() as client:
out = await _drain(tts_stream_long(
text, voice="donut", dials=pad_to_dials(None), client=client, url=_URL, budget=18,
))
assert out == self._WAV_CHUNK # INV-TTS-4 degrade: keep chunk 1, drop the tail, no raise
@respx.mock
async def test_later_chunk_missing_data_degrades_keeps_prior(self) -> None:
# chunk 1 = valid WAV; chunk 2 = a 200 non-WAV body (no `data` chunk) -> degrade.
respx.post(_URL).mock(side_effect=[
httpx.Response(200, content=self._WAV_CHUNK),
httpx.Response(200, content=b"xxxxx no marker present xxxxx"),
])
text = "First part here. Second part here." # budget 18 -> 2 chunks
async with httpx.AsyncClient() as client:
out = await _drain(tts_stream_long(
text, voice="donut", dials=pad_to_dials(None), client=client, url=_URL, budget=18,
))
# INV-TTS-4 degrade: chunk 1 audio retained verbatim, chunk 2 dropped (no raise, no
# garbage bytes emitted from the malformed body).
assert out == self._WAV_CHUNK
@respx.mock
async def test_first_chunk_gateway_failure_raises(self) -> None:
respx.post(_URL).mock(return_value=httpx.Response(500, content=b"boom"))
async with httpx.AsyncClient() as client:
with pytest.raises(TtsUnavailable):
await _drain(tts_stream_long(
"hi", voice="Cora", dials=pad_to_dials(None),
client=client, url=_URL, budget=100,
))
async def test_pcm_after_header_reassembles_data_marker_across_reads(self) -> None:
# The `data` marker can straddle two network reads; _pcm_after_header must accumulate
# until it lands, then yield only the PCM after it. Pins the docstring's straddle claim.
from ratatoskr.tts import _pcm_after_header
async def _split_stream():
yield b"RIFF\xff\xff\xff\xffWAVEfmt \x10\x00\x00\x00" + b"\x00" * 20 + b"da"
yield b"ta\xff\xff\xff\xff" + b"\x11\x22" * 4 # rest of 'data' + size + PCM
out = await _drain(_pcm_after_header(_split_stream()))
assert out == b"\x11\x22" * 4 # PCM only; marker reassembled across the read boundary
async def test_pcm_after_header_no_data_marker_raises(self) -> None:
from ratatoskr.tts import _pcm_after_header
async def _no_marker():
yield b"xxxxx no marker present xxxxx"
with pytest.raises(TtsUnavailable):
await _drain(_pcm_after_header(_no_marker()))
tests/test_web_server.py
+119 -25
View File
@@ -1458,10 +1458,11 @@ class TestMemoryChunksEndpoint:
class TestTtsEndpoint:
"""tts_endpoint FN — GET /api/tts → audio/wav STREAMED (chunked) from the Zonos
gateway. Voice per-character (DEC-8), emotion dials from p/a query floats (DEC-7),
"""tts_endpoint FN — POST /api/tts → audio/wav STREAMED (chunked) from the Zonos
gateway. Voice per-character (DEC-8), emotion dials from p/a body floats (DEC-7),
the gateway host never reaches the browser (INV-TTS-1), gateway open-failure → 503
(INV-TTS-4). GET so a browser <audio src> plays it progressively."""
(INV-TTS-4). POST so an arbitrarily long turn rides the body; the server chunk-and-
concatenates it under the model's 71.2s/call cap (DEC-10)."""
# Streaming WAV bytes (placeholder 0xFFFFFFFF sizes) — proxied through verbatim.
_WAV = b"RIFF\xff\xff\xff\xffWAVEdata\xff\xff\xff\xff" + b"\x11\x22" * 64
@@ -1475,17 +1476,18 @@ class TestTtsEndpoint:
return_value=httpx.Response(200, content=self._WAV)
)
app = create_app(_mock_client_factory(), tts_url=self._TTS)
resp = TestClient(app).get(
resp = TestClient(app).post(
"/api/tts",
params={"text": "Carl is a softie.", "agent_id": "ratatoskr:donut",
"p": "0.6", "a": "0.3"},
json={"text": "Carl is a softie.", "agent_id": "ratatoskr:donut",
"p": 0.6, "a": 0.3},
)
assert resp.status_code == 200
assert resp.headers["content-type"].startswith("audio/wav")
assert resp.content == self._WAV # streamed through verbatim, no header rewrite
assert resp.content == self._WAV # one chunk → streamed through verbatim
body = json.loads(route.calls.last.request.content)
assert body["voice"] == "donut"
assert body["response_format"] == "wav"
assert body["language"] == "en-us" # DEC-9 pinned per chunk
assert body["emotion_valence"] == pytest.approx(0.6)
@respx.mock
@@ -1496,7 +1498,7 @@ class TestTtsEndpoint:
return_value=httpx.Response(200, content=self._WAV)
)
app = create_app(_mock_client_factory(), tts_url=self._TTS)
resp = TestClient(app).get("/api/tts", params={"text": "hello", "agent_id": "mimir"})
resp = TestClient(app).post("/api/tts", json={"text": "hello", "agent_id": "mimir"})
assert resp.status_code == 200
body = json.loads(route.calls.last.request.content)
assert body["voice"] == "Cora" # gateway default (DEC-8)
@@ -1506,7 +1508,7 @@ class TestTtsEndpoint:
from ratatoskr.web.server import create_app
app = create_app(_mock_client_factory(), tts_url=self._TTS)
resp = TestClient(app).get("/api/tts", params={"agent_id": "ratatoskr:donut"})
resp = TestClient(app).post("/api/tts", json={"agent_id": "ratatoskr:donut"})
assert resp.status_code == 400
assert resp.json()["error_code"] == "missing_text"
@@ -1516,32 +1518,124 @@ class TestTtsEndpoint:
respx.post(self._TTS).mock(return_value=httpx.Response(500, content=b"boom"))
app = create_app(_mock_client_factory(), tts_url=self._TTS)
resp = TestClient(app).get("/api/tts", params={"text": "hi"})
resp = TestClient(app).post("/api/tts", json={"text": "hi"})
assert resp.status_code == 503
assert resp.json()["error_code"] == "tts_unavailable"
@respx.mock
def test_long_text_truncated_at_cap(self) -> None:
from ratatoskr.web.server import _TTS_MAX_TEXT_CHARS, 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).get("/api/tts", params={"text": "word " * 1000})
assert resp.status_code == 200
body = json.loads(route.calls.last.request.content)
assert len(body["input"]) <= _TTS_MAX_TEXT_CHARS # truncated for the URL / lock
@respx.mock
def test_malformed_pad_query_degrades_no_500(self) -> None:
def test_long_text_is_chunked_each_call_within_budget(self) -> None:
from ratatoskr.tts import _TTS_CHUNK_CHAR_BUDGET
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).get("/api/tts", params={"text": "hi", "p": "notafloat", "a": "0.1"})
# ~5000 chars of distinct sentences → several gateway calls (DEC-10 chunking).
text = " ".join(f"Sentence number {i} about the dungeon." for i in range(200))
resp = TestClient(app).post("/api/tts", json={"text": text})
assert resp.status_code == 200
assert len(route.calls) >= 2 # chunk-and-concatenate, not one giant call
for call in route.calls:
body = json.loads(call.request.content)
assert len(body["input"]) <= _TTS_CHUNK_CHAR_BUDGET # each chunk under cap
@respx.mock
def test_malformed_pad_body_degrades_no_500(self) -> None:
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", "p": "notafloat", "a": 0.1})
assert resp.status_code == 200
body = json.loads(route.calls.last.request.content)
assert "emotion_valence" not in body # malformed p → neutral read, not a 500
@respx.mock
def test_huge_int_pad_degrades_no_500(self) -> None:
# A 400-digit JSON int → float() OverflowError (an ArithmeticError, not a
# ValueError). POST re-opened this path that GET's string query params couldn't.
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", "p": 10**400, "a": 0.1})
assert resp.status_code == 200
body = json.loads(route.calls.last.request.content)
assert "emotion_valence" not in body # OverflowError → neutral read, not a 500
@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().
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": ["donut"]})
assert resp.status_code == 200
body = json.loads(route.calls.last.request.content)
assert body["voice"] == "Cora" # non-str agent_id → default voice, not a 500
def test_whitespace_text_returns_400(self) -> None:
from ratatoskr.web.server import create_app
app = create_app(_mock_client_factory(), tts_url=self._TTS)
resp = TestClient(app).post("/api/tts", json={"text": " \n\t "})
assert resp.status_code == 400
assert resp.json()["error_code"] == "missing_text"
def test_lock_released_when_peek_raises_unexpectedly(self, monkeypatch) -> None:
# An unmapped error during the first-byte peek (e.g. httpx.InvalidURL, or a
# browser abort/CancelledError before byte 1) must still release the process-wide
# serialize lock, or every later /api/tts deadlocks at lock.acquire() (DEC-5).
from ratatoskr.web import server as _server
from ratatoskr.web.server import create_app
async def _boom(*a, **k):
raise RuntimeError("boom during peek")
yield b"" # unreachable — marks this an async generator
monkeypatch.setattr(_server, "tts_stream_long", _boom)
app = create_app(_mock_client_factory(), tts_url=self._TTS)
with pytest.raises(RuntimeError):
TestClient(app).post("/api/tts", json={"text": "hi"})
assert app.state.tts_lock.locked() is False
@respx.mock
def test_lone_surrogate_text_scrubbed_no_500(self) -> None:
# A lone surrogate rides JSON (\udXXX) but would UnicodeEncodeError when httpx
# serializes the gateway body → a 500. It's scrubbed at the text guard instead.
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)
# Send the surrogate as an escaped JSON literal — TestClient's json= can't encode it.
resp = TestClient(app).post(
"/api/tts", content='{"text": "\\ud800hello"}',
headers={"Content-Type": "application/json"},
)
assert resp.status_code == 200
body = json.loads(route.calls.last.request.content)
assert body["input"] == "hello" # surrogate dropped, rest intact — not a 500
@respx.mock
def test_non_wav_200_body_rejected_503(self) -> None:
# A 200 whose first bytes are not a RIFF header (a proxy error page) must not be
# mislabeled audio/wav — the peek RIFF-sniff rejects it as 503.
from ratatoskr.web.server import create_app
respx.post(self._TTS).mock(
return_value=httpx.Response(200, content=b"<html>error: data unavailable</html>")
)
app = create_app(_mock_client_factory(), tts_url=self._TTS)
resp = TestClient(app).post("/api/tts", json={"text": "hi"})
assert resp.status_code == 503
assert resp.json()["error_code"] == "tts_unavailable"