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

tests/tools/test_tts_streaming.py

@@ -6,8 +6,11 @@
 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
@@ -221,3 +224,123 @@ def _endless():
     out = list(ts._capped(_endless(), "test"))
     assert len(out) == 1  # 64 ok, 128 > cap → stop
     assert sum(len(c) for c in out) <= 100
+
+
+# ── 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}"

tools/tts_tool.py

@@ -50,6 +50,7 @@
 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, 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
@@ -3382,6 +3476,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:
@@ -3434,9 +3533,11 @@ def _speak_sentence(sentence: str):
             # 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:
@@ -3469,29 +3570,6 @@ def mark_audio_output_active(_active):
             except Exception as exc:
                 logger.warning("Streaming TTS sentence failed: %s", exc)
 
-        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 _play_via_tempfile(audio_iter, stop_evt, sample_rate=24000):
             """Write PCM chunks to a temp WAV file and play it."""
             tmp = None
@@ -3563,6 +3641,14 @@ def _play_via_tempfile(audio_iter, stop_evt, sample_rate=24000):
     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
         # Always close the audio output stream to avoid locking the device
         if output_stream is not None:
             try: