feat(tts): pin English, stream long turns via chunking, dialogue-only Donut
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:
@@ -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()))
|
||||
|
||||
Reference in New Issue
Block a user