Refactor speech queue: producer-consumer with priority and timeouts

Replace asyncio.Lock with bounded PriorityQueue + dedicated consumer
coroutine. Addresses Hamilton review findings:

- Playback timeout: pw-play subprocess killed if it exceeds 2x
  expected duration + margin (prevents deadlock on PipeWire hang)
- Bounded queue: max 20 items with backpressure rejection
- Priority tiers: urgent messages jump ahead of normal items
- Graceful shutdown: consumer cancelled, pending items drained,
  in-flight subprocess killed on lifespan teardown
- Structured errors: all failure modes return dicts, not raw exceptions

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:
+class PlaybackError(RuntimeError):
-    """Play a WAV file through PipeWire (pw-play). Async wrapper."""
+    """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()}"
+        )

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
+import sys
-from contextlib import asynccontextmanager
+import time
-from dataclasses import dataclass
+from dataclasses import dataclass, field
-from typing import AsyncIterator
+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 Priority(IntEnum):
-class _Waiter:
+    URGENT = 0
-    caller_id: str
+    NORMAL = 1
-    text_preview: str
+
+
+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
+    Call start() to launch the consumer, stop() to cancel it and
-    someone is speaking wait in line and get progress updates via
+    drain pending items.
-    their MCP context.
     """
 
-    def __init__(self) -> None:
+    def __init__(self, max_depth: int = MAX_QUEUE_DEPTH) -> None:
-        self._lock = asyncio.Lock()
+        self._queue: asyncio.PriorityQueue[_WorkItem] = asyncio.PriorityQueue(
-        self._current: _Waiter | None = None
+            maxsize=max_depth
-        self._waiters: deque[_Waiter] = deque()
+        )
+        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,
+            "queue_depth": self._queue.qsize(),
-            "waiting": [w.caller_id for w in self._waiters],
+            "max_depth": self._max_depth,
         }
 
-    @asynccontextmanager
+    def start(self) -> None:
-    async def acquire(
+        """Launch the consumer coroutine."""
-        self,
+        if self._consumer_task is None or self._consumer_task.done():
-        caller_id: str | None = None,
+            self._stopped = False
-        text_preview: str = "",
+            self._consumer_task = asyncio.create_task(
-    ) -> AsyncIterator[str]:
+                self._consumer(), name="speech-queue-consumer"
-        """Context manager that waits for the speaker's turn.
+            )
-
-        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:
-            async with self._lock:
+                await self._consumer_task
-                # We're up — remove ourselves from the waiting list
+            except asyncio.CancelledError:
-                if waiter in self._waiters:
+                pass
-                    self._waiters.remove(waiter)
+
-                self._current = waiter
+        # Drain remaining items
-                yield cid
+        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:
-            if self._current is waiter:
                 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()
+        if self._stopped:
-        preview = f"{result.engine}/{result.voice}"
-
-        # Show queue position before acquiring
-        if self._lock.locked():
-            pos = self.depth + 1
-            msg = f"Queued at position {pos}"
-            if self._current:
-                msg += f" (currently playing: {self._current.caller_id})"
-            if info_callback:
-                await info_callback(msg)
-
-        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,
+                "played": False,
-            "file": str(result.audio_path),
+                "error": "Queue is shut down",
-            "duration_seconds": result.duration_seconds,
                 "engine": result.engine,
                 "voice": result.voice,
             }
+
+        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})"
+            try:
+                await info_callback(msg)
+            except Exception:
+                pass  # Context may not be ready yet — non-fatal
+
+        return await future

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 <laugh>, <sigh>, 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,
     )