Graceful shutdown, blocking speak() with progress, and cancel support

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

@@ -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.

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

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

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}

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

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()}

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,
 }