feat(audio): lower other audio while dictating (#861)

* feat(audio): lower other audio while dictating

Opt-in setting lowers the default PipeWire or PulseAudio sink while the
microphone is open, then restores that exact level. It stays off by default.

* fix(test): keep pytest.approx working when numpy is mocked

Other test modules replace sys.modules["numpy"] with a MagicMock at import
and leave it there. pytest.approx then crashes on isinstance(..., numpy.bool_).

* fix(audio): restore every channel and do not duck if it cannot be undone

A failed restore-point write no longer lowers the sink. PulseAudio puts
each channel back, and a dead microphone restores playback instead of
waiting for an explicit stop. The recognition test patches the real
audio module when an older test left a mock behind.

* fix(test): put the audio-feedback mock back after the ducking test

The recognition test has to patch the real module, but later tests still
expect the mock that test_text_injector installs. Restore that mock when
the test ends, and annotate the new helpers.

* fix(test): annotate the playback-duck tests

AGENTS.md requires a type hint on every new function. These tests were the
ones still missing them.

* fix(audio): skip arming playback duck after recording already failed

* chore: retrigger review after rebase

AGENTS.md

@@ -135,6 +135,8 @@ Optional extras: `vosk`, `whisper`, `vad`, `dev`.
 ```
 src/vocalinux/
 ├── main.py, version.py, common_types.py
+├── audio/
+│   └── playback_ducker.py      # lower default sink while dictating
 ├── single_instance.py          # $XDG_DATA_HOME/vocalinux/instance.lock
 ├── auto_pause_monitor.py       # unload model while configured apps run
 ├── model_keepalive.py          # idle unload

docs/USER_GUIDE.md

@@ -34,6 +34,8 @@ When Vocalinux injects through the clipboard (the usual Wayland / ydotool path),
 
 On non-US layouts such as German Neo, that chord uses the key that types **v** on the active layout (not physical KEY_V). Nested terminal panels inside an IDE are often invisible to window-class detection. If paste lands as a literal `^V` or does nothing, open **Settings → Dictation → Clipboard Paste Shortcut** and choose **Ctrl+Shift+V**.
 
+**Other audio.** Settings → Audio → Other audio can lower speakers and headphones while the microphone is open, then put that volume back. It stays off until you turn it on. Level while dictating is a percent of the current volume (0 is silent). Quitting mid-dictation puts the volume back; if the app crashes first, the next launch does.
+
 ## Shortcuts
 
 Configure under **Settings → Shortcuts**:

pyproject.toml

@@ -124,6 +124,7 @@ required-version = ">=0.12,<0.13"
 # Resource dirs carry no .py files, but setuptools still treats them as packages.
 packages = [
     "vocalinux",
+    "vocalinux.audio",
     "vocalinux.resources",
     "vocalinux.resources.icons",
     "vocalinux.resources.icons.scalable",

src/vocalinux/audio/__init__.py

@@ -0,0 +1 @@
+"""Playback helpers that are not the dictation microphone."""

src/vocalinux/audio/playback_ducker.py

@@ -0,0 +1,804 @@
+"""Lower the default speaker volume while dictating, then put it back.
+
+PipeWire (``wpctl``) and PulseAudio (``pactl``) can read a sink and restore
+that exact level. The setting stays off until the user opts in, so an upgrade
+does not change anyone's volume. A file in the config directory remembers an
+in-progress duck: if the process quits or crashes with the microphone open,
+the next launch puts the same sink back and then deletes the file.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import math
+import os
+import re
+import shutil
+import subprocess
+import threading
+from dataclasses import dataclass
+from typing import Callable, Optional
+
+from ..common_types import CancelableTimer, SinkVolumeControl
+from ..utils.host_process import host_env
+from ..utils.paths import config_dir
+
+logger = logging.getLogger(__name__)
+
+# wpctl prints volume with two decimal places (``%.2f``). A restore that
+# required an exact match would skip itself on that rounding. One and a half
+# percent is enough for the printed value and for ordinary float noise, and
+# still smaller than a normal volume-key step.
+VOLUME_MATCH_TOLERANCE = 0.015
+
+# How long the start cue is allowed to finish before other audio is lowered.
+# The tail is a little longer than the WAV so the last samples are not cut.
+CUE_TAIL_SECONDS = 0.05
+MAX_DUCK_DELAY_SECONDS = 1.5
+
+PENDING_RECORD_NAME = "playback-duck.json"
+_COMMAND_TIMEOUT_SECONDS = 2.0
+_MAX_LINEAR_VOLUME = 10.0  # 1000%; anything past this is not a real sink level
+_SINK_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:+~-]{0,200}$")
+_WPCTL_VOLUME_RE = re.compile(r"Volume:\s*(\d+(?:\.\d+)?)")
+_WPCTL_ID_RE = re.compile(r"(?m)^id\s+(\d+)\b")
+_PACTL_PERCENT_RE = re.compile(r"(\d+(?:\.\d+)?)\s*%")
+
+CommandResult = tuple[int, str, str]
+CommandRunner = Callable[[list[str]], CommandResult]
+Which = Callable[[str], Optional[str]]
+Channels = tuple[float, ...]
+
+
+@dataclass(frozen=True)
+class PendingDuck:
+    """Per-channel volume to put back, and the level we left the sink at."""
+
+    sink_id: str
+    original_channels: Channels
+    ducked_channels: Channels
+
+    @property
+    def original_volume(self) -> float:
+        """First channel, kept so a one-channel record stays easy to log."""
+        return self.original_channels[0]
+
+    @property
+    def ducked_volume(self) -> float:
+        """First channel of the level we applied."""
+        return self.ducked_channels[0]
+
+
+def duck_delay_seconds(
+    *,
+    sound_effects_enabled: bool,
+    tone: str,
+    cue_duration_seconds: float,
+) -> float:
+    """Seconds to wait before ducking so the start cue can be heard.
+
+    Sound effects off, or the Off tone, duck immediately. Otherwise the cue's
+    WAV duration plus a short tail, and never longer than 1.5 seconds.
+    """
+    if not sound_effects_enabled or tone == "off":
+        return 0.0
+    try:
+        duration = float(cue_duration_seconds)
+    except (TypeError, ValueError):
+        duration = 0.0
+    if math.isnan(duration) or duration < 0.0:
+        duration = 0.0
+    return min(MAX_DUCK_DELAY_SECONDS, duration + CUE_TAIL_SECONDS)
+
+
+def volumes_match(left: float, right: float) -> bool:
+    """Whether two linear volumes are the same aside from readout noise."""
+    return math.isclose(left, right, rel_tol=0.0, abs_tol=VOLUME_MATCH_TOLERANCE)
+
+
+def channels_match(left: Channels, right: Channels) -> bool:
+    """Whether every channel matches, so a stereo balance change is visible."""
+    if len(left) != len(right):
+        return False
+    return all(volumes_match(a, b) for a, b in zip(left, right))
+
+
+def parse_wpctl_volume(text: str) -> Optional[float]:
+    """Parse ``wpctl get-volume`` stdout. None when it is not a volume line."""
+    match = _WPCTL_VOLUME_RE.search(text or "")
+    if match is None:
+        return None
+    return _finite_volume(float(match.group(1)))
+
+
+def parse_wpctl_sink_id(text: str) -> Optional[str]:
+    """Parse the leading ``id N`` line from ``wpctl inspect``."""
+    match = _WPCTL_ID_RE.search(text or "")
+    if match is None:
+        return None
+    sink_id = match.group(1)
+    if not _valid_sink_id(sink_id):
+        return None
+    return sink_id
+
+
+def parse_pactl_channels(text: str) -> Optional[Channels]:
+    """Every channel percent on the ``Volume:`` line. Ignores Base Volume.
+
+    Left and right are often different. Saving only the first one and writing
+    it back to the whole sink would flatten that balance.
+    """
+    for line in (text or "").splitlines():
+        stripped = line.strip()
+        if not stripped.lower().startswith("volume:"):
+            continue
+        matches = _PACTL_PERCENT_RE.findall(stripped)
+        if not matches:
+            return None
+        channels: list[float] = []
+        for raw in matches:
+            value = _finite_volume(float(raw) / 100.0)
+            if value is None:
+                return None
+            channels.append(value)
+        return tuple(channels)
+    return None
+
+
+def parse_pactl_volume(text: str) -> Optional[float]:
+    """First channel of :func:`parse_pactl_channels`, or None."""
+    channels = parse_pactl_channels(text)
+    if not channels:
+        return None
+    return channels[0]
+
+
+def parse_pactl_short_sink_ids(text: str) -> set[str]:
+    """Sink indexes and names from ``pactl list short sinks``."""
+    found: set[str] = set()
+    for line in (text or "").splitlines():
+        parts = line.split()
+        if not parts:
+            continue
+        if _valid_sink_id(parts[0]):
+            found.add(parts[0])
+        if len(parts) > 1 and _valid_sink_id(parts[1]):
+            found.add(parts[1])
+    return found
+
+
+def _finite_volume(value: float) -> Optional[float]:
+    if math.isnan(value) or math.isinf(value) or value < 0.0 or value > _MAX_LINEAR_VOLUME:
+        return None
+    return value
+
+
+def _valid_sink_id(sink_id: str) -> bool:
+    """Reject ids that are empty, a flag, or not a single argv token."""
+    return bool(_SINK_ID_RE.fullmatch(sink_id))
+
+
+def _valid_volume(value: float) -> bool:
+    return _finite_volume(value) is not None
+
+
+def _finite_channels(channels: Channels) -> Optional[Channels]:
+    """A non-empty tuple of finite volumes, or None."""
+    if not channels:
+        return None
+    checked: list[float] = []
+    for channel in channels:
+        value = _finite_volume(channel)
+        if value is None:
+            return None
+        checked.append(value)
+    return tuple(checked)
+
+
+def _output_says_missing(stdout: str, stderr: str) -> bool:
+    text = f"{stdout}\n{stderr}".lower()
+    return any(
+        needle in text
+        for needle in ("not found", "no such", "does not exist", "no sink", "unknown entity")
+    )
+
+
+def _run_host_command(args: list[str]) -> CommandResult:
+    """Run a host volume tool. The caller handles a non-zero status."""
+    completed = subprocess.run(
+        args,
+        capture_output=True,
+        text=True,
+        timeout=_COMMAND_TIMEOUT_SECONDS,
+        check=False,
+        env=host_env(),
+    )
+    return completed.returncode, completed.stdout or "", completed.stderr or ""
+
+
+def _default_playback_duck_settings() -> tuple[bool, int]:
+    """Read the saved switch and percent without creating a config on first import.
+
+    Constructing ConfigManager on a machine with no config file seeds the
+    language and shells out. Dictation tests, and a first run that has not
+    opened Settings, should see the packaged default (off, 20%) instead.
+    """
+    from ..ui.config_manager import (
+        CONFIG_FILE,
+        DEFAULT_PLAYBACK_DUCK_PERCENT,
+        get_shared_config_manager,
+        peek_shared_config_manager,
+    )
+
+    manager = peek_shared_config_manager()
+    if manager is None and os.path.isfile(CONFIG_FILE):
+        manager = get_shared_config_manager()
+    if manager is None:
+        return False, DEFAULT_PLAYBACK_DUCK_PERCENT
+    return manager.is_playback_duck_enabled(), manager.get_playback_duck_percent()
+
+
+def _default_enabled() -> bool:
+    enabled, _percent = _default_playback_duck_settings()
+    return enabled
+
+
+def _default_percent() -> int:
+    _enabled, percent = _default_playback_duck_settings()
+    return percent
+
+
+class SystemSinkVolume:
+    """Default-sink volume through ``wpctl``, or ``pactl`` when ``wpctl`` is absent.
+
+    The sink id captured at duck time is the only id later commands receive.
+    ``@DEFAULT_*`` is used to find that sink, not to restore it: the default
+    may have moved, and a missing sink must not spill onto another output.
+    """
+
+    def __init__(
+        self,
+        runner: Optional[CommandRunner] = None,
+        which: Optional[Which] = None,
+    ) -> None:
+        self._runner = runner or _run_host_command
+        self._which = which or shutil.which
+        if self._which("wpctl"):
+            self._backend: Optional[str] = "wpctl"
+        elif self._which("pactl"):
+            self._backend = "pactl"
+        else:
+            self._backend = None
+
+    def default_sink(self) -> Optional[tuple[str, Channels]]:
+        """Identify the default sink and read each channel's linear volume."""
+        if self._backend == "wpctl":
+            return self._wpctl_default()
+        if self._backend == "pactl":
+            return self._pactl_default()
+        logger.warning("Neither wpctl nor pactl is available; not changing playback volume")
+        return None
+
+    def volume_of(self, sink_id: str) -> Optional[Channels]:
+        """Read ``sink_id``. None on a bad id, a missing sink, or a parse failure."""
+        if not _valid_sink_id(sink_id):
+            logger.warning("Refusing to read sink id %r", sink_id)
+            return None
+        if self._backend == "wpctl":
+            return self._wpctl_volume(sink_id)
+        if self._backend == "pactl":
+            return self._pactl_volume(sink_id)
+        return None
+
+    def sink_exists(self, sink_id: str) -> Optional[bool]:
+        """Whether ``sink_id`` is still this machine's sink."""
+        if not _valid_sink_id(sink_id) or self._backend is None:
+            return False if self._backend is not None else None
+        if self._backend == "wpctl":
+            result = self._run(["wpctl", "inspect", sink_id])
+            if result is None:
+                return None
+            code, stdout, stderr = result
+            if code == 0 and parse_wpctl_sink_id(stdout) == sink_id:
+                return True
+            if _output_says_missing(stdout, stderr):
+                return False
+            return None
+        result = self._run(["pactl", "list", "short", "sinks"])
+        if result is None:
+            return None
+        code, stdout, stderr = result
+        if code != 0:
+            if _output_says_missing(stdout, stderr):
+                return False
+            return None
+        return sink_id in parse_pactl_short_sink_ids(stdout)
+
+    def set_volume(self, sink_id: str, channels: Channels) -> bool:
+        """Set one sink, one argument per channel. A failure leaves the server as it was."""
+        if not _valid_sink_id(sink_id):
+            logger.warning("Refusing to set volume on sink id %r", sink_id)
+            return False
+        checked = _finite_channels(channels)
+        if checked is None:
+            logger.warning("Refusing to set sink %s to an invalid volume %r", sink_id, channels)
+            return False
+        if self._backend == "wpctl":
+            # wpctl reports one sink volume. Per-channel balance is not in that reading.
+            result = self._run(["wpctl", "set-volume", sink_id, f"{checked[0]:.6f}"])
+        elif self._backend == "pactl":
+            percents = [f"{channel * 100.0:.4f}%" for channel in checked]
+            result = self._run(["pactl", "set-sink-volume", sink_id, *percents])
+        else:
+            logger.warning("Neither wpctl nor pactl is available; not changing playback volume")
+            return False
+        if result is None or result[0] != 0:
+            detail = ""
+            if result is not None:
+                detail = (result[2] or result[1]).strip()
+            logger.warning(
+                "Could not set volume on sink %s%s", sink_id, f": {detail}" if detail else ""
+            )
+            return False
+        return True
+
+    def _wpctl_default(self) -> Optional[tuple[str, Channels]]:
+        inspected = self._run(["wpctl", "inspect", "@DEFAULT_AUDIO_SINK@"])
+        if inspected is None or inspected[0] != 0:
+            logger.warning("Could not identify the default audio sink")
+            return None
+        sink_id = parse_wpctl_sink_id(inspected[1])
+        if sink_id is None:
+            logger.warning("Could not parse the default sink id from wpctl inspect")
+            return None
+        volume = self._wpctl_volume(sink_id)
+        if volume is None:
+            return None
+        return sink_id, volume
+
+    def _wpctl_volume(self, sink_id: str) -> Optional[Channels]:
+        result = self._run(["wpctl", "get-volume", sink_id])
+        if result is None or result[0] != 0:
+            logger.warning("wpctl get-volume failed for sink %s", sink_id)
+            return None
+        volume = parse_wpctl_volume(result[1])
+        if volume is None:
+            logger.warning("Could not parse wpctl volume %r", result[1].strip()[:200])
+            return None
+        return (volume,)
+
+    def _pactl_default(self) -> Optional[tuple[str, Channels]]:
+        named = self._run(["pactl", "get-default-sink"])
+        if named is None or named[0] != 0:
+            logger.warning("Could not identify the default PulseAudio sink")
+            return None
+        lines = [line.strip() for line in named[1].splitlines() if line.strip()]
+        sink_id = lines[-1] if lines else ""
+        if not _valid_sink_id(sink_id):
+            logger.warning("Unexpected default sink name %r", sink_id)
+            return None
+        volume = self._pactl_volume(sink_id)
+        if volume is None:
+            return None
+        return sink_id, volume
+
+    def _pactl_volume(self, sink_id: str) -> Optional[Channels]:
+        result = self._run(["pactl", "get-sink-volume", sink_id])
+        if result is None or result[0] != 0:
+            logger.warning("pactl get-sink-volume failed for sink %s", sink_id)
+            return None
+        channels = parse_pactl_channels(result[1])
+        if channels is None:
+            logger.warning("Could not parse pactl volume %r", result[1].strip()[:200])
+            return None
+        return channels
+
+    def _run(self, args: list[str]) -> Optional[CommandResult]:
+        try:
+            return self._runner(args)
+        except (OSError, subprocess.SubprocessError) as exc:
+            logger.warning("Volume command %s failed: %s", args[0] if args else "?", exc)
+            return None
+
+
+class PlaybackDucker:
+    """Duck the default sink for one dictation and restore that same sink.
+
+    A second ``duck`` while one is active does nothing, so the percent is not
+    applied twice. ``restore`` writes the saved volume only when the sink is
+    still at the ducked level; if the user moved it, their level is kept.
+    Either way a finished restore deletes the on-disk record.
+    """
+
+    def __init__(
+        self,
+        control: Optional[SinkVolumeControl] = None,
+        directory: Optional[str] = None,
+        *,
+        enabled: Optional[Callable[[], bool]] = None,
+        percent: Optional[Callable[[], int]] = None,
+        recover: bool = True,
+    ) -> None:
+        self._control = control if control is not None else SystemSinkVolume()
+        self._record_path = os.path.join(
+            directory if directory is not None else config_dir(),
+            PENDING_RECORD_NAME,
+        )
+        self._enabled_fn = enabled if enabled is not None else _default_enabled
+        self._percent_fn = percent if percent is not None else _default_percent
+        self._lock = threading.Lock()
+        self._pending: Optional[PendingDuck] = None
+        if recover:
+            # A previous process may have crashed between duck and restore.
+            try:
+                self.restore()
+            except Exception:
+                logger.error("Could not restore playback volume from a previous run", exc_info=True)
+
+    def is_enabled(self) -> bool:
+        """Whether the user asked to lower other audio. Failures count as off."""
+        try:
+            return bool(self._enabled_fn())
+        except Exception:
+            logger.warning("Could not read whether playback ducking is enabled", exc_info=True)
+            return False
+
+    def duck(self) -> None:
+        """Lower the default sink to the configured percent of its current volume."""
+        with self._lock:
+            if not self.is_enabled():
+                return
+            if self._load_pending() is not None:
+                logger.info("Playback is already lowered; not stacking another duck")
+                return
+            try:
+                snapshot = self._control.default_sink()
+            except Exception:
+                logger.warning("Could not read the default sink", exc_info=True)
+                return
+            if snapshot is None:
+                logger.warning("Not lowering playback: the default sink volume could not be read")
+                return
+            sink_id, original = snapshot
+            original_channels = _finite_channels(original)
+            if not _valid_sink_id(sink_id) or original_channels is None:
+                logger.warning("Not lowering playback: the default sink reading was unusable")
+                return
+            percent = self._current_percent()
+            target = _finite_channels(
+                tuple(max(0.0, channel * (percent / 100.0)) for channel in original_channels)
+            )
+            if target is None:
+                logger.warning("Not lowering playback: computed volume %r is unusable", target)
+                return
+            # Persist the restore point before changing the sink. If that write
+            # fails, do not lower anything: a crash would otherwise forget the
+            # original level. A crash after a successful write can still put it back.
+            record = PendingDuck(sink_id, original_channels, target)
+            try:
+                self._write_pending(record)
+            except OSError:
+                logger.warning(
+                    "Could not save the volume to restore for sink %s; not lowering it",
+                    sink_id,
+                )
+                return
+            self._pending = record
+            # Percent 100 (or an already-matching level) must not touch the sink.
+            if channels_match(original_channels, target):
+                logger.info(
+                    "Playback duck level is %d%%; leaving sink %s at %.3f",
+                    percent,
+                    sink_id,
+                    original_channels[0],
+                )
+                return
+            try:
+                applied = self._control.set_volume(sink_id, target)
+            except Exception:
+                logger.warning("Could not lower sink %s", sink_id, exc_info=True)
+                return
+            if not applied:
+                # The sink was not changed, so the record would restore nothing
+                # useful and might later skip a real user volume.
+                self._forget()
+                return
+            logger.info(
+                "Lowered sink %s from %s to %s (%d%%) while dictating",
+                sink_id,
+                original_channels,
+                target,
+                percent,
+            )
+
+    def restore(self) -> None:
+        """Put the ducked sink back, unless the user changed it or it is gone."""
+        with self._lock:
+            record = self._load_pending()
+            if record is None:
+                return
+            try:
+                exists = self._control.sink_exists(record.sink_id)
+            except Exception:
+                logger.warning("Could not check sink %s", record.sink_id, exc_info=True)
+                return
+            if exists is False:
+                logger.info(
+                    "Ducked sink %s is gone; dropping the saved volume without touching another",
+                    record.sink_id,
+                )
+                self._forget()
+                return
+            if exists is None:
+                logger.warning(
+                    "Could not check sink %s; leaving the saved volume in place", record.sink_id
+                )
+                return
+            try:
+                current = self._control.volume_of(record.sink_id)
+            except Exception:
+                logger.warning("Could not read sink %s", record.sink_id, exc_info=True)
+                return
+            if current is None:
+                logger.warning(
+                    "Could not read sink %s; leaving the saved volume in place", record.sink_id
+                )
+                return
+            if not channels_match(current, record.ducked_channels):
+                logger.info(
+                    "Playback volume changed during dictation (now %s, ducked %s); "
+                    "not overwriting it",
+                    current,
+                    record.ducked_channels,
+                )
+                self._forget()
+                return
+            if not channels_match(current, record.original_channels):
+                try:
+                    restored = self._control.set_volume(record.sink_id, record.original_channels)
+                except Exception:
+                    logger.warning("Could not restore sink %s", record.sink_id, exc_info=True)
+                    return
+                if not restored:
+                    return
+                logger.info("Restored sink %s to %s", record.sink_id, record.original_channels)
+            else:
+                logger.info("Sink %s is already at the saved volume", record.sink_id)
+            self._forget()
+
+    def _current_percent(self) -> int:
+        from ..ui.config_manager import (
+            DEFAULT_PLAYBACK_DUCK_PERCENT,
+            clamp_playback_duck_percent,
+        )
+
+        try:
+            return clamp_playback_duck_percent(self._percent_fn())
+        except Exception:
+            logger.warning(
+                "Could not read the playback-duck level; using %d",
+                DEFAULT_PLAYBACK_DUCK_PERCENT,
+                exc_info=True,
+            )
+            return DEFAULT_PLAYBACK_DUCK_PERCENT
+
+    def _load_pending(self) -> Optional[PendingDuck]:
+        if self._pending is not None:
+            return self._pending
+        record = self._read_pending()
+        self._pending = record
+        return record
+
+    def _read_pending(self) -> Optional[PendingDuck]:
+        try:
+            with open(self._record_path, encoding="utf-8") as handle:
+                payload = json.load(handle)
+        except FileNotFoundError:
+            return None
+        except (OSError, UnicodeError, json.JSONDecodeError) as exc:
+            logger.warning("Could not read the playback-duck record: %s", exc)
+            return None
+        record = _record_from_payload(payload)
+        if record is None:
+            logger.warning("Dropping an unreadable playback-duck record")
+            self._clear_file()
+            return None
+        return record
+
+    def _write_pending(self, record: PendingDuck) -> None:
+        directory = os.path.dirname(self._record_path)
+        os.makedirs(directory, exist_ok=True)
+        temporary = self._record_path + ".tmp"
+        payload = {
+            "sink_id": record.sink_id,
+            "original_volume": record.original_volume,
+            "ducked_volume": record.ducked_volume,
+            "original_channels": list(record.original_channels),
+            "ducked_channels": list(record.ducked_channels),
+        }
+        try:
+            with open(temporary, "w", encoding="utf-8") as handle:
+                json.dump(payload, handle)
+                handle.write("\n")
+                handle.flush()
+                os.fsync(handle.fileno())
+            os.replace(temporary, self._record_path)
+        except OSError:
+            try:
+                os.unlink(temporary)
+            except OSError:
+                pass
+            raise
+
+    def _forget(self) -> None:
+        self._pending = None
+        self._clear_file()
+
+    def _clear_file(self) -> None:
+        try:
+            os.remove(self._record_path)
+        except FileNotFoundError:
+            return
+        except OSError as exc:
+            logger.warning("Could not remove the playback-duck record: %s", exc)
+
+
+def _record_from_payload(payload: object) -> Optional[PendingDuck]:
+    if not isinstance(payload, dict):
+        return None
+    sink_id = payload.get("sink_id")
+    original_raw = payload.get("original_volume")
+    ducked_raw = payload.get("ducked_volume")
+    if (
+        isinstance(original_raw, bool)
+        or isinstance(ducked_raw, bool)
+        or not isinstance(original_raw, (int, float))
+        or not isinstance(ducked_raw, (int, float))
+    ):
+        return None
+    if not isinstance(sink_id, str) or not _valid_sink_id(sink_id):
+        return None
+    original = float(original_raw)
+    ducked = float(ducked_raw)
+    if not _valid_volume(original) or not _valid_volume(ducked):
+        return None
+    original_channels = _channels_from_payload(payload.get("original_channels"), original)
+    ducked_channels = _channels_from_payload(payload.get("ducked_channels"), ducked)
+    if original_channels is None or ducked_channels is None:
+        return None
+    return PendingDuck(sink_id, original_channels, ducked_channels)
+
+
+def _channels_from_payload(raw: object, fallback: float) -> Optional[Channels]:
+    """Use a saved channel list, or the single legacy volume when it is absent."""
+    if raw is None:
+        checked = _finite_channels((fallback,))
+        return checked
+    if not isinstance(raw, list) or isinstance(raw, bool):
+        return None
+    numbers: list[float] = []
+    for item in raw:
+        if isinstance(item, bool) or not isinstance(item, (int, float)):
+            return None
+        numbers.append(float(item))
+    return _finite_channels(tuple(numbers))
+
+
+Scheduler = Callable[[float, Callable[[], None]], CancelableTimer]
+
+
+def _daemon_timer(delay: float, callback: Callable[[], None]) -> threading.Timer:
+    """Fire ``callback`` on a daemon timer so a missed cancel cannot pin the process."""
+    timer = threading.Timer(delay, callback)
+    timer.daemon = True
+    timer.start()
+    return timer
+
+
+class DictationDuckSession:
+    """When a dictation starts, duck after the start cue; stop cancels or restores.
+
+    The scheduler is injectable so tests can fire or skip the callback without
+    sleeping. ``cancel`` invalidates a callback that is already running too:
+    the timer thread and ``cancel`` share a lock, and the duck itself happens
+    inside that lock, so a stop cannot be followed by a late duck.
+    """
+
+    def __init__(
+        self,
+        ducker: PlaybackDucker,
+        *,
+        enabled: Callable[[], bool],
+        schedule: Optional[Scheduler] = None,
+    ) -> None:
+        self._ducker = ducker
+        self._enabled_fn = enabled
+        self._schedule = schedule or _daemon_timer
+        self._lock = threading.Lock()
+        self._generation = 0
+        self._armed = False
+        self._timer: Optional[CancelableTimer] = None
+
+    def enabled(self) -> bool:
+        """Whether this dictation should duck. A settings failure means no."""
+        try:
+            return bool(self._enabled_fn())
+        except Exception:
+            logger.warning("Could not read the playback-duck setting", exc_info=True)
+            return False
+
+    def start(self, delay_seconds: float) -> None:
+        """Arm a duck. ``delay_seconds`` of zero still goes through the scheduler."""
+        if not self.enabled():
+            return
+        try:
+            delay = float(delay_seconds)
+        except (TypeError, ValueError):
+            delay = 0.0
+        if math.isnan(delay) or delay < 0.0:
+            delay = 0.0
+        delay = min(MAX_DUCK_DELAY_SECONDS, delay)
+
+        with self._lock:
+            self._generation += 1
+            generation = self._generation
+            self._armed = True
+            previous = self._timer
+            self._timer = None
+        self._cancel_timer(previous)
+        try:
+            timer = self._schedule(delay, lambda: self._fire(generation))
+        except Exception:
+            logger.error("Could not schedule playback duck", exc_info=True)
+            with self._lock:
+                if self._generation == generation:
+                    self._armed = False
+            return
+        with self._lock:
+            if self._generation != generation or not self._armed:
+                self._cancel_timer(timer)
+                return
+            self._timer = timer
+
+    def cancel(self) -> None:
+        """Drop a duck that has not run. Safe if nothing was armed."""
+        with self._lock:
+            self._generation += 1
+            self._armed = False
+            timer = self._timer
+            self._timer = None
+        self._cancel_timer(timer)
+
+    def restore(self) -> None:
+        """Restore now. Does not cancel; call ``cancel`` first on the stop path."""
+        try:
+            self._ducker.restore()
+        except Exception:
+            logger.error("Could not restore playback volume", exc_info=True)
+
+    def _fire(self, generation: int) -> None:
+        with self._lock:
+            if generation != self._generation or not self._armed:
+                return
+            self._armed = False
+            self._timer = None
+            try:
+                self._ducker.duck()
+            except Exception:
+                logger.error("Could not lower playback volume", exc_info=True)
+
+    @staticmethod
+    def _cancel_timer(timer: Optional[CancelableTimer]) -> None:
+        if timer is None:
+            return
+        try:
+            timer.cancel()
+        except Exception:
+            logger.warning("Could not cancel the playback-duck timer", exc_info=True)
+
+
+def default_dictation_duck_session() -> DictationDuckSession:
+    """The session the recognition manager uses, including crash recovery."""
+    ducker = PlaybackDucker()
+    return DictationDuckSession(ducker, enabled=ducker.is_enabled)

src/vocalinux/common_types.py

@@ -99,3 +99,31 @@ class _EvdevCaptureDevice(Protocol):
     def active_keys(self) -> Iterable[int]:
         """Return evdev codes currently down on this device."""
         ...
+
+
+class SinkVolumeControl(Protocol):
+    """Default playback sink. Tests supply a fake; production uses wpctl or pactl."""
+
+    def default_sink(self) -> Optional[tuple[str, tuple[float, ...]]]:
+        """``(sink_id, per_channel_linear_volume)``, or None if it cannot be read."""
+        ...
+
+    def volume_of(self, sink_id: str) -> Optional[tuple[float, ...]]:
+        """Per-channel linear volume of ``sink_id``, or None if it cannot be read."""
+        ...
+
+    def sink_exists(self, sink_id: str) -> Optional[bool]:
+        """True if present, False if gone, None if presence could not be checked."""
+        ...
+
+    def set_volume(self, sink_id: str, channels: tuple[float, ...]) -> bool:
+        """Set ``sink_id`` only. False on failure. Never substitute another sink."""
+        ...
+
+
+class CancelableTimer(Protocol):
+    """A scheduled call that can be dropped before it runs."""
+
+    def cancel(self) -> None:
+        """Drop the scheduled call if it has not started."""
+        ...

src/vocalinux/speech_recognition/recognition_manager.py

@@ -23,6 +23,7 @@ from typing import TYPE_CHECKING, Callable, Optional
 if TYPE_CHECKING:
     import numpy as np
 
+from ..audio.playback_ducker import default_dictation_duck_session, duck_delay_seconds
 from ..common_types import RecognitionState
 from ..ui.audio_feedback import play_error_sound, play_start_sound, play_stop_sound
 from ..ui.config_manager import _multilingual_sibling
@@ -1221,6 +1222,16 @@ class SpeechRecognitionManager:
         self._capture_channels = 1  # Default, updated when device is opened
         self._capture_downmix_channel = None  # Speech-gated sticky N>=3 channel for open stream
 
+        # Recover a sink left quiet by a crash before doing anything slow.
+        # Tests inject a session so this never touches a real audio server.
+        injected_duck = kwargs.get("playback_duck")
+        self._playback_duck = (
+            default_dictation_duck_session() if injected_duck is None else injected_duck
+        )
+        # Serializes arm vs release: a failure path that already released the
+        # duck must not be followed by an arm that re-lowers the sink.
+        self._playback_duck_lock = threading.Lock()
+
         # Create models directory if it doesn't exist
         os.makedirs(MODELS_DIR, exist_ok=True)
 
@@ -3329,6 +3340,85 @@ class SpeechRecognitionManager:
         chunk_duration_ms = (1024 / 16000) * 1000
         return int(guard_ms / chunk_duration_ms)
 
+    def _playback_duck_delay_seconds(self) -> float:
+        """How long to let the start cue play before lowering other audio.
+
+        Sound effects off, or the Off tone, duck immediately. A mocked or
+        broken feedback module also ducks immediately rather than guessing.
+        """
+        try:
+            from ..ui import audio_feedback
+
+            enabled_fn = getattr(audio_feedback, "_is_sound_effects_enabled", None)
+            tone_fn = getattr(audio_feedback, "_resolved_tone", None)
+            duration_fn = getattr(audio_feedback, "_wav_duration_seconds", None)
+            path_fn = getattr(audio_feedback, "tone_sound_path", None)
+            # Separate checks so the type checker treats each as callable.
+            if not callable(enabled_fn) or not callable(tone_fn):
+                return 0.0
+            if not callable(duration_fn) or not callable(path_fn):
+                return 0.0
+            enabled = enabled_fn()
+            tone = tone_fn()
+            if not isinstance(enabled, bool) or not isinstance(tone, str):
+                return 0.0
+            if not enabled or tone == "off":
+                return duck_delay_seconds(
+                    sound_effects_enabled=False, tone=tone, cue_duration_seconds=0.0
+                )
+            path = path_fn(tone, "start")
+            if not isinstance(path, str):
+                return 0.0
+            duration = duration_fn(path)
+            if isinstance(duration, bool) or not isinstance(duration, (int, float)):
+                return 0.0
+            return duck_delay_seconds(
+                sound_effects_enabled=True,
+                tone=tone,
+                cue_duration_seconds=float(duration),
+            )
+        except Exception:
+            logger.warning(
+                "Could not measure the start cue; ducking other audio immediately",
+                exc_info=True,
+            )
+            return 0.0
+
+    def _arm_playback_duck(self) -> None:
+        """Schedule a duck once dictation is actually listening."""
+        try:
+            if not self._playback_duck.enabled():
+                return
+            # The capture thread may already have failed and released the duck;
+            # arming now would lower playback in the error state with nothing
+            # left to put it back.
+            with self._playback_duck_lock:
+                if not self.should_record or self.state != RecognitionState.LISTENING:
+                    return
+                self._playback_duck.start(self._playback_duck_delay_seconds())
+        except Exception:
+            logger.error("Could not schedule playback duck", exc_info=True)
+
+    def _cancel_pending_playback_duck(self) -> None:
+        """Drop a duck whose timer has not fired. Does not change the volume."""
+        try:
+            self._playback_duck.cancel()
+        except Exception:
+            logger.error("Could not cancel playback duck", exc_info=True)
+
+    def release_playback_duck(self) -> None:
+        """Cancel a not-yet-applied duck and restore the sink. Safe to call twice.
+
+        Used when dictation ends, including error exits and quitting while the
+        microphone is still open. Failures are logged and never raised.
+        """
+        with self._playback_duck_lock:
+            self._cancel_pending_playback_duck()
+            try:
+                self._playback_duck.restore()
+            except Exception:
+                logger.error("Could not restore playback volume", exc_info=True)
+
     def start_recognition(self, mode: str = "toggle") -> bool:
         """Start the speech recognition process.
 
@@ -3406,6 +3496,8 @@ class SpeechRecognitionManager:
         self.recognition_thread = threading.Thread(target=self._perform_recognition)
         self.recognition_thread.daemon = True
         self.recognition_thread.start()
+        # After the threads exist: a start that returned early must not duck.
+        self._arm_playback_duck()
         return True
 
     def stop_recognition(self):
@@ -3415,8 +3507,11 @@ class SpeechRecognitionManager:
 
         logger.info("Stopping speech recognition")
 
-        # Stop recording FIRST to prevent capturing the stop sound
+        # Mark recording over before touching the duck so a concurrent arm
+        # sees the dictation as ended. The volume goes back only once the
+        # microphone thread has left the device, and before the stop cue.
         self.should_record = False
+        self._cancel_pending_playback_duck()
 
         # Wait for audio thread to finish recording and enqueue any pending audio
         # This is critical to prevent race condition where recognition thread exits
@@ -3424,6 +3519,8 @@ class SpeechRecognitionManager:
         if self.audio_thread and self.audio_thread.is_alive():
             self.audio_thread.join(timeout=2.0)
 
+        self.release_playback_duck()
+
         # Play stop sound now that the audio thread is done and cannot capture it.
         # Kept before buffer processing so the cue still feels immediate.
         play_stop_sound()
@@ -3476,6 +3573,8 @@ class SpeechRecognitionManager:
         except ImportError as e:
             logger.error(f"Failed to import required audio libraries: {e}")
             logger.error("Please install required dependencies: pip install pyaudio numpy")
+            self.should_record = False
+            self.release_playback_duck()
             play_error_sound()
             self._update_state(RecognitionState.ERROR)
             return
@@ -3582,6 +3681,8 @@ class SpeechRecognitionManager:
                         stream = self._audio_stream
                         CHANNELS = self._capture_channels
                     else:
+                        self.should_record = False
+                        self.release_playback_duck()
                         play_error_sound()
                         audio.terminate()
                         self._update_state(RecognitionState.ERROR)
@@ -3595,6 +3696,7 @@ class SpeechRecognitionManager:
             self._recording_segment_has_speech = False
             log_level_interval = 0  # Counter for periodic level logging
             max_level_seen = 0.0
+            capture_failed = False
             # Accumulator for 512-sample Silero chunks.  When the capture rate
             # is higher than 16 kHz (e.g. 48 kHz), resampling produces fewer
             # than 1024 samples per read (~341 at 48 kHz), so the buffer may
@@ -3761,16 +3863,27 @@ class SpeechRecognitionManager:
                             continue  # Continue recording with new stream
                         else:
                             logger.error("Audio reconnection failed, stopping recording")
+                            capture_failed = True
                             break
                     else:
                         logger.warning(
                             "Audio error occurred too soon after last error, stopping recording"
                         )
+                        capture_failed = True
                         break
                 except Exception as e:
                     logger.error(f"Unexpected error reading audio data: {e}")
+                    capture_failed = True
                     break
 
+            # A dead microphone must not leave other audio lowered, or the
+            # session stuck in LISTENING, until the user happens to stop.
+            if capture_failed:
+                self.should_record = False
+                self.release_playback_duck()
+                play_error_sound()
+                self._update_state(RecognitionState.ERROR)
+
             # Clean up
             _safe_close_stream(stream)
 
@@ -3799,6 +3912,8 @@ class SpeechRecognitionManager:
 
         except Exception as e:
             logger.error(f"Error in audio recording: {e}")
+            self.should_record = False
+            self.release_playback_duck()
             play_error_sound()
             self._update_state(RecognitionState.ERROR)
 

src/vocalinux/ui/config_manager.py

@@ -41,6 +41,7 @@ SOUND_EFFECT_TONES: tuple[tuple[str, str], ...] = (
 )
 SOUND_EFFECT_TONE_IDS = frozenset(tone_id for tone_id, _label in SOUND_EFFECT_TONES)
 DEFAULT_SOUND_EFFECT_TONE = "voca"
+DEFAULT_PLAYBACK_DUCK_PERCENT = 20
 
 PASTE_SHORTCUTS: tuple[tuple[str, str], ...] = (
     ("auto", "Auto-detect"),
@@ -85,6 +86,21 @@ def _backend_from_config(config: Any) -> tuple[bool, Any]:
     return True, text_injection.get("backend", DEFAULT_TEXT_INJECTION_BACKEND)
 
 
+def clamp_playback_duck_percent(value: Any) -> int:
+    """Return ``value`` as an integer percent in 0..100.
+
+    Missing or unusable values become the default (20). Booleans are rejected
+    because ``bool`` is an ``int`` subclass and must not become 0 or 1.
+    """
+    if isinstance(value, bool) or value is None:
+        return DEFAULT_PLAYBACK_DUCK_PERCENT
+    try:
+        percent = int(round(float(value)))
+    except (TypeError, ValueError, OverflowError):
+        return DEFAULT_PLAYBACK_DUCK_PERCENT
+    return max(0, min(100, percent))
+
+
 # Default configuration
 DEFAULT_CONFIG = {
     "speech_recognition": {  # Changed section name
@@ -124,6 +140,11 @@ DEFAULT_CONFIG = {
         "enabled": True,  # Master mute for start/stop/error cues
         "tone": "voca",  # Family catalog id; missing/unknown also resolve to voca
     },
+    # Off by default so an upgrade does not change the user's speaker volume.
+    "playback_duck": {
+        "enabled": False,  # Lower the default sink while the microphone is open
+        "percent": 20,  # Percent of the current volume; 0 is silent
+    },
     "shortcuts": {
         "toggle_recognition": "right_alt+right_alt",
         "mode": "push_to_talk",  # "toggle" or "push_to_talk"
@@ -731,6 +752,29 @@ class ConfigManager:
             self.config["sound_effects"] = {}
         self.config["sound_effects"]["tone"] = normalize_sound_effect_tone(tone)
 
+    def is_playback_duck_enabled(self) -> bool:
+        """Whether other audio is lowered while dictating. Off when unset."""
+        enabled = self.config.get("playback_duck", {}).get("enabled", False)
+        return enabled if isinstance(enabled, bool) else False
+
+    def set_playback_duck_enabled(self, enabled: bool) -> None:
+        """Enable or disable lowering other audio while dictating."""
+        if "playback_duck" not in self.config:
+            self.config["playback_duck"] = {}
+        self.config["playback_duck"]["enabled"] = bool(enabled)
+
+    def get_playback_duck_percent(self) -> int:
+        """Percent of the current volume to use while dictating, clamped 0–100."""
+        return clamp_playback_duck_percent(
+            self.config.get("playback_duck", {}).get("percent", DEFAULT_PLAYBACK_DUCK_PERCENT)
+        )
+
+    def set_playback_duck_percent(self, percent: int) -> None:
+        """Save the duck level. Values outside 0–100, and non-numbers, are clamped."""
+        if "playback_duck" not in self.config:
+            self.config["playback_duck"] = {}
+        self.config["playback_duck"]["percent"] = clamp_playback_duck_percent(percent)
+
     def get_paste_shortcut(self) -> str:
         """Return the clipboard-paste shortcut id (auto when unset or unknown)."""
         return normalize_paste_shortcut(self.config.get("text_injection", {}).get("paste_shortcut"))
@@ -765,6 +809,16 @@ _shared_instance: Optional[ConfigManager] = None
 _shared_instance_lock = threading.Lock()
 
 
+def peek_shared_config_manager() -> Optional[ConfigManager]:
+    """Return the process-wide ConfigManager, or None if nothing has created it.
+
+    Callers that only need a setting during dictation use this so a missing
+    config file does not construct a manager (that seeds language and can
+    shell out) just to discover the packaged default.
+    """
+    return _shared_instance
+
+
 def get_shared_config_manager() -> ConfigManager:
     """Return the process-wide ConfigManager, creating it on first use.
 

src/vocalinux/ui/settings_dialog.py

@@ -2954,6 +2954,50 @@ class SettingsDialog(Gtk.Dialog):
         self.sound_tone_combo.connect("changed", self._on_sound_tone_changed)
         self.preview_tone_btn.connect("clicked", self._on_preview_tone_clicked)
 
+        # Other audio: opt-in duck of the default sink for the length of a dictation.
+        other_audio = PreferencesGroup(
+            title="Other audio",
+            keywords=("duck", "volume", "speakers", "music"),
+        )
+        self.duck_playback_switch = Gtk.Switch()
+        self.duck_playback_switch.set_tooltip_text(
+            "Turn speakers and headphones down while the microphone is open, "
+            "then put the volume back"
+        )
+        duck_switch_row = PreferenceRow(
+            title="Lower other audio while dictating",
+            subtitle=(
+                "Turn speakers and headphones down while the microphone is open, "
+                "then put the volume back."
+            ),
+            widget=self.duck_playback_switch,
+            keywords=("duck", "volume", "speakers", "music", "headphones"),
+        )
+        other_audio.add_row(duck_switch_row)
+
+        self.duck_level_scale = Gtk.Scale.new_with_range(Gtk.Orientation.HORIZONTAL, 0, 100, 1)
+        self.duck_level_scale.set_digits(0)
+        self.duck_level_scale.set_round_digits(0)
+        self.duck_level_scale.set_value_pos(Gtk.PositionType.RIGHT)
+        self.duck_level_scale.set_value(20)
+        self.duck_level_scale.set_draw_value(True)
+        self.duck_level_scale.set_size_request(180, -1)
+        self.duck_level_scale.set_hexpand(False)
+        self.duck_level_scale.set_halign(Gtk.Align.END)
+        self.duck_level_scale.set_sensitive(False)
+        self.duck_level_scale.set_tooltip_text("Percent of the current volume. 0 is silent.")
+        _prevent_scroll_on_hover(self.duck_level_scale)
+        duck_level_row = PreferenceRow(
+            title="Level while dictating",
+            subtitle="Percent of the current volume. 0 is silent.",
+            widget=self.duck_level_scale,
+            keywords=("duck", "volume", "speakers", "music"),
+        )
+        other_audio.add_row(duck_level_row)
+        self.audio_tab.pack_start(other_audio, False, False, 0)
+        self.duck_playback_switch.connect("state-set", self._on_duck_playback_toggled)
+        self.duck_level_scale.connect("value-changed", self._on_duck_level_changed)
+
         # Populate devices
         self._populate_audio_devices()
         self.audio_device_combo.connect("changed", self._on_audio_device_changed)
@@ -3435,6 +3479,24 @@ class SettingsDialog(Gtk.Dialog):
         logger.info(f"Sound effects {'enabled' if enabled else 'disabled'}")
         return False
 
+    def _on_duck_playback_toggled(self, _widget: Gtk.Widget, state: bool) -> bool:
+        enabled = bool(state)
+        self.duck_level_scale.set_sensitive(enabled)
+        if self._initializing or self._applying_settings:
+            return False
+        logger.info("Lower other audio while dictating toggled: %s", enabled)
+        self.config_manager.set_playback_duck_enabled(enabled)
+        self.config_manager.save_settings()
+        return False
+
+    def _on_duck_level_changed(self, scale: Gtk.Scale) -> None:
+        if self._initializing or self._applying_settings:
+            return
+        percent = int(round(scale.get_value()))
+        logger.info("Playback duck level set to %s%%", percent)
+        self.config_manager.set_playback_duck_percent(percent)
+        self.config_manager.save_settings()
+
     def _sync_tone_preview_button(self, tone_id: str) -> None:
         """Update the two-stage preview button next to the tone combo."""
         if tone_id == "off":
@@ -5519,6 +5581,10 @@ class SettingsDialog(Gtk.Dialog):
         if not self.paste_shortcut_combo.set_active_id(paste_shortcut):
             self.paste_shortcut_combo.set_active_id(DEFAULT_PASTE_SHORTCUT)
         self.sound_effects_switch.set_active(self.config_manager.is_sound_effects_enabled())
+        duck_enabled = self.config_manager.is_playback_duck_enabled()
+        self.duck_level_scale.set_value(self.config_manager.get_playback_duck_percent())
+        self.duck_playback_switch.set_active(duck_enabled)
+        self.duck_level_scale.set_sensitive(duck_enabled)
         tone_id = self.config_manager.get_sound_effects_tone()
         if not self.sound_tone_combo.set_active_id(tone_id):
             self.sound_tone_combo.set_active_id(DEFAULT_SOUND_EFFECT_TONE)

src/vocalinux/ui/tray_indicator.py

@@ -1136,6 +1136,17 @@ class TrayIndicator:
         """Quit the application."""
         logger.info("Quitting application")
 
+        # stop_recognition is not called here (it would play the stop cue and
+        # join the capture thread). Put the speakers back before the process
+        # exits; a crash that skips this still restores on the next launch.
+        engine = getattr(self, "speech_engine", None)
+        release = getattr(engine, "release_playback_duck", None)
+        if callable(release):
+            try:
+                release()
+            except Exception:
+                logger.error("Could not restore playback volume while quitting", exc_info=True)
+
         if self._suspend_handler is not None:
             self._suspend_handler.shutdown()
 

tests/test_playback_ducker.py

@@ -0,0 +1,747 @@
+"""Playback ducking: lower the default sink while dictating, then restore it.
+
+These tests never spawn wpctl or pactl. The system backend is driven with a
+fake command runner, and the recognition hook with a fake clock.
+"""
+
+import json
+import sys
+from pathlib import Path
+from typing import Any, Callable, Iterator, Optional, Union
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from vocalinux.audio import playback_ducker as duck
+from vocalinux.audio.playback_ducker import (
+    DictationDuckSession,
+    PlaybackDucker,
+    SystemSinkVolume,
+    duck_delay_seconds,
+)
+from vocalinux.ui.config_manager import (
+    DEFAULT_CONFIG,
+    DEFAULT_PLAYBACK_DUCK_PERCENT,
+    ConfigManager,
+    clamp_playback_duck_percent,
+)
+
+
+@pytest.fixture(autouse=True)
+def _numpy_is_real_while_comparing() -> Iterator[None]:
+    """Keep pytest.approx working in the full suite.
+
+    Older tests assign ``sys.modules["numpy"] = MagicMock()`` at import and
+    never put the real module back. ``pytest.approx`` then does
+    ``isinstance(value, numpy.bool_)`` and raises TypeError.
+    """
+    current = sys.modules.get("numpy")
+    real = getattr(sys, "_vocalinux_real_numpy", None)
+    if isinstance(current, MagicMock) and real is not None:
+        sys.modules["numpy"] = real
+        try:
+            yield
+        finally:
+            sys.modules["numpy"] = current
+    else:
+        yield
+
+
+class FakeSink:
+    """In-memory default sink. ``present`` False means that sink is gone."""
+
+    def __init__(
+        self,
+        sink_id: str = "sink-a",
+        volume: float = 0.5,
+        channels: Optional[tuple[float, ...]] = None,
+    ) -> None:
+        self.sink_id = sink_id
+        self._channels = channels if channels is not None else (float(volume),)
+        self.present = True
+        self.fail_read = False
+        self.sets: list[tuple[str, Union[float, tuple[float, ...]]]] = []
+
+    @property
+    def volume(self) -> float:
+        return self._channels[0]
+
+    @volume.setter
+    def volume(self, value: float) -> None:
+        # A single knob moves every channel together. Tests that assign
+        # ``sink.volume`` are describing that.
+        self._channels = tuple(float(value) for _ in self._channels)
+
+    def default_sink(self) -> Optional[tuple[str, tuple[float, ...]]]:
+        if not self.present or self.fail_read:
+            return None
+        return self.sink_id, self._channels
+
+    def volume_of(self, sink_id: str) -> Optional[tuple[float, ...]]:
+        if sink_id != self.sink_id or not self.present or self.fail_read:
+            return None
+        return self._channels
+
+    def sink_exists(self, sink_id: str) -> bool:
+        return sink_id == self.sink_id and self.present
+
+    def set_volume(self, sink_id: str, linear: tuple[float, ...]) -> bool:
+        recorded: Union[float, tuple[float, ...]] = linear[0] if len(linear) == 1 else linear
+        self.sets.append((sink_id, recorded))
+        if sink_id != self.sink_id or not self.present:
+            return False
+        self._channels = tuple(linear)
+        return True
+
+
+def _ducker(
+    tmp_path: Path,
+    sink: FakeSink,
+    *,
+    enabled: bool = True,
+    percent: int = 20,
+    recover: bool = True,
+) -> PlaybackDucker:
+    return PlaybackDucker(
+        sink,
+        str(tmp_path),
+        enabled=lambda: enabled,
+        percent=lambda: percent,
+        recover=recover,
+    )
+
+
+def _record(tmp_path: Path) -> dict:
+    return json.loads((tmp_path / duck.PENDING_RECORD_NAME).read_text(encoding="utf-8"))
+
+
+def _real_audio_feedback(monkeypatch: pytest.MonkeyPatch) -> Any:
+    """Module object the recognition manager's ``import audio_feedback`` returns.
+
+    ``test_audio_feedback`` leaves a MagicMock in ``sys.modules``. The package
+    attribute can still be the real module, and that is the one a later import
+    uses. Patching the mock measures the real cue file instead of the stub.
+    """
+    import importlib
+
+    import vocalinux.ui as ui_pkg
+
+    key = "vocalinux.ui.audio_feedback"
+    bound = getattr(ui_pkg, "audio_feedback", None)
+    leaked = sys.modules.get(key)
+    if isinstance(bound, MagicMock) or isinstance(leaked, MagicMock):
+        # Import the real module, then put the leak back so monkeypatch
+        # restores it. Leaving the real module in place breaks later tests
+        # that expect the mock (test_text_injector).
+        if key in sys.modules:
+            del sys.modules[key]
+        if isinstance(bound, MagicMock):
+            delattr(ui_pkg, "audio_feedback")
+        real = importlib.import_module(key)
+        if leaked is None:
+            sys.modules.pop(key, None)
+        else:
+            sys.modules[key] = leaked
+        if isinstance(bound, MagicMock):
+            setattr(ui_pkg, "audio_feedback", bound)
+    elif bound is not None:
+        real = bound
+    else:
+        real = importlib.import_module(key)
+    monkeypatch.setattr(ui_pkg, "audio_feedback", real, raising=False)
+    monkeypatch.setitem(sys.modules, key, real)
+    return real
+
+
+class _Timer:
+    def __init__(self, delay: float, callback: Callable[[], None]) -> None:
+        self.delay = delay
+        self.callback = callback
+        self.cancelled = False
+
+    def cancel(self) -> None:
+        self.cancelled = True
+
+    def fire(self) -> None:
+        # Invoke even after cancel so the test proves the generation guard,
+        # not merely that cancel dropped the callback.
+        self.callback()
+
+
+class _Clock:
+    def __init__(self) -> None:
+        self.timers: list[_Timer] = []
+
+    def __call__(self, delay: float, callback: Callable[[], None]) -> "_Timer":
+        timer = _Timer(delay, callback)
+        self.timers.append(timer)
+        return timer
+
+
+class _FakeDucker:
+    def __init__(self) -> None:
+        self.ducks = 0
+        self.restores = 0
+
+    def duck(self) -> None:
+        self.ducks += 1
+
+    def restore(self) -> None:
+        self.restores += 1
+
+
+class _RecordingSession:
+    def __init__(self, enabled: bool = True) -> None:
+        self._enabled = enabled
+        self.events: list = []
+
+    def enabled(self) -> bool:
+        return self._enabled
+
+    def start(self, delay_seconds: float) -> None:
+        self.events.append(("start", delay_seconds))
+
+    def cancel(self) -> None:
+        self.events.append("cancel")
+
+    def restore(self) -> None:
+        self.events.append("restore")
+
+
+def test_duck_scales_saved_volume_and_does_not_stack(tmp_path: Path) -> None:
+    sink = FakeSink(volume=0.5)
+    ducker = _ducker(tmp_path, sink, percent=20)
+
+    ducker.duck()
+
+    assert sink.volume == pytest.approx(0.1)
+    assert sink.sets == [("sink-a", pytest.approx(0.1))]
+    saved = _record(tmp_path)
+    assert saved["sink_id"] == "sink-a"
+    assert saved["original_volume"] == pytest.approx(0.5)
+    assert saved["ducked_volume"] == pytest.approx(0.1)
+    assert saved["original_channels"] == pytest.approx((0.5,))
+    assert saved["ducked_channels"] == pytest.approx((0.1,))
+
+    sink.volume = 0.9
+    ducker.duck()
+
+    assert sink.volume == pytest.approx(0.9)
+    assert len(sink.sets) == 1
+
+
+def test_zero_percent_silences_and_disabled_does_nothing(tmp_path: Path) -> None:
+    sink = FakeSink(volume=0.8)
+    silent = _ducker(tmp_path, sink, percent=0)
+    silent.duck()
+    assert sink.volume == 0.0
+    assert sink.sets == [("sink-a", 0.0)]
+
+    untouched = FakeSink(volume=0.8)
+    disabled = _ducker(tmp_path / "off", untouched, enabled=False, percent=0)
+    (tmp_path / "off").mkdir()
+    disabled.duck()
+    assert untouched.volume == 0.8
+    assert untouched.sets == []
+    assert not (tmp_path / "off" / duck.PENDING_RECORD_NAME).exists()
+
+
+def test_unwritable_restore_point_does_not_lower_the_sink(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    sink = FakeSink(volume=0.5)
+    ducker = _ducker(tmp_path, sink, percent=20)
+
+    def refuse(_record: duck.PendingDuck) -> None:
+        raise OSError("read-only config")
+
+    monkeypatch.setattr(ducker, "_write_pending", refuse)
+    ducker.duck()
+
+    assert sink.volume == pytest.approx(0.5)
+    assert sink.sets == []
+    assert ducker._pending is None
+    assert not (tmp_path / duck.PENDING_RECORD_NAME).exists()
+
+
+def test_percent_100_does_not_change_the_sink(tmp_path: Path) -> None:
+    sink = FakeSink(volume=0.5)
+    ducker = _ducker(tmp_path, sink, percent=100)
+
+    ducker.duck()
+
+    assert sink.volume == pytest.approx(0.5)
+    assert sink.sets == []
+    assert _record(tmp_path)["ducked_volume"] == pytest.approx(0.5)
+    ducker.duck()
+    assert sink.sets == []
+
+
+def test_restore_writes_saved_volume_unless_the_user_moved_it(tmp_path: Path) -> None:
+    sink = FakeSink(volume=0.5)
+    ducker = _ducker(tmp_path, sink, percent=20)
+    ducker.duck()
+
+    sink.volume = 0.1 + 0.01  # readout noise, still our ducked level
+    ducker.restore()
+    assert sink.volume == pytest.approx(0.5)
+    assert not (tmp_path / duck.PENDING_RECORD_NAME).exists()
+
+    ducker.duck()
+    sink.volume = 0.42  # user moved it while dictating
+    sets_before = len(sink.sets)
+    ducker.restore()
+    assert sink.volume == pytest.approx(0.42)
+    assert len(sink.sets) == sets_before
+    assert not (tmp_path / duck.PENDING_RECORD_NAME).exists()
+
+
+def test_unreadable_volume_keeps_the_record_and_a_missing_sink_drops_it(tmp_path: Path) -> None:
+    sink = FakeSink(volume=0.5)
+    ducker = _ducker(tmp_path, sink, percent=20)
+    ducker.duck()
+    sink.fail_read = True
+
+    ducker.restore()
+
+    assert (tmp_path / duck.PENDING_RECORD_NAME).exists()
+    assert sink.sets == [("sink-a", pytest.approx(0.1))]
+
+    sink.fail_read = False
+    sink.present = False
+    ducker.restore()
+
+    assert sink.volume == pytest.approx(0.1)
+    assert [target for target, _level in sink.sets] == ["sink-a"]
+    assert len(sink.sets) == 1
+    assert not (tmp_path / duck.PENDING_RECORD_NAME).exists()
+
+
+def test_new_instance_restores_pending_record_once(tmp_path: Path) -> None:
+    sink = FakeSink(volume=0.8)
+    first = _ducker(tmp_path, sink, percent=25)
+    first.duck()
+    assert sink.volume == pytest.approx(0.2)
+    assert (tmp_path / duck.PENDING_RECORD_NAME).is_file()
+
+    # Setting off must not skip crash recovery: the previous process already ducked.
+    second = _ducker(tmp_path, sink, enabled=False, percent=25)
+    assert sink.volume == pytest.approx(0.8)
+    assert not (tmp_path / duck.PENDING_RECORD_NAME).exists()
+    sets_after_restore = len(sink.sets)
+
+    third = _ducker(tmp_path, sink, percent=25)
+    assert sink.volume == pytest.approx(0.8)
+    assert len(sink.sets) == sets_after_restore
+    assert third is not None
+
+
+def test_corrupt_record_is_dropped_without_touching_volume(tmp_path: Path) -> None:
+    sink = FakeSink(volume=0.5)
+    path = tmp_path / duck.PENDING_RECORD_NAME
+    path.write_text(
+        json.dumps({"sink_id": "-rf", "original_volume": 0.5, "ducked_volume": 0.1}),
+        encoding="utf-8",
+    )
+
+    _ducker(tmp_path, sink)
+
+    assert sink.sets == []
+    assert sink.volume == pytest.approx(0.5)
+    assert not path.exists()
+
+
+def test_duck_delay_waits_for_the_cue_and_caps() -> None:
+    assert (
+        duck_delay_seconds(sound_effects_enabled=False, tone="voca", cue_duration_seconds=0.4)
+        == 0.0
+    )
+    assert (
+        duck_delay_seconds(sound_effects_enabled=True, tone="off", cue_duration_seconds=0.4) == 0.0
+    )
+    assert duck_delay_seconds(
+        sound_effects_enabled=True, tone="voca", cue_duration_seconds=0.4
+    ) == pytest.approx(0.45)
+    assert (
+        duck_delay_seconds(sound_effects_enabled=True, tone="voca", cue_duration_seconds=10) == 1.5
+    )
+    assert duck_delay_seconds(
+        sound_effects_enabled=True, tone="voca", cue_duration_seconds=-1
+    ) == pytest.approx(0.05)
+
+
+def test_stop_before_the_timer_does_not_duck_and_stop_after_restores() -> None:
+    clock = _Clock()
+    backend = _FakeDucker()
+    session = DictationDuckSession(backend, enabled=lambda: True, schedule=clock)
+
+    session.start(0.45)
+    assert len(clock.timers) == 1
+    assert clock.timers[0].delay == pytest.approx(0.45)
+    assert backend.ducks == 0
+
+    session.cancel()
+    session.restore()
+    clock.timers[0].fire()
+
+    assert backend.ducks == 0
+    assert backend.restores == 1
+
+    clock = _Clock()
+    backend = _FakeDucker()
+    session = DictationDuckSession(backend, enabled=lambda: True, schedule=clock)
+    session.start(0.2)
+    clock.timers[0].fire()
+    assert backend.ducks == 1
+    session.cancel()
+    session.restore()
+    clock.timers[0].fire()
+    assert backend.ducks == 1
+    assert backend.restores == 1
+
+
+def test_disabled_session_does_not_schedule() -> None:
+    clock = _Clock()
+    session = DictationDuckSession(_FakeDucker(), enabled=lambda: False, schedule=clock)
+    session.start(0.4)
+    assert clock.timers == []
+
+
+def test_daemon_timer_is_a_daemon_and_can_be_cancelled(monkeypatch: pytest.MonkeyPatch) -> None:
+    created = {}
+
+    class FakeTimer:
+        def __init__(self, delay: float, callback: Callable[[], None]) -> None:
+            created["delay"] = delay
+            created["callback"] = callback
+            self.daemon = False
+
+        def start(self) -> None:
+            created["started"] = True
+
+        def cancel(self) -> None:
+            created["cancelled"] = True
+
+    monkeypatch.setattr(duck.threading, "Timer", FakeTimer)
+    timer = duck._daemon_timer(0.2, lambda: None)
+    assert created["delay"] == pytest.approx(0.2)
+    assert created["started"] is True
+    assert timer.daemon is True
+    timer.cancel()
+    assert created["cancelled"] is True
+
+
+def test_wpctl_parse_and_commands_do_not_set_on_parse_failure() -> None:
+    assert duck.parse_wpctl_volume("Volume: 0.40\n") == pytest.approx(0.40)
+    assert duck.parse_wpctl_volume("Volume: 0.40 [MUTED]\n") == pytest.approx(0.40)
+    assert duck.parse_wpctl_volume("Volume: 1.25\n") == pytest.approx(1.25)
+    assert duck.parse_wpctl_volume("not a volume") is None
+    assert duck.parse_wpctl_sink_id("id 48, type PipeWire:Interface:Node\n") == "48"
+
+    pactl = (
+        "Volume: front-left: 32768 /  50% / -18.06 dB,   "
+        "front-right: 32768 /  40% / -23.88 dB\n"
+        "        balance 0.00\n"
+        "Base Volume: 65536 / 100% / 0.00 dB\n"
+    )
+    assert duck.parse_pactl_volume(pactl) == pytest.approx(0.50)
+
+
+def test_wpctl_is_preferred_and_parse_failure_does_not_set(tmp_path: Path) -> None:
+    calls = []
+
+    def runner(args: list[str]) -> tuple[int, str, str]:
+        calls.append(list(args))
+        if args[:3] == ["wpctl", "inspect", "@DEFAULT_AUDIO_SINK@"]:
+            return 0, "id 7, type PipeWire:Interface:Node\n", ""
+        if args[:3] == ["wpctl", "get-volume", "7"]:
+            return 0, "nope\n", ""
+        if args[1] == "set-volume":
+            raise AssertionError(f"set-volume must not run after a parse failure: {args}")
+        return 1, "", "unexpected"
+
+    control = SystemSinkVolume(
+        runner=runner,
+        which=lambda name: "/usr/bin/wpctl" if name == "wpctl" else "/usr/bin/pactl",
+    )
+    PlaybackDucker(control, str(tmp_path), enabled=lambda: True, percent=lambda: 20).duck()
+
+    assert calls[0][0] == "wpctl"
+    assert not any(args[1] == "set-volume" for args in calls)
+    assert not (tmp_path / duck.PENDING_RECORD_NAME).exists()
+
+
+def test_wpctl_sets_the_resolved_sink_and_a_missing_sink_is_not_replaced(tmp_path: Path) -> None:
+    calls = []
+
+    def runner(args: list[str]) -> tuple[int, str, str]:
+        calls.append(list(args))
+        if args[:3] == ["wpctl", "inspect", "@DEFAULT_AUDIO_SINK@"]:
+            return 0, "id 7, type PipeWire:Interface:Node\n", ""
+        if args[:3] == ["wpctl", "get-volume", "7"]:
+            return 0, "Volume: 0.50\n", ""
+        if args[:3] == ["wpctl", "set-volume", "7"]:
+            return 0, "", ""
+        return 1, "", "unexpected"
+
+    control = SystemSinkVolume(runner=runner, which=lambda name: "/usr/bin/" + name)
+    PlaybackDucker(control, str(tmp_path), enabled=lambda: True, percent=lambda: 20).duck()
+    assert ["wpctl", "set-volume", "7", "0.100000"] in calls
+
+    def missing(args: list[str]) -> tuple[int, str, str]:
+        if args[:3] == ["wpctl", "inspect", "7"]:
+            return 3, "", "Object '7' not found\n"
+        if args[1] == "set-volume":
+            raise AssertionError(args)
+        return 1, "", "unexpected"
+
+    SystemSink = SystemSinkVolume(runner=missing, which=lambda name: "/usr/bin/" + name)
+    PlaybackDucker(SystemSink, str(tmp_path), enabled=lambda: True, percent=lambda: 20)
+    assert not (tmp_path / duck.PENDING_RECORD_NAME).exists()
+
+
+def test_pactl_is_used_when_wpctl_is_absent(tmp_path: Path) -> None:
+    calls = []
+    state = {"volume": 0.50}
+    sink_name = "alsa_output.pci-0000_00_1f.3.analog-stereo"
+
+    def runner(args: list[str]) -> tuple[int, str, str]:
+        calls.append(list(args))
+        if args[:2] == ["pactl", "get-default-sink"]:
+            return 0, sink_name + "\n", ""
+        if args[:2] == ["pactl", "get-sink-volume"]:
+            percent = state["volume"] * 100.0
+            text = (
+                f"Volume: front-left: 1 / {percent:.0f}% / 0 dB,   "
+                f"front-right: 1 / {percent:.0f}% / 0 dB\n"
+                "Base Volume: 65536 / 100% / 0.00 dB\n"
+            )
+            return 0, text, ""
+        if args[:2] == ["pactl", "set-sink-volume"]:
+            state["volume"] = float(args[3].rstrip("%")) / 100.0
+            return 0, "", ""
+        if args[:3] == ["pactl", "list", "short"]:
+            return 0, f"7\t{sink_name}\tPipeWire\n", ""
+        return 1, "", "unexpected"
+
+    control = SystemSinkVolume(
+        runner=runner,
+        which=lambda name: None if name == "wpctl" else "/usr/bin/pactl",
+    )
+    ducker = PlaybackDucker(
+        control, str(tmp_path), enabled=lambda: True, percent=lambda: 20, recover=False
+    )
+    ducker.duck()
+    assert ["pactl", "set-sink-volume", sink_name, "10.0000%", "10.0000%"] in calls
+    ducker.restore()
+    assert ["pactl", "set-sink-volume", sink_name, "50.0000%", "50.0000%"] in calls
+    assert state["volume"] == pytest.approx(0.50)
+
+
+def test_pactl_restores_each_channel(tmp_path: Path) -> None:
+    calls = []
+    channels = [0.50, 0.40]
+
+    def runner(args: list[str]) -> tuple[int, str, str]:
+        calls.append(list(args))
+        if args[:2] == ["pactl", "get-default-sink"]:
+            return 0, "analog-stereo\n", ""
+        if args[:2] == ["pactl", "get-sink-volume"]:
+            text = (
+                "Volume: front-left: 1 / "
+                f"{channels[0] * 100:.0f}% / 0 dB,   front-right: 1 / "
+                f"{channels[1] * 100:.0f}% / 0 dB\n"
+                "Base Volume: 65536 / 100% / 0.00 dB\n"
+            )
+            return 0, text, ""
+        if args[:2] == ["pactl", "set-sink-volume"]:
+            channels[0] = float(args[3].rstrip("%")) / 100.0
+            channels[1] = float(args[4].rstrip("%")) / 100.0
+            return 0, "", ""
+        if args[:3] == ["pactl", "list", "short"]:
+            return 0, "7\tanalog-stereo\tPipeWire\n", ""
+        return 1, "", "unexpected"
+
+    control = SystemSinkVolume(
+        runner=runner,
+        which=lambda name: None if name == "wpctl" else "/usr/bin/pactl",
+    )
+    ducker = PlaybackDucker(
+        control, str(tmp_path), enabled=lambda: True, percent=lambda: 20, recover=False
+    )
+    ducker.duck()
+    assert ["pactl", "set-sink-volume", "analog-stereo", "10.0000%", "8.0000%"] in calls
+    ducker.restore()
+    assert ["pactl", "set-sink-volume", "analog-stereo", "50.0000%", "40.0000%"] in calls
+    assert channels == pytest.approx([0.50, 0.40])
+
+
+def test_playback_duck_percent_clamps_and_round_trips(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    config_dir = tmp_path / "vocalinux"
+    config_dir.mkdir()
+    config_file = config_dir / "config.json"
+    monkeypatch.setattr("vocalinux.ui.config_manager.CONFIG_DIR", str(config_dir))
+    monkeypatch.setattr("vocalinux.ui.config_manager.CONFIG_FILE", str(config_file))
+    monkeypatch.setattr(
+        "vocalinux.utils.system_language.detect_system_language", lambda *_a, **_k: None
+    )
+
+    assert DEFAULT_CONFIG["playback_duck"] == {"enabled": False, "percent": 20}
+    assert DEFAULT_PLAYBACK_DUCK_PERCENT == 20
+    assert clamp_playback_duck_percent(True) == 20
+    assert clamp_playback_duck_percent(None) == 20
+    assert clamp_playback_duck_percent("nope") == 20
+    assert clamp_playback_duck_percent(-4) == 0
+    assert clamp_playback_duck_percent(150) == 100
+    assert clamp_playback_duck_percent(20.4) == 20
+
+    fresh = ConfigManager()
+    assert fresh.is_playback_duck_enabled() is False
+    assert fresh.get_playback_duck_percent() == 20
+
+    config_file.write_text(
+        json.dumps({"playback_duck": {"enabled": True, "percent": 250}}),
+        encoding="utf-8",
+    )
+    loaded = ConfigManager()
+    assert loaded.is_playback_duck_enabled() is True
+    assert loaded.get_playback_duck_percent() == 100
+    loaded.set_playback_duck_percent(-10)
+    assert loaded.config["playback_duck"]["percent"] == 0
+    loaded.set_playback_duck_percent(101)
+    assert loaded.config["playback_duck"]["percent"] == 100
+    loaded.set_playback_duck_enabled(True)
+    loaded.set_playback_duck_percent(35)
+    loaded.save_config()
+
+    again = ConfigManager()
+    assert again.is_playback_duck_enabled() is True
+    assert again.get_playback_duck_percent() == 35
+    assert again.config["playback_duck"]["percent"] == 35
+
+
+def test_recognition_hook_schedules_after_start_and_restores_before_stop_cue(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+) -> None:
+    """Dictation that never starts does not duck; stop restores before the cue."""
+    from vocalinux.common_types import RecognitionState
+    from vocalinux.speech_recognition.recognition_manager import SpeechRecognitionManager
+
+    timeline = []
+    session = _RecordingSession(enabled=True)
+    session.start = lambda delay: timeline.append(("start", delay))
+    session.cancel = lambda: timeline.append("cancel")
+    session.restore = lambda: timeline.append("restore")
+
+    feedback = _real_audio_feedback(monkeypatch)
+    monkeypatch.setattr(feedback, "_is_sound_effects_enabled", lambda: True)
+    monkeypatch.setattr(feedback, "_resolved_tone", lambda: "voca")
+    monkeypatch.setattr(feedback, "tone_sound_path", lambda _tone, _kind: "/cue.wav")
+    monkeypatch.setattr(feedback, "_wav_duration_seconds", lambda _path: 0.4)
+
+    model_dir = tmp_path / "vosk-model"
+    model_dir.mkdir()
+    with patch.dict(sys.modules, {"vosk": MagicMock()}):
+        with patch.object(
+            SpeechRecognitionManager, "_get_vosk_model_path", return_value=str(model_dir)
+        ):
+            manager = SpeechRecognitionManager(engine="vosk", playback_duck=session)
+
+    assert manager._playback_duck_delay_seconds() == pytest.approx(0.45)
+    monkeypatch.setattr(feedback, "_resolved_tone", lambda: "off")
+    assert manager._playback_duck_delay_seconds() == 0.0
+    monkeypatch.setattr(feedback, "_is_sound_effects_enabled", lambda: False)
+    assert manager._playback_duck_delay_seconds() == 0.0
+    monkeypatch.setattr(feedback, "_is_sound_effects_enabled", lambda: True)
+    monkeypatch.setattr(feedback, "_resolved_tone", lambda: "voca")
+    monkeypatch.setattr(feedback, "_wav_duration_seconds", lambda _path: 2.0)
+    assert manager._playback_duck_delay_seconds() == 1.5
+    monkeypatch.setattr(feedback, "_wav_duration_seconds", lambda _path: 0.4)
+
+    manager.stop_recognition()
+    assert timeline == []
+
+    manager.state = RecognitionState.LISTENING
+    assert manager.start_recognition() is False
+    assert timeline == []
+
+    manager._model_initialized = False
+    manager.model = None
+    manager.state = RecognitionState.IDLE
+    assert manager.start_recognition() is False
+    assert timeline == []
+
+    manager._model_initialized = True
+    manager.model = object()
+    with (
+        patch(
+            "vocalinux.speech_recognition.recognition_manager.play_start_sound",
+            side_effect=lambda: timeline.append("start_sound"),
+        ),
+        patch(
+            "vocalinux.speech_recognition.recognition_manager.play_stop_sound",
+            side_effect=lambda: timeline.append("stop_sound"),
+        ),
+        patch("vocalinux.speech_recognition.recognition_manager.threading.Thread") as thread_cls,
+    ):
+        thread_cls.return_value = MagicMock()
+        assert manager.start_recognition() is True
+        manager.stop_recognition()
+
+    start_at = timeline.index("start_sound")
+    scheduled = next(item for item in timeline if isinstance(item, tuple))
+    assert scheduled[0] == "start"
+    assert scheduled[1] == pytest.approx(0.45)
+    assert start_at < timeline.index(scheduled)
+    assert timeline.index("restore") < timeline.index("stop_sound")
+    assert "cancel" in timeline[: timeline.index("restore")]
+
+    session._enabled = False
+    timeline.clear()
+    with patch("vocalinux.speech_recognition.recognition_manager.threading.Thread") as thread_cls:
+        thread_cls.return_value = MagicMock()
+        assert manager.start_recognition() is True
+    assert not any(isinstance(item, tuple) for item in timeline)
+
+
+def test_arm_after_capture_failure_does_not_schedule(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+) -> None:
+    """A capture thread that already failed must not be followed by a duck."""
+    from vocalinux.common_types import RecognitionState
+    from vocalinux.speech_recognition.recognition_manager import SpeechRecognitionManager
+
+    session = _RecordingSession(enabled=True)
+    feedback = _real_audio_feedback(monkeypatch)
+    monkeypatch.setattr(feedback, "_is_sound_effects_enabled", lambda: False)
+    monkeypatch.setattr(feedback, "_resolved_tone", lambda: "voca")
+    monkeypatch.setattr(feedback, "tone_sound_path", lambda _tone, _kind: "/cue.wav")
+    monkeypatch.setattr(feedback, "_wav_duration_seconds", lambda _path: 0.4)
+
+    model_dir = tmp_path / "vosk-model"
+    model_dir.mkdir()
+    with patch.dict(sys.modules, {"vosk": MagicMock()}):
+        with patch.object(
+            SpeechRecognitionManager, "_get_vosk_model_path", return_value=str(model_dir)
+        ):
+            manager = SpeechRecognitionManager(engine="vosk", playback_duck=session)
+
+    # The capture thread failed before start_recognition reached the arm: it
+    # released the session and ended the dictation. The late arm must not
+    # schedule a duck that nothing will restore.
+    manager.should_record = False
+    manager.release_playback_duck()
+    manager._arm_playback_duck()
+    assert session.events == ["cancel", "restore"]
+
+    manager.state = RecognitionState.ERROR
+    manager.should_record = True
+    session.events.clear()
+    manager._arm_playback_duck()
+    assert session.events == []
+
+    manager.state = RecognitionState.LISTENING
+    manager._arm_playback_duck()
+    assert session.events == [("start", 0.0)]