Add media ducking: fade external audio during TTS playback

Duck all PulseAudio sink-inputs (Firefox, Spotify, etc.) before speech,
crossfade the entry tone over the fading media, then restore volumes
after the exit tone. Uses pactl subprocesses for reliable PipeWire
compatibility — pulsectl-asyncio's native protocol writes are silently
dropped by PipeWire's PA compat layer.

New module media_duck.py with MediaDucker class. Unduck is consolidated
into the consumer's finally block to guarantee restoration on all exit
paths (normal, error, cancel, shutdown).

CLAUDE.md

@@ -82,11 +82,59 @@ Voices are interleaved for perceptual diversity: American female → British mal
 
 - `voice_identity.py` — Pool filtering, interleaving, round-robin assignment, JSON persistence
 
+## Media Ducking
+
+When `speak()` is called, external audio streams (Firefox, Spotify, etc.) are automatically faded down via PulseAudio before speech begins, then faded back up after the exit tone. This creates a radio-broadcast-interruption effect where the entry tone crossfades over the fading media.
+
+### Audio Timeline
+
+```
+0ms     ─ Media starts fading (vol 100%)
+150ms   ─ Entry tone starts! Media at ~70%    ← crossfade overlap
+294ms   ─ Entry tone ends, media at ~40%
+500ms   ─ Media at 0% (ducked)
+        ─ [synthesis + playback — media stays silent]
+        ─ Exit tone plays (roger/standby)
+        ─ Media fades back in over 1000ms
+```
+
+The duck fires in `speak()` (before synthesis), the unduck fires in the consumer (after exit/cancel tone). Media stays ducked across multiple queued items — only unducks when the queue is empty or on cancel/shutdown.
+
+### Configuration
+
+```env
+TTS_DUCK_MEDIA=true             # Enable/disable (on by default)
+TTS_DUCK_FADE_OUT_MS=500        # Fade-out duration (ms)
+TTS_DUCK_FADE_IN_MS=1000        # Fade-in duration (slower = natural)
+```
+
+### Docker Requirements
+
+The PulseAudio compatibility socket must be mounted in the container:
+
+```yaml
+volumes:
+  - /run/user/1000/pulse:/run/user/1000/pulse
+```
+
+If the socket is missing or pactl fails, ducking silently no-ops — TTS still works normally.
+
+**Why pactl, not pulsectl-asyncio?** PipeWire's PulseAudio compat layer silently drops `sink_input_volume_set` operations from pulsectl's native protocol connection, while `pactl` (using libpulse C library) works reliably. The `pulseaudio-utils` package is installed in the container for this reason.
+
+### Non-fatal Design
+
+All pactl errors are caught and logged. If PulseAudio is unavailable (no socket, wrong permissions, pactl not installed), duck/unduck become no-ops. This ensures TTS never breaks due to ducking failures.
+
+### Files
+
+- `media_duck.py` — `MediaDucker` class (async pactl subprocess volume control with stepped fades)
+
 ## Architecture
 
 - `server.py` — FastMCP lifespan, tool definitions, engine setup
 - `queue.py` — Producer-consumer speech queue with priority tiers and outcome tracking
 - `tones.py` — Tone WAV generator (entry/exit/standby)
+- `media_duck.py` — Async PulseAudio volume control for media ducking
 - `audio.py` — WAV writing and `pw-play` async wrapper
 - `settings.py` — Pydantic settings from env vars (prefix: `TTS_`)
 - `engines/` — TTSEngine implementations (kokoro, piper, orpheus)

Dockerfile

@@ -1,8 +1,10 @@
 FROM ghcr.io/astral-sh/uv:python3.13-bookworm-slim
 
 # PipeWire client tools — pw-play connects to the host's PipeWire socket
+# pulseaudio-utils provides pactl for media ducking (volume control via PulseAudio compat)
 RUN apt-get update && apt-get install -y --no-install-recommends \
     pipewire-bin \
