diff --git a/src/vocalinux/common_types.py b/src/vocalinux/common_types.py index 070931f..2fab15d 100644 --- a/src/vocalinux/common_types.py +++ b/src/vocalinux/common_types.py @@ -72,6 +72,10 @@ class SpeechRecognitionManagerProtocol(Protocol): """Register a callback for recognized text.""" ... + def register_segment_callback(self, callback: Callable[[str, float], None]) -> None: + """Register a callback for recognized text with its audio-capture start time.""" + ... + class TextInjectorProtocol(Protocol): """Protocol defining the interface for TextInjector.""" diff --git a/src/vocalinux/main.py b/src/vocalinux/main.py index 358326b..d66f709 100644 --- a/src/vocalinux/main.py +++ b/src/vocalinux/main.py @@ -7,6 +7,7 @@ import argparse import atexit import logging import sys +import threading from typing import Optional from .utils.vosk_model_info import SUPPORTED_LANGUAGES @@ -388,6 +389,11 @@ def main(): from .ui.action_handler import ActionHandler from .ui.config_manager import get_shared_config_manager from .ui.logging_manager import initialize_logging + from .ui.transcription_history import ( + DEFAULT_MAX_ITEMS, + TranscriptionHistory, + sanitize_max_items, + ) # Initialize logging manager early initialize_logging() @@ -483,6 +489,12 @@ def main(): advanced_settings = config_manager.get_settings().get("advanced", {}) + history_settings = config_manager.get_settings().get("history", {}) + history_enabled = bool(history_settings.get("enabled", True)) + # A hand-edited config.json can hold a non-numeric or out-of-range value; + # sanitize on load so a bad preference cannot abort startup. + history_max_items = sanitize_max_items(history_settings.get("max_items", DEFAULT_MAX_ITEMS)) + logger.info(f"Final settings: engine={engine}, language={language}, model={model_size}") if audio_device_index is not None: logger.info( @@ -530,10 +542,49 @@ def main(): # Initialize action handler action_handler = ActionHandler(text_system) + # Transcription history: an in-memory, newest-first list of recent + # dictation snippets surfaced in the tray menu. One snippet == one + # dictation session (everything said between start and stop). + transcription_history = TranscriptionHistory( + max_items=history_max_items, enabled=history_enabled + ) + # Segments dictated during the open session, joined and committed to + # history when the session ends (state returns to IDLE). Each entry + # keeps the monotonic time its audio capture began: a history.clear() + # lands mid-decode, so capture time — not arrival order — decides + # whether a segment counts as pre-clear speech (dropped) or new + # dictation (kept). + # + # Session association: stop_recognition() emits IDLE after only a + # bounded wait on the recognition worker, so a slow final segment can + # still fire its text callback afterwards. Each segment must land in + # the session that produced it: while a session is open, segments + # accumulate in session_segments; a segment arriving on a worker that + # is not the open session's worker (or while no session is open) is a + # leftover of the just-ended session and is folded into its snippet + # instead of leaking into the next one. + session_lock = threading.Lock() + session_segments: list[tuple[str, float]] = [] + session_open = False + # Worker thread that produced the open session's segments; used to + # detect callbacks from a previous session's still-running worker. + session_worker: Optional[threading.Thread] = None + # True when the newest history entry is the just-closed session's + # snippet, so late segments can still merge into it. + latest_snippet_extendable = False + # Clear epoch the most-recently-ended session was committed under; + # late segments merging into its snippet are judged against it. + ended_session_epoch = transcription_history.epoch + # --- Callback wiring --------------------------------------------------- - # The speech engine emits three kinds of events, each handled by a + # The speech engine emits four kinds of events, each handled by a # dedicated callback registered below: # + # segment_callback(text: str, started_at: float) + # Fires for each finalised segment with the monotonic time its + # audio capture began, just before the text callbacks. Records + # the segment into the transcription history. + # # text_callback(text: str) # Called on the recognition thread when a transcription segment # is finalised. The wrapper below strips whitespace, optionally @@ -550,6 +601,82 @@ def main(): # listening session ends. # ------------------------------------------------------------------ + def _normalize_segment_text(text: str) -> str: + """Segment text as it is injected: stripped, auto-capitalized (Vosk).""" + normalized = text.lstrip().rstrip(" \t") + # Auto-capitalize sentences if enabled (Vosk only - Whisper outputs proper casing) + if ( + normalized + and config_manager.get("text_injection", "auto_capitalize") + and speech_engine.engine == "vosk" + ): + from vocalinux.speech_recognition.command_processor import capitalize_sentences + + normalized = capitalize_sentences(normalized) + return normalized + + def record_history_segment(segment: str, started_at: float) -> None: + """File a recognized segment under the dictation session it came from. + + Runs on the recognition worker thread as a segment callback, so + ``started_at`` is the moment the segment's audio capture began. + While a session is open, segments accumulate into that session's + pending snippet; a segment delivered after the session was + finalized — the worker can outlive the manager's bounded stop + wait and emit text after IDLE — is folded into its own session's + snippet rather than the next one. + + Segments whose capture began before the most recent + history.clear() are refused: their audio can only contain + pre-clear speech, however late the decode finishes. Segments + captured after the clear are kept, so dictation that continues + across a clear is still recoverable. + """ + nonlocal session_worker, latest_snippet_extendable + if not transcription_history.enabled: + return + segment = _normalize_segment_text(segment) + if not segment: + return + worker = threading.current_thread() + # The engine's live worker, when it exposes one: a segment from + # any other thread is a leftover from an older session. + current_worker = getattr(speech_engine, "recognition_thread", None) + with session_lock: + if started_at <= transcription_history.cleared_at: + # Captured before the last clear — must not re-enter. + return + # Segments captured while the Settings mic test runs are test + # speech, not dictation; the floor is stamped before the test + # starts recognition, so leftovers from an earlier session — + # whose capture began before it — still file normally. + test_floor = getattr(speech_engine, "test_capture_floor", None) + if isinstance(test_floor, (int, float)) and started_at >= test_floor: + return + if session_open and ( + worker is session_worker + or worker is current_worker + # When the engine exposes no worker (tests, mocks), the + # first segment of a session tags it. + or (session_worker is None and not isinstance(current_worker, threading.Thread)) + ): + session_worker = worker + session_segments.append((segment, started_at)) + return + # Late segment from a session that already ended: merge into + # its committed snippet when there is one. Both writes are + # guarded by the ended session's epoch, so a clear() landing + # between that session's commit and this delivery still + # refuses the text. + if latest_snippet_extendable and transcription_history.extend_latest( + segment, expected_epoch=ended_session_epoch + ): + return + # Otherwise the late segments are the session's only output + # and form their own snippet. + if transcription_history.add(segment, expected_epoch=ended_session_epoch): + latest_snippet_extendable = True + def text_callback_wrapper(text: str) -> None: """Bridge between speech engine text events and the text injector. @@ -564,17 +691,10 @@ def main(): """ # Preserve trailing newlines ("new line" / "new paragraph"); only # strip spaces/tabs that whisper sometimes wraps around tokens. - text_to_inject = text.lstrip().rstrip(" \t") + text_to_inject = _normalize_segment_text(text) if not text_to_inject: return - # Auto-capitalize sentences if enabled (Vosk only - Whisper outputs proper casing) - auto_capitalize = config_manager.get("text_injection", "auto_capitalize") - if auto_capitalize and speech_engine.engine == "vosk": - from vocalinux.speech_recognition.command_processor import capitalize_sentences - - text_to_inject = capitalize_sentences(text_to_inject) - # Read from disk so the Settings toggle applies without restart. append_trailing_space = _should_append_trailing_space() @@ -596,11 +716,62 @@ def main(): action_handler.set_last_injected_text(text_to_inject) def on_state_change(state: RecognitionState) -> None: - """Reset the last-injected buffer when a listening session ends.""" - if state == RecognitionState.IDLE: - action_handler.set_last_injected_text("") + """Reset the last-injected buffer when a listening session ends. - # Connect speech recognition to text injection and action handling + Also commits the just-finished dictation session to the + transcription history as a single snippet. + """ + nonlocal session_open, session_worker, latest_snippet_extendable + nonlocal ended_session_epoch + if state in (RecognitionState.IDLE, RecognitionState.ERROR): + if state == RecognitionState.IDLE: + action_handler.set_last_injected_text("") + with session_lock: + session_open = False + session_worker = None + ended_session_epoch = transcription_history.epoch + # Only segments captured after the last clear() join the + # snippet — speech captured before it is gone for good, + # while dictation continued across the clear is kept. + cleared_at = transcription_history.cleared_at + joined = " ".join( + text for text, started_at in session_segments if started_at > cleared_at + ) + session_segments.clear() + # Guarded by the epoch observed here: a clear() landing + # between this read and the add still refuses the + # snippet. The committed snippet stays open to late + # segments still trickling out of the worker; a session + # that produced no text leaves no entry to merge into. + latest_snippet_extendable = transcription_history.add( + joined, expected_epoch=ended_session_epoch + ) + else: + with session_lock: + if not session_open: + session_open = True + session_worker = None + # Segments left over by a session that ended without a + # closing state commit as their own snippet rather + # than leaking into the new session's — still only + # those captured after the last clear(). + if session_segments: + cleared_at = transcription_history.cleared_at + latest_snippet_extendable = transcription_history.add( + " ".join( + text + for text, started_at in session_segments + if started_at > cleared_at + ), + expected_epoch=transcription_history.epoch, + ) + session_segments.clear() + + # Connect speech recognition to text injection and action handling. + # Segments reach history with their capture-start time so a mid- + # session clear() can separate pre-clear speech from new dictation; + # injection keeps the plain text callback. + speech_engine.register_segment_callback(record_history_segment) speech_engine.register_text_callback(text_callback_wrapper) speech_engine.register_action_callback(action_handler.handle_action) speech_engine.register_state_callback(on_state_change) @@ -609,6 +780,7 @@ def main(): indicator = tray_indicator.TrayIndicator( speech_engine=speech_engine, text_injector=text_system, + transcription_history=transcription_history, ) # Start the GTK main loop diff --git a/src/vocalinux/speech_recognition/recognition_manager.py b/src/vocalinux/speech_recognition/recognition_manager.py index 6d8e9b9..53f6f28 100644 --- a/src/vocalinux/speech_recognition/recognition_manager.py +++ b/src/vocalinux/speech_recognition/recognition_manager.py @@ -73,6 +73,21 @@ from .command_processor import CommandProcessor from .silero_vad import SILERO_CHUNK_SIZE, load_silero_vad +class _AudioSegment(list): + """A queued audio segment that knows when its capture began. + + Stays a plain ``list`` of chunks so consumers (including tests that put + or compare raw buffers) see an ordinary segment; ``started_at`` is the + ``time.monotonic()`` timestamp of its first captured chunk. That stamp + is what lets a listener tell speech captured before a point in time + from speech captured after it, however late the decode finishes. + """ + + def __init__(self, chunks: list[bytes], started_at: Optional[float]) -> None: + super().__init__(chunks) + self.started_at = started_at + + def _pywhispercpp_distribution_version() -> Optional[tuple[int, ...]]: """Installed pywhispercpp version, or None if it cannot be read. @@ -487,6 +502,12 @@ class SpeechRecognitionManager: self._voice_commands_enabled = self._resolve_voice_commands_enabled() self.text_callbacks: list[Callable[[str], None]] = [] + # (text, capture-started_at) consumers, e.g. transcription history. + self.segment_callbacks: list[Callable[[str, float], None]] = [] + # While the Settings microphone test runs, segments whose capture + # began at or after this monotonic time are test speech; consumers + # (e.g. history) read the floor to skip them. + self.test_capture_floor: Optional[float] = None self.state_callbacks: list[Callable[[RecognitionState], None]] = [] self.action_callbacks: list[Callable[[str], None]] = [] @@ -560,6 +581,9 @@ class SpeechRecognitionManager: self.should_record = False self._recognition_mode = "toggle" # "toggle" or "push_to_talk" self.audio_buffer = [] + # Monotonic time the current buffer's first chunk was captured, + # propagated onto each enqueued _AudioSegment. + self._segment_started_at: Optional[float] = None self._recording_segment_has_speech = False self._buffer_lock = threading.Lock() # Thread safety for audio_buffer self._model_lock = threading.Lock() # Thread safety for model/recognizer access @@ -2654,6 +2678,32 @@ class SpeechRecognitionManager: """Set the text callbacks list (used for temporarily replacing callbacks).""" self.text_callbacks = list(callbacks) + def register_segment_callback(self, callback: Callable[[str, float], None]) -> None: + """ + Register a callback invoked with ``(text, started_at)`` per segment. + + ``started_at`` is the ``time.monotonic()`` timestamp when the + segment's audio capture began. Segment callbacks run on the + recognition thread just before the text callbacks. + + Args: + callback: A function taking (recognized_text, capture_started_at) + """ + self.segment_callbacks.append(callback) + + def unregister_segment_callback(self, callback: Callable[[str, float], None]) -> None: + """ + Unregister a segment callback function. + + Args: + callback: The callback function to remove. + """ + try: + self.segment_callbacks.remove(callback) + logger.debug(f"Unregistered segment callback: {callback}") + except ValueError: + logger.warning(f"Callback {callback} not found in segment_callbacks.") + def register_state_callback(self, callback: Callable[[RecognitionState], None]): """ Register a callback function that will be called when the recognition state changes. @@ -2909,6 +2959,7 @@ class SpeechRecognitionManager: self.should_record = True self._recognition_mode = mode self.audio_buffer = [] + self._segment_started_at = None self._segment_queue = queue.Queue(maxsize=32) # Start the audio recording thread @@ -3160,6 +3211,12 @@ class SpeechRecognitionManager: data = source.read_chunk() + if not self.audio_buffer: + # First chunk of a new segment: remember when its + # capture began. A mid-session history clear + # compares against this to separate speech + # captured before it from speech captured after. + self._segment_started_at = time.monotonic() self.audio_buffer.append(data) # Voice Activity Detection (VAD) @@ -3407,6 +3464,11 @@ class SpeechRecognitionManager: f"processed_text='{processed_text[:50] if processed_text else '(empty)'}...', callbacks={len(self.text_callbacks)}" ) if processed_text: + started_at = getattr(audio_buffer, "started_at", None) + if started_at is None: + started_at = time.monotonic() + for segment_callback in self.segment_callbacks: + segment_callback(processed_text, started_at) for callback in self.text_callbacks: logger.debug( f"invoking text callback: {callback.__name__ if hasattr(callback, '__name__') else callback}" @@ -3478,7 +3540,7 @@ class SpeechRecognitionManager: def _enqueue_audio_segment(self, audio_buffer: list[bytes]): """Queue an audio segment for asynchronous transcription.""" - segment = audio_buffer.copy() + segment = _AudioSegment(audio_buffer.copy(), self._segment_started_at) if not segment: logger.warning("_enqueue_audio_segment called with empty buffer") return diff --git a/src/vocalinux/ui/config_manager.py b/src/vocalinux/ui/config_manager.py index bb3f992..81bd0f6 100644 --- a/src/vocalinux/ui/config_manager.py +++ b/src/vocalinux/ui/config_manager.py @@ -196,6 +196,10 @@ DEFAULT_CONFIG = { # VOCALINUX_FORCE_BACKEND overrides this for a single run. "backend": "auto", }, + "history": { + "enabled": True, # Keep recent dictation snippets in the tray menu + "max_items": 10, # How many snippets to retain (in-memory only, cleared on quit) + }, "advanced": { "power_user_mode": False, "debug_logging": False, diff --git a/src/vocalinux/ui/settings_dialog.py b/src/vocalinux/ui/settings_dialog.py index d4b70a2..7f97242 100644 --- a/src/vocalinux/ui/settings_dialog.py +++ b/src/vocalinux/ui/settings_dialog.py @@ -115,6 +115,7 @@ from .keyboard_backends import ( # noqa: E402 parse_shortcut_spec, ) from .keyboard_backends.evdev_backend import MODIFIER_KEY_CODES # noqa: E402 +from .transcription_history import sanitize_max_items # noqa: E402 from ..utils.faster_whisper_model_info import ( # isort:skip FASTER_WHISPER_MODEL_INFO, @@ -3220,10 +3221,43 @@ class SettingsDialog(Gtk.Dialog): self.general_tab.pack_start(group, False, False, 0) + # Transcription history group + history_group = PreferencesGroup( + title="Transcription History", + description="Recent dictation snippets are kept in memory and shown " + "in the tray menu. Nothing is written to disk; history clears on quit. " + "Changes take effect after restarting Vocalinux.", + ) + + self.history_enabled_switch = Gtk.Switch() + self.history_enabled_switch.set_tooltip_text( + "Keep recent dictation snippets in the tray menu for quick re-copying" + ) + history_enabled_row = PreferenceRow( + title="Keep History", + subtitle="Show recent snippets in the tray menu", + widget=self.history_enabled_switch, + ) + history_group.add_row(history_enabled_row) + + self.history_max_items_spin = Gtk.SpinButton.new_with_range(1, 50, 1) + self.history_max_items_spin.set_tooltip_text("How many recent snippets to keep") + _prevent_scroll_on_hover(self.history_max_items_spin) + history_max_items_row = PreferenceRow( + title="Snippets to Keep", + subtitle="Number of recent snippets retained", + widget=self.history_max_items_spin, + ) + history_group.add_row(history_max_items_row) + + self.general_tab.pack_start(history_group, False, False, 0) + self.autostart_switch.connect("state-set", self._on_autostart_toggled) self.start_minimized_switch.connect("state-set", self._on_start_minimized_toggled) self.missing_tray_warning_switch.connect("state-set", self._on_missing_tray_warning_toggled) self.show_overlay_switch.connect("state-set", self._on_show_overlay_toggled) + self.history_enabled_switch.connect("state-set", self._on_history_enabled_toggled) + self.history_max_items_spin.connect("value-changed", self._on_history_max_items_changed) def _build_auto_pause_section(self): """Build Auto-Pause settings: enable toggle + process name list.""" @@ -3580,6 +3614,27 @@ class SettingsDialog(Gtk.Dialog): self.config_manager.set("model_keepalive", "idle_timeout_seconds", seconds) self.config_manager.save_settings() + def _on_history_enabled_toggled(self, widget: Gtk.Switch, state: bool) -> bool: + """Handle toggle of the keep-history switch.""" + if self._initializing or self._applying_settings: + return False + + enabled = bool(state) + logger.info(f"Transcription history toggled: {enabled}") + self.config_manager.set("history", "enabled", enabled) + self.config_manager.save_settings() + return False + + def _on_history_max_items_changed(self, widget: Gtk.SpinButton) -> None: + """Handle change of the snippets-to-keep spin button.""" + if self._initializing or self._applying_settings: + return + + max_items = widget.get_value_as_int() + logger.info(f"Transcription history max items: {max_items}") + self.config_manager.set("history", "max_items", max_items) + self.config_manager.save_settings() + def _on_autostart_toggled(self, widget, state): """Handle toggle of the autostart switch.""" if _handlers_suppressed(self): @@ -6096,6 +6151,7 @@ class SettingsDialog(Gtk.Dialog): general_settings = self.config_manager.get_settings().get("general", {}) ui_settings = self.config_manager.get_settings().get("ui", {}) text_injection_settings = self.config_manager.get_settings().get("text_injection", {}) + history_settings = self.config_manager.get_settings().get("history", {}) autostart_enabled = general_settings.get("autostart", False) start_minimized = ui_settings.get("start_minimized", False) @@ -6105,6 +6161,10 @@ class SettingsDialog(Gtk.Dialog): auto_capitalize = text_injection_settings.get("auto_capitalize", True) append_trailing_space = text_injection_settings.get("append_trailing_space", True) paste_shortcut = self.config_manager.get_paste_shortcut() + history_enabled = bool(history_settings.get("enabled", True)) + # A hand-edited config.json can hold a non-numeric value; sanitize so a + # bad preference cannot crash the settings dialog. + history_max_items = sanitize_max_items(history_settings.get("max_items", 10)) self.autostart_switch.set_active(autostart_enabled) self.start_minimized_switch.set_active(start_minimized) @@ -6145,6 +6205,9 @@ class SettingsDialog(Gtk.Dialog): if not self.model_keepalive_timeout_combo.set_active_id(str(timeout_seconds)): self.model_keepalive_timeout_combo.set_active_id("300") + self.history_enabled_switch.set_active(history_enabled) + self.history_max_items_spin.set_value(history_max_items) + disable_internal_hotkey = self.config_manager.get_bool( "shortcuts", "disable_internal_hotkey", False ) @@ -7927,9 +7990,14 @@ class SettingsDialog(Gtk.Dialog): self._saved_text_callbacks = self.speech_engine.get_text_callbacks() self.speech_engine.set_text_callbacks([self._test_text_callback]) + # History recording runs on segment callbacks keyed by capture time: + # stamp the floor so test speech stays out of Recent Snippets while + # leftover segments from a dictation still file normally. + self.speech_engine.test_capture_floor = time.monotonic() if not self.speech_engine.start_recognition(): self.speech_engine.set_text_callbacks(self._saved_text_callbacks) + self.speech_engine.test_capture_floor = None del self._saved_text_callbacks self.test_output_revealer.set_reveal_child(True) if getattr(self.speech_engine, "is_auto_paused", False): @@ -8135,6 +8203,7 @@ class SettingsDialog(Gtk.Dialog): if hasattr(self, "_saved_text_callbacks"): self.speech_engine.set_text_callbacks(self._saved_text_callbacks) del self._saved_text_callbacks + self.speech_engine.test_capture_floor = None # Check result after giving time for final callbacks to complete GLib.timeout_add(300, self._check_test_result) diff --git a/src/vocalinux/ui/transcription_history.py b/src/vocalinux/ui/transcription_history.py new file mode 100644 index 0000000..ac996aa --- /dev/null +++ b/src/vocalinux/ui/transcription_history.py @@ -0,0 +1,212 @@ +"""In-memory transcription history for Vocalinux. + +Keeps a bounded, newest-first list of recent dictation snippets so the user +can review and re-copy recent voice input from the tray menu. + +The history lives only for the lifetime of the running process — nothing is +written to disk — so dictated text never persists past the current session. +This is a deliberate privacy choice: a dictation tool sees everything the user +types by voice, and that should not silently accumulate in a file. +""" + +import logging +import threading +import time +from collections import deque +from typing import Any, Callable, List, Optional + +logger = logging.getLogger(__name__) + +# Default number of snippets to retain. Kept small so the tray menu stays +# readable; configurable via the "history" config section. +DEFAULT_MAX_ITEMS = 10 + + +def sanitize_max_items(value: Any) -> int: + """Coerce a configured snippet cap to a positive int. + + config.json is user-editable, so a saved ``history.max_items`` may be + missing, non-numeric, or out of range. An unusable value falls back to + ``DEFAULT_MAX_ITEMS`` rather than raising: a malformed preference must + never prevent the app from starting. + """ + try: + return max(1, int(value)) + except (TypeError, ValueError): + logger.warning( + "Invalid transcription history max_items %r; falling back to %d", + value, + DEFAULT_MAX_ITEMS, + ) + return DEFAULT_MAX_ITEMS + + +class TranscriptionHistory: + """A bounded, thread-safe, in-memory store of recent dictation snippets. + + A "snippet" is the text of a single dictation session (everything said + between starting and stopping voice typing). Entries are stored oldest to + newest internally and returned newest-first for display. + + Recording happens on the speech-recognition thread while the tray menu is + rebuilt on the GTK main thread, so all access is guarded by a lock. The + optional change callback lets the UI refresh when entries are added, + cleared, or trimmed; callers are responsible for marshalling that callback + onto the correct thread (e.g. via ``GLib.idle_add``). + """ + + def __init__(self, max_items: int = DEFAULT_MAX_ITEMS, enabled: bool = True) -> None: + self._max_items = sanitize_max_items(max_items) + self._enabled = bool(enabled) + self._entries: deque = deque(maxlen=self._max_items) + self._lock = threading.Lock() + self._change_callback: Optional[Callable[[], None]] = None + # Bumped every time the entries are wiped; lets callers refuse text + # produced before a clear so it cannot reappear afterwards. + self._epoch = 0 + # Monotonic time of the last wipe. Segments carry the moment their + # audio capture began, so text decoded from speech captured before + # this point can be refused however late it arrives. + self._cleared_at = 0.0 + + def set_change_callback(self, callback: Optional[Callable[[], None]]) -> None: + """Register a callback invoked whenever the history changes.""" + self._change_callback = callback + + @property + def enabled(self) -> bool: + """Whether new snippets are being recorded.""" + return self._enabled + + @property + def max_items(self) -> int: + """The maximum number of snippets retained.""" + return self._max_items + + @property + def epoch(self) -> int: + """Clear generation, incremented each time the entries are wiped. + + ``add`` and ``extend_latest`` take an ``expected_epoch`` and check + it atomically under the lock, so a caller holding the epoch from + before a ``clear`` can be refused instead of re-entering history + as fresh text. + """ + with self._lock: + return self._epoch + + @property + def cleared_at(self) -> float: + """``time.monotonic()`` of the most recent wipe, or 0.0 if never. + + A recognized segment whose audio capture began at or before this + timestamp can only contain pre-clear speech and must not re-enter + history; anything captured afterwards is genuinely new dictation. + """ + with self._lock: + return self._cleared_at + + def set_max_items(self, max_items: int) -> None: + """Change the retained-snippet cap, trimming oldest entries if needed.""" + max_items = sanitize_max_items(max_items) + with self._lock: + if max_items == self._max_items: + return + self._max_items = max_items + # deque(maxlen=...) keeps the rightmost (newest) items on trim. + self._entries = deque(self._entries, maxlen=max_items) + self._notify() + + def set_enabled(self, enabled: bool) -> None: + """Enable or disable recording. Disabling also clears existing entries.""" + enabled = bool(enabled) + with self._lock: + if enabled == self._enabled: + return + self._enabled = enabled + if not enabled: + self._entries.clear() + self._epoch += 1 + self._cleared_at = time.monotonic() + self._notify() + + def add(self, text: str, *, expected_epoch: Optional[int] = None) -> bool: + """Add a snippet, returning True when it was recorded. + + No-op when disabled or when text is empty. With ``expected_epoch`` + the add is also refused once the epoch has advanced — i.e. the + history was cleared since that epoch was observed — so text + produced before the clear cannot reappear as a new entry. + """ + if not text: + return False + text = text.strip() + if not text: + return False + with self._lock: + if not self._enabled: + return False + if expected_epoch is not None and expected_epoch != self._epoch: + return False + self._entries.append(text) + self._notify() + return True + + def extend_latest(self, text: str, *, expected_epoch: Optional[int] = None) -> bool: + """Append a late-arriving segment to the most recent snippet. + + The recognition worker can emit a final segment after its session + already ended (the manager stops waiting for it after a bounded + timeout and reports IDLE anyway). That text belongs to the just-ended + session's snippet, so it is merged into the newest entry instead of + becoming a snippet of its own or leaking into the next session. + + Returns False when there is nothing to extend (empty or disabled + history, or empty text) or when ``expected_epoch`` no longer matches + — the history was cleared since the caller observed that epoch, and + the cleared snippet must not grow back. + """ + if not text or not text.strip(): + return False + with self._lock: + if not self._enabled or not self._entries: + return False + if expected_epoch is not None and expected_epoch != self._epoch: + return False + self._entries[-1] = f"{self._entries[-1]} {text.strip()}" + self._notify() + return True + + def get_all(self) -> List[str]: + """Return all snippets, newest first.""" + with self._lock: + return list(reversed(self._entries)) + + def clear(self) -> None: + """Remove all snippets. + + The epoch advances even when the history is already empty: the + call still expresses "forget everything dictated so far", so + text still in flight from before it must not re-enter. + """ + with self._lock: + self._epoch += 1 + self._cleared_at = time.monotonic() + if not self._entries: + return + self._entries.clear() + self._notify() + + def __len__(self) -> int: + with self._lock: + return len(self._entries) + + def _notify(self) -> None: + callback = self._change_callback + if callback is None: + return + try: + callback() + except (RuntimeError, TypeError, ValueError) as e: + # A misbehaving UI callback must never break recording. + logger.exception("Transcription history change callback failed: %s", e) diff --git a/src/vocalinux/ui/tray_indicator.py b/src/vocalinux/ui/tray_indicator.py index 7b3edc2..ec893b9 100644 --- a/src/vocalinux/ui/tray_indicator.py +++ b/src/vocalinux/ui/tray_indicator.py @@ -50,6 +50,7 @@ from .config_manager import get_shared_config_manager from .keyboard_backends import DEFAULT_SHORTCUT, DEFAULT_SHORTCUT_MODE from .keyboard_shortcuts import KeyboardShortcutManager from .settings_dialog import SettingsDialog, recommended_model_for_engine +from .transcription_history import TranscriptionHistory logger = logging.getLogger(__name__) @@ -101,6 +102,9 @@ def _themed_icon_names() -> dict: } +# Maximum characters shown for a history snippet label before truncation. +_HISTORY_LABEL_MAX_CHARS = 50 + # /dev/input settle-detection tuning (used after resume) _INPUT_SETTLE_SECONDS = 2 _INPUT_MONITOR_CAP_SECONDS = 10 @@ -119,16 +123,20 @@ class TrayIndicator: self, speech_engine: SpeechRecognitionManagerProtocol, text_injector: TextInjectorProtocol, - ): + transcription_history: Optional[TranscriptionHistory] = None, + ) -> None: """ Initialize the system tray indicator. Args: speech_engine: The speech recognition manager instance text_injector: The text injector instance + transcription_history: Optional in-memory store of recent dictation + snippets. When provided, a "Recent Snippets" submenu is shown. """ self.speech_engine = speech_engine self.text_injector = text_injector + self.transcription_history = transcription_history # Shared with main() and the settings dialog: separate instances would # overwrite each other's saves with stale in-memory copies. self.config_manager = get_shared_config_manager() @@ -138,6 +146,15 @@ class TrayIndicator: # than reapplying a disable_internal_hotkey setting D-Bus cannot serve. self._external_activation_unavailable = False self._syncing_autostart_menu = False + self._history_menu_item = None + + # Refresh the history submenu whenever the history changes. The change + # callback fires on the recognition thread, so marshal onto the GTK + # main thread before touching widgets. + if self.transcription_history is not None: + self.transcription_history.set_change_callback( + lambda: GLib.idle_add(self._refresh_history_menu) + ) # Get configured shortcut and mode from config shortcut = self.config_manager.get_str("shortcuts", "toggle_recognition", DEFAULT_SHORTCUT) @@ -404,6 +421,13 @@ class TrayIndicator: self._add_menu_item("Stop Voice Typing", self._on_stop_clicked) self._add_menu_separator() + # Recent snippets submenu (only when history is enabled) + if self.transcription_history is not None and self.transcription_history.enabled: + self._history_menu_item = Gtk.MenuItem.new_with_label("Recent Snippets") + self.menu.append(self._history_menu_item) + self._refresh_history_menu() + self._add_menu_separator() + self._autostart_menu_item = self._add_menu_checkbox( "Start on Login", self._on_autostart_toggled ) @@ -896,6 +920,61 @@ class TrayIndicator: logger.debug("Stop Voice Typing clicked") self.speech_engine.stop_recognition() + def _refresh_history_menu(self) -> bool: + """Rebuild the Recent Snippets submenu from the current history.""" + if self._history_menu_item is None or self.transcription_history is None: + return False # Remove idle callback + + submenu = Gtk.Menu() + entries = self.transcription_history.get_all() + + if not entries: + empty_item = Gtk.MenuItem.new_with_label("(no snippets yet)") + empty_item.set_sensitive(False) + submenu.append(empty_item) + else: + for entry in entries: + item = Gtk.MenuItem.new_with_label(self._truncate_label(entry)) + item.set_tooltip_text(entry) + # The full snippet is passed as connect user-data, so each item + # copies its own text (no late-binding closure pitfall). + item.connect("activate", self._on_history_item_clicked, entry) + submenu.append(item) + + submenu.append(Gtk.SeparatorMenuItem()) + clear_item = Gtk.MenuItem.new_with_label("Clear History") + clear_item.connect("activate", self._on_clear_history_clicked) + submenu.append(clear_item) + + submenu.show_all() + self._history_menu_item.set_submenu(submenu) + return False # Remove idle callback + + @staticmethod + def _truncate_label(text: str) -> str: + """Collapse whitespace and truncate a snippet for menu display.""" + single_line = " ".join(text.split()) + if len(single_line) > _HISTORY_LABEL_MAX_CHARS: + return single_line[: _HISTORY_LABEL_MAX_CHARS - 1].rstrip() + "…" + return single_line + + def _on_history_item_clicked(self, widget: Gtk.MenuItem, text: str) -> None: + """Copy the selected snippet to the clipboard.""" + logger.debug("History snippet clicked, copying to clipboard") + # Lazy import keeps the module importable when gi.repository is a + # minimal test stand-in (e.g. the AppIndicator fallback tests). + from gi.repository import Gdk + + clipboard = Gtk.Clipboard.get(Gdk.SELECTION_CLIPBOARD) + clipboard.set_text(text, -1) + clipboard.store() + + def _on_clear_history_clicked(self, widget: Gtk.MenuItem) -> None: + """Clear all stored snippets.""" + logger.debug("Clear history clicked") + if self.transcription_history is not None: + self.transcription_history.clear() + def _on_settings_clicked(self, widget): """Handle click on the Settings menu item.""" logger.debug("Settings clicked") diff --git a/tests/test_main.py b/tests/test_main.py index b561a89..e262765 100644 --- a/tests/test_main.py +++ b/tests/test_main.py @@ -4,8 +4,12 @@ Tests for the main module functionality. import argparse import sys +import threading +import time import unittest -from unittest.mock import MagicMock, patch +from contextlib import ExitStack +from typing import Any, Callable, Dict, Optional, Tuple +from unittest.mock import ANY, MagicMock, patch # Mock GTK modules before importing vocalinux.main sys.modules["gi"] = MagicMock() @@ -14,6 +18,7 @@ sys.modules["gi.repository"] = MagicMock() # Update import to use the new package structure from vocalinux.common_types import RecognitionState from vocalinux.main import check_dependencies, main, parse_arguments +from vocalinux.ui.transcription_history import TranscriptionHistory class TestMainModule(unittest.TestCase): @@ -244,11 +249,14 @@ class TestMainModule(unittest.TestCase): mock_text.assert_called_once_with(wayland_mode=True) mock_action_handler.assert_called_once_with(mock_text_instance) mock_tray.assert_called_once_with( - speech_engine=mock_speech_instance, text_injector=mock_text_instance + speech_engine=mock_speech_instance, + text_injector=mock_text_instance, + transcription_history=ANY, ) # Verify callbacks were registered mock_speech_instance.register_text_callback.assert_called_once() + mock_speech_instance.register_segment_callback.assert_called_once() mock_speech_instance.register_action_callback.assert_called_once_with( mock_action_instance.handle_action ) @@ -1353,5 +1361,257 @@ class TestMainCallbackTrailingSpaceEdges(unittest.TestCase): stack.close() +class TestSessionHistoryRecording(unittest.TestCase): + """Segments commit to transcription history as one snippet per session.""" + + def _boot(self, *, extra_settings: Optional[Dict[str, Any]] = None) -> Tuple[ + ExitStack, + Callable[[str], None], + Callable[[str, float], None], + Callable[[RecognitionState], None], + TranscriptionHistory, + ]: + """Boot main() under mocks; return (stack, text_cb, segment_cb, state_cb, history). + + The history is the real TranscriptionHistory instance handed to the + (mocked) TrayIndicator, so assertions observe actual recording. + segment_cb(text, started_at) delivers a recognized segment with the + monotonic time its audio capture began — history recording consumes + capture time so tests control pre-/post-clear placement precisely. + """ + stack = ExitStack() + stack.enter_context(patch("vocalinux.main.check_dependencies", return_value=True)) + mock_config_cls = stack.enter_context( + patch("vocalinux.ui.config_manager.get_shared_config_manager") + ) + mock_config = MagicMock() + settings = { + "speech_recognition": {}, + "general": {"first_run": False}, + } + if extra_settings: + settings.update(extra_settings) + mock_config.get_settings.return_value = settings + mock_config.get.return_value = False # auto_capitalize off + mock_config_cls.return_value = mock_config + + mock_speech_cls = stack.enter_context( + patch("vocalinux.speech_recognition.recognition_manager.SpeechRecognitionManager") + ) + mock_speech = MagicMock() + mock_speech.engine = "whisper_cpp" + mock_speech_cls.return_value = mock_speech + + mock_text_cls = stack.enter_context( + patch("vocalinux.text_injection.text_injector.TextInjector") + ) + mock_text = MagicMock() + mock_text.inject_text.return_value = True + mock_text_cls.return_value = mock_text + + mock_tray_cls = stack.enter_context(patch("vocalinux.ui.tray_indicator.TrayIndicator")) + mock_tray_cls.return_value = MagicMock() + + stack.enter_context(patch("vocalinux.ui.logging_manager.initialize_logging")) + mock_parse = stack.enter_context(patch("vocalinux.main.parse_arguments")) + stack.enter_context(patch("sys.argv", ["vocalinux"])) + + mock_args = MagicMock() + mock_args.debug = False + mock_args.model = "tiny" + mock_args.engine = "whisper_cpp" + mock_args.language = "en-us" + mock_args.wayland = False + mock_args.start_minimized = False + mock_parse.return_value = mock_args + try: + main() + text_cb = mock_speech.register_text_callback.call_args.args[0] + segment_cb = mock_speech.register_segment_callback.call_args.args[0] + state_cb = mock_speech.register_state_callback.call_args.args[0] + history = mock_tray_cls.call_args.kwargs["transcription_history"] + except BaseException: + stack.close() + raise + + return stack, text_cb, segment_cb, state_cb, history + + def test_session_segments_commit_as_single_snippet(self) -> None: + stack, _, segment_cb, state_cb, history = self._boot() + try: + state_cb(RecognitionState.LISTENING) + segment_cb("Hello.", time.monotonic()) + state_cb(RecognitionState.PROCESSING) + state_cb(RecognitionState.LISTENING) + segment_cb("World", time.monotonic()) + state_cb(RecognitionState.IDLE) + self.assertEqual(history.get_all(), ["Hello. World"]) + finally: + stack.close() + + def test_successive_sessions_record_separate_snippets_newest_first(self) -> None: + stack, _, segment_cb, state_cb, history = self._boot() + try: + state_cb(RecognitionState.LISTENING) + segment_cb("first session", time.monotonic()) + state_cb(RecognitionState.IDLE) + + state_cb(RecognitionState.LISTENING) + segment_cb("second session", time.monotonic()) + state_cb(RecognitionState.IDLE) + + self.assertEqual(history.get_all(), ["second session", "first session"]) + finally: + stack.close() + + def test_late_segment_merges_into_its_own_sessions_snippet(self) -> None: + """A worker that outlives the bounded stop wait lands in its snippet.""" + stack, _, segment_cb, state_cb, history = self._boot() + try: + state_cb(RecognitionState.LISTENING) + segment_cb("hello", time.monotonic()) + segment_cb("world", time.monotonic()) + # IDLE fires while the worker still holds a final segment. + state_cb(RecognitionState.IDLE) + segment_cb("late tail", time.monotonic()) + self.assertEqual(history.get_all(), ["hello world late tail"]) + + # The next session's snippet is not contaminated. + state_cb(RecognitionState.LISTENING) + segment_cb("next session", time.monotonic()) + state_cb(RecognitionState.IDLE) + self.assertEqual(history.get_all(), ["next session", "hello world late tail"]) + finally: + stack.close() + + def test_late_segment_from_old_worker_during_next_session(self) -> None: + """A leftover worker delivering on its own thread stays out of the open session.""" + stack, _, segment_cb, state_cb, history = self._boot() + try: + state_cb(RecognitionState.LISTENING) + segment_cb("one", time.monotonic()) + state_cb(RecognitionState.IDLE) + + state_cb(RecognitionState.LISTENING) + segment_cb("two", time.monotonic()) + # The previous session's worker finally delivers on its own thread. + t = threading.Thread(target=segment_cb, args=("trailing", time.monotonic())) + t.start() + t.join() + self.assertEqual(history.get_all(), ["one trailing"]) + + state_cb(RecognitionState.IDLE) + self.assertEqual(history.get_all(), ["two", "one trailing"]) + finally: + stack.close() + + def test_session_without_segments_creates_no_snippet(self) -> None: + stack, _, segment_cb, state_cb, history = self._boot() + try: + state_cb(RecognitionState.LISTENING) + state_cb(RecognitionState.IDLE) + self.assertEqual(history.get_all(), []) + + # A segment arriving now belongs to that empty session: it becomes + # its own snippet rather than merging into an older entry. + segment_cb("orphan", time.monotonic()) + self.assertEqual(history.get_all(), ["orphan"]) + finally: + stack.close() + + def test_error_state_also_commits_partial_snippet(self) -> None: + stack, _, segment_cb, state_cb, history = self._boot() + try: + state_cb(RecognitionState.LISTENING) + segment_cb("partial", time.monotonic()) + state_cb(RecognitionState.ERROR) + self.assertEqual(history.get_all(), ["partial"]) + finally: + stack.close() + + def test_history_disabled_records_nothing(self) -> None: + stack, _, segment_cb, state_cb, history = self._boot( + extra_settings={"history": {"enabled": False}} + ) + try: + state_cb(RecognitionState.LISTENING) + segment_cb("hello", time.monotonic()) + state_cb(RecognitionState.IDLE) + self.assertEqual(history.get_all(), []) + finally: + stack.close() + + def test_invalid_history_max_items_uses_default(self) -> None: + """A corrupted saved limit must not prevent startup.""" + stack, _, _, _, history = self._boot(extra_settings={"history": {"max_items": "abc"}}) + try: + self.assertEqual(history.max_items, 10) + finally: + stack.close() + + def test_late_segment_after_clear_does_not_reappear(self) -> None: + """A straggler from an ended session must not resurrect cleared history.""" + stack, _, segment_cb, state_cb, history = self._boot() + try: + state_cb(RecognitionState.LISTENING) + segment_cb("before clear", time.monotonic()) + state_cb(RecognitionState.IDLE) + self.assertEqual(history.get_all(), ["before clear"]) + + history.clear() + # Its audio was captured before the clear; the decode only + # finished afterwards, so it must still be refused. + segment_cb("late tail", history.cleared_at - 1.0) + self.assertEqual(history.get_all(), []) + finally: + stack.close() + + def test_late_orphan_segment_after_clear_does_not_reappear(self) -> None: + """Late-only output of an empty session is refused after a clear too.""" + stack, _, segment_cb, state_cb, history = self._boot() + try: + state_cb(RecognitionState.LISTENING) + state_cb(RecognitionState.IDLE) # empty session: no snippet + history.clear() + segment_cb("orphan", history.cleared_at - 1.0) + self.assertEqual(history.get_all(), []) + finally: + stack.close() + + def test_clear_during_session_drops_its_snippet(self) -> None: + """A clear issued while a session runs keeps its whole snippet out.""" + stack, _, segment_cb, state_cb, history = self._boot() + try: + state_cb(RecognitionState.LISTENING) + segment_cb("dictated", time.monotonic()) + history.clear() + state_cb(RecognitionState.IDLE) + self.assertEqual(history.get_all(), []) + finally: + stack.close() + + def test_dictation_after_mid_session_clear_is_kept(self) -> None: + """A mid-session clear splits segments by capture time, not arrival. + + Speech captured before the clear is refused no matter when its + decode lands; speech captured after it joins the session's snippet. + """ + stack, _, segment_cb, state_cb, history = self._boot() + try: + state_cb(RecognitionState.LISTENING) + segment_cb("before", time.monotonic()) + history.clear() + boundary = history.cleared_at + # A decode of pre-clear audio finishing after the clear is still + # refused — this is the segment the epoch check used to keep or + # drop wholesale. + segment_cb("pre-clear tail", boundary - 1.0) + segment_cb("after clear", time.monotonic()) + state_cb(RecognitionState.IDLE) + self.assertEqual(history.get_all(), ["after clear"]) + finally: + stack.close() + + if __name__ == "__main__": unittest.main() diff --git a/tests/test_transcription_history.py b/tests/test_transcription_history.py new file mode 100644 index 0000000..4c20919 --- /dev/null +++ b/tests/test_transcription_history.py @@ -0,0 +1,241 @@ +""" +Tests for the in-memory transcription history. +""" + +import unittest + +from vocalinux.ui.transcription_history import DEFAULT_MAX_ITEMS, TranscriptionHistory + + +class TestTranscriptionHistory(unittest.TestCase): + """Test cases for the TranscriptionHistory store.""" + + def test_defaults(self) -> None: + history = TranscriptionHistory() + self.assertEqual(history.max_items, DEFAULT_MAX_ITEMS) + self.assertTrue(history.enabled) + self.assertEqual(len(history), 0) + self.assertEqual(history.get_all(), []) + + def test_add_and_newest_first(self) -> None: + history = TranscriptionHistory() + history.add("first") + history.add("second") + history.add("third") + self.assertEqual(history.get_all(), ["third", "second", "first"]) + self.assertEqual(len(history), 3) + + def test_add_strips_whitespace(self) -> None: + history = TranscriptionHistory() + history.add(" hello world ") + self.assertEqual(history.get_all(), ["hello world"]) + + def test_empty_and_whitespace_ignored(self) -> None: + history = TranscriptionHistory() + history.add("") + history.add(" ") + history.add(None) # type: ignore[arg-type] + self.assertEqual(len(history), 0) + + def test_max_items_cap_drops_oldest(self) -> None: + history = TranscriptionHistory(max_items=3) + for text in ["a", "b", "c", "d"]: + history.add(text) + # "a" dropped; newest first. + self.assertEqual(history.get_all(), ["d", "c", "b"]) + self.assertEqual(len(history), 3) + + def test_set_max_items_trims_keeping_newest(self) -> None: + history = TranscriptionHistory(max_items=5) + for text in ["a", "b", "c", "d", "e"]: + history.add(text) + history.set_max_items(2) + self.assertEqual(history.max_items, 2) + self.assertEqual(history.get_all(), ["e", "d"]) + + def test_max_items_floor_is_one(self) -> None: + history = TranscriptionHistory(max_items=0) + self.assertEqual(history.max_items, 1) + history.add("a") + history.add("b") + self.assertEqual(history.get_all(), ["b"]) + + def test_invalid_max_items_falls_back_to_default(self) -> None: + self.assertEqual(TranscriptionHistory(max_items="abc").max_items, DEFAULT_MAX_ITEMS) + self.assertEqual(TranscriptionHistory(max_items=None).max_items, DEFAULT_MAX_ITEMS) + self.assertEqual(TranscriptionHistory(max_items=[1]).max_items, DEFAULT_MAX_ITEMS) + + def test_numeric_string_max_items_accepted(self) -> None: + history = TranscriptionHistory(max_items="4") + self.assertEqual(history.max_items, 4) + + def test_set_max_items_invalid_falls_back_to_default(self) -> None: + history = TranscriptionHistory(max_items=3) + history.set_max_items("bogus") + self.assertEqual(history.max_items, DEFAULT_MAX_ITEMS) + + def test_extend_latest_appends_to_newest_entry(self) -> None: + history = TranscriptionHistory() + history.add("one") + self.assertTrue(history.extend_latest("late tail")) + self.assertEqual(history.get_all(), ["one late tail"]) + + def test_extend_latest_only_touches_newest_entry(self) -> None: + history = TranscriptionHistory() + history.add("one") + history.add("two") + history.extend_latest("tail") + self.assertEqual(history.get_all(), ["two tail", "one"]) + + def test_extend_latest_strips_segment_whitespace(self) -> None: + history = TranscriptionHistory() + history.add("one") + history.extend_latest(" tail ") + self.assertEqual(history.get_all(), ["one tail"]) + + def test_extend_latest_empty_history_returns_false(self) -> None: + self.assertFalse(TranscriptionHistory().extend_latest("x")) + + def test_extend_latest_disabled_returns_false(self) -> None: + self.assertFalse(TranscriptionHistory(enabled=False).extend_latest("x")) + + def test_extend_latest_blank_text_returns_false(self) -> None: + history = TranscriptionHistory() + history.add("one") + self.assertFalse(history.extend_latest(" ")) + self.assertEqual(history.get_all(), ["one"]) + + def test_extend_latest_after_clear_returns_false(self) -> None: + history = TranscriptionHistory() + history.add("a") + history.clear() + self.assertFalse(history.extend_latest("x")) + + def test_extend_latest_fires_change_callback(self) -> None: + history = TranscriptionHistory() + history.add("a") + calls = [] + history.set_change_callback(lambda: calls.append(1)) + history.extend_latest("b") + self.assertEqual(calls, [1]) + + def test_clear(self) -> None: + history = TranscriptionHistory() + history.add("a") + history.add("b") + history.clear() + self.assertEqual(history.get_all(), []) + self.assertEqual(len(history), 0) + + def test_disabled_does_not_record(self) -> None: + history = TranscriptionHistory(enabled=False) + self.assertFalse(history.enabled) + history.add("ignored") + self.assertEqual(len(history), 0) + + def test_set_enabled_false_clears_entries(self) -> None: + history = TranscriptionHistory() + history.add("a") + history.set_enabled(False) + self.assertFalse(history.enabled) + self.assertEqual(len(history), 0) + history.add("b") # still ignored while disabled + self.assertEqual(len(history), 0) + history.set_enabled(True) + history.add("c") + self.assertEqual(history.get_all(), ["c"]) + + def test_change_callback_fires_on_mutations(self) -> None: + history = TranscriptionHistory() + calls = [] + history.set_change_callback(lambda: calls.append(1)) + + history.add("a") # fires + history.clear() # fires + history.clear() # no-op, already empty -> no fire + self.assertEqual(len(calls), 2) + + def test_change_callback_not_fired_when_disabled_add(self) -> None: + history = TranscriptionHistory(enabled=False) + calls = [] + history.set_change_callback(lambda: calls.append(1)) + history.add("a") + self.assertEqual(calls, []) + + def test_change_callback_exception_is_swallowed(self) -> None: + history = TranscriptionHistory() + + def boom() -> None: + raise RuntimeError("callback failure") + + history.set_change_callback(boom) + # Must not propagate. + history.add("a") + self.assertEqual(history.get_all(), ["a"]) + + # --- Clear epoch ------------------------------------------------------- + + def test_epoch_advances_on_every_clear(self) -> None: + history = TranscriptionHistory() + first = history.epoch + history.clear() + self.assertNotEqual(history.epoch, first) + second = history.epoch + history.clear() + self.assertNotEqual(history.epoch, second) + + def test_epoch_advances_on_disable(self) -> None: + history = TranscriptionHistory() + epoch = history.epoch + history.set_enabled(False) + self.assertNotEqual(history.epoch, epoch) + + def test_add_refused_from_stale_epoch(self) -> None: + """Text produced before a clear must not re-enter afterwards.""" + history = TranscriptionHistory() + epoch = history.epoch + history.clear() + self.assertFalse(history.add("stale", expected_epoch=epoch)) + self.assertEqual(history.get_all(), []) + + def test_add_accepted_within_same_epoch(self) -> None: + history = TranscriptionHistory() + self.assertTrue(history.add("a", expected_epoch=history.epoch)) + self.assertEqual(history.get_all(), ["a"]) + + def test_extend_latest_refused_from_stale_epoch(self) -> None: + """A cleared snippet must not grow back via a stale epoch.""" + history = TranscriptionHistory() + history.add("a") + epoch = history.epoch + history.clear() + history.add("b") + self.assertFalse(history.extend_latest("tail", expected_epoch=epoch)) + self.assertEqual(history.get_all(), ["b"]) + + def test_unguarded_writes_ignore_epoch(self) -> None: + history = TranscriptionHistory() + history.add("a") + history.clear() + self.assertTrue(history.add("b")) + self.assertEqual(history.get_all(), ["b"]) + + # --- Clear timestamp --------------------------------------------------- + + def test_cleared_at_advances_on_clear(self) -> None: + history = TranscriptionHistory() + self.assertEqual(history.cleared_at, 0.0) + history.clear() + self.assertGreater(history.cleared_at, 0.0) + first = history.cleared_at + history.clear() + self.assertGreaterEqual(history.cleared_at, first) + + def test_cleared_at_advances_on_disable(self) -> None: + history = TranscriptionHistory() + history.set_enabled(False) + self.assertGreater(history.cleared_at, 0.0) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_tray_indicator.py b/tests/test_tray_indicator.py index d5a30b9..8c9a032 100644 --- a/tests/test_tray_indicator.py +++ b/tests/test_tray_indicator.py @@ -982,6 +982,87 @@ class TestTrayIndicator(unittest.TestCase): self.assertIsNone(self.tray_indicator.overlay) patched_gtk.main_quit.assert_called_once() + # --- Recent Snippets history menu ------------------------------------- + + def test_truncate_label_collapses_and_truncates(self) -> None: + from vocalinux.ui.tray_indicator import TrayIndicator + + self.assertEqual(TrayIndicator._truncate_label(" a b\nc "), "a b c") + long = "x" * 200 + truncated = TrayIndicator._truncate_label(long) + self.assertTrue(truncated.endswith("…")) + self.assertEqual(len(truncated), 50) + + def test_refresh_history_menu_populated(self) -> None: + from vocalinux.ui.transcription_history import TranscriptionHistory + + history = TranscriptionHistory() + history.add("first snippet") + history.add("second snippet") + self.tray_indicator.transcription_history = history + self.tray_indicator._history_menu_item = MagicMock() + + result = self.tray_indicator._refresh_history_menu() + + self.assertFalse(result) + self.tray_indicator._history_menu_item.set_submenu.assert_called_once() + + def test_refresh_history_menu_empty(self) -> None: + from vocalinux.ui.transcription_history import TranscriptionHistory + + self.tray_indicator.transcription_history = TranscriptionHistory() + self.tray_indicator._history_menu_item = MagicMock() + + result = self.tray_indicator._refresh_history_menu() + + self.assertFalse(result) + self.tray_indicator._history_menu_item.set_submenu.assert_called_once() + + def test_refresh_history_menu_noop_without_item(self) -> None: + self.tray_indicator.transcription_history = None + self.tray_indicator._history_menu_item = None + + # Should return False and not raise. + self.assertFalse(self.tray_indicator._refresh_history_menu()) + + def test_on_history_item_clicked_copies_to_clipboard(self) -> None: + self.tray_indicator._on_history_item_clicked(MagicMock(), "some snippet") + + clipboard = mock_gtk.Clipboard.get.return_value + clipboard.set_text.assert_called_once_with("some snippet", -1) + clipboard.store.assert_called_once() + + def test_on_clear_history_clicked_clears(self) -> None: + from vocalinux.ui.transcription_history import TranscriptionHistory + + history = TranscriptionHistory() + history.add("a") + history.add("b") + self.tray_indicator.transcription_history = history + + self.tray_indicator._on_clear_history_clicked(MagicMock()) + + self.assertEqual(len(history), 0) + + def test_construct_with_history_creates_submenu_and_wires_callback(self) -> None: + from vocalinux.ui.transcription_history import TranscriptionHistory + from vocalinux.ui.tray_indicator import TrayIndicator + + history = TranscriptionHistory() + tray = TrayIndicator( + speech_engine=self.mock_speech_engine, + text_injector=self.mock_text_injector, + transcription_history=history, + ) + tray.shortcut_manager = self.mock_ksm + + # The submenu item is created during _init_indicator. + self.assertIsNotNone(tray._history_menu_item) + # Adding a snippet fires the change callback, which refreshes the menu + # (GLib.idle_add runs synchronously under the test mocks). + history.add("live snippet") + self.assertTrue(tray._history_menu_item.set_submenu.called) + class TestTrayIndicatorFlatpakIcons(unittest.TestCase): """Tray icon names must adapt to the Flatpak runtime.