diff --git a/tests/tools/test_tts_streaming.py b/tests/tools/test_tts_streaming.py index ed9da09a074cd..79ee2c8441609 100644 --- a/tests/tools/test_tts_streaming.py +++ b/tests/tools/test_tts_streaming.py @@ -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}" diff --git a/tools/tts_tool.py b/tools/tts_tool.py index 67a0f48851f8f..341d94c643b2d 100644 --- a/tools/tts_tool.py +++ b/tools/tts_tool.py @@ -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: