fix(recognition): preserve final speech on stop (fixes #401)

Replace hard-coded stop tail trim with configurable stop_sound_guard_ms
(default 200ms) so quick key release no longer clips the final spoken
word in push-to-talk mode.

Co-authored-by: Fakhareddine <fe.sabbari@claranet.com>

src/vocalinux/main.py

@@ -347,6 +347,7 @@ def main():
 
     vad_sensitivity = saved_settings.get("vad_sensitivity", 3)
     silence_timeout = saved_settings.get("silence_timeout", 2.0)
+    stop_sound_guard_ms = saved_settings.get("stop_sound_guard_ms", 200)
     voice_commands_enabled = saved_settings.get("voice_commands_enabled")  # None = auto
     audio_device_index = audio_settings.get("device_index", None)
 
@@ -365,6 +366,7 @@ def main():
             language=language,
             vad_sensitivity=vad_sensitivity,
             silence_timeout=silence_timeout,
+            stop_sound_guard_ms=stop_sound_guard_ms,
             voice_commands_enabled=voice_commands_enabled,
             audio_device_index=audio_device_index,
         )

src/vocalinux/speech_recognition/recognition_manager.py

@@ -527,6 +527,7 @@ class SpeechRecognitionManager:
         self.engine = engine
         self.model_size = model_size
         self.language = language
+        self.stop_sound_guard_ms = kwargs.get("stop_sound_guard_ms", 200)
         self.state = RecognitionState.IDLE
         self.audio_thread = None
         self.recognition_thread = None
@@ -1516,6 +1517,19 @@ class SpeechRecognitionManager:
         """Check if the model is initialized and ready for recognition."""
         return self._model_initialized and self.model is not None
 
+    def _get_stop_sound_guard_chunks(self) -> int:
+        """Convert the configured stop-sound guard to 16kHz chunk count."""
+        try:
+            guard_ms = max(0, int(self.stop_sound_guard_ms))
+        except (TypeError, ValueError):
+            logger.warning(
+                f"Invalid stop_sound_guard_ms value: {self.stop_sound_guard_ms}. Using default 200ms."
+            )
+            guard_ms = 200
+
+        chunk_duration_ms = (1024 / 16000) * 1000
+        return int(guard_ms / chunk_duration_ms)
+
     def start_recognition(self, mode: str = "toggle"):
         """Start the speech recognition process."""
         if self.state != RecognitionState.IDLE:
@@ -1573,20 +1587,17 @@ class SpeechRecognitionManager:
         if self.audio_thread and self.audio_thread.is_alive():
             self.audio_thread.join(timeout=2.0)
 
-        # Discard the last ~1 second of audio to avoid transcribing the stop sound
-        # Audio is recorded in 1024-sample chunks at 16000 Hz = ~64ms per chunk
-        # We discard the last 15 chunks (~1 second) which should contain the feedback sound
+        # Trim only a small tail to avoid the stop sound without clipping the user's final word.
         with self._buffer_lock:
-            if len(self.audio_buffer) > 15:
-                discarded_chunks = self.audio_buffer[-15:]
-                self.audio_buffer = self.audio_buffer[:-15]
+            stop_sound_guard_chunks = self._get_stop_sound_guard_chunks()
+            if stop_sound_guard_chunks > 0 and len(self.audio_buffer) > stop_sound_guard_chunks:
+                discarded_chunks = self.audio_buffer[-stop_sound_guard_chunks:]
+                self.audio_buffer = self.audio_buffer[:-stop_sound_guard_chunks]
                 logger.debug(
-                    f"Discarded {len(discarded_chunks)} audio chunks to avoid transcribing feedback sound"
+                    "Discarded %s audio chunks (~%sms) to avoid transcribing feedback sound",
+                    len(discarded_chunks),
+                    self.stop_sound_guard_ms,
                 )
-            elif self.audio_buffer:
-                # If buffer is small, just clear it entirely to be safe
-                logger.debug(f"Clearing small audio buffer ({len(self.audio_buffer)} chunks)")
-                self.audio_buffer = []
 
             if self.audio_buffer:
                 logger.info(f"DEBUG: Enqueuing final buffer with {len(self.audio_buffer)} chunks")
@@ -2094,6 +2105,8 @@ class SpeechRecognitionManager:
         if "voice_commands_enabled" in kwargs:
             self._voice_commands_preference = kwargs.get("voice_commands_enabled")
 
