🎨 Palette: Add persistent CLI status indicator

.jules/palette.md

@@ -0,0 +1,3 @@
+## 2024-05-23 - Async Status Indicators
+**Learning:** When using `rich.console.Status` in an event-driven app with background threads, simple `start()`/`stop()` calls can cause race conditions. If a background task finishes and calls `stop()` while the main thread has already started a new state (e.g. "Recording..."), the UI breaks.
+**Action:** Always guard `status.stop()` in background threads with a state check (e.g. `if not self.active: status.stop()`) to ensure we don't clear a valid, newer status.

src/chirp/main.py

@@ -53,16 +53,18 @@ def __init__(self, *, verbose: bool = False) -> None:
             volume=self.config.audio_feedback_volume,
         )
 
-        console = None
+        self.console = None
         for handler in self.logger.handlers:
             if isinstance(handler, RichHandler):
-                console = handler.console
+                self.console = handler.console
                 break
-        if not console:
-            console = Console(stderr=True)
+        if not self.console:
+            self.console = Console(stderr=True)
+
+        self.status_indicator = self.console.status("Idle")
 
         try:
-            with console.status("[bold green]Initializing Parakeet model...[/bold green]", spinner="dots"):
+            with self.console.status("[bold green]Initializing Parakeet model...[/bold green]", spinner="dots"):
                 self.parakeet = ParakeetManager(
                     model_name=self.config.parakeet_model,
                     quantization=self.config.parakeet_quantization,
@@ -125,6 +127,9 @@ def _start_recording(self) -> None:
         self.audio_feedback.play_start(self.config.start_sound_path)
         self.logger.info("Recording started")
 
+        self.status_indicator.update("[bold red]Recording...[/bold red]", spinner="point")
+        self.status_indicator.start()
+
         if self.config.max_recording_duration > 0:
             self._stop_timer = threading.Timer(
                 self.config.max_recording_duration, self._handle_timeout
@@ -145,26 +150,34 @@ def _stop_recording(self) -> None:
         self._recording = False
         self.audio_feedback.play_stop(self.config.stop_sound_path)
         self.logger.info("Recording stopped (%s samples)", waveform.size)
+
+        self.status_indicator.update("[bold green]Transcribing...[/bold green]", spinner="dots")
+
         self._executor.submit(self._transcribe_and_inject, waveform)
 
     def _transcribe_and_inject(self, waveform) -> None:
-        start_time = time.perf_counter()
-        if waveform.size == 0:
-            self.logger.warning("No audio samples captured")
-            return
         try:
-            text = self.parakeet.transcribe(waveform, sample_rate=16_000, language=self.config.language)
-        except Exception as exc:
-            self.logger.exception("Transcription failed: %s", exc)
-            self.audio_feedback.play_error(self.config.error_sound_path)
-            return
-        duration = time.perf_counter() - start_time
-        self.logger.debug("Transcription finished in %.2fs (chars=%s)", duration, len(text))
-        if not text.strip():
-            self.logger.info("Transcription empty; skipping paste")
-            return
-        self.logger.debug("Transcription: %s", text)
-        self.text_injector.inject(text)
+            start_time = time.perf_counter()
+            if waveform.size == 0:
+                self.logger.warning("No audio samples captured")
+                return
+            try:
+                text = self.parakeet.transcribe(waveform, sample_rate=16_000, language=self.config.language)
+            except Exception as exc:
+                self.logger.exception("Transcription failed: %s", exc)
+                self.audio_feedback.play_error(self.config.error_sound_path)
+                return
+            duration = time.perf_counter() - start_time
+            self.logger.debug("Transcription finished in %.2fs (chars=%s)", duration, len(text))
+            if not text.strip():
+                self.logger.info("Transcription empty; skipping paste")
+                return
+            self.logger.debug("Transcription: %s", text)
+            self.text_injector.inject(text)
+        finally:
+            # Only stop the status if we haven't started a new recording in the meantime.
+            if not self._recording:
+                self.status_indicator.stop()
 
     def _log_capture_status(self, message: str) -> None:
         self.logger.debug("Audio status: %s", message)

tests/test_ui_status.py

@@ -0,0 +1,116 @@
+import unittest
+from unittest.mock import MagicMock, patch
+import sys
+import types
+
+# Mock sounddevice globally because it's missing in the test environment
+# and imported at top-level by chirp.audio_capture
+if "sounddevice" not in sys.modules:
+    mock_sd = types.ModuleType("sounddevice")
+    mock_sd.InputStream = MagicMock()
+    sys.modules["sounddevice"] = mock_sd
+
+# Mock winsound if on Windows
+if sys.platform == "win32" and "winsound" not in sys.modules:
+    mock_winsound = types.ModuleType("winsound")
+    sys.modules["winsound"] = mock_winsound
+
+# Mock keyboard (imported by main.py -> keyboard_shortcuts.py)
+if "keyboard" not in sys.modules:
+    mock_keyboard_lib = types.ModuleType("keyboard")
+    sys.modules["keyboard"] = mock_keyboard_lib
+
+# Mock pyperclip (imported by text_injector.py)
+if "pyperclip" not in sys.modules:
+    mock_pyperclip = types.ModuleType("pyperclip")
+    mock_pyperclip.copy = MagicMock()
+    mock_pyperclip.paste = MagicMock()
+    mock_pyperclip.PyperclipException = Exception
+    sys.modules["pyperclip"] = mock_pyperclip
+
+from chirp.main import ChirpApp
+
+class TestUIStatus(unittest.TestCase):
+    @patch("chirp.main.ParakeetManager")
+    @patch("chirp.main.AudioCapture")
+    @patch("chirp.main.AudioFeedback")
+    @patch("chirp.main.KeyboardShortcutManager")
+    @patch("chirp.main.ConfigManager")
+    @patch("chirp.main.Console")
+    @patch("chirp.main.get_logger")
+    def test_status_lifecycle(self, mock_get_logger, MockConsole, MockConfigManager, MockKeyboard, MockAudioFeedback, MockAudioCapture, MockParakeet):
+        # Setup mocks
+        mock_console_instance = MockConsole.return_value
+        mock_status = MagicMock()
+        mock_console_instance.status.return_value = mock_status
+
+        # Mock logger to ensure no handlers so ChirpApp uses patched Console
+        mock_logger = MagicMock()
+        mock_logger.handlers = []
+        mock_get_logger.return_value = mock_logger
+
+        # Mock config
+        mock_config = MockConfigManager.return_value.load.return_value
+        mock_config.parakeet_model = "test-model"
+        mock_config.audio_feedback = True
+        mock_config.error_sound_path = None
+        mock_config.start_sound_path = None
+        mock_config.stop_sound_path = None
+        mock_config.max_recording_duration = 0
+        mock_config.clipboard_clear_delay = 0.5
+        mock_config.paste_mode = "ctrl"
+        mock_config.word_overrides = {}
+        mock_config.post_processing = ""
+        mock_config.clipboard_behavior = False
+        mock_config.threads = 1
+        mock_config.onnx_providers = "cpu"
+        mock_config.parakeet_quantization = None
+        mock_config.model_timeout = 300.0
+        mock_config.language = "en"
+        mock_config.primary_shortcut = "ctrl+space"
+
+        # Initialize app
+        app = ChirpApp()
+
+        # Verify status created (but not necessarily started for idle, or maybe "Idle" status created)
+        # In my plan: self.status_indicator = self.console.status("Idle")
+        mock_console_instance.status.assert_any_call("Idle")
+
+        # Reset mock to clear initialization calls
+        mock_status.reset_mock()
+
+        # 1. Start Recording
+        app._start_recording()
+
+        # Verify status updated to recording
+        mock_status.update.assert_called_with("[bold red]Recording...[/bold red]", spinner="point")
+        mock_status.start.assert_called()
+
+        # 2. Stop Recording
+        # Mock waveform
+        mock_waveform = MagicMock()
+        mock_waveform.size = 16000
+        app.audio_capture.stop.return_value = mock_waveform
+
+        app._stop_recording()
+
+        # Verify status updated to transcribing
+        mock_status.update.assert_called_with("[bold green]Transcribing...[/bold green]", spinner="dots")
+
+        # 3. Transcribe
+        # This is usually called by executor, but we call directly for test
+        app._transcribe_and_inject(mock_waveform)
+
+        # Verify status stopped
+        mock_status.stop.assert_called()
+
+        # 4. Race condition check: If recording started again during transcribe
+        mock_status.reset_mock()
+        app._recording = True
+        app._transcribe_and_inject(mock_waveform)
+
+        # Verify status was NOT stopped
+        mock_status.stop.assert_not_called()
+
+if __name__ == "__main__":
+    unittest.main()