perf(tts): pipeline sync per-sentence synthesis with playback

The universal sync fallback in stream_tts_to_speaker ran strictly serially
per sentence — synthesize, play, and only then start synthesizing the next
sentence — so every sentence boundary added a full synthesis-time of dead
air. Chunked streamers (elevenlabs/openai/gemini/xai) already avoid this;
every other provider (edge, piper, plugin providers) paid it on each reply
in voice mode and the wake-word loop.

_SyncSentencePipeline overlaps the two: one single-threaded synthesis
worker (sentences stay FIFO; providers never see concurrent calls from
this loop — same effective concurrency as before) feeds one playback
worker through a small bounded queue, so sentence n+1 synthesizes while
sentence n plays. Lookahead is bounded (backpressure + at most a couple of
temp files), stop_event short-circuits both stages, synthesis failures are
isolated per sentence, temp files are always unlinked, and the finally
block flushes the pipeline BEFORE tts_done_event fires so continuous voice
mode never reopens the mic over its own voice. synthesize/play are
resolved late so existing monkeypatch-based tests work unchanged.

Measured with a real local model provider (OmniVoice plugin, Apple
Silicon), same 3-sentence reply, playback simulated at the produced clips'
true durations, best-of-2 interleaved runs under identical load:

                     serial   pipelined
  time to first word  10.8s        4.4s
  mid-reply dead air  11.2s        1.8s   (second gap: 0.03s)
  full reply wall     33.2s       17.0s

Tests: 4 new (timestamp-proven overlap, order + per-sentence failure
isolation, stop skips queued playback, temp-file hygiene); the existing
sync-fallback and display-callback tests pass unchanged.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Mahdi Hedhli 2026-07-29 14:38:24 -04:00 committed by kshitij
parent dd600d1ace
commit 75901a295d
2 changed files with 235 additions and 26 deletions

View File

