fix(audio): stop sink-wake from clipping start/stop cues (#804)

* fix(audio): prepend silent preroll so sink-wake does not clip cues

Short catalog tones start in a fresh paplay process while PipeWire is
still waking a suspended sink, so the first tens of milliseconds of the
cue are missing. Prepend 100ms of silence in the same WAV, cache the
result by source path, mtime, and preroll length, and fall back to the
original file if the preroll cannot be built. CI mock players keep the
raw path.

Fixes #800

* fix(audio): cache sink-wake preroll without tempfile.mkstemp

Some tests replace sys.modules["tempfile"] with a MagicMock, so
mkstemp unpacking fails and playback falls back to the raw cue.
Name the cache file from a short sha256 of the source path, mtime,
and preroll length, write a sibling .partial, then os.replace it
into place. Reload real tempfile and wave in the audio feedback
tests so they still pass after those mocks leak.

Fixes #800

* fix(tests): pin real wave for preroll without reloading tempfile

Popping tempfile out of sys.modules in the audio feedback fixture
made preroll tests pass, but later whisper tests then bound
TemporaryDirectory from a MagicMock. Import real tempfile and wave
at module top, pin only wave while audio_feedback is reimported,
and bind TemporaryDirectory from the real stdlib if a sibling test
already replaced tempfile.

Fixes #800

* fix(audio): catch specific preroll errors and type test helpers

The new preroll path swallowed every exception, which hid bugs and
disagreed with AGENTS.md. Catch OSError, ValueError, and wave.Error
on the inner partial cleanup and the outer fallback so a bad WAV or
cache write still falls back to the original cue, and other errors
still propagate. Annotate the two test helpers the same review
flagged.

* fix(audio): fall back when truncated cues raise EOFError

wave.open can raise EOFError on a truncated RIFF before playback.
Catch it with the other preroll failures so start/stop still play the
original file.

Fixes #800

src/vocalinux/ui/audio_feedback.py

@@ -4,6 +4,7 @@ Audio feedback module for Vocalinux.
 This module provides audio feedback for various recognition states.
 """
 
+import hashlib
 import logging
 import os
 import shutil
@@ -58,6 +59,14 @@ START_SOUND = _resource_manager.get_sound_path("start_recording")
 STOP_SOUND = _resource_manager.get_sound_path("stop_recording")
 ERROR_SOUND = _resource_manager.get_sound_path("error")
 
+# Silent lead-in so a suspended PipeWire/WirePlumber sink can wake before the
+# audible cue. Applied in the same WAV (one player process), not as a second
+# stream. See #800.
+_SINK_WAKE_PREROLL_MS = 100
+_REAL_AUDIO_PLAYERS = frozenset({"paplay", "aplay", "play", "mplayer"})
+_preroll_lock = threading.Lock()
+_preroll_cache: dict[tuple[str, int, int], str] = {}
+
 
 def _is_sound_effects_enabled() -> bool:
     try:
@@ -92,6 +101,78 @@ def _wav_duration_seconds(sound_path: str) -> float:
         return 0.35
 
 
+def _pcm_silence(nframes: int, nchannels: int, sampwidth: int) -> bytes:
+    """Return PCM bytes of silence matching the source sample format."""
+    if nframes < 1 or nchannels < 1 or sampwidth < 1:
+        raise ValueError("invalid PCM parameters for silence")
+    if sampwidth == 1:
+        # 8-bit WAV is unsigned; 0x80 is the zero point.
+        return bytes([0x80]) * (nframes * nchannels)
+    return b"\x00" * (nframes * nchannels * sampwidth)
+
+
+def _preroll_cache_dir() -> str:
+    cache_home = os.environ.get("XDG_CACHE_HOME") or os.path.join(os.path.expanduser("~"), ".cache")
+    path = os.path.join(cache_home, "vocalinux", "audio-preroll")
+    os.makedirs(path, exist_ok=True)
+    return path
+
+
+def _write_preroll_wav(source_path: str, dest_path: str, preroll_ms: int) -> None:
+    """Write ``source_path`` with ``preroll_ms`` of leading silence to ``dest_path``."""
+    with wave.open(source_path, "rb") as src:
+        nchannels = src.getnchannels()
+        sampwidth = src.getsampwidth()
+        framerate = src.getframerate()
+        if nchannels < 1 or sampwidth < 1 or framerate < 1:
+            raise ValueError(
+                f"invalid WAV parameters: channels={nchannels} "
+                f"sampwidth={sampwidth} framerate={framerate}"
+            )
+        audio = src.readframes(src.getnframes())
+    preroll_frames = max(1, int(round(framerate * preroll_ms / 1000.0)))
+    silence = _pcm_silence(preroll_frames, nchannels, sampwidth)
+    with wave.open(dest_path, "wb") as dst:
+        dst.setnchannels(nchannels)
+        dst.setsampwidth(sampwidth)
+        dst.setframerate(framerate)
+        dst.writeframes(silence + audio)
+
+
+def _prerolled_sound_path(sound_path: str, preroll_ms: int = _SINK_WAKE_PREROLL_MS) -> str:
+    """Return a cached WAV with silent preroll, or ``sound_path`` on failure."""
+    try:
+        abs_path = os.path.abspath(sound_path)
+        mtime_ns = os.stat(abs_path).st_mtime_ns
+        key = (abs_path, mtime_ns, preroll_ms)
+        with _preroll_lock:
+            cached = _preroll_cache.get(key)
+            if cached is not None and os.path.isfile(cached):
+                return cached
+            digest = hashlib.sha256(
+                f"{abs_path}\0{mtime_ns}\0{preroll_ms}".encode("utf-8")
+            ).hexdigest()
+            dest = os.path.join(_preroll_cache_dir(), f"{digest[:16]}.wav")
+            if os.path.isfile(dest):
+                _preroll_cache[key] = dest
+                return dest
+            partial = dest + ".partial"
+            try:
+                _write_preroll_wav(abs_path, partial, preroll_ms)
+                os.replace(partial, dest)
+            except (OSError, ValueError, wave.Error, EOFError):
+                try:
+                    os.unlink(partial)
+                except OSError:
+                    pass
+                raise
+            _preroll_cache[key] = dest
+            return dest
+    except (OSError, ValueError, wave.Error, EOFError) as exc:
+        logger.warning("Could not prepend audio preroll for %s: %s", sound_path, exc)
+        return sound_path
+
+
 def _get_audio_player():
     """
     Determine the best available audio player on the system.
@@ -126,7 +207,7 @@ def _get_audio_player():
     return None, []
 
 
-def _play_sound_file(sound_path):
+def _play_sound_file(sound_path: str) -> bool:
     """
     Play a sound file using the best available player.
 
@@ -158,31 +239,35 @@ def _play_sound_file(sound_path):
         logger.info(f"CI mode: Simulating playing sound {sound_path}")
         return True
 
+    playback_path = sound_path
+    if player in _REAL_AUDIO_PLAYERS:
+        playback_path = _prerolled_sound_path(sound_path)
+
     try:
         if player == "paplay":
             subprocess.Popen(
-                [player, sound_path],
+                [player, playback_path],
                 stdout=subprocess.DEVNULL,
                 stderr=subprocess.DEVNULL,
                 env=host_env(),
             )
         elif player == "aplay":
             subprocess.Popen(
-                [player, "-q", sound_path],
+                [player, "-q", playback_path],
                 stdout=subprocess.DEVNULL,
                 stderr=subprocess.DEVNULL,
                 env=host_env(),
             )
         elif player == "mplayer":
             subprocess.Popen(
-                [player, "-really-quiet", sound_path],
+                [player, "-really-quiet", playback_path],
                 stdout=subprocess.DEVNULL,
                 stderr=subprocess.DEVNULL,
                 env=host_env(),
             )
         elif player == "play":
             subprocess.Popen(
-                [player, "-q", sound_path],
+                [player, "-q", playback_path],
                 stdout=subprocess.DEVNULL,
                 stderr=subprocess.DEVNULL,
                 env=host_env(),

tests/test_audio_feedback.py

@@ -4,7 +4,10 @@ Tests for the audio feedback functionality.
 
 import os
 import sys
+import tempfile
 import unittest
+import wave
+from typing import Any
 from unittest.mock import patch
 
 import pytest
@@ -12,20 +15,77 @@ import pytest
 # We need to use absolute paths for patching in module scope
 AUDIO_FEEDBACK_MODULE = "vocalinux.ui.audio_feedback"
 
+# Pytest collects files in command-line order, so this module may import after
+# tests that replace tempfile/wave in sys.modules with MagicMock. Rebind to
+# the real stdlib modules for WAV fixtures. Leave real tempfile in sys.modules
+# so a later `from tempfile import TemporaryDirectory` is not a MagicMock.
+# Do not pop tempfile in the per-test fixture below.
+if getattr(tempfile, "__name__", None) != "tempfile":
+    sys.modules.pop("tempfile", None)
+    import tempfile
+if getattr(wave, "__name__", None) != "wave":
+    sys.modules.pop("wave", None)
+    import wave
+
 
 @pytest.fixture(autouse=True)
 def reset_audio_module():
     """Reset the audio_feedback module before each test to allow proper testing."""
+    # Sibling tests replace sys.modules["wave"] with a MagicMock. Pin the real
+    # module this file imported so a reimported audio_feedback gets wave.open.
+    previous_wave = sys.modules.get("wave")
+    sys.modules["wave"] = wave
+
     # Remove the mock that conftest installs
     if AUDIO_FEEDBACK_MODULE in sys.modules:
         del sys.modules[AUDIO_FEEDBACK_MODULE]
 
-    yield
+    try:
+        yield
+    finally:
+        if previous_wave is None:
+            sys.modules.pop("wave", None)
+        else:
+            sys.modules["wave"] = previous_wave
 
-    # Restore the mock after test for other tests that need it
-    from conftest import mock_audio_feedback
+        # Restore the mock after test for other tests that need it
+        from conftest import mock_audio_feedback
 
-    sys.modules[AUDIO_FEEDBACK_MODULE] = mock_audio_feedback
+        sys.modules[AUDIO_FEEDBACK_MODULE] = mock_audio_feedback
+
+
+def _write_test_wav(
+    path: str,
+    *,
+    nframes: int = 80,
+    framerate: int = 8000,
+    nchannels: int = 1,
+    sampwidth: int = 2,
+    payload: bytes | None = None,
+) -> str:
+    """Write a tiny PCM WAV fixture and return ``path``."""
+    if payload is None:
+        payload = b"\x01\x00" * (nframes * nchannels)
+    with wave.open(path, "wb") as wav_file:
+        wav_file.setnchannels(nchannels)
+        wav_file.setsampwidth(sampwidth)
+        wav_file.setframerate(framerate)
+        wav_file.writeframes(payload)
+    return path
+
+
+def _popen_argv(mock_popen: Any) -> list[str]:
+    args, _kwargs = mock_popen.call_args
+    return list(args[0])
+
+
+def _assert_played_wav(argv: list[str], player: str, extra_flags: tuple[str, ...] = ()) -> str:
+    """Assert player argv shape and return the wav path that was played."""
+    assert argv[0] == player
+    assert argv[1 : 1 + len(extra_flags)] == list(extra_flags)
+    played = argv[-1]
+    assert str(played).endswith(".wav")
+    return played
 
 
 class TestAudioFeedback(unittest.TestCase):
@@ -190,95 +250,84 @@ class TestAudioFeedback(unittest.TestCase):
         # Import the module first
         import vocalinux.ui.audio_feedback as audio_feedback
 
-        with (
-            patch.object(audio_feedback.os.path, "exists", return_value=True),
-            patch.object(
-                audio_feedback,
-                "_get_audio_player",
-                return_value=("paplay", ["wav"]),
-            ),
-            patch.object(audio_feedback.subprocess, "Popen") as mock_popen,
-        ):
-            # Call the function
-            result = audio_feedback._play_sound_file("test.wav")
+        with tempfile.TemporaryDirectory() as tmp:
+            cue = _write_test_wav(os.path.join(tmp, "cue.wav"))
+            with (
+                patch.object(
+                    audio_feedback,
+                    "_get_audio_player",
+                    return_value=("paplay", ["wav"]),
+                ),
+                patch.object(audio_feedback.subprocess, "Popen") as mock_popen,
+            ):
+                result = audio_feedback._play_sound_file(cue)
 
-            # Verify the function returned True and called Popen correctly
             self.assertTrue(result)
             mock_popen.assert_called_once()
-            args, kwargs = mock_popen.call_args
-            self.assertEqual(args[0][0], "paplay")
-            self.assertEqual(args[0][1], "test.wav")
+            played = _assert_played_wav(_popen_argv(mock_popen), "paplay")
+            self.assertTrue(os.path.isfile(played), played)
 
     def test_play_sound_file_aplay(self):
         """Test playing sound with aplay."""
         # Import the module first
         import vocalinux.ui.audio_feedback as audio_feedback
 
-        with (
-            patch.object(audio_feedback.os.path, "exists", return_value=True),
-            patch.object(
-                audio_feedback,
-                "_get_audio_player",
-                return_value=("aplay", ["wav"]),
-            ),
-            patch.object(audio_feedback.subprocess, "Popen") as mock_popen,
-        ):
-            # Call the function
-            result = audio_feedback._play_sound_file("test.wav")
+        with tempfile.TemporaryDirectory() as tmp:
+            cue = _write_test_wav(os.path.join(tmp, "cue.wav"))
+            with (
+                patch.object(
+                    audio_feedback,
+                    "_get_audio_player",
+                    return_value=("aplay", ["wav"]),
+                ),
+                patch.object(audio_feedback.subprocess, "Popen") as mock_popen,
+            ):
+                result = audio_feedback._play_sound_file(cue)
 
-            # Verify the function returned True and called Popen correctly
             self.assertTrue(result)
             mock_popen.assert_called_once()
-            args, kwargs = mock_popen.call_args
-            self.assertEqual(args[0][0], "aplay")
-            self.assertEqual(args[0][1], "-q")
-            self.assertEqual(args[0][2], "test.wav")
+            played = _assert_played_wav(_popen_argv(mock_popen), "aplay", ("-q",))
+            self.assertTrue(os.path.isfile(played), played)
 
     def test_play_sound_file_mplayer(self):
         """Test playing sound with mplayer."""
         # Import the module first
         import vocalinux.ui.audio_feedback as audio_feedback
 
-        with (
-            patch.object(audio_feedback.os.path, "exists", return_value=True),
-            patch.object(
-                audio_feedback,
-                "_get_audio_player",
-                return_value=("mplayer", ["wav"]),
-            ),
-            patch.object(audio_feedback.subprocess, "Popen") as mock_popen,
-        ):
-            # Call the function
-            result = audio_feedback._play_sound_file("test.wav")
+        with tempfile.TemporaryDirectory() as tmp:
+            cue = _write_test_wav(os.path.join(tmp, "cue.wav"))
+            with (
+                patch.object(
+                    audio_feedback,
+                    "_get_audio_player",
+                    return_value=("mplayer", ["wav"]),
+                ),
+                patch.object(audio_feedback.subprocess, "Popen") as mock_popen,
+            ):
+                result = audio_feedback._play_sound_file(cue)
 
-            # Verify the function returned True and called Popen correctly
             self.assertTrue(result)
             mock_popen.assert_called_once()
-            args, kwargs = mock_popen.call_args
-            self.assertEqual(args[0][0], "mplayer")
-            self.assertEqual(args[0][1], "-really-quiet")
-            self.assertEqual(args[0][2], "test.wav")
+            played = _assert_played_wav(_popen_argv(mock_popen), "mplayer", ("-really-quiet",))
+            self.assertTrue(os.path.isfile(played), played)
 
     def test_play_sound_file_play(self):
         """Test playing sound with play (SoX)."""
         # Import the module first
         import vocalinux.ui.audio_feedback as audio_feedback
 
-        with (
-            patch.object(audio_feedback.os.path, "exists", return_value=True),
-            patch.object(audio_feedback, "_get_audio_player", return_value=("play", ["wav"])),
-            patch.object(audio_feedback.subprocess, "Popen") as mock_popen,
-        ):
-            # Call the function
-            result = audio_feedback._play_sound_file("test.wav")
+        with tempfile.TemporaryDirectory() as tmp:
+            cue = _write_test_wav(os.path.join(tmp, "cue.wav"))
+            with (
+                patch.object(audio_feedback, "_get_audio_player", return_value=("play", ["wav"])),
+                patch.object(audio_feedback.subprocess, "Popen") as mock_popen,
+            ):
+                result = audio_feedback._play_sound_file(cue)
 
-            # Verify the function returned True and called Popen correctly
             self.assertTrue(result)
             mock_popen.assert_called_once()
-            args, kwargs = mock_popen.call_args
-            self.assertEqual(args[0][0], "play")
-            self.assertEqual(args[0][1], "-q")
-            self.assertEqual(args[0][2], "test.wav")
+            played = _assert_played_wav(_popen_argv(mock_popen), "play", ("-q",))
+            self.assertTrue(os.path.isfile(played), played)
 
     def test_play_sound_file_exception(self):
         """Test handling exception when playing sound."""
@@ -304,6 +353,124 @@ class TestAudioFeedback(unittest.TestCase):
             # Verify the function returned False
             self.assertFalse(result)
 
+    def test_play_sound_file_prepends_silent_preroll(self):
+        """Real players play a WAV with leading silence of the preroll duration."""
+        import vocalinux.ui.audio_feedback as audio_feedback
+
+        framerate = 8000
+        nchannels = 1
+        sampwidth = 2
+        nframes = 80
+        payload = b"\x01\x00" * nframes
+
+        with tempfile.TemporaryDirectory() as tmp:
+            cue = _write_test_wav(
+                os.path.join(tmp, "cue.wav"),
+                nframes=nframes,
+                framerate=framerate,
+                nchannels=nchannels,
+                sampwidth=sampwidth,
+                payload=payload,
+            )
+            with (
+                patch.object(
+                    audio_feedback,
+                    "_get_audio_player",
+                    return_value=("paplay", ["wav"]),
+                ),
+                patch.object(audio_feedback.subprocess, "Popen") as mock_popen,
+            ):
+                result = audio_feedback._play_sound_file(cue)
+
+            self.assertTrue(result)
+            mock_popen.assert_called_once()
+            played = _assert_played_wav(_popen_argv(mock_popen), "paplay")
+            self.assertTrue(os.path.isfile(played), played)
+            self.assertNotEqual(os.path.realpath(played), os.path.realpath(cue))
+
+            with wave.open(cue, "rb") as src, wave.open(played, "rb") as dst:
+                self.assertEqual(dst.getnchannels(), nchannels)
+                self.assertEqual(dst.getsampwidth(), sampwidth)
+                self.assertEqual(dst.getframerate(), framerate)
+                src_frames = src.getnframes()
+                dst_frames = dst.getnframes()
+                dst_bytes = dst.readframes(dst_frames)
+
+            preroll_ms = audio_feedback._SINK_WAKE_PREROLL_MS
+            expected_silence = max(1, int(round(framerate * preroll_ms / 1000.0)))
+            self.assertEqual(dst_frames, src_frames + expected_silence)
+
+            frame_size = nchannels * sampwidth
+            silence = dst_bytes[: expected_silence * frame_size]
+            rest = dst_bytes[expected_silence * frame_size :]
+            self.assertEqual(silence, b"\x00" * len(silence))
+            self.assertEqual(rest, payload)
+
+    def test_preroll_cache_reuses_built_file(self):
+        """The same source path and mtime reuse the prerolled WAV."""
+        import vocalinux.ui.audio_feedback as audio_feedback
+
+        with tempfile.TemporaryDirectory() as tmp:
+            cue = _write_test_wav(os.path.join(tmp, "cue.wav"))
+            with (
+                patch.object(
+                    audio_feedback,
+                    "_get_audio_player",
+                    return_value=("paplay", ["wav"]),
+                ),
+                patch.object(audio_feedback.subprocess, "Popen") as mock_popen,
+            ):
+                self.assertTrue(audio_feedback._play_sound_file(cue))
+                self.assertTrue(audio_feedback._play_sound_file(cue))
+
+            self.assertEqual(mock_popen.call_count, 2)
+            first = _assert_played_wav(list(mock_popen.call_args_list[0].args[0]), "paplay")
+            second = _assert_played_wav(list(mock_popen.call_args_list[1].args[0]), "paplay")
+            self.assertEqual(first, second)
+
+    def test_preroll_build_failure_falls_back_to_original(self):
+        """A preroll build error plays the original cue instead of failing closed."""
+        import vocalinux.ui.audio_feedback as audio_feedback
+
+        with tempfile.TemporaryDirectory() as tmp:
+            cue = _write_test_wav(os.path.join(tmp, "cue.wav"))
+            with (
+                patch.object(
+                    audio_feedback,
+                    "_get_audio_player",
+                    return_value=("paplay", ["wav"]),
+                ),
+                patch.object(audio_feedback.wave, "open", side_effect=wave.Error("boom")),
+                patch.object(audio_feedback.subprocess, "Popen") as mock_popen,
+            ):
+                result = audio_feedback._play_sound_file(cue)
+
+            self.assertTrue(result)
+            played = _assert_played_wav(_popen_argv(mock_popen), "paplay")
+            self.assertEqual(os.path.realpath(played), os.path.realpath(cue))
+
+    def test_preroll_eoferror_falls_back_to_original(self):
+        """A truncated/corrupt WAV raising EOFError still plays the original cue."""
+        import vocalinux.ui.audio_feedback as audio_feedback
+
+        with tempfile.TemporaryDirectory() as tmp:
+            cue = os.path.join(tmp, "cue.wav")
+            with open(cue, "wb") as handle:
+                handle.write(b"RIFF")
+            with (
+                patch.object(
+                    audio_feedback,
+                    "_get_audio_player",
+                    return_value=("paplay", ["wav"]),
+                ),
+                patch.object(audio_feedback.subprocess, "Popen") as mock_popen,
+            ):
+                result = audio_feedback._play_sound_file(cue)
+
+            self.assertTrue(result)
+            played = _assert_played_wav(_popen_argv(mock_popen), "paplay")
+            self.assertEqual(os.path.realpath(played), os.path.realpath(cue))
+
     def test_play_start_sound(self):
         """Test playing start sound (default tone is voca)."""
         # Import the module first
@@ -347,24 +514,23 @@ class TestAudioFeedback(unittest.TestCase):
         # Import the module first
         import vocalinux.ui.audio_feedback as audio_feedback
 
-        with (
-            patch.object(audio_feedback.os.path, "exists", return_value=True),
-            patch.object(
-                audio_feedback,
-                "_get_audio_player",
-                return_value=("ci_test_player", ["wav"]),
-            ),
-            patch.object(audio_feedback.subprocess, "Popen") as mock_popen,
-        ):
-            # Call the function
-            result = audio_feedback._play_sound_file("test.wav")
+        with tempfile.TemporaryDirectory() as tmp:
+            cue = _write_test_wav(os.path.join(tmp, "cue.wav"))
+            with (
+                patch.object(
+                    audio_feedback,
+                    "_get_audio_player",
+                    return_value=("ci_test_player", ["wav"]),
+                ),
+                patch.object(audio_feedback.subprocess, "Popen") as mock_popen,
+            ):
+                result = audio_feedback._play_sound_file(cue)
 
-            # Verify the function returned True and called Popen correctly
             self.assertTrue(result)
             mock_popen.assert_called_once()
-            args, kwargs = mock_popen.call_args
-            self.assertEqual(args[0][0], "ci_test_player")
-            self.assertEqual(args[0][1], "test.wav")
+            argv = _popen_argv(mock_popen)
+            self.assertEqual(argv[0], "ci_test_player")
+            self.assertEqual(argv[1], cue)
 
     def test_get_audio_player_github_actions_fallback(self):
         """Test ci_test_player assignment in GitHub Actions without audio player."""

tests/test_whisper_model_info.py

@@ -8,11 +8,19 @@ dialog that reported "large" as missing no matter how often it was fetched.
 """
 
 import os
+import sys
 import unittest
 from pathlib import Path
 from tempfile import TemporaryDirectory
 from unittest.mock import patch
 
+# test_recognition_manager / test_speech_recognition replace sys.modules
+# ["tempfile"] with a MagicMock at import time and never put it back. Bind
+# TemporaryDirectory from the real stdlib module even when that happens.
+if not isinstance(TemporaryDirectory, type):
+    sys.modules.pop("tempfile", None)
+    from tempfile import TemporaryDirectory
+
 from vocalinux.utils.whisper_model_info import (
     WHISPER_MODEL_SIZES,
     migrate_legacy_checkpoint_names,