Ryan Malloy
7959efbd51
Chunked synthesis splits long texts into sentences and pipelines synthesis with playback — first audio plays within ~1-2s instead of waiting for full synthesis. Queue supports suppress_exit_tone between chunks for seamless playback. Tool annotations (ToolAnnotations) signal behavioral hints to MCP clients: speak() is openWorldHint=True (physical speakers), cancel_speech() is destructiveHint=True, read-only tools marked accordingly. Progress notifications now include message= strings that tell LLM clients they can continue working during playback. Server instructions corrected from "returns immediately" to accurately describe blocking behavior with parallel-safe guidance. speak() docstring updated to encourage concurrent tool calls.
420 lines
16 KiB
Python
420 lines
16 KiB
Python
"""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 call enqueue() which returns immediately with a speech_id.
|
|
Use get_status() to check playback outcomes asynchronously.
|
|
|
|
Priority tiers:
|
|
0 = urgent (preempts normal items in the queue)
|
|
1 = normal (default)
|
|
"""
|
|
|
|
import asyncio
|
|
import sys
|
|
import time
|
|
from collections import OrderedDict
|
|
from dataclasses import dataclass, field
|
|
from enum import IntEnum
|
|
from pathlib import Path
|
|
|
|
from .audio import PlaybackError, play_audio
|
|
from .engines.base import TTSResult
|
|
|
|
|
|
class Priority(IntEnum):
|
|
URGENT = 0
|
|
NORMAL = 1
|
|
|
|
|
|
MAX_QUEUE_DEPTH = 20
|
|
MAX_OUTCOMES = 100
|
|
|
|
|
|
@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)
|
|
speech_id: str = field(default="", compare=False)
|
|
enqueued_at: float = field(default_factory=time.time, compare=False)
|
|
suppress_exit_tone: bool = field(default=False, compare=False)
|
|
|
|
|
|
class SpeechQueue:
|
|
"""Bounded priority queue with a dedicated playback consumer.
|
|
|
|
Call start() to launch the consumer, stop() to cancel it and
|
|
drain pending items.
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
max_depth: int = MAX_QUEUE_DEPTH,
|
|
exit_tone: Path | None = None,
|
|
standby_tone: Path | None = None,
|
|
cancel_tone: Path | None = None,
|
|
shutdown_timeout: float = 30.0,
|
|
) -> None:
|
|
self._queue: asyncio.PriorityQueue[_WorkItem] = asyncio.PriorityQueue(
|
|
maxsize=max_depth
|
|
)
|
|
self._consumer_task: asyncio.Task | None = None
|
|
self._current: _WorkItem | None = None
|
|
self._item_cancelled = False
|
|
self._counter = 0
|
|
self._max_depth = max_depth
|
|
self._stopped = False
|
|
self._exit_tone = exit_tone
|
|
self._standby_tone = standby_tone
|
|
self._cancel_tone = cancel_tone
|
|
self._shutdown_timeout = shutdown_timeout
|
|
self._outcomes: OrderedDict[str, dict] = OrderedDict()
|
|
self._futures: dict[str, asyncio.Future] = {}
|
|
|
|
def _next_speech_id(self) -> str:
|
|
self._counter += 1
|
|
return f"speech-{self._counter}"
|
|
|
|
def _record_outcome(self, speech_id: str, result_dict: dict) -> None:
|
|
"""Store a completed outcome, evicting the oldest if over limit."""
|
|
result_dict["speech_id"] = speech_id
|
|
result_dict["completed_at"] = time.time()
|
|
self._outcomes[speech_id] = result_dict
|
|
self._futures.pop(speech_id, None)
|
|
while len(self._outcomes) > MAX_OUTCOMES:
|
|
self._outcomes.popitem(last=False)
|
|
|
|
@property
|
|
def current_speaker(self) -> str | None:
|
|
return self._current.speech_id if self._current else None
|
|
|
|
def status(self) -> dict:
|
|
return {
|
|
"current_speaker": self.current_speaker,
|
|
"queue_depth": self._queue.qsize(),
|
|
"max_depth": self._max_depth,
|
|
}
|
|
|
|
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"
|
|
)
|
|
|
|
def _drain_remaining(self) -> None:
|
|
"""Drain pending items, recording shutdown outcomes. Sync — safe for finally."""
|
|
while not self._queue.empty():
|
|
try:
|
|
item = self._queue.get_nowait()
|
|
outcome = {
|
|
"played": False,
|
|
"error": "Queue shut down",
|
|
"engine": item.result.engine,
|
|
"voice": item.result.voice,
|
|
}
|
|
if not item.future.done():
|
|
item.future.set_result(outcome)
|
|
self._record_outcome(item.speech_id, outcome)
|
|
except asyncio.QueueEmpty:
|
|
break
|
|
|
|
def _force_cancel_consumer(self) -> None:
|
|
"""Cancel consumer task if still running."""
|
|
if self._consumer_task and not self._consumer_task.done():
|
|
self._consumer_task.cancel()
|
|
|
|
async def stop(self) -> None:
|
|
"""Gracefully stop the consumer — let current speech finish, then drain.
|
|
|
|
Waits up to shutdown_timeout seconds for the currently-playing item to
|
|
complete. If it doesn't finish in time, falls back to hard cancel.
|
|
Drain always runs, even if stop() itself is cancelled.
|
|
"""
|
|
self._stopped = True
|
|
try:
|
|
if self._consumer_task and not self._consumer_task.done():
|
|
# Log what we're waiting for
|
|
if self._current:
|
|
elapsed = time.time() - self._current.enqueued_at
|
|
remaining = max(0, self._current.result.duration_seconds - elapsed)
|
|
print(
|
|
f"Speech queue: waiting for current audio "
|
|
f"(~{remaining:.0f}s remaining, budget {self._shutdown_timeout:.0f}s)",
|
|
file=sys.stderr,
|
|
)
|
|
else:
|
|
print("Speech queue: consumer idle, shutting down", file=sys.stderr)
|
|
|
|
# Consumer polls _stopped every 0.5s between items, so it will
|
|
# exit the while loop after finishing the current item.
|
|
try:
|
|
await asyncio.wait_for(
|
|
asyncio.shield(self._consumer_task),
|
|
timeout=self._shutdown_timeout,
|
|
)
|
|
print(
|
|
"Speech queue: graceful shutdown complete",
|
|
file=sys.stderr,
|
|
)
|
|
except asyncio.TimeoutError:
|
|
print(
|
|
f"Speech queue: shutdown timeout ({self._shutdown_timeout:.0f}s) "
|
|
f"expired, force-cancelling consumer",
|
|
file=sys.stderr,
|
|
)
|
|
self._consumer_task.cancel()
|
|
try:
|
|
await self._consumer_task
|
|
except asyncio.CancelledError:
|
|
pass
|
|
except asyncio.CancelledError:
|
|
# stop() itself was cancelled — force-kill consumer before draining
|
|
self._force_cancel_consumer()
|
|
raise
|
|
finally:
|
|
self._drain_remaining()
|
|
|
|
def _resolve_outcome(self, item: _WorkItem, outcome: dict) -> None:
|
|
"""Set the future result and record the outcome."""
|
|
if not item.future.done():
|
|
item.future.set_result(outcome)
|
|
self._record_outcome(item.speech_id, outcome)
|
|
|
|
async def _consumer(self) -> None:
|
|
"""Pull work items and play them sequentially.
|
|
|
|
The _item_cancelled flag distinguishes cancel(speech_id) (item-level,
|
|
continue loop) from shutdown cancellation (exit loop).
|
|
"""
|
|
while not self._stopped:
|
|
try:
|
|
item = await asyncio.wait_for(self._queue.get(), timeout=0.5)
|
|
except asyncio.TimeoutError:
|
|
continue # Re-check _stopped flag
|
|
except asyncio.CancelledError:
|
|
break
|
|
|
|
if item.future.cancelled():
|
|
self._queue.task_done()
|
|
continue
|
|
|
|
self._current = item
|
|
self._item_cancelled = False
|
|
|
|
try:
|
|
await play_audio(
|
|
item.result.audio_path,
|
|
expected_seconds=item.result.duration_seconds,
|
|
)
|
|
|
|
# Play exit tone — "standby" if more queued, "roger" if done
|
|
# suppress_exit_tone skips tones between chunks of the same message
|
|
if not item.suppress_exit_tone:
|
|
exit_t = (
|
|
self._standby_tone if self._queue.qsize() > 0
|
|
else self._exit_tone
|
|
)
|
|
if exit_t:
|
|
try:
|
|
await play_audio(exit_t, expected_seconds=0.3)
|
|
except PlaybackError:
|
|
pass # Non-fatal
|
|
|
|
self._resolve_outcome(item, {
|
|
"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:
|
|
self._resolve_outcome(item, {
|
|
"played": False,
|
|
"error": str(e),
|
|
"file": str(item.result.audio_path),
|
|
"engine": item.result.engine,
|
|
"voice": item.result.voice,
|
|
})
|
|
|
|
except asyncio.CancelledError:
|
|
if self._item_cancelled:
|
|
# Item-level cancel via cancel() — play cancel tone, continue
|
|
self._item_cancelled = False
|
|
if self._cancel_tone:
|
|
try:
|
|
await play_audio(self._cancel_tone, expected_seconds=0.3)
|
|
except (PlaybackError, asyncio.CancelledError):
|
|
pass
|
|
self._resolve_outcome(item, {
|
|
"played": False,
|
|
"error": "Cancelled by client",
|
|
"engine": item.result.engine,
|
|
"voice": item.result.voice,
|
|
})
|
|
# Don't break — continue to next item
|
|
else:
|
|
# Shutdown — record and exit loop
|
|
self._resolve_outcome(item, {
|
|
"played": False,
|
|
"error": "Playback cancelled",
|
|
"engine": item.result.engine,
|
|
"voice": item.result.voice,
|
|
})
|
|
break
|
|
|
|
except Exception as e:
|
|
self._resolve_outcome(item, {
|
|
"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 enqueue(
|
|
self,
|
|
result: TTSResult,
|
|
priority: Priority = Priority.NORMAL,
|
|
suppress_exit_tone: bool = False,
|
|
) -> dict:
|
|
"""Enqueue a TTSResult for playback. Returns immediately with enqueue metadata.
|
|
|
|
Does NOT block until playback finishes — use get_status(speech_id) to
|
|
check the outcome later.
|
|
|
|
Raises QueueFull if the queue is at max capacity (backpressure).
|
|
"""
|
|
if self._stopped:
|
|
return {
|
|
"queued": False,
|
|
"error": "Queue is shut down",
|
|
"engine": result.engine,
|
|
"voice": result.voice,
|
|
}
|
|
|
|
speech_id = self._next_speech_id()
|
|
loop = asyncio.get_running_loop()
|
|
future: asyncio.Future = loop.create_future()
|
|
|
|
item = _WorkItem(
|
|
priority=priority,
|
|
sequence=self._counter,
|
|
result=result,
|
|
future=future,
|
|
speech_id=speech_id,
|
|
suppress_exit_tone=suppress_exit_tone,
|
|
)
|
|
|
|
try:
|
|
self._queue.put_nowait(item)
|
|
except asyncio.QueueFull:
|
|
return {
|
|
"queued": False,
|
|
"error": f"Queue full ({self._max_depth} items). Try again later.",
|
|
"speech_id": speech_id,
|
|
"engine": result.engine,
|
|
"voice": result.voice,
|
|
}
|
|
|
|
self._futures[speech_id] = future
|
|
|
|
return {
|
|
"speech_id": speech_id,
|
|
"queued": True,
|
|
"position": self._queue.qsize(),
|
|
"file": str(result.audio_path),
|
|
"duration_seconds": result.duration_seconds,
|
|
"engine": result.engine,
|
|
"voice": result.voice,
|
|
}
|
|
|
|
async def wait_for_completion(self, speech_id: str) -> dict:
|
|
"""Await playback completion for a specific speech_id.
|
|
|
|
Returns the outcome dict when playback finishes (or fails/is cancelled).
|
|
If the speech_id is unknown or already completed, returns immediately.
|
|
"""
|
|
future = self._futures.get(speech_id)
|
|
if future is None:
|
|
# Already completed and cleaned up, or unknown
|
|
return self._outcomes.get(speech_id, {"status": "unknown", "speech_id": speech_id})
|
|
return await future
|
|
|
|
def cancel(self, speech_id: str) -> dict:
|
|
"""Cancel a queued or playing speech item.
|
|
|
|
If the item is currently playing, kills pw-play and the consumer plays
|
|
the cancel tone. If queued, marks the future as cancelled so the
|
|
consumer skips it. Returns cancellation status.
|
|
"""
|
|
# Currently playing — cancel via the consumer task
|
|
if self._current and self._current.speech_id == speech_id:
|
|
if self._consumer_task and not self._consumer_task.done():
|
|
self._item_cancelled = True
|
|
self._consumer_task.cancel()
|
|
return {"cancelled": True, "was": "playing", "speech_id": speech_id}
|
|
|
|
# Queued — cancel the future (consumer skips cancelled futures)
|
|
future = self._futures.get(speech_id)
|
|
if future and not future.done():
|
|
future.cancel()
|
|
self._futures.pop(speech_id, None)
|
|
self._record_outcome(speech_id, {
|
|
"played": False,
|
|
"error": "Cancelled by client",
|
|
})
|
|
return {"cancelled": True, "was": "queued", "speech_id": speech_id}
|
|
|
|
return {
|
|
"cancelled": False,
|
|
"speech_id": speech_id,
|
|
"reason": "Not found or already completed",
|
|
}
|
|
|
|
def get_status(self, speech_id: str) -> dict:
|
|
"""Check the status of a speech item by its ID.
|
|
|
|
Returns a dict with 'status' key: 'completed', 'playing', 'queued', or 'unknown'.
|
|
"""
|
|
# Check completed outcomes
|
|
if speech_id in self._outcomes:
|
|
outcome = self._outcomes[speech_id]
|
|
return {"status": "completed", **outcome}
|
|
|
|
# Check currently playing
|
|
if self._current and self._current.speech_id == speech_id:
|
|
return {
|
|
"status": "playing",
|
|
"speech_id": speech_id,
|
|
"engine": self._current.result.engine,
|
|
"voice": self._current.result.voice,
|
|
"file": str(self._current.result.audio_path),
|
|
"duration_seconds": self._current.result.duration_seconds,
|
|
}
|
|
|
|
# Scan queue (PriorityQueue._queue is the underlying heap list)
|
|
try:
|
|
for pos, item in enumerate(self._queue._queue, start=1):
|
|
if item.speech_id == speech_id:
|
|
return {
|
|
"status": "queued",
|
|
"speech_id": speech_id,
|
|
"position": pos,
|
|
"engine": item.result.engine,
|
|
"voice": item.result.voice,
|
|
}
|
|
except AttributeError:
|
|
pass # Internal API — degrade gracefully
|
|
|
|
return {"status": "unknown", "speech_id": speech_id}
|