+    pulseaudio-utils \
     && rm -rf /var/lib/apt/lists/*
 
 WORKDIR /app

docker-compose.yml

@@ -13,6 +13,8 @@ services:
       TTS_ORPHEUS_URL: http://llama-server:8081
       # PipeWire client config
       XDG_RUNTIME_DIR: /run/user/1000
+      # PulseAudio socket for media ducking (bypass XDG_RUNTIME_DIR ownership check)
+      PULSE_SERVER: unix:/run/user/1000/pulse/native
       # Force unbuffered Python output so stderr shows up in docker logs immediately
       PYTHONUNBUFFERED: "1"
     volumes:
@@ -24,6 +26,8 @@ services:
       - tts-data:/data
       # PipeWire socket for audio playback through host speakers
       - /run/user/1000/pipewire-0:/run/user/1000/pipewire-0
+      # PulseAudio compat socket for media ducking (volume control)
+      - /run/user/1000/pulse:/run/user/1000/pulse
     depends_on:
       llama-server:
         condition: service_healthy

src/tts_mcp/media_duck.py

@@ -0,0 +1,177 @@
+"""Media ducking — fade external audio during TTS playback.
+
+Uses pactl (PulseAudio CLI) to control sink-input volumes through PipeWire's
+PulseAudio compatibility layer. When duck() is called, all active sink-inputs
+(Firefox, Spotify, etc.) have their volumes stepped down over a configurable
+fade period. unduck() restores them.
+
+We use pactl as a subprocess rather than pulsectl-asyncio because PipeWire's
+PA compat layer silently drops volume-set operations from pulsectl's native
+protocol connection, while pactl (using libpulse) works reliably.
+
+If pactl is not installed or the PulseAudio socket is unavailable, all
+operations silently no-op — TTS still works, just without ducking.
+"""
+
+import asyncio
+import re
+import shutil
+import sys
+from dataclasses import dataclass, field
+
+_HAS_PACTL = shutil.which("pactl") is not None
+
+
+@dataclass
+class _SavedStream:
+    """Original volume state for a single sink-input."""
+
+    index: int
+    app_name: str
+    volume_pct: int  # percentage (0-100+), per PulseAudio's volume scale
+
+
+@dataclass
+class MediaDucker:
+    """Async PulseAudio volume controller for media ducking.
+
+    Fade-out steps down all external sink-input volumes over `fade_out_ms`.
+    Fade-in restores saved volumes over `fade_in_ms`. Both use ~10 steps
+    for smooth animation.
+
+    All errors are caught and logged — never blocks TTS playback.
+    """
+
+    fade_out_ms: int = 500
+    fade_in_ms: int = 1000
+    _is_ducked: bool = field(default=False, init=False)
+    _saved_streams: list[_SavedStream] = field(default_factory=list, init=False)
+    _lock: asyncio.Lock = field(default_factory=asyncio.Lock, init=False)
+
+    async def duck(self) -> None:
+        """Fade down all external audio streams. No-op if already ducked."""
+        if not _HAS_PACTL:
+            return
+        async with self._lock:
+            if self._is_ducked:
+                return
+            try:
+                await self._do_duck()
+                self._is_ducked = True
+            except Exception as e:
+                print(f"  Media duck failed (non-fatal): {e}", file=sys.stderr)
+                # Rollback partially-ducked streams to avoid the "volume ratchet"
+                try:
+                    await self._do_unduck()
+                except Exception:
+                    pass
+                self._saved_streams.clear()
+
+    async def unduck(self) -> None:
+        """Restore all audio streams to original volumes. No-op if not ducked."""
+        if not _HAS_PACTL:
+            return
+        async with self._lock:
+            if not self._is_ducked:
+                return
+            try:
+                await self._do_unduck()
+            except Exception as e:
+                print(f"  Media unduck failed (non-fatal): {e}", file=sys.stderr)
+            finally:
+                self._is_ducked = False
+                self._saved_streams.clear()
+
+    async def _pactl_set_volume(self, sink_input_idx: int, volume_pct: int) -> None:
+        """Set a sink-input's volume via pactl subprocess."""
+        proc = await asyncio.create_subprocess_exec(
+            "pactl", "set-sink-input-volume", str(sink_input_idx), f"{volume_pct}%",
+            stdout=asyncio.subprocess.DEVNULL,
+            stderr=asyncio.subprocess.DEVNULL,
+        )
+        await proc.wait()
+
+    async def _list_sink_inputs(self) -> list[_SavedStream]:
+        """Parse pactl list sink-inputs to get index, app name, and volume."""
+        proc = await asyncio.create_subprocess_exec(
+            "pactl", "list", "sink-inputs",
+            stdout=asyncio.subprocess.PIPE,
+            stderr=asyncio.subprocess.DEVNULL,
+        )
+        stdout, _ = await proc.communicate()
+        if proc.returncode != 0:
+            return []
+
+        streams = []
+        text = stdout.decode(errors="replace")
+
+        # Split into per-sink-input blocks
+        blocks = re.split(r"^Sink Input #", text, flags=re.MULTILINE)
+        for block in blocks[1:]:  # skip preamble before first match
+            # Extract index
+            idx_match = re.match(r"(\d+)", block)
+            if not idx_match:
+                continue
+            idx = int(idx_match.group(1))
+
+            # Extract volume percentage (take first channel)
+            vol_match = re.search(r"Volume:.*?(\d+)%", block)
+            vol_pct = int(vol_match.group(1)) if vol_match else 100
+
+            # Extract app name
+            app_match = re.search(r'application\.name\s*=\s*"([^"]*)"', block)
+            app_name = app_match.group(1) if app_match else "unknown"
+
+            streams.append(_SavedStream(index=idx, app_name=app_name, volume_pct=vol_pct))
+
+        return streams
+
+    async def _do_duck(self) -> None:
+        """Internal: enumerate sink-inputs and step volumes to zero."""
+        streams = await self._list_sink_inputs()
+        if not streams:
+            print("  Media duck: no sink-inputs found", file=sys.stderr)
+            return
+
+        self._saved_streams = streams
+        apps = [s.app_name for s in streams]
+        print(
+            f"  Media duck: fading {len(streams)} streams ({', '.join(apps)})",
+            file=sys.stderr,
+        )
+
+        # Step down in ~10 steps
+        n_steps = 10
+        step_delay = self.fade_out_ms / 1000.0 / n_steps
+
+        for step in range(1, n_steps + 1):
+            fraction = 1.0 - (step / n_steps)  # 0.9 → 0.0
+            tasks = []
+            for s in self._saved_streams:
+                target_pct = max(int(s.volume_pct * fraction), 0)
+                tasks.append(self._pactl_set_volume(s.index, target_pct))
+            await asyncio.gather(*tasks, return_exceptions=True)
+            await asyncio.sleep(step_delay)
+
+    async def _do_unduck(self) -> None:
+        """Internal: restore saved volumes with fade-in."""
+        if not self._saved_streams:
+            return
+
+        print(
+            f"  Media unduck: restoring {len(self._saved_streams)} streams",
+            file=sys.stderr,
+        )
+
+        # Step up in ~10 steps
+        n_steps = 10
+        step_delay = self.fade_in_ms / 1000.0 / n_steps
+
+        for step in range(1, n_steps + 1):
+            fraction = step / n_steps  # 0.1 → 1.0
+            tasks = []
+            for s in self._saved_streams:
+                target_pct = int(s.volume_pct * fraction)
+                tasks.append(self._pactl_set_volume(s.index, target_pct))
+            await asyncio.gather(*tasks, return_exceptions=True)
+            await asyncio.sleep(step_delay)

src/tts_mcp/queue.py

@@ -20,6 +20,7 @@ from pathlib import Path
 
 from .audio import PlaybackError, play_audio
 from .engines.base import TTSResult
+from .media_duck import MediaDucker
 
 
 class Priority(IntEnum):
@@ -58,6 +59,7 @@ class SpeechQueue:
         standby_tone: Path | None = None,
         cancel_tone: Path | None = None,
         shutdown_timeout: float = 30.0,
+        ducker: MediaDucker | None = None,
     ) -> None:
         self._queue: asyncio.PriorityQueue[_WorkItem] = asyncio.PriorityQueue(
             maxsize=max_depth
@@ -72,6 +74,7 @@ class SpeechQueue:
         self._standby_tone = standby_tone
         self._cancel_tone = cancel_tone
         self._shutdown_timeout = shutdown_timeout
+        self._ducker = ducker
         self._outcomes: OrderedDict[str, dict] = OrderedDict()
         self._futures: dict[str, asyncio.Future] = {}
 
@@ -178,6 +181,12 @@ class SpeechQueue:
             self._force_cancel_consumer()
             raise
         finally:
+            # Always restore media volumes on shutdown
+            if self._ducker:
+                try:
+                    await self._ducker.unduck()
+                except Exception:
+                    pass  # Non-fatal
             self._drain_remaining()
 
     def _resolve_outcome(self, item: _WorkItem, outcome: dict) -> None:
@@ -217,8 +226,8 @@ class SpeechQueue:
                 # 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
+                        self._exit_tone if self._queue.qsize() == 0
+                        else self._standby_tone
                     )
                     if exit_t:
                         try:
@@ -280,6 +289,13 @@ class SpeechQueue:
             finally:
                 self._current = None
                 self._queue.task_done()
+                # Unduck media when queue is empty — covers ALL exit paths
+                # (normal completion, PlaybackError, cancel, generic exception)
+                if self._queue.qsize() == 0 and self._ducker and not self._stopped:
+                    try:
+                        await self._ducker.unduck()
+                    except Exception:
+                        pass  # Non-fatal
 
     async def enqueue(
         self,

src/tts_mcp/server.py

@@ -17,6 +17,7 @@ from .engines.base import TTSEngine
 from .engines.kokoro import KokoroEngine
 from .engines.orpheus import OrpheusEngine
 from .engines.piper import PiperEngine
+from .media_duck import MediaDucker
 from .queue import Priority, SpeechQueue
 from .settings import settings
 from .tones import generate_tones, resolve_tone
@@ -65,20 +66,31 @@ async def app_lifespan(server: FastMCP):
     cancel_tone = resolve_tone(settings.cancel_tone, tone_paths, "cancel_tone", default="scratch")
     standby_tone = tone_paths.get("standby")
 
+    # --- Media ducker ---
+    ducker: MediaDucker | None = None
+    if settings.duck_media:
+        ducker = MediaDucker(
+            fade_out_ms=settings.duck_fade_out_ms,
+            fade_in_ms=settings.duck_fade_in_ms,
+        )
+
     queue = SpeechQueue(
         exit_tone=exit_tone,
         standby_tone=standby_tone,
         cancel_tone=cancel_tone,
         shutdown_timeout=settings.shutdown_timeout,
+        ducker=ducker,
     )
     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"
+    duck_label = "on" if ducker else "off"
     print(
         f"TTS MCP server ready on {settings.host}:{settings.port} "
-        f"with {len(engines)} engines, tones={entry_label}/{exit_label}/{cancel_label}",
+        f"with {len(engines)} engines, tones={entry_label}/{exit_label}/{cancel_label}, "
+        f"duck={duck_label}",
         file=sys.stderr,
     )
 
@@ -101,6 +113,7 @@ async def app_lifespan(server: FastMCP):
             "queue": queue,
             "voice_cache": voice_cache,
             "entry_tone": entry_tone,
+            "ducker": ducker,
         }
     finally:
         print("TTS MCP server shutting down", file=sys.stderr)
@@ -132,10 +145,18 @@ mcp = FastMCP(
 )
 
 
-def _get_state(ctx: Context) -> tuple[dict[str, TTSEngine], SpeechQueue, VoiceIdentityCache, Path | None]:
-    """Extract engines, queue, voice cache, and entry tone from lifespan context."""
+def _get_state(
+    ctx: Context,
+) -> tuple[dict[str, TTSEngine], SpeechQueue, VoiceIdentityCache, Path | None, MediaDucker | None]:
+    """Extract engines, queue, voice cache, entry tone, and ducker from lifespan context."""
     state = ctx.lifespan_context
-    return state["engines"], state["queue"], state["voice_cache"], state["entry_tone"]
+    return (
+        state["engines"],
+        state["queue"],
+        state["voice_cache"],
+        state["entry_tone"],
+        state["ducker"],
+    )
 
 
 async def _play_tone(tone_path: Path) -> None:
@@ -221,6 +242,7 @@ async def _speak_single(
     entry_tone: Path | None,
     priority: Priority,
     ctx: Context,
+    ducker: MediaDucker | None = None,
 ) -> dict:
     """Single-shot speak path — synthesize full text, then enqueue.
 
@@ -228,7 +250,14 @@ async def _speak_single(
     no sentence boundaries.
     """
     try:
-        if entry_tone:
+        # Duck external media with crossfade into entry tone
+        if ducker:
+            duck_task = asyncio.create_task(ducker.duck())
+            await asyncio.sleep(0.15)  # Let fade start — media dips to ~70%
+            if entry_tone:
+                await _play_tone(entry_tone)  # Tone crossfades over fading media
+            await duck_task  # Ensure duck completes
+        elif entry_tone:
             await _play_tone(entry_tone)
         await ctx.report_progress(progress=5, total=100, message="Entry tone")
 
@@ -273,6 +302,7 @@ async def _speak_chunked(
     entry_tone: Path | None,
     priority: Priority,
     ctx: Context,
+    ducker: MediaDucker | None = None,
 ) -> dict:
     """Chunked speak path — pipeline synthesis with playback.
 
@@ -286,7 +316,14 @@ async def _speak_chunked(
     first_enqueue_time: float | None = None
 
     try:
-        if entry_tone:
+        # Duck external media with crossfade into entry tone
+        if ducker:
+            duck_task = asyncio.create_task(ducker.duck())
+            await asyncio.sleep(0.15)  # Let fade start — media dips to ~70%
+            if entry_tone:
+                await _play_tone(entry_tone)  # Tone crossfades over fading media
+            await duck_task  # Ensure duck completes
+        elif entry_tone:
             await _play_tone(entry_tone)
         await ctx.report_progress(progress=5, total=100, message="Entry tone")
 
@@ -431,7 +468,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, entry_tone = _get_state(ctx)
+    engines, queue, voice_cache, entry_tone, ducker = _get_state(ctx)
 
     if engine not in engines:
         return {"error": f"Unknown engine: {engine}. Available: {list(engines.keys())}"}
@@ -445,8 +482,12 @@ async def speak(
 
     chunks = split_text(text)
     if len(chunks) <= 1:
-        return await _speak_single(eng, engine, text, voice, queue, entry_tone, priority, ctx)
-    return await _speak_chunked(eng, engine, chunks, voice, queue, entry_tone, priority, ctx)
+        return await _speak_single(
+            eng, engine, text, voice, queue, entry_tone, priority, ctx, ducker,
+        )
+    return await _speak_chunked(
+        eng, engine, chunks, voice, queue, entry_tone, priority, ctx, ducker,
+    )
 
 
 @mcp.tool(annotations=ToolAnnotations(readOnlyHint=True))
@@ -462,7 +503,7 @@ async def speech_status(
     Args:
         speech_id: The speech_id returned by speak().
     """
-    _, queue, _, _ = _get_state(ctx)
+    _, queue, _, _, _ = _get_state(ctx)
     return queue.get_status(speech_id)
 
 
@@ -483,7 +524,7 @@ async def cancel_speech(
     Args:
         speech_id: The speech_id returned by speak().
     """
-    _, queue, _, _ = _get_state(ctx)
+    _, queue, _, _, _ = _get_state(ctx)
     return queue.cancel(speech_id)
 
 
@@ -510,7 +551,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())}"}
@@ -545,7 +586,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 []
@@ -561,7 +602,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

@@ -49,6 +49,11 @@ class Settings(BaseSettings):
     # Cancel tone: "scratch", "reverse-roger", "none", or path to custom WAV
     cancel_tone: str = "scratch"
 
+    # Media ducking
+    duck_media: bool = True
+    duck_fade_out_ms: int = 500
+    duck_fade_in_ms: int = 1000
+
     # Graceful shutdown: max seconds to wait for current speech to finish
     shutdown_timeout: float = 30.0