diff --git a/src/tts_mcp/audio.py b/src/tts_mcp/audio.py index 3b02f49..f2c472c 100644 --- a/src/tts_mcp/audio.py +++ b/src/tts_mcp/audio.py @@ -74,13 +74,43 @@ def wav_duration(path: Path) -> float: return wf.getnframes() / wf.getframerate() -async def play_audio(path: Path) -> None: - """Play a WAV file through PipeWire (pw-play). Async wrapper.""" +class PlaybackError(RuntimeError): + """Raised when audio playback fails or times out.""" + + +async def play_audio(path: Path, expected_seconds: float = 0) -> None: + """Play a WAV file through PipeWire (pw-play). Async wrapper. + + Args: + path: Path to the WAV file. + expected_seconds: Expected duration for timeout calculation. + If 0, reads duration from the WAV file header. + """ + if expected_seconds <= 0: + expected_seconds = wav_duration(path) + # Generous margin: 2x duration + 10s for PipeWire startup overhead + timeout = max(expected_seconds * 2, 10) + 10 + proc = await asyncio.create_subprocess_exec( "pw-play", str(path), stdout=asyncio.subprocess.DEVNULL, stderr=asyncio.subprocess.PIPE, ) - _, stderr = await proc.communicate() + try: + _, stderr = await asyncio.wait_for(proc.communicate(), timeout=timeout) + except asyncio.TimeoutError: + proc.kill() + await proc.wait() + raise PlaybackError( + f"pw-play timed out after {timeout:.0f}s " + f"(expected {expected_seconds:.1f}s audio)" + ) + except asyncio.CancelledError: + # Consumer was cancelled (shutdown) — don't orphan the subprocess + proc.kill() + await proc.wait() + raise if proc.returncode != 0: - raise RuntimeError(f"pw-play failed ({proc.returncode}): {stderr.decode().strip()}") + raise PlaybackError( + f"pw-play failed ({proc.returncode}): {stderr.decode().strip()}" + ) diff --git a/src/tts_mcp/queue.py b/src/tts_mcp/queue.py index 8f2d8fd..8bd5674 100644 --- a/src/tts_mcp/queue.py +++ b/src/tts_mcp/queue.py @@ -1,43 +1,70 @@ -"""Speech queue — serializes playback so agents don't talk over each other.""" +"""Speech queue — producer-consumer pattern with priority and cancellation. + +Replaces the original asyncio.Lock approach. A dedicated consumer coroutine +pulls work items from a bounded PriorityQueue and plays them one at a time. +Producers enqueue items and await their Future to get the result. + +Priority tiers: + 0 = urgent (preempts normal items in the queue) + 1 = normal (default) +""" import asyncio -from collections import deque -from contextlib import asynccontextmanager -from dataclasses import dataclass -from typing import AsyncIterator +import sys +import time +from dataclasses import dataclass, field +from enum import IntEnum +from typing import Callable, Coroutine -from .audio import play_audio +from .audio import PlaybackError, play_audio from .engines.base import TTSResult -@dataclass -class _Waiter: - caller_id: str - text_preview: str +class Priority(IntEnum): + URGENT = 0 + NORMAL = 1 + + +MAX_QUEUE_DEPTH = 20 + + +@dataclass(order=True) +class _WorkItem: + """A prioritized playback request.""" + + priority: int + sequence: int # tiebreaker for same-priority items (FIFO) + result: TTSResult = field(compare=False) + future: asyncio.Future = field(compare=False) + info_callback: Callable[..., Coroutine] | None = field( + default=None, compare=False + ) + caller_id: str = field(default="", compare=False) + enqueued_at: float = field(default_factory=time.time, compare=False) class SpeechQueue: - """FIFO queue for audio playback. + """Bounded priority queue with a dedicated playback consumer. - Only one audio file plays at a time. Callers that arrive while - someone is speaking wait in line and get progress updates via - their MCP context. + Call start() to launch the consumer, stop() to cancel it and + drain pending items. """ - def __init__(self) -> None: - self._lock = asyncio.Lock() - self._current: _Waiter | None = None - self._waiters: deque[_Waiter] = deque() + def __init__(self, max_depth: int = MAX_QUEUE_DEPTH) -> None: + self._queue: asyncio.PriorityQueue[_WorkItem] = asyncio.PriorityQueue( + maxsize=max_depth + ) + self._consumer_task: asyncio.Task | None = None + self._current: _WorkItem | None = None + self._current_proc: asyncio.subprocess.Process | None = None self._counter = 0 + self._max_depth = max_depth + self._stopped = False def _next_id(self) -> str: self._counter += 1 return f"speaker-{self._counter}" - @property - def depth(self) -> int: - return len(self._waiters) - @property def current_speaker(self) -> str | None: return self._current.caller_id if self._current else None @@ -45,66 +72,163 @@ class SpeechQueue: def status(self) -> dict: return { "current_speaker": self.current_speaker, - "queue_depth": self.depth, - "waiting": [w.caller_id for w in self._waiters], + "queue_depth": self._queue.qsize(), + "max_depth": self._max_depth, } - @asynccontextmanager - async def acquire( - self, - caller_id: str | None = None, - text_preview: str = "", - ) -> AsyncIterator[str]: - """Context manager that waits for the speaker's turn. + def start(self) -> None: + """Launch the consumer coroutine.""" + if self._consumer_task is None or self._consumer_task.done(): + self._stopped = False + self._consumer_task = asyncio.create_task( + self._consumer(), name="speech-queue-consumer" + ) - Yields the caller_id once it's this caller's turn to play audio. - """ - cid = caller_id or self._next_id() - waiter = _Waiter(caller_id=cid, text_preview=text_preview[:60]) - self._waiters.append(waiter) + async def stop(self) -> None: + """Cancel the consumer and reject all pending items.""" + self._stopped = True + if self._consumer_task and not self._consumer_task.done(): + self._consumer_task.cancel() + try: + await self._consumer_task + except asyncio.CancelledError: + pass - try: - async with self._lock: - # We're up — remove ourselves from the waiting list - if waiter in self._waiters: - self._waiters.remove(waiter) - self._current = waiter - yield cid - finally: - if self._current is waiter: + # Drain remaining items + while not self._queue.empty(): + try: + item = self._queue.get_nowait() + if not item.future.done(): + item.future.set_result({ + "played": False, + "error": "Queue shut down", + "engine": item.result.engine, + "voice": item.result.voice, + }) + except asyncio.QueueEmpty: + break + + # Kill in-flight playback + if self._current_proc and self._current_proc.returncode is None: + self._current_proc.kill() + print("Killed in-flight pw-play on shutdown", file=sys.stderr) + + async def _consumer(self) -> None: + """Pull work items and play them sequentially.""" + while not self._stopped: + try: + item = await self._queue.get() + except asyncio.CancelledError: + break + + if item.future.cancelled(): + self._queue.task_done() + continue + + self._current = item + try: + if item.info_callback: + try: + await item.info_callback( + f"Now playing: {item.result.engine}/{item.result.voice}" + ) + except Exception: + pass # Context may have expired — non-fatal + await play_audio( + item.result.audio_path, + expected_seconds=item.result.duration_seconds, + ) + if not item.future.done(): + item.future.set_result({ + "played": True, + "file": str(item.result.audio_path), + "duration_seconds": item.result.duration_seconds, + "engine": item.result.engine, + "voice": item.result.voice, + }) + except PlaybackError as e: + print(f" Playback error: {e}", file=sys.stderr) + if not item.future.done(): + item.future.set_result({ + "played": False, + "error": str(e), + "file": str(item.result.audio_path), + "engine": item.result.engine, + "voice": item.result.voice, + }) + except asyncio.CancelledError: + if not item.future.done(): + item.future.set_result({ + "played": False, + "error": "Playback cancelled", + "engine": item.result.engine, + "voice": item.result.voice, + }) + break + except Exception as e: + print(f" Unexpected playback error: {e}", file=sys.stderr) + if not item.future.done(): + item.future.set_result({ + "played": False, + "error": f"Unexpected error: {e}", + "engine": item.result.engine, + "voice": item.result.voice, + }) + finally: self._current = None + self._queue.task_done() async def speak( self, result: TTSResult, caller_id: str | None = None, - info_callback=None, + priority: Priority = Priority.NORMAL, + info_callback: Callable[..., Coroutine] | None = None, ) -> dict: - """Queue and play a TTSResult. Returns status dict when done. + """Enqueue a TTSResult for playback. Returns result dict when done. - info_callback: async callable(message) for progress updates (e.g. ctx.info). + Raises QueueFull if the queue is at max capacity (backpressure). """ - cid = caller_id or self._next_id() - preview = f"{result.engine}/{result.voice}" + if self._stopped: + return { + "played": False, + "error": "Queue is shut down", + "engine": result.engine, + "voice": result.voice, + } - # Show queue position before acquiring - if self._lock.locked(): - pos = self.depth + 1 - msg = f"Queued at position {pos}" + cid = caller_id or self._next_id() + loop = asyncio.get_running_loop() + future: asyncio.Future = loop.create_future() + + item = _WorkItem( + priority=priority, + sequence=self._counter, + result=result, + future=future, + info_callback=info_callback, + caller_id=cid, + ) + + try: + self._queue.put_nowait(item) + except asyncio.QueueFull: + return { + "played": False, + "error": f"Queue full ({self._max_depth} items). Try again later.", + "engine": result.engine, + "voice": result.voice, + } + + # Notify caller of queue position + depth = self._queue.qsize() + if depth > 1 and info_callback: + msg = f"Queued at position {depth}" if self._current: msg += f" (currently playing: {self._current.caller_id})" - if info_callback: + try: await info_callback(msg) + except Exception: + pass # Context may not be ready yet — non-fatal - async with self.acquire(cid, preview): - if info_callback: - await info_callback(f"Now playing: {result.engine}/{result.voice}") - await play_audio(result.audio_path) - - return { - "played": True, - "file": str(result.audio_path), - "duration_seconds": result.duration_seconds, - "engine": result.engine, - "voice": result.voice, - } + return await future diff --git a/src/tts_mcp/server.py b/src/tts_mcp/server.py index dc93ccf..ed4f55f 100644 --- a/src/tts_mcp/server.py +++ b/src/tts_mcp/server.py @@ -13,7 +13,7 @@ from .engines.base import TTSEngine from .engines.kokoro import KokoroEngine from .engines.orpheus import OrpheusEngine from .engines.piper import PiperEngine -from .queue import SpeechQueue +from .queue import Priority, SpeechQueue from .settings import settings ENGINE_NAMES = Literal["piper", "kokoro", "orpheus"] @@ -53,6 +53,7 @@ async def app_lifespan(server: FastMCP): print(f" {name}: {health['status']}", file=sys.stderr) queue = SpeechQueue() + queue.start() print( f"TTS MCP server ready on {settings.host}:{settings.port} " @@ -64,6 +65,7 @@ async def app_lifespan(server: FastMCP): yield {"engines": engines, "queue": queue} finally: print("TTS MCP server shutting down", file=sys.stderr) + await queue.stop() # --------------------------------------------------------------------------- @@ -98,17 +100,20 @@ async def speak( text: str, engine: ENGINE_NAMES = "kokoro", voice: str | None = None, + urgent: bool = False, ctx: Context = CurrentContext(), ) -> dict: """Synthesize text and play it through the host speakers. Audio is queued — if another agent is currently speaking, you'll wait - your turn and get notified when playback starts. + your turn and get notified when playback starts. Urgent messages + jump ahead of normal-priority items in the queue. Args: text: Text to speak. Orpheus supports emotion tags like , , etc. engine: TTS engine to use. kokoro is fastest, orpheus is most expressive. voice: Voice name (use list_voices to see options). None = engine default. + urgent: If True, this message jumps ahead of normal-priority items. """ engines, queue = _get_state(ctx) @@ -121,9 +126,10 @@ async def speak( await ctx.info(f"Synthesizing with {engine}...") result = await eng.synthesize(text, voice) - # Queue for playback (serialized) + # Queue for playback (serialized, priority-ordered) return await queue.speak( result, + priority=Priority.URGENT if urgent else Priority.NORMAL, info_callback=ctx.info, )