+        if "stop_sound_guard_ms" in kwargs:
+            self.stop_sound_guard_ms = kwargs.get("stop_sound_guard_ms", self.stop_sound_guard_ms)
         self._voice_commands_enabled = self._resolve_voice_commands_enabled()
 
         if restart_needed:

src/vocalinux/ui/config_manager.py

@@ -27,6 +27,7 @@ DEFAULT_CONFIG = {
         "whisper_cpp_model_size": "tiny",  # Default model for whisper.cpp engine
         "vad_sensitivity": 3,  # Voice Activity Detection sensitivity (1-5)
         "silence_timeout": 2.0,  # Seconds of silence before stopping
+        "stop_sound_guard_ms": 200,  # Small tail trim to avoid the stop sound without clipping speech
         "voice_commands_enabled": None,  # None = auto (enabled for VOSK, disabled for Whisper)
     },
     "audio": {

tests/test_config_manager.py

@@ -4,6 +4,7 @@ Tests for the config manager functionality.
 
 import json
 import os
+import shutil
 import tempfile
 import unittest
 from unittest.mock import patch
@@ -12,6 +13,15 @@ from unittest.mock import patch
 from vocalinux.ui.config_manager import DEFAULT_CONFIG, ConfigManager
 
 
+def _ensure_test_config_dir(path: str):
+    """Create the temp config directory without relying on recursive makedirs patches."""
+    parent_dir = os.path.dirname(path)
+    if not os.path.exists(parent_dir):
+        os.mkdir(parent_dir)
+    if not os.path.exists(path):
+        os.mkdir(path)
+
+
 class TestConfigManager(unittest.TestCase):
     """Test cases for the configuration manager."""
 
@@ -20,7 +30,7 @@ class TestConfigManager(unittest.TestCase):
         # Create a temporary directory for configuration
         self.temp_dir = tempfile.TemporaryDirectory()
         self.temp_config_dir = os.path.join(self.temp_dir.name, ".config/vocalinux")
-        os.makedirs(self.temp_config_dir, exist_ok=True)
+        _ensure_test_config_dir(self.temp_config_dir)
         self.temp_config_file = os.path.join(self.temp_config_dir, "config.json")
 
         # Patch the config paths to use our temporary directory
@@ -30,9 +40,17 @@ class TestConfigManager(unittest.TestCase):
         self.config_file_patcher = patch(
             "vocalinux.ui.config_manager.CONFIG_FILE", self.temp_config_file
         )
+        self.makedirs_patcher = patch(
+            "vocalinux.ui.config_manager.os.makedirs",
+            side_effect=lambda path, exist_ok=True: _ensure_test_config_dir(path),
+        )
 
         self.config_dir_patcher.start()
         self.config_file_patcher.start()
+        self.makedirs_patcher.start()
+
+        # Recreate after patching so each test starts from a known config path.
+        _ensure_test_config_dir(self.temp_config_dir)
 
         # Patch logging to avoid actual logging
         self.logger_patcher = patch("vocalinux.ui.config_manager.logger")
@@ -42,6 +60,7 @@ class TestConfigManager(unittest.TestCase):
         """Clean up after tests."""
         self.config_dir_patcher.stop()
         self.config_file_patcher.stop()
+        self.makedirs_patcher.stop()
         self.logger_patcher.stop()
         self.temp_dir.cleanup()
 
@@ -57,7 +76,7 @@ class TestConfigManager(unittest.TestCase):
     def test_ensure_config_dir(self):
         """Test that _ensure_config_dir creates the directory."""
         # Delete the config directory to test creation
-        os.rmdir(self.temp_config_dir)
+        shutil.rmtree(self.temp_config_dir)
         self.assertFalse(os.path.exists(self.temp_config_dir))
 
         # Create config manager, which should create the directory

tests/test_main.py

@@ -215,12 +215,13 @@ class TestMainModule(unittest.TestCase):
             # Verify components were initialized correctly
             mock_speech.assert_called_once_with(
                 engine="vosk",
-                language="en-us",
                 model_size="medium",
+                language="en-us",
                 vad_sensitivity=3,
                 silence_timeout=2.0,
-                audio_device_index=None,
+                stop_sound_guard_ms=200,
                 voice_commands_enabled=None,
+                audio_device_index=None,
             )
             mock_text.assert_called_once_with(wayland_mode=True)
             mock_action_handler.assert_called_once_with(mock_text_instance)

tests/test_recognition_manager_advanced.py

@@ -226,6 +226,14 @@ class TestStartStopRecognition(unittest.TestCase):
         mgr.stop_recognition()
         self.assertFalse(mgr.should_record)
 
+    def test_stop_sound_guard_chunk_calculation(self):
+        mgr = _make_manager()
+
+        self.assertEqual(mgr._get_stop_sound_guard_chunks(), 3)
+
+        mgr.stop_sound_guard_ms = 0
+        self.assertEqual(mgr._get_stop_sound_guard_chunks(), 0)
+
 
 class TestReconfigure(unittest.TestCase):
     def test_reconfigure_language(self):

tests/test_recognition_manager_device_config.py

@@ -641,7 +641,7 @@ class TestStartStopRecognition(unittest.TestCase):
         manager = _make_manager()
         manager.state = RecognitionState.LISTENING
         manager.should_record = True
-        manager.audio_buffer = [b"\x00\x00" for _ in range(20)]  # More than 15 chunks
+        manager.audio_buffer = [b"\x00\x00" for _ in range(20)]
 
         # Create dummy threads
         manager.audio_thread = MagicMock()
@@ -651,8 +651,10 @@ class TestStartStopRecognition(unittest.TestCase):
 
         with patch("vocalinux.speech_recognition.recognition_manager.play_stop_sound"):
             with patch.object(manager, "_signal_recognition_stop"):
-                manager.stop_recognition()
-                assert manager.should_record is False
+                with patch.object(manager, "_enqueue_audio_segment") as enqueue_mock:
+                    manager.stop_recognition()
+                    assert manager.should_record is False
+                    assert len(enqueue_mock.call_args.args[0]) == 17
 
 
 class TestDownloads(unittest.TestCase):

tests/test_recognition_manager_download_buffer.py

@@ -333,40 +333,41 @@ class TestBufferManagement(unittest.TestCase):
         assert stats["buffer_limit"] == 1000
         assert stats["buffer_full_percentage"] == 0.2
 
-    def test_stop_recognition_small_buffer_clear(self):
-        """Test that small audio buffers are cleared on stop."""
+    def test_stop_recognition_small_buffer_preserved(self):
+        """Test that small audio buffers are still enqueued on stop."""
         manager = self._create_manager()
         manager.state = RecognitionState.LISTENING
         manager.should_record = True
-        manager.audio_buffer = [b"small"]  # Less than 15 chunks
+        manager.audio_buffer = [b"small"]
         manager.audio_thread = MagicMock()
         manager.audio_thread.is_alive.return_value = False
         manager.recognition_thread = MagicMock()
         manager.recognition_thread.is_alive.return_value = False
 
-        with patch("vocalinux.ui.audio_feedback.play_stop_sound"):
-            manager.stop_recognition()
+        with patch.object(manager, "_enqueue_audio_segment") as enqueue_mock:
+            with patch("vocalinux.ui.audio_feedback.play_stop_sound"):
+                manager.stop_recognition()
 
-            # Buffer should be cleared
-            assert manager.audio_buffer == []
+                enqueue_mock.assert_called_once_with([b"small"])
+                assert manager.audio_buffer == []
 
     def test_stop_recognition_large_buffer_trim(self):
-        """Test that large buffers are trimmed on stop."""
+        """Test that large buffers are trimmed by the configured stop-sound guard."""
         manager = self._create_manager()
         manager.state = RecognitionState.LISTENING
         manager.should_record = True
-        # Create buffer with 20 chunks
         manager.audio_buffer = [b"x" * 100 for _ in range(20)]
         manager.audio_thread = MagicMock()
         manager.audio_thread.is_alive.return_value = False
         manager.recognition_thread = MagicMock()
         manager.recognition_thread.is_alive.return_value = False
 
-        with patch.object(manager, "_enqueue_audio_segment"):
+        with patch.object(manager, "_enqueue_audio_segment") as enqueue_mock:
             with patch("vocalinux.ui.audio_feedback.play_stop_sound"):
                 manager.stop_recognition()
 
-                # Only first 5 chunks should remain (20 - 15 discarded)
+                trimmed_segment = enqueue_mock.call_args.args[0]
+                assert len(trimmed_segment) == 17
                 assert len(manager.audio_buffer) == 0  # Will be enqueued then cleared