@ -6,8 +6,11 @@ synth path are all mocked. Covers the registry/resolver, provider availability,
the chunked-streamer playback path, and the universal per-sentence sync fallback.
"""
import os
import queue
import tempfile
import threading
import time
from unittest.mock import MagicMock, patch
import pytest
@ -792,3 +795,123 @@ def test_display_callback_not_called_when_streaming_enabled(monkeypatch):
assert done.is_set()
# No assertion on display — the point is no crash and done is set.
# ── Sync fallback: one-ahead synthesis/playback pipeline ─────────────────
#
# The universal per-sentence sync path pipelines synthesis with playback:
# while sentence n plays, sentence n+1 is already synthesizing. For local
# model providers (RTF near 1) the serial path spent as long silent between
# sentences as speaking; these pin the overlap, ordering, stop, failure
# isolation, and temp-file hygiene of the pipelined path.
def _timed_sync_run(monkeypatch, sentences, *, synth_s=0.12, play_s=0.12,
synth_fail_on=None, stop_after_plays=None):
"""Drive stream_tts_to_speaker over the sync path with timed fakes.
Returns (events, stop, done): events is [(kind, sentence, t_start, t_end)]
with kinds "synth"/"play", timestamps from a shared monotonic origin.
"""
from tools import tts_tool
origin = time.monotonic()
events = []
lock = threading.Lock()
stop, done = threading.Event(), threading.Event()
def fake_synth(text, output_path):
t0 = time.monotonic() - origin
if synth_fail_on and synth_fail_on in text:
raise RuntimeError("synth exploded")
time.sleep(synth_s)
with open(output_path, "wb") as fh:
fh.write(b"x" * 100)
with lock:
events.append(("synth", text, t0, time.monotonic() - origin))
def fake_play(path):
t0 = time.monotonic() - origin
time.sleep(play_s)
with lock:
events.append(("play", path, t0, time.monotonic() - origin))
plays = sum(1 for e in events if e[0] == "play")
if stop_after_plays is not None and plays >= stop_after_plays:
stop.set()
monkeypatch.setattr(tts_tool, "text_to_speech_tool", fake_synth)
fake_vm = MagicMock()
fake_vm.play_audio_file.side_effect = fake_play
monkeypatch.setitem(__import__("sys").modules, "tools.voice_mode", fake_vm)
q = _drain_queue(sentences)
with patch("tools.tts_streaming.resolve_streaming_provider", return_value=None):
tts_tool.stream_tts_to_speaker(q, stop, done)
return events, stop, done
def test_sync_pipeline_overlaps_synthesis_with_playback(monkeypatch):
sentences = ["First full sentence here. ", "Second full sentence here. ",
"Third full sentence here. "]
events, _stop, done = _timed_sync_run(monkeypatch, sentences)
synths = [e for e in events if e[0] == "synth"]
plays = [e for e in events if e[0] == "play"]
assert len(synths) == 3 and len(plays) == 3
assert done.is_set()
# The point of the pipeline: sentence 2's synthesis STARTS before
# sentence 1's playback ENDS (serial code could never do this).
synth2_start = synths[1][2]
play1_end = plays[0][3]
assert synth2_start < play1_end, (
f"no overlap: synth2 started at {synth2_start:.3f}, "
f"play1 ended at {play1_end:.3f}"
)
def test_sync_pipeline_preserves_order_and_isolates_failures(monkeypatch):
sentences = ["Alpha sentence spoken first. ", "Bravo sentence explodes here. ",
"Charlie sentence still plays. "]
events, _stop, done = _timed_sync_run(monkeypatch, sentences,
synth_fail_on="Bravo")
synths = [e[1] for e in events if e[0] == "synth"]
plays = [e for e in events if e[0] == "play"]
# Bravo's synth raised: never synthesized-to-file, never played — but
# Alpha and Charlie both played, in submission order.
assert [s.split()[0] for s in synths] == ["Alpha", "Charlie"]
assert len(plays) == 2
assert done.is_set()
def test_sync_pipeline_stop_skips_queued_playback(monkeypatch):
sentences = ["First full sentence here. ", "Second full sentence here. ",
"Third full sentence here. ", "Fourth full sentence here. "]
events, stop, done = _timed_sync_run(monkeypatch, sentences,
stop_after_plays=1)
plays = [e for e in events if e[0] == "play"]
assert len(plays) == 1, f"stop after first play must skip the rest, got {len(plays)}"
assert stop.is_set() and done.is_set()
def test_sync_pipeline_cleans_temp_files(monkeypatch):
from tools import tts_tool
created = []
real_mkstemp = tempfile.mkstemp
def tracking_mkstemp(*a, **k):
fd, path = real_mkstemp(*a, **k)
created.append(path)
return fd, path
monkeypatch.setattr(tts_tool.tempfile, "mkstemp", tracking_mkstemp)
events, _stop, done = _timed_sync_run(monkeypatch,
["First full sentence here. ",
"Second full sentence here. "])
assert len([e for e in events if e[0] == "play"]) == 2
assert created, "expected temp files to be created via mkstemp"
leftovers = [p for p in created if os.path.exists(p)]
assert not leftovers, f"temp files not cleaned: {leftovers}"

View File

@ -50,6 +50,7 @@ import tempfile
import threading
import time
import uuid
from concurrent.futures import Future, ThreadPoolExecutor
from dataclasses import dataclass, field
from pathlib import Path
from typing import Callable, Dict, Any, Iterator, Optional
@ -3348,6 +3349,98 @@ def _strip_markdown_for_tts(text: str) -> str:
return text.strip()
class _SyncSentencePipeline:
"""Overlap per-sentence synthesis with playback for non-streaming providers.
The universal sync fallback used to run strictly serially per sentence
synthesize, play, and only then start synthesizing the next sentence so
every sentence boundary added a full synthesis-time of dead air. For local
model providers that cost dominates the conversation: a provider at
real-time-factor ~1 spends as long silent between sentences as it does
speaking. Chunked streamers already avoid this; this closes the same gap
for everyone else (edge, piper, plugin providers, ) without touching the
provider contract.
Shape: one synthesis worker (single-threaded executor, so sentences are
synthesized FIFO and providers never see concurrent calls from this loop
same effective concurrency as the serial path) feeding one playback worker
through a small bounded queue. While sentence *n* plays, sentence *n+1* is
already synthesizing. The bound keeps lookahead and the temp files it
implies small, and gives natural backpressure to the caller.
``synthesize``/``play`` are resolved late (module global / import inside
the worker) so tests that monkeypatch ``text_to_speech_tool`` or
``tools.voice_mode`` keep working unchanged.
"""
def __init__(self, stop_event: threading.Event, *, lookahead: int = 2):
self._stop = stop_event
self._queue: "queue.Queue[Optional[tuple[str, Future]]]" = queue.Queue(
maxsize=max(1, lookahead)
)
self._executor = ThreadPoolExecutor(
max_workers=1, thread_name_prefix="tts-sync-synth"
)
self._player = threading.Thread(
target=self._drain, name="tts-sync-play", daemon=True
)
self._player.start()
def speak(self, cleaned: str) -> None:
"""Queue one sentence. Blocks only when the lookahead bound is full."""
if self._stop.is_set():
return
future = self._executor.submit(self._synthesize_to_tmp, cleaned)
self._queue.put((cleaned, future))
def close(self) -> None:
"""Flush queued sentences in order (skipped if stopped), then join."""
self._queue.put(None)
self._player.join()
self._executor.shutdown(wait=True)
def _synthesize_to_tmp(self, cleaned: str) -> Optional[str]:
if self._stop.is_set():
return None
tmp_path = None
try:
fd, tmp_path = tempfile.mkstemp(suffix=".mp3")
os.close(fd)
text_to_speech_tool(text=cleaned, output_path=tmp_path)
return tmp_path
except Exception as exc:
logger.warning("Sync per-sentence TTS synthesis failed: %s", exc)
if tmp_path:
try:
os.unlink(tmp_path)
except OSError:
pass
return None
def _drain(self) -> None:
while True:
item = self._queue.get()
if item is None:
return
_sentence, future = item
tmp_path = None
try:
tmp_path = future.result()
if (tmp_path and not self._stop.is_set()
and os.path.isfile(tmp_path)
and os.path.getsize(tmp_path) > 0):
from tools.voice_mode import play_audio_file
play_audio_file(tmp_path)
except Exception as exc:
logger.warning("Sync per-sentence TTS failed: %s", exc)
finally:
if tmp_path:
try:
os.unlink(tmp_path)
except OSError:
pass
def stream_tts_to_speaker(
text_queue: queue.Queue,
stop_event: threading.Event,
@ -3372,6 +3465,7 @@ def stream_tts_to_speaker(
waiting on it (continuous voice mode) know playback is finished.
"""
tts_done_event.clear()
sync_pipeline: Optional[_SyncSentencePipeline] = None
try:
output_stream = None
@ -3386,6 +3480,11 @@ def stream_tts_to_speaker(
from tools.tts_streaming import SentenceChunker, resolve_streaming_provider
streamer = resolve_streaming_provider(tts_config, preferred=provider)
# No chunked streamer: per-sentence sync synthesis, pipelined so the
# next sentence synthesizes while the current one plays (closed in the
# finally block, which flushes anything still queued).
sync_pipeline = _SyncSentencePipeline(stop_event) if streamer is None else None
stream_max_len = 0
if streamer is not None:
try:
@ -3640,9 +3739,11 @@ def stream_tts_to_speaker(
# Display raw sentence on screen before TTS processing
if display_callback is not None:
display_callback(sentence)
# No chunked streamer → per-sentence sync synthesis (universal).
if streamer is None:
_speak_via_sync(cleaned)
# No chunked streamer → per-sentence sync synthesis (universal),
# pipelined: this enqueues and returns, so sentence n+1 is already
# synthesizing while sentence n is still playing.
if sync_pipeline is not None:
sync_pipeline.speak(cleaned)
return
# Truncate very long sentences to the provider's per-request cap.
if stream_max_len and len(cleaned) > stream_max_len:
@ -3652,29 +3753,6 @@ def stream_tts_to_speaker(
# sentence N+1 is already buffering while sentence N plays.
_enqueue_audio(cleaned)
def _speak_via_sync(cleaned: str):
"""Synthesize one sentence via the proven sync tool, then block on
playback. No chunked API, but per-*sentence* granularity keeps the
flow conversational for edge and every other non-streaming provider.
"""
tmp_path = None
try:
fd, tmp_path = tempfile.mkstemp(suffix=".mp3")
os.close(fd)
text_to_speech_tool(text=cleaned, output_path=tmp_path)
if (not stop_event.is_set() and os.path.isfile(tmp_path)
and os.path.getsize(tmp_path) > 0):
from tools.voice_mode import play_audio_file
play_audio_file(tmp_path)
except Exception as exc:
logger.warning("Sync per-sentence TTS failed: %s", exc)
finally:
if tmp_path:
try:
os.unlink(tmp_path)
except OSError:
pass
def _align_int16_chunks(chunks, stop_evt):
"""Yield int16-aligned byte chunks from an iterable."""
leftover = b""
@ -3758,6 +3836,14 @@ def stream_tts_to_speaker(
except Exception as exc:
logger.warning("Streaming TTS pipeline error: %s", exc)
finally:
# Flush the sync pipeline first: queued sentences finish playing (or
# are skipped when stop_event is set) BEFORE tts_done_event fires, so
# continuous voice mode never reopens the mic over its own voice.
if sync_pipeline is not None:
try:
sync_pipeline.close()
except Exception:
pass
# Signal the playback worker that no more audio is coming. This lives
# in finally: so an exception in the text pump still sends the sentinel.
if streamer is not None and _worker_thread is not None: