Header Logo

How to Add Reliability Tests for TTS Chunking and Streaming in WebSocket Avatar Apps

How to Add Reliability Tests for TTS Chunking and Streaming in WebSocket Avatar Apps

Test TTS chunking and WebSocket audio streaming with deterministic unit, transport, and timing-based CI reliability checks.

Introduction


TTS chunking is one of those details that only shows up when it breaks. In a realtime avatar app, your model may emit text in partial sentences, your TTS service may stream audio as it synthesizes, and your WebSocket or WebRTC pipeline may forward that audio to a video face that has to stay lip-synced. If the chunk boundaries are wrong, you get awkward pauses, duplicated words, truncated phrases, or audio that arrives after the avatar has already moved on.


This post is about testing that entire path, not just the happy path. By the end, you should be able to build reliable tests for:


  • text chunking before TTS synthesis,

  • streaming audio delivery over a socketed realtime pipeline, and

  • the timing and ordering invariants that keep an avatar speaking naturally.


We’ll focus on practical tests you can run locally and in CI, with a few failure modes that are easy to miss until production.


What “reliability” means in a streamed avatar pipeline


In a typical voice-agent avatar app, there are at least three asynchronous boundaries:


  1. The language model emits partial text incrementally.

  2. The TTS layer consumes those chunks and produces audio frames or segments.

  3. The avatar renderer consumes audio and video timing cues to produce synchronized playback.


Each boundary introduces state. That state can drift unless you test for explicit contracts.


The core contracts worth testing are:


  • Chunk completeness: no word should be dropped or duplicated when a stream is split.

  • Boundary correctness: punctuation, sentence ends, and mid-word splits should not create invalid TTS requests.

  • Ordering: audio chunks must be emitted and consumed in the same order they were synthesized.

  • Cancellation behavior: if the user interrupts, older chunks must stop cleanly and not leak into the new turn.

  • Latency budget: the system should stay responsive under realistic network jitter and model delays.


Most bugs in this area are not “the stream crashed”; they are “the stream still worked, but the spoken output was subtly wrong.” That is exactly the kind of thing you want tests to catch.


Test the chunker as a deterministic state machine


Start with the text chunker. Do not test this by calling a real TTS provider first. The chunker should be a deterministic state machine that ingests text deltas and yields safe synthesis units.


Good unit tests cover the obvious and the annoying cases:


  • plain sentence endings

  • abbreviations like “Dr.” or “e.g.”

  • numbers, URLs, and timestamps

  • quotes and parentheses

  • mid-word splits across deltas


A useful invariant is: if you concatenate all emitted chunks and normalize whitespace, you should recover the original text stream. That does not guarantee good prosody, but it does catch loss and duplication.


def test_chunker_never_drops_text():

assert reconstructed == "Hello world. How are you?"
def test_chunker_never_drops_text():

assert reconstructed == "Hello world. How are you?"
def test_chunker_never_drops_text():

assert reconstructed == "Hello world. How are you?"


That test is intentionally simple. Real chunkers often need a buffer and punctuation-aware flush rules, so add cases that force state transitions:


def test_chunker_handles_abbreviation():

assert chunks == ["We met Dr. Smith yesterday."]
def test_chunker_handles_abbreviation():

assert chunks == ["We met Dr. Smith yesterday."]
def test_chunker_handles_abbreviation():

assert chunks == ["We met Dr. Smith yesterday."]


For this layer, property-based testing is very effective. Generate random splits of a known sentence, feed them through the chunker, and assert that the normalized output matches the input. This catches edge cases you would never handwrite.


Test streaming audio like a transport, not like a string


Once you have a chunker, test the audio stream as a transport protocol. The relevant questions are not “did audio arrive?” but “did every chunk arrive exactly once, in order, and within the expected time window?”


When testing a WebSocket-based pipeline, simulate three classes of faults:


  1. Delayed delivery: frames arrive later than expected.

  2. Out-of-order sequencing: one chunk is held back and delivered after a later one.

  3. Disconnect/reconnect: the client or server drops mid-turn.


You do not need a full browser or a real microphone for this. A fake socket that records writes and replays them with controlled delays is enough to validate your sequencing logic.


async def test_audio_chunks_are_ordered(fake_socket):

assert [frame["seq"] for frame in frames] == [1, 2]
async def test_audio_chunks_are_ordered(fake_socket):

assert [frame["seq"] for frame in frames] == [1, 2]
async def test_audio_chunks_are_ordered(fake_socket):

assert [frame["seq"] for frame in frames] == [1, 2]


If your transport uses sequence numbers or turn IDs, assert them aggressively. In realtime voice systems, stale audio is often worse than missing audio because it sounds like the assistant is talking over the user or responding to the wrong prompt.


Also test cancellation. A new user utterance should invalidate the in-flight turn. The old audio may still be in a buffer, so your pipeline needs a hard check before playback or forwarding:


def test_cancellation_stops_old_turn():
assert not should_play(turn_id)
def test_cancellation_stops_old_turn():
assert not should_play(turn_id)
def test_cancellation_stops_old_turn():
assert not should_play(turn_id)


Build end-to-end tests around timing, not exact waveforms


Full waveform comparison is usually the wrong goal for streamed TTS. Minor synthesis differences, codec changes, and provider updates make exact audio snapshots brittle. Instead, validate timing and structure.


For a realistic end-to-end test, measure:


  • time from first text delta to first playable audio

  • time between chunk emission and audio availability

  • gap between consecutive audio segments

  • whether the session remains synchronized after interruption


If you have a staging environment, run a scripted conversation with a known prompt and compare the resulting event log, not the raw waveform. Event logs are much easier to stabilize across environments.


def test_turn_latency_within_budget(metrics):
assert metrics["orphan_audio_frames"] == 0
def test_turn_latency_within_budget(metrics):
assert metrics["orphan_audio_frames"] == 0
def test_turn_latency_within_budget(metrics):
assert metrics["orphan_audio_frames"] == 0


Those numbers are examples, not universal targets. Your budget depends on the TTS provider, codec, model, and whether you synthesize in the same region as your app. The important part is to track the same metrics over time and fail the build when they regress.


One subtle issue: if your chunking logic tries to optimize for lower latency by flushing too early, you may reduce the time to first audio but worsen prosody. Tests should encode the trade-off you actually want. For example, you may accept slightly higher first-audio latency if it prevents sentence fragments being synthesized separately.


Useful fixtures for CI


Reliable tests are much easier when you control time, randomness, and network behavior. A few fixtures go a long way:


  • Fake clock: avoid sleeping in tests; advance time explicitly.

  • Mock TTS backend: return predictable audio chunks keyed by input text.

  • Socket recorder: capture outbound frames and assert ordering/metadata.

  • Replay logs: feed recorded model deltas back into the chunker.


In integration tests, prefer “golden event traces” over “golden audio files.” A trace can capture chunk boundaries, sequence numbers, turn IDs, and timing events. That gives you a stable signal when something regresses without making the test suite fragile.


Another good practice is to test failure injection explicitly:


  • drop the last chunk of a sentence

  • duplicate a chunk

  • insert a 500 ms delay before the second audio frame

  • disconnect the socket during playback


If your system remains correct under those conditions, it is usually correct under normal conditions too.


How Protoface fits into this


This is exactly the kind of problem space where a developer platform like Protoface is useful: you can keep the avatar layer attached to your voice agent while you test the parts that actually matter for reliability, namely chunking, streaming, and turn handling. The LiveKit Agents plugin is a good fit if your app already uses LiveKit for realtime voice; the avatar becomes another participant in the pipeline, so your tests can focus on synchronized turn boundaries instead of video plumbing.


If you are using the Python ecosystem, the published plugin and SDK examples are a practical reference point. For the LiveKit path, see the plugin repo and examples at https://github.com/protoface-ai/protoface-plugin-pipecat and the Pipecat integration guide at https://docs.pipecat.ai/api-reference/server/services/video/protoface. For API and SDK details, check https://docs.protoface.com and the Python SDK repo at https://github.com/protoface-ai/protoface-sdk-python.


A minimal REST call is enough to create a session in a test harness; the exact fields depend on the docs, but the shape is familiar:


curl -X POST https://api.protoface.com/v1/sessions \
-d '{"avatar_id":"avt_123","voice":"default"}'
curl -X POST https://api.protoface.com/v1/sessions \
-d '{"avatar_id":"avt_123","voice":"default"}'
curl -X POST https://api.protoface.com/v1/sessions \
-d '{"avatar_id":"avt_123","voice":"default"}'


In practice, you can use that session in integration tests to verify that your chunking logic produces the right realtime behavior when an avatar is actually attached, without baking provider-specific details into your core text-stream tests.


Conclusion


If your avatar app streams text, synthesizes speech incrementally, and forwards audio over a realtime transport, then reliability lives at the seams. Test the chunker as a deterministic state machine, test the socket as an ordered transport, and test the end-to-end path with timing-based assertions rather than brittle waveform snapshots.


That combination catches the bugs users actually hear: duplicated phrases, missing words, awkward pauses, stale audio, and broken interruption handling.


If you want implementation details, integration references, or the current API shapes, start with docs.protoface.com. If you already have a voice agent, add a small reliability harness around your existing TTS path before you scale the avatar rollout. The sooner you make chunking and streaming measurable, the less likely you are to debug them in production.

Add a face to your AI.

No credit card needed.

Add a face to your AI.

No credit card needed.

Add a face to your AI.

No credit card needed.