From 7cae72b936bcc9ef08f6dc1efb91db147e68cdd6 Mon Sep 17 00:00:00 2001 From: Ryan Malloy Date: Mon, 2 Mar 2026 17:45:48 -0700 Subject: [PATCH] Graceful shutdown, blocking speak() with progress, and cancel support MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit speak() now blocks until playback finishes, reporting progress via SSE (5% entry tone → 30% synthesis → 35-99% playing → 100% done). Entry tone fires immediately on call to cover synthesis latency. Queue shutdown waits for current speech to finish (configurable timeout, default 30s) before draining pending items — no more mid-sentence cutoffs on container restart. Cancellation via cancel_speech() tool or MCP notifications/cancelled kills pw-play and plays a vinyl scratch tone. Consumer continues to next item after cancel. Progress tracking uses a background ticker task instead of asyncio.wait_for polling — the latter causes stale CancelledError propagation to the consumer under Python 3.13. --- CLAUDE.md | 54 ++++++- docker-compose.yml | 3 + src/tts_mcp/audio.py | 2 +- src/tts_mcp/queue.py | 332 ++++++++++++++++++++++++++++------------ src/tts_mcp/server.py | 157 ++++++++++++++++--- src/tts_mcp/settings.py | 6 + src/tts_mcp/tones.py | 40 +++++ 7 files changed, 470 insertions(+), 124 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index 40c7b8b..e56233d 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -19,9 +19,10 @@ Queued speech playback (`speak()`) is bookended by short alert tones. `generate_ | Position | When | Purpose | |----------|------|---------| -| **Entry tone** | Before speech starts | "Incoming transmission" alert | +| **Entry tone** | Immediately when `speak()` is called (before synthesis) | "I heard you" acknowledgement — covers synthesis latency | | **Exit tone** | After speech, queue empty | "Over and out" — channel clear | | **Standby tone** | After speech, more queued | "Standby" — more messages coming | +| **Cancel tone** | After `cancel_speech()` or MCP cancellation kills playback | "Nevermind" — speech was aborted | ### Available Tones @@ -32,12 +33,16 @@ Queued speech playback (`speak()`) is bookended by short alert tones. `generate_ | `roger` | 1400-1000 Hz | ~100 ms | Classic CB radio descending two-tone roger beep | | `quindar-out` | 2475 Hz | 250 ms | NASA quindar unkey tone (distinct frequency from intro) | | `standby` | 1000-1400 Hz | ~60 ms | Ascending blip — inverse of roger, signals "more coming" | +| `scratch` | 2000-300 Hz sweep + noise | ~120 ms | Vinyl record scratch — needle yanked off the platter | +| `reverse-roger` | 1000-1400 Hz | ~100 ms | Ascending two-tone — mathematical inverse of roger beep | ### Configuration ```env TTS_ENTRY_TONE=chirp # before speech (chirp, apollo, none, or /path/to/custom.wav) TTS_EXIT_TONE=roger # after speech, queue empty (roger, quindar-out, none, or path) +TTS_CANCEL_TONE=scratch # on cancel (scratch, reverse-roger, none, or /path/to/custom.wav) +TTS_SHUTDOWN_TIMEOUT=30 # max seconds to wait for current speech on container stop ``` The standby tone is always the built-in ascending blip. It plays instead of the exit tone when more items are queued. @@ -80,15 +85,60 @@ Voices are interleaved for perceptual diversity: American female → British mal ## Architecture - `server.py` — FastMCP lifespan, tool definitions, engine setup -- `queue.py` — Producer-consumer speech queue with priority tiers +- `queue.py` — Producer-consumer speech queue with priority tiers and outcome tracking - `tones.py` — Tone WAV generator (entry/exit/standby) - `audio.py` — WAV writing and `pw-play` async wrapper - `settings.py` — Pydantic settings from env vars (prefix: `TTS_`) - `engines/` — TTSEngine implementations (kokoro, piper, orpheus) +## `speak()` Progress Lifecycle + +`speak()` blocks until playback finishes, reporting progress throughout: + +| Progress | Phase | +|----------|-------| +| 5% | Entry tone played — audible "I heard you" | +| 30% | Synthesis complete | +| 35% | Enqueued for playback | +| 35-99% | Playing — progress tracks elapsed time vs expected duration | +| 100% | Playback finished | + +MCP-aware clients see a live progress bar. The entry tone fires before synthesis, covering the 1-3s latency gap. + +`speech_status(speech_id)` is still available for checking outcomes after the fact. Possible statuses: `completed`, `playing`, `queued`, `unknown`. + +Outcomes are stored in a bounded ring (last 100 items) — old entries are evicted automatically. + +## Cancellation + +Speech can be cancelled two ways: + +1. **MCP cancellation** — the client sends `notifications/cancelled` for an in-flight `speak()` call. FastMCP throws `CancelledError` into the tool, which triggers `queue.cancel(speech_id)`. +2. **Explicit `cancel_speech(speech_id)`** — a separate tool that cancels any queued or playing item. + +When a currently-playing item is cancelled, pw-play is killed immediately and the cancel tone plays. When a queued item is cancelled, it's removed from the queue without ever playing. The consumer continues to the next item in both cases. + +## Graceful Shutdown + +On `docker compose down` or `make restart`, the server lets the currently-playing speech finish before stopping — no more mid-sentence cutoffs. + +**How it works:** `queue.stop()` sets a flag and waits up to `shutdown_timeout` seconds for the consumer to finish the current item. If the timeout expires, it falls back to hard cancel. Docker's `stop_grace_period: 35s` gives the 30s shutdown timeout room to complete before SIGKILL. + +**Entry tone timing:** The Nextel chirp fires immediately when `speak()` is called (before synthesis), acting as an audible "I heard you" that covers the 1-3s synthesis latency. Exit/standby tones still play from the consumer (they depend on queue state after playback). + ## Key Design Decisions - Speech queue is serialized (one playback at a time) but synthesis is parallel +- `speak()` blocks until playback finishes with live progress (5% → 30% → 35-99% → 100%) +- Progress uses a background ticker task, NOT `asyncio.wait_for` polling (see below) +- Entry tone is awaited in `speak()` before synthesis — covers latency gap +- Cancellation via `cancel_speech()` or MCP `notifications/cancelled` kills pw-play + plays cancel tone +- Consumer directly awaits `play_audio()`; cancel() targets the consumer task with `_item_cancelled` flag +- Graceful shutdown waits for current speech, then drains pending items with error outcomes - Tones are non-fatal: if `pw-play` fails on a tone, speech still plays - Orpheus uses llama-server (not Ollama) for 15x throughput via continuous batching - SNAC decoder is lazy-loaded on first Orpheus call to reduce idle memory + +### Python 3.13 asyncio.wait_for pitfall + +**Do NOT use `asyncio.wait_for(future, timeout)` in a polling loop to track progress.** In Python 3.13, `wait_for` cancels its inner task on timeout. When the inner task is awaiting the same `asyncio.Future` that the consumer will resolve, repeated cancel/re-await cycles cause a stale `CancelledError` to propagate to the consumer task — killing pw-play mid-playback. Instead, use a background `asyncio.create_task` ticker for progress and directly `await` the future for completion. diff --git a/docker-compose.yml b/docker-compose.yml index 054483e..e65e61e 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -3,6 +3,7 @@ services: build: . container_name: tts-mcp restart: unless-stopped + stop_grace_period: 35s env_file: .env environment: # Override for Docker networking (container DNS instead of IPs) @@ -10,6 +11,8 @@ services: TTS_ORPHEUS_URL: http://llama-server:8081 # PipeWire client config XDG_RUNTIME_DIR: /run/user/1000 + # Force unbuffered Python output so stderr shows up in docker logs immediately + PYTHONUNBUFFERED: "1" volumes: # Kokoro ONNX models (read-only) - ./models:/app/models:ro diff --git a/src/tts_mcp/audio.py b/src/tts_mcp/audio.py index f2c472c..8eb328b 100644 --- a/src/tts_mcp/audio.py +++ b/src/tts_mcp/audio.py @@ -106,7 +106,7 @@ async def play_audio(path: Path, expected_seconds: float = 0) -> None: f"(expected {expected_seconds:.1f}s audio)" ) except asyncio.CancelledError: - # Consumer was cancelled (shutdown) — don't orphan the subprocess + # Consumer was cancelled (shutdown or cancel) — don't orphan the subprocess proc.kill() await proc.wait() raise diff --git a/src/tts_mcp/queue.py b/src/tts_mcp/queue.py index e585623..62af471 100644 --- a/src/tts_mcp/queue.py +++ b/src/tts_mcp/queue.py @@ -2,7 +2,8 @@ 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. +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) @@ -12,10 +13,10 @@ Priority tiers: import asyncio import sys import time +from collections import OrderedDict from dataclasses import dataclass, field from enum import IntEnum from pathlib import Path -from typing import Callable, Coroutine from .audio import PlaybackError, play_audio from .engines.base import TTSResult @@ -27,6 +28,7 @@ class Priority(IntEnum): MAX_QUEUE_DEPTH = 20 +MAX_OUTCOMES = 100 @dataclass(order=True) @@ -37,10 +39,7 @@ class _WorkItem: 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) + speech_id: str = field(default="", compare=False) enqueued_at: float = field(default_factory=time.time, compare=False) @@ -54,30 +53,43 @@ class SpeechQueue: def __init__( self, max_depth: int = MAX_QUEUE_DEPTH, - entry_tone: Path | None = None, 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._current_proc: asyncio.subprocess.Process | None = None + self._item_cancelled = False self._counter = 0 self._max_depth = max_depth self._stopped = False - self._entry_tone = entry_tone 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_id(self) -> str: + def _next_speech_id(self) -> str: self._counter += 1 - return f"speaker-{self._counter}" + 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.caller_id if self._current else None + return self._current.speech_id if self._current else None def status(self) -> dict: return { @@ -94,40 +106,84 @@ class SpeechQueue: self._consumer(), name="speech-queue-consumer" ) - 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 - - # Drain remaining items + 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({ - "played": False, - "error": "Queue shut down", - "engine": item.result.engine, - "voice": item.result.voice, - }) + item.future.set_result(outcome) + self._record_outcome(item.speech_id, outcome) 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) + 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(): + # 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}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.""" + """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 self._queue.get() + item = await asyncio.wait_for(self._queue.get(), timeout=0.5) + except asyncio.TimeoutError: + continue # Re-check _stopped flag except asyncio.CancelledError: break @@ -136,98 +192,101 @@ class SpeechQueue: continue self._current = item + self._item_cancelled = False + 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 - - # Play entry tone (chirp/quindar) before speech - if self._entry_tone: - try: - await play_audio(self._entry_tone, expected_seconds=0.3) - except PlaybackError: - pass # Non-fatal — continue with speech - await play_audio( item.result.audio_path, expected_seconds=item.result.duration_seconds, ) # Play exit tone — "standby" if more queued, "roger" if done - exit = ( + exit_t = ( self._standby_tone if self._queue.qsize() > 0 else self._exit_tone ) - if exit: + if exit_t: try: - await play_audio(exit, expected_seconds=0.3) + await play_audio(exit_t, expected_seconds=0.3) except PlaybackError: pass # Non-fatal - 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, - }) + 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: - print(f" Playback error: {e}", file=sys.stderr) - if not item.future.done(): - item.future.set_result({ + 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": str(e), - "file": str(item.result.audio_path), + "error": "Cancelled by client", "engine": item.result.engine, "voice": item.result.voice, }) - except asyncio.CancelledError: - if not item.future.done(): - item.future.set_result({ + # 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 + 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, - }) + 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 speak( + async def enqueue( self, result: TTSResult, - caller_id: str | None = None, priority: Priority = Priority.NORMAL, - info_callback: Callable[..., Coroutine] | None = None, ) -> dict: - """Enqueue a TTSResult for playback. Returns result dict when done. + """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 { - "played": False, + "queued": False, "error": "Queue is shut down", "engine": result.engine, "voice": result.voice, } - cid = caller_id or self._next_id() + speech_id = self._next_speech_id() loop = asyncio.get_running_loop() future: asyncio.Future = loop.create_future() @@ -236,29 +295,108 @@ class SpeechQueue: sequence=self._counter, result=result, future=future, - info_callback=info_callback, - caller_id=cid, + speech_id=speech_id, ) try: self._queue.put_nowait(item) except asyncio.QueueFull: return { - "played": False, + "queued": False, "error": f"Queue full ({self._max_depth} items). Try again later.", + "speech_id": speech_id, "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 + 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} diff --git a/src/tts_mcp/server.py b/src/tts_mcp/server.py index 5a511e0..dc81bed 100644 --- a/src/tts_mcp/server.py +++ b/src/tts_mcp/server.py @@ -4,11 +4,13 @@ import asyncio import sys import time from contextlib import asynccontextmanager +from pathlib import Path from typing import Literal from fastmcp import Context, FastMCP from fastmcp.server.dependencies import CurrentContext +from .audio import play_audio from .engines.base import TTSEngine from .engines.kokoro import KokoroEngine from .engines.orpheus import OrpheusEngine @@ -58,16 +60,23 @@ async def app_lifespan(server: FastMCP): tone_paths = generate_tones(settings.output_dir) entry_tone = resolve_tone(settings.entry_tone, tone_paths, "entry_tone", default="chirp") exit_tone = resolve_tone(settings.exit_tone, tone_paths, "exit_tone", default="roger") + cancel_tone = resolve_tone(settings.cancel_tone, tone_paths, "cancel_tone", default="scratch") standby_tone = tone_paths.get("standby") - queue = SpeechQueue(entry_tone=entry_tone, exit_tone=exit_tone, standby_tone=standby_tone) + queue = SpeechQueue( + exit_tone=exit_tone, + standby_tone=standby_tone, + cancel_tone=cancel_tone, + shutdown_timeout=settings.shutdown_timeout, + ) queue.start() entry_label = settings.entry_tone if entry_tone else "none" exit_label = settings.exit_tone if exit_tone else "none" + cancel_label = settings.cancel_tone if cancel_tone else "none" print( f"TTS MCP server ready on {settings.host}:{settings.port} " - f"with {len(engines)} engines, tones={entry_label}/{exit_label}", + f"with {len(engines)} engines, tones={entry_label}/{exit_label}/{cancel_label}", file=sys.stderr, ) @@ -81,7 +90,12 @@ async def app_lifespan(server: FastMCP): ) try: - yield {"engines": engines, "queue": queue, "voice_cache": voice_cache} + yield { + "engines": engines, + "queue": queue, + "voice_cache": voice_cache, + "entry_tone": entry_tone, + } finally: print("TTS MCP server shutting down", file=sys.stderr) await queue.stop() @@ -97,8 +111,9 @@ async def app_lifespan(server: FastMCP): mcp = FastMCP( "tts-mcp", instructions=( - "Multi-engine text-to-speech server. Use 'speak' to synthesize and play audio " - "through the host speakers (queued so agents don't talk over each other). " + "Multi-engine text-to-speech server. Use 'speak' to synthesize and enqueue audio " + "for playback through the host speakers — returns immediately with a speech_id. " + "Use 'speech_status' to check if playback completed, is in progress, or still queued. " "Use 'generate_audio' to synthesize without playing. " "Engines: kokoro (fast ONNX, ~50 voices), piper (Wyoming/Docker), " "orpheus (LLM via llama-server, supports etc.). " @@ -109,10 +124,18 @@ mcp = FastMCP( ) -def _get_state(ctx: Context) -> tuple[dict[str, TTSEngine], SpeechQueue, VoiceIdentityCache]: - """Extract engines, queue, and voice cache from lifespan context.""" +def _get_state(ctx: Context) -> tuple[dict[str, TTSEngine], SpeechQueue, VoiceIdentityCache, Path | None]: + """Extract engines, queue, voice cache, and entry tone from lifespan context.""" state = ctx.lifespan_context - return state["engines"], state["queue"], state["voice_cache"] + return state["engines"], state["queue"], state["voice_cache"], state["entry_tone"] + + +async def _play_tone(tone_path: Path) -> None: + """Fire-and-forget tone playback. Non-fatal — errors are swallowed.""" + try: + await play_audio(tone_path, expected_seconds=0.3) + except Exception as e: + print(f" Entry tone failed (non-fatal): {e}", file=sys.stderr) async def _resolve_project_voice( @@ -155,9 +178,11 @@ async def speak( ) -> 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. Urgent messages - jump ahead of normal-priority items in the queue. + Blocks until playback finishes, reporting progress throughout: + entry tone → synthesis → queued → playing → done. + + Audio is queued so agents don't talk over each other. Urgent messages + jump ahead of normal-priority items. Args: text: Text to speak. Orpheus supports emotion tags like , , etc. @@ -166,7 +191,7 @@ async def speak( urgent: If True, this message jumps ahead of normal-priority items. project: Project name for voice identity (auto-detected from MCP roots if omitted). """ - engines, queue, voice_cache = _get_state(ctx) + engines, queue, voice_cache, entry_tone = _get_state(ctx) if engine not in engines: return {"error": f"Unknown engine: {engine}. Available: {list(engines.keys())}"} @@ -176,16 +201,100 @@ async def speak( ctx, engine, eng, voice, text, voice_cache, project, ) - # Synthesize audio (not queued — multiple agents can synthesize simultaneously) - await ctx.info(f"Synthesizing with {engine}...") - result = await eng.synthesize(text, voice) + speech_id = None + try: + # Entry tone — audible "I heard you" before synthesis begins + if entry_tone: + await _play_tone(entry_tone) + await ctx.report_progress(progress=5, total=100) - # Queue for playback (serialized, priority-ordered) - return await queue.speak( - result, - priority=Priority.URGENT if urgent else Priority.NORMAL, - info_callback=ctx.info, - ) + # Synthesize audio (multiple agents can synthesize simultaneously) + await ctx.info(f"Synthesizing with {engine}...") + result = await eng.synthesize(text, voice) + await ctx.report_progress(progress=30, total=100) + + # Enqueue for serialized playback + await ctx.info(f"Synthesized {result.duration_seconds:.1f}s audio, enqueueing...") + enqueue_result = await queue.enqueue( + result, + priority=Priority.URGENT if urgent else Priority.NORMAL, + ) + + if not enqueue_result.get("queued"): + return enqueue_result + + speech_id = enqueue_result["speech_id"] + duration = result.duration_seconds + await ctx.report_progress(progress=35, total=100) + + # Wait for playback with progress via a background ticker. + # NOTE: Do NOT use asyncio.wait_for() with the future — Python 3.13's + # wait_for cancels its inner task on timeout, and repeated cancel/re-await + # on the same Future causes a stale CancelledError to propagate to the + # consumer task, killing pw-play mid-sentence. + await ctx.info("Playing...") + + async def _progress_ticker(): + """Send progress updates while playback runs.""" + t0 = time.time() + while True: + await asyncio.sleep(0.5) + elapsed = time.time() - t0 + pct = min(35 + int(64 * elapsed / max(duration, 0.1)), 99) + await ctx.report_progress(progress=pct, total=100) + + ticker = asyncio.create_task(_progress_ticker()) + try: + outcome = await queue.wait_for_completion(speech_id) + finally: + ticker.cancel() + try: + await ticker + except asyncio.CancelledError: + pass + + await ctx.report_progress(progress=100, total=100) + return outcome + + except asyncio.CancelledError: + # MCP client cancelled the tool call — cancel in-flight playback + if speech_id: + queue.cancel(speech_id) + raise + + +@mcp.tool +async def speech_status( + speech_id: str, + ctx: Context = CurrentContext(), +) -> dict: + """Check the status of a previously enqueued speech item. + + Returns the current state: 'completed' (with playback result), 'playing', + 'queued' (with position), or 'unknown' (expired from history or invalid ID). + + Args: + speech_id: The speech_id returned by speak(). + """ + _, queue, _, _ = _get_state(ctx) + return queue.get_status(speech_id) + + +@mcp.tool +async def cancel_speech( + speech_id: str, + ctx: Context = CurrentContext(), +) -> dict: + """Cancel a queued or currently-playing speech item. + + If the item is playing, stops playback immediately and plays a cancel tone. + If the item is queued, removes it from the queue. + + Args: + speech_id: The speech_id returned by speak(). + """ + _, queue, _, _ = _get_state(ctx) + return queue.cancel(speech_id) @mcp.tool @@ -207,7 +316,7 @@ async def generate_audio( voice: Voice name (use list_voices to see options). None = auto-assigned by project. project: Project name for voice identity (auto-detected from MCP roots if omitted). """ - engines, _, voice_cache = _get_state(ctx) + engines, _, voice_cache, _ = _get_state(ctx) if engine not in engines: return {"error": f"Unknown engine: {engine}. Available: {list(engines.keys())}"} @@ -242,7 +351,7 @@ async def list_voices( Args: engine: Which engine to list voices for. """ - engines, _, _ = _get_state(ctx) + engines, _, _, _ = _get_state(ctx) if engine not in engines: return [] @@ -258,7 +367,7 @@ async def list_engines( Returns engine name, default voice, and health check results. """ - engines, queue, _ = _get_state(ctx) + engines, queue, _, _ = _get_state(ctx) results = [] for name, eng in engines.items(): diff --git a/src/tts_mcp/settings.py b/src/tts_mcp/settings.py index bc15262..eb0092e 100644 --- a/src/tts_mcp/settings.py +++ b/src/tts_mcp/settings.py @@ -46,6 +46,12 @@ class Settings(BaseSettings): # When more items are queued, plays "standby" (ascending blip) instead exit_tone: str = "roger" + # Cancel tone: "scratch", "reverse-roger", "none", or path to custom WAV + cancel_tone: str = "scratch" + + # Graceful shutdown: max seconds to wait for current speech to finish + shutdown_timeout: float = 30.0 + @property def blacklisted_voices(self) -> set[str]: return {v.strip().lower() for v in self.voice_blacklist.split(",") if v.strip()} diff --git a/src/tts_mcp/tones.py b/src/tts_mcp/tones.py index 4b8570f..de93dce 100644 --- a/src/tts_mcp/tones.py +++ b/src/tts_mcp/tones.py @@ -72,6 +72,44 @@ def _build_standby() -> np.ndarray: return np.concatenate(segments) +def _build_scratch() -> np.ndarray: + """Vinyl record scratch — descending sweep with noise, ~120ms.""" + duration_ms = 120 + n_samples = int(SAMPLE_RATE * duration_ms / 1000) + t = np.arange(n_samples, dtype=np.float32) / SAMPLE_RATE + + # Descending frequency sweep: 2000 Hz -> 300 Hz + f0, f1 = 2000.0, 300.0 + phase = 2 * np.pi * (f0 * t + (f1 - f0) / (2 * duration_ms / 1000) * t**2) + sweep = np.sin(phase) + + # Mix in noise (pink-ish via low-pass filtered white noise) + rng = np.random.default_rng(42) + noise = rng.standard_normal(n_samples).astype(np.float32) + # Simple low-pass: cumulative average over 8-sample windows + kernel = np.ones(8, dtype=np.float32) / 8 + noise = np.convolve(noise, kernel, mode="same") + + # Mix: 60% sweep + 40% noise, with amplitude envelope (fade out) + envelope = np.linspace(1.0, 0.0, n_samples, dtype=np.float32) ** 0.5 + mixed = (0.6 * sweep + 0.4 * noise) * envelope + + # Click at start (needle hit) — 2ms burst + click_samples = int(SAMPLE_RATE * 0.002) + mixed[:click_samples] += 0.5 * rng.standard_normal(click_samples).astype(np.float32) + + return mixed + + +def _build_reverse_roger() -> np.ndarray: + """Ascending two-tone — inverse of roger beep: 1000 Hz -> 1400 Hz.""" + segments = [ + _sine_segment(1000, 40), + _sine_segment(1400, 60), + ] + return np.concatenate(segments) + + def _write_tone(samples: np.ndarray, path: Path) -> Path: """Write float32 samples to 16-bit PCM WAV.""" peak = max(abs(samples.max()), abs(samples.min()), 1e-8) @@ -92,6 +130,8 @@ _TONE_BUILDERS = { "roger": _build_roger, "quindar-out": _build_quindar_out, "standby": _build_standby, + "scratch": _build_scratch, + "reverse-roger": _build_reverse_roger, }