diff --git a/src/vocalinux/main.py b/src/vocalinux/main.py index 4a23d03..632e813 100644 --- a/src/vocalinux/main.py +++ b/src/vocalinux/main.py @@ -8,6 +8,7 @@ import atexit import logging import sys import threading +import time from typing import Optional from .utils.vosk_model_info import SUPPORTED_LANGUAGES @@ -352,8 +353,6 @@ def main(): if not single_instance.acquire_lock(): # Another instance is already running - show notification and exit try: - import time - from gi.repository import Notify Notify.init("Vocalinux") @@ -601,16 +600,45 @@ def main(): session_lock = threading.Lock() session_segments: list[tuple[str, float]] = [] session_open = False + # Monotonic time the open session started; a segment delivered on the + # live worker whose capture predates it belongs to an earlier session. + session_started_floor = 0.0 # 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 + # Id of the just-closed session's snippet (from + # TranscriptionHistory.add), so late segments extend that entry + # rather than whichever snippet happens to be newest. + latest_snippet_id: Optional[int] = None # 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 + # In-app dictation pad: the Wayland-proof fallback target that + # receives dictated text when "dictate to pad" capture is enabled + # (#726). Constructed before the tray so the tray menu can open it. + from .ui.dictation_pad import DictationPad + + dictation_pad = DictationPad( + enabled=config_manager.is_dictate_to_pad_enabled(), + config_manager=config_manager, + ) + + def dictate_to_pad_enabled() -> bool: + """Read the live capture toggle from the shared config manager. + + The shared manager's in-memory cache is updated by Settings and + the pad checkbox before save_config rewrites config.json, so this + can never flip on a torn mid-write disk read and send dictated + text to whichever application holds focus. + """ + return bool(config_manager.is_dictate_to_pad_enabled()) + + # Where the last delivered segment went. "delete that" follows the + # text, not the current toggle: the capture setting may have flipped + # since the segment was delivered. + last_injected = {"to_pad": False} + # --- Callback wiring --------------------------------------------------- # The speech engine emits four kinds of events, each handled by a # dedicated callback registered below: @@ -667,7 +695,7 @@ def main(): captured after the clear are kept, so dictation that continues across a clear is still recoverable. """ - nonlocal session_worker, latest_snippet_extendable + nonlocal session_worker, latest_snippet_id if not transcription_history.enabled: return segment = _normalize_segment_text(segment) @@ -682,35 +710,49 @@ def main(): # 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. + # speech, not dictation. The [floor, ceiling) window survives + # the test's end, so segments still decoding after it cannot + # leak into history. test_floor = getattr(speech_engine, "test_capture_floor", None) - if isinstance(test_floor, (int, float)) and started_at >= test_floor: + test_ceiling = getattr(speech_engine, "test_capture_ceiling", None) + if ( + isinstance(test_floor, (int, float)) + and started_at >= test_floor + and (not isinstance(test_ceiling, (int, float)) or started_at < test_ceiling) + ): 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)) + if ( + session_open + and started_at >= session_started_floor + 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 + # its own snippet by id — a newer session may already have + # committed on top, so the newest entry is not the target. + # Both writes are guarded by the ended session's epoch, so a + # clear() landing between that commit and this delivery + # still refuses the text. + if latest_snippet_id is not None and transcription_history.extend_entry( + latest_snippet_id, 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 + snippet_id = transcription_history.add(segment, expected_epoch=ended_session_epoch) + if snippet_id is not None: + latest_snippet_id = snippet_id def text_callback_wrapper(text: str) -> None: """Bridge between speech engine text events and the text injector. @@ -746,9 +788,17 @@ def main(): text_to_inject = " " + text_to_inject logger.debug("Added space separator before new segment") - success = text_system.inject_text(text_to_inject) + captured = dictate_to_pad_enabled() + if captured: + # In-app capture: skip cross-application injection entirely + # and land the text in the pad instead (#726). + dictation_pad.append_text(text_to_inject) + success = True + else: + success = text_system.inject_text(text_to_inject) if success: action_handler.set_last_injected_text(text_to_inject) + last_injected["to_pad"] = captured def on_state_change(state: RecognitionState) -> None: """Reset the last-injected buffer when a listening session ends. @@ -756,11 +806,12 @@ def main(): 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 + nonlocal session_open, session_worker, latest_snippet_id + nonlocal ended_session_epoch, session_started_floor if state in (RecognitionState.IDLE, RecognitionState.ERROR): if state == RecognitionState.IDLE: action_handler.set_last_injected_text("") + last_injected["to_pad"] = False with session_lock: session_open = False session_worker = None @@ -775,10 +826,10 @@ def main(): 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 + # snippet. The committed snippet's id 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( + latest_snippet_id = transcription_history.add( joined, expected_epoch=ended_session_epoch ) else: @@ -786,13 +837,14 @@ def main(): if not session_open: session_open = True session_worker = None + session_started_floor = time.monotonic() # 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( + latest_snippet_id = transcription_history.add( " ".join( text for text, started_at in session_segments @@ -807,8 +859,43 @@ def main(): # session clear() can separate pre-clear speech from new dictation; # injection keeps the plain text callback. speech_engine.register_segment_callback(record_history_segment) + + def action_callback_wrapper(action: str) -> bool: + """Route editing voice commands to the pad while capture is on. + + "delete that" follows the segment it removes: the pad when the + last delivered text went there (even if capture was toggled off + since) and the focused application when it was injected (even if + capture was toggled on). Every other editing command is handled + pad-side while capturing so its shortcuts never leak into + whichever application holds focus. + """ + if action == "delete_last": + if not action_handler.last_injected_text: + return True + if last_injected["to_pad"]: + # The pad's own segment bookkeeping wins over the recorded + # text: a pad "undo" may have popped that segment, leaving + # its length stale. + target = dictation_pad.last_segment or action_handler.last_injected_text + deleted = dictation_pad.delete_last_chars(len(target)) + if deleted: + action_handler.set_last_injected_text("") + last_injected["to_pad"] = False + return True + return bool(action_handler.handle_action(action)) + if dictate_to_pad_enabled(): + handled = bool(dictation_pad.handle_action(action)) + if handled and action in ("undo", "redo") and last_injected["to_pad"]: + # Pad history moved: retarget "delete that" at the segment + # now at the pad's tail. Keep the pad destination even + # when the tail is empty — a redo can restore it. + action_handler.set_last_injected_text(dictation_pad.last_segment or "") + return handled + return bool(action_handler.handle_action(action)) + speech_engine.register_text_callback(text_callback_wrapper) - speech_engine.register_action_callback(action_handler.handle_action) + speech_engine.register_action_callback(action_callback_wrapper) speech_engine.register_state_callback(on_state_change) # Initialize and start the system tray indicator @@ -816,6 +903,7 @@ def main(): speech_engine=speech_engine, text_injector=text_system, transcription_history=transcription_history, + dictation_pad=dictation_pad, ) # Start the GTK main loop diff --git a/src/vocalinux/speech_recognition/recognition_manager.py b/src/vocalinux/speech_recognition/recognition_manager.py index 41ce068..853d96b 100644 --- a/src/vocalinux/speech_recognition/recognition_manager.py +++ b/src/vocalinux/speech_recognition/recognition_manager.py @@ -415,8 +415,12 @@ class SpeechRecognitionManager: 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. + # (e.g. history) read the floor to skip them. When the test ends the + # ceiling closes the window — segments captured during it but still + # decoding must keep failing the check, so the floor alone is never + # lifted back to None. self.test_capture_floor: Optional[float] = None + self.test_capture_ceiling: Optional[float] = None self.state_callbacks: list[Callable[[RecognitionState], None]] = [] self.action_callbacks: list[Callable[[str], None]] = [] diff --git a/src/vocalinux/ui/config_manager.py b/src/vocalinux/ui/config_manager.py index b3379c1..976a767 100644 --- a/src/vocalinux/ui/config_manager.py +++ b/src/vocalinux/ui/config_manager.py @@ -199,6 +199,9 @@ DEFAULT_CONFIG = { # "ydotool"/"xdotool" when autodetection is wrong (#476). # VOCALINUX_FORCE_BACKEND overrides this for a single run. "backend": "auto", + # Route dictation into the in-app Dictation Pad window instead of + # injecting into other apps — the Wayland-safe fallback (#726). + "dictate_to_pad": False, }, "history": { "enabled": True, # Keep recent dictation snippets in the tray menu @@ -881,6 +884,14 @@ class ConfigManager: """Enable or disable the floating dictation overlay.""" self.set("ui", "show_overlay", bool(enabled)) + def is_dictate_to_pad_enabled(self) -> bool: + """Check if dictation is routed into the in-app pad (default False).""" + return self.get_bool("text_injection", "dictate_to_pad", False) + + def set_dictate_to_pad(self, enabled: bool) -> None: + """Route dictation into the in-app Dictation Pad instead of injecting.""" + self.set("text_injection", "dictate_to_pad", bool(enabled)) + def _update_dict_recursive(self, target: dict, source: dict): """ Update a dictionary recursively. diff --git a/src/vocalinux/ui/dictation_pad.py b/src/vocalinux/ui/dictation_pad.py new file mode 100644 index 0000000..d1d5fd0 --- /dev/null +++ b/src/vocalinux/ui/dictation_pad.py @@ -0,0 +1,677 @@ +""" +In-app dictation pad for Vocalinux. + +A small persistent window with an editable text view that receives dictated +text directly, bypassing cross-application text injection entirely. It is the +explicit fallback path for desktops where injecting keystrokes into another +window is unreliable or blocked (notably Wayland compositors): the user +dictates into the pad, then selects and copies the text into the real target. + +Like DictationOverlay, the pure buffer logic lives in DictationPadController +so unit tests can drive it without a display; the GTK wrapper degrades to a +headless no-op when GTK is unavailable. +""" + +from __future__ import annotations + +import logging +from typing import TYPE_CHECKING, Any, Callable, Optional + +if TYPE_CHECKING: + from .config_manager import ConfigManager + +logger = logging.getLogger(__name__) + +_PAD_WIDTH = 480 +_PAD_HEIGHT = 360 +_COPIED_FEEDBACK_MS = 1200 +_UNDO_LIMIT = 200 + + +def _ui_errors() -> tuple[type[BaseException], ...]: + """Exception types tolerated by the pad's UI glue. + + GTK and config calls fail with a bounded set of runtime exceptions — + naming them (instead of ``Exception``) keeps real bugs visible while + the pad still degrades gracefully on systems without a usable display. + ``GLib.Error`` joins only when gi is importable, so the module stays + GTK-free for tests. + """ + errors: tuple[type[BaseException], ...] = ( + ImportError, + RuntimeError, + ValueError, + TypeError, + AttributeError, + OSError, + ) + try: + from gi.repository import GLib + except (ImportError, ValueError): + return errors + # Under gi-mocked test runs GLib.Error is a MagicMock, not a class — + # only real exception types may join the tuple. + glib_error = getattr(GLib, "Error", None) + if isinstance(glib_error, type) and issubclass(glib_error, BaseException): + return errors + (glib_error,) + return errors + + +# Voice commands the pad performs itself while capture mode is on. They are +# the same actions ActionHandler would otherwise send as keystrokes into +# whichever application happens to hold focus. +_HISTORY_ACTIONS = frozenset({"undo", "redo"}) +_WIDGET_ACTIONS = frozenset( + {"select_all", "select_line", "select_word", "select_paragraph", "cut", "copy", "paste"} +) + + +class DictationPadController: + """ + Pure buffer state for the dictation pad: the accumulated text plus a + bounded undo history. + + Separated from GTK so unit tests can exercise append/delete semantics + without a display. + """ + + def __init__(self, enabled: bool = False) -> None: + self._enabled = bool(enabled) + self._text = "" + # Dictated segments still present in the buffer, oldest first. Manual + # widget edits blur the boundaries (the list clears); undo snapshots + # restore them, so "delete that" can retarget after history moves. + self._segments: list[str] = [] + self._undo_stack: list[tuple[str, list[str]]] = [] + self._redo_stack: list[tuple[str, list[str]]] = [] + + @property + def enabled(self) -> bool: + """Whether dictation is routed into the pad instead of injected.""" + return self._enabled + + def set_enabled(self, enabled: bool) -> None: + """Enable or disable capture mode (does not touch the buffer).""" + self._enabled = bool(enabled) + + @property + def text(self) -> str: + """The full contents of the pad.""" + return self._text + + @property + def last_segment(self) -> Optional[str]: + """The last dictated segment still in the buffer. + + ``None`` when the buffer holds no tracked dictation — empty, cleared, + or rewritten by a manual edit. Undo/redo restore the tracked tail + along with the text snapshot. + """ + return self._segments[-1] if self._segments else None + + def append(self, text: str) -> None: + """Append a transcription segment to the end of the buffer.""" + if not text: + return + self._record_undo() + self._text += text + self._segments.append(text) + + def set_text(self, text: str) -> None: + """Replace the buffer (e.g. edits made directly in the widget).""" + if text == self._text: + return + self._record_undo() + self._text = text + # An arbitrary replacement blurs segment boundaries; undo restores + # them from the snapshot. + self._segments = [] + + def delete_last(self, count: int) -> int: + """ + Delete up to ``count`` characters from the end of the buffer. + + Returns the number of characters actually removed. + """ + if count <= 0 or not self._text: + return 0 + deleted = min(count, len(self._text)) + self._record_undo() + self._text = self._text[:-deleted] + self._trim_segments(deleted) + return deleted + + def clear(self) -> None: + """Drop everything in the pad.""" + if not self._text: + return + self._record_undo() + self._text = "" + self._segments = [] + + def undo(self) -> bool: + """Revert the last buffer mutation. Returns False with empty history.""" + if not self._undo_stack: + return False + self._redo_stack.append((self._text, list(self._segments))) + self._text, self._segments = self._undo_stack.pop() + return True + + def redo(self) -> bool: + """Re-apply the last undone mutation. Returns False with no redo lane.""" + if not self._redo_stack: + return False + self._undo_stack.append((self._text, list(self._segments))) + self._text, self._segments = self._redo_stack.pop() + return True + + def _trim_segments(self, count: int) -> None: + """Drop ``count`` characters from the tracked segment tail.""" + remaining = count + while remaining > 0 and self._segments: + tail = self._segments[-1] + if len(tail) <= remaining: + self._segments.pop() + remaining -= len(tail) + else: + self._segments[-1] = tail[:-remaining] + remaining = 0 + + def _record_undo(self) -> None: + """Snapshot the buffer before a mutation; new edits drop the redo lane.""" + self._undo_stack.append((self._text, list(self._segments))) + if len(self._undo_stack) > _UNDO_LIMIT: + del self._undo_stack[0] + self._redo_stack.clear() + + +class DictationPad: + """ + The dictation pad window: a GTK window backed by DictationPadController. + + Text arrives on the recognition thread, so every mutation of the visible + widget is marshalled onto the GTK main loop with GLib.idle_add. Closing + the window hides it — the buffer survives for the session. + """ + + def __init__( + self, + enabled: bool = False, + config_manager: Optional["ConfigManager"] = None, + ) -> None: + self.controller = DictationPadController(enabled=enabled) + self._config_manager = config_manager + self._window: Any = None + self._textview: Any = None + self._buffer: Any = None + self._capture_check: Any = None + self._copy_button: Any = None + self._gtk_ready = False + self._syncing_capture_check = False + self._syncing_widget = False + # Monotonic tag stamped on every queued widget op. A Clear (or a full + # refresh) bumps it, so appends still waiting in the GTK idle queue + # can tell they are stale and must not resurrect removed text. + self._generation = 0 + # Widget ops waiting in the GTK idle queue, in queue order. GTK input + # events outrank idle callbacks, so a manual edit can land before a + # queued append — _flush_idle_ops replays this list to keep the view + # in the order the controller already applied. + self._pending_idle: list[tuple[Callable[..., None], tuple[Any, ...]]] = [] + self._copied_feedback_id: Optional[int] = None + + try: + self._init_gtk_window() + self._gtk_ready = True + except _ui_errors() as e: + # Headless / missing display: keep the controller usable for tests. + logger.warning("Dictation pad window unavailable: %s", e) + + def _init_gtk_window(self) -> None: + """Construct the pad window, text view and action buttons.""" + import gi + + gi.require_version("Gtk", "3.0") + gi.require_version("Gdk", "3.0") + from gi.repository import Gdk, GLib, Gtk # noqa: F401 + + self._GLib = GLib + self._Gdk = Gdk + self._Gtk = Gtk + + window = Gtk.Window(type=Gtk.WindowType.TOPLEVEL) + window.set_title("Vocalinux Dictation Pad") + window.set_default_size(_PAD_WIDTH, _PAD_HEIGHT) + # Closing only hides: the pad keeps its buffer for the whole session + # and must not take the app down with it. + window.connect("delete-event", self._on_delete_event) + + outer = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=8) + outer.set_margin_start(12) + outer.set_margin_end(12) + outer.set_margin_top(12) + outer.set_margin_bottom(12) + window.add(outer) + + self._capture_check = Gtk.CheckButton( + label="Dictate into this window instead of injecting into other apps" + ) + self._capture_check.set_tooltip_text( + "Fallback for desktops where text injection does not work " + "(e.g. Wayland): dictated text lands in this box, then you copy " + "it out by hand." + ) + self._capture_check.set_active(self.controller.enabled) + self._capture_check.connect("toggled", self._on_capture_toggled) + outer.pack_start(self._capture_check, False, False, 0) + + scrolled = Gtk.ScrolledWindow() + scrolled.set_policy(Gtk.PolicyType.AUTOMATIC, Gtk.PolicyType.AUTOMATIC) + scrolled.set_hexpand(True) + scrolled.set_vexpand(True) + outer.pack_start(scrolled, True, True, 0) + + self._textview = Gtk.TextView() + self._textview.set_editable(True) + self._textview.set_wrap_mode(Gtk.WrapMode.WORD_CHAR) + self._textview.set_margin_start(8) + self._textview.set_margin_end(8) + self._textview.set_margin_top(8) + self._textview.set_margin_bottom(8) + self._buffer = self._textview.get_buffer() + # Manual edits in the widget feed back into the controller so + # deletion history never computes against stale dictated text. + self._buffer.connect("changed", self._on_buffer_changed) + # insert-text/delete-range fire BEFORE the edit is applied, so the + # queued dictation ops replay ahead of it in true chronological order. + self._buffer.connect("insert-text", self._on_buffer_user_edit) + self._buffer.connect("delete-range", self._on_buffer_user_edit) + scrolled.add(self._textview) + + button_row = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=8) + button_row.set_halign(Gtk.Align.END) + outer.pack_start(button_row, False, False, 0) + + self._copy_button = Gtk.Button(label="Copy All") + self._copy_button.set_tooltip_text("Copy the entire pad to the clipboard") + self._copy_button.connect("clicked", self._on_copy_all_clicked) + button_row.pack_start(self._copy_button, False, False, 0) + + clear_button = Gtk.Button(label="Clear") + clear_button.set_tooltip_text("Erase everything in the pad") + clear_button.connect("clicked", self._on_clear_clicked) + button_row.pack_start(clear_button, False, False, 0) + + self._window = window + + # -- public API --------------------------------------------------------- + + def append_text(self, text: str) -> None: + """ + Append a transcription segment (safe to call from any thread). + + The buffer is updated immediately so a headless pad still captures + text; the widget refresh and auto-show happen on the GTK main loop, + tagged with the current generation so a later Clear can drop it. + """ + self.controller.append(text) + self._idle_add(self._apply_append, text, self._generation) + + def delete_last_chars(self, count: int) -> int: + """ + Delete up to ``count`` characters from the end (any thread). + + Used by the "delete that" voice command while capture mode is on. + Returns the number of characters actually removed. + """ + deleted = self.controller.delete_last(count) + if deleted: + self._idle_add(self._apply_delete, deleted, self._generation) + return deleted + + @property + def last_segment(self) -> Optional[str]: + """The last dictated segment still in the pad (None if untracked).""" + return self.controller.last_segment + + def show_pad(self) -> None: + """Show the pad window, resyncing capture state from live config.""" + if not self._gtk_ready or self._window is None: + return + self._sync_capture_check() + if not self._window.get_visible(): + self._window.show_all() + try: + self._window.present_with_time(self._Gtk.get_current_event_time()) + except _ui_errors(): + pass + + def set_capture_enabled(self, enabled: bool) -> None: + """Programmatically flip capture mode (keeps the checkbox in sync).""" + self.controller.set_enabled(enabled) + self._sync_capture_check() + + def destroy(self) -> None: + """Tear down the window and any pending feedback timer.""" + self._bump_generation() + self._buffer = None + self._textview = None + if self._copied_feedback_id is not None: + try: + self._GLib.source_remove(self._copied_feedback_id) + except _ui_errors(): + pass + self._copied_feedback_id = None + if self._window is not None: + try: + self._window.destroy() + except _ui_errors(): + pass + self._window = None + + # -- internals ---------------------------------------------------------- + + def _idle_add(self, func: Callable[..., None], *args: Any) -> None: + """Schedule ``func`` on the GTK main loop when GTK is available.""" + glib = getattr(self, "_GLib", None) + if glib is None: + return + entry = (func, args) + + def _call() -> bool: + # A flush may have already run this op; GLib still dispatches the + # callback, so only act while this entry is still pending. Match + # by identity — an equal payload must not cancel a later twin. + idx = next( + (i for i, pending in enumerate(self._pending_idle) if pending is entry), + None, + ) + if idx is not None: + del self._pending_idle[idx] + func(*args) + return False + + self._pending_idle.append(entry) + glib.idle_add(_call) + + def _bump_generation(self) -> None: + """Invalidate queued widget ops and forget them. + + Stale ops still in the GLib queue drop themselves by generation; + clearing the list keeps a later flush from replaying them first. + """ + self._generation += 1 + self._pending_idle = [] + + def _flush_idle_ops(self, keep_user_view: bool = False) -> None: + """Replay queued widget ops so the view catches up with the controller. + + GTK input events outrank idle callbacks: a keystroke can land in the + buffer before appends queued earlier, and the "changed" handler would + then write the pre-append view back over the controller — losing the + queued dictation. Running the queue first keeps the view (and the + controller sync) in the order the controller already applied. + + With ``keep_user_view`` — a manual edit being synced back — ops that + replace the whole buffer are dropped instead of run: the keystroke + already landed on the newer view, so a stale refresh would erase it. + """ + pending = self._pending_idle + self._pending_idle = [] + for func, args in pending: + if keep_user_view and func == self._apply_set_text: + continue + func(*args) + + def _apply_append(self, text: str, generation: int) -> None: + """Insert a segment at the end of the widget and keep the tail visible.""" + if self._buffer is None or generation != self._generation: + # Cleared or refreshed while this insert sat in the idle queue. + return + try: + self._syncing_widget = True + self._buffer.insert(self._buffer.get_end_iter(), text) + # Scroll to the end without place_cursor: moving the caret would + # collapse a selection the user is making for copy-out. + self._textview.scroll_to_iter(self._buffer.get_end_iter(), 0.0, True, 0.0, 1.0) + except _ui_errors() as e: + logger.debug("Could not append text to dictation pad: %s", e) + finally: + self._syncing_widget = False + # Reveal the pad on the first incoming segment; once visible, further + # appends update silently so the window never re-raises mid-selection. + if self._capture_enabled() and self._window is not None and not self._window.get_visible(): + self.show_pad() + + def _apply_delete(self, deleted: int, generation: int) -> None: + """Remove ``deleted`` characters from the end of the widget.""" + if self._buffer is None or generation != self._generation: + return + try: + self._syncing_widget = True + end = self._buffer.get_end_iter() + start = end.copy() + start.backward_chars(deleted) + self._buffer.delete(start, end) + except _ui_errors() as e: + logger.debug("Could not delete text in dictation pad: %s", e) + finally: + self._syncing_widget = False + + def _apply_set_text(self, text: str, generation: int) -> None: + """Replace the widget contents with ``text``.""" + if self._buffer is None or generation != self._generation: + return + try: + self._syncing_widget = True + self._buffer.set_text(text) + except _ui_errors() as e: + logger.debug("Could not update dictation pad view: %s", e) + finally: + self._syncing_widget = False + + def _apply_action(self, action: str, generation: int) -> None: + """Execute a selection or clipboard command on the GTK main loop.""" + if self._buffer is None or generation != self._generation: + return + try: + if action == "cut": + self._buffer.cut_clipboard(self._clipboard(), True) + elif action == "copy": + self._buffer.copy_clipboard(self._clipboard()) + elif action == "paste": + self._buffer.paste_clipboard(self._clipboard(), None, True) + else: + start, end = self._selection_bounds(action) + if start is not None: + self._buffer.select_range(start, end) + except _ui_errors() as e: + logger.debug("Could not perform %s in dictation pad: %s", action, e) + + def _clipboard(self) -> Any: + """Return the shared CLIPBOARD selection.""" + return self._Gtk.Clipboard.get(self._Gdk.SELECTION_CLIPBOARD) + + def _selection_bounds(self, action: str) -> tuple[Optional[Any], Optional[Any]]: + """Iter pair for a selection ``action`` taken at the insert mark. + + Mirrors the shortcuts ActionHandler would inject: select_line is + Home+Shift+End, select_word is Ctrl+Shift+Right, and + select_paragraph extends to the next blank line (``\\n\\n`` delimits + paragraphs, matching the "new paragraph" voice command). + """ + if action == "select_all": + return self._buffer.get_start_iter(), self._buffer.get_end_iter() + it = self._buffer.get_iter_at_mark(self._buffer.get_insert()) + start = it.copy() + end = it.copy() + if action == "select_line": + start.set_line_offset(0) + if not end.ends_line(): + end.forward_to_line_end() + return start, end + if action == "select_word": + end.forward_word_end() + return start, end + if action == "select_paragraph": + text = self._buffer.get_text( + self._buffer.get_start_iter(), self._buffer.get_end_iter(), False + ) + boundary = text.find("\n\n", it.get_offset()) + end = self._buffer.get_iter_at_offset(len(text) if boundary < 0 else boundary) + return start, end + return None, None + + def _sync_widget_from_controller(self) -> None: + """Queue a full widget refresh, dropping appends queued before it.""" + self._bump_generation() + self._idle_add(self._apply_set_text, self.controller.text, self._generation) + + def _on_buffer_user_edit(self, *_args: Any) -> None: + """Replay queued dictation ops before a manual edit reaches the buffer. + + GTK input events outrank idle callbacks, so a keystroke can land + before an append queued earlier. insert-text/delete-range fire before + the edit is applied, so flushing here keeps the widget (and the + controller sync that follows) in the order the controller applied. + """ + if self._syncing_widget: + return + self._flush_idle_ops() + + def _on_buffer_changed(self, buffer: Any) -> None: + """Mirror edits made directly in the widget back into the controller.""" + if self._syncing_widget: + return + # Fallback for edit paths that bypass insert-text/delete-range. A + # stale whole-view refresh must not erase the edit that just landed. + self._flush_idle_ops(keep_user_view=True) + try: + text = buffer.get_text(buffer.get_start_iter(), buffer.get_end_iter(), False) + except _ui_errors() as e: + logger.debug("Could not sync dictation pad edits: %s", e) + return + self.controller.set_text(text) + + def _capture_enabled(self) -> bool: + """Return whether dictation is currently routed into the pad. + + Prefers the live config value so a Settings change applies even + before the pad is ever shown; falls back to the controller flag + when no config manager is attached. + """ + if self._config_manager is None: + return self.controller.enabled + try: + enabled = bool(self._config_manager.get_bool("text_injection", "dictate_to_pad", False)) + except (AttributeError, TypeError, KeyError): + return self.controller.enabled + self.controller.set_enabled(enabled) + return enabled + + def _sync_capture_check(self) -> None: + """Mirror the persisted capture setting onto the checkbox.""" + if self._capture_check is None: + return + try: + enabled = self._capture_enabled() + self._syncing_capture_check = True + self._capture_check.set_active(enabled) + except _ui_errors() as e: + logger.debug("Could not sync dictation pad toggle: %s", e) + finally: + self._syncing_capture_check = False + + def _on_delete_event(self, *_args: Any) -> bool: + """Hide on close instead of destroying — the buffer is session-scoped.""" + if self._window is not None: + self._window.hide() + return True + + def _on_capture_toggled(self, widget: Any) -> None: + """Persist the pad's own capture checkbox.""" + if self._syncing_capture_check: + return + enabled = bool(widget.get_active()) + self.controller.set_enabled(enabled) + if self._config_manager is not None: + try: + self._config_manager.set("text_injection", "dictate_to_pad", enabled) + self._config_manager.save_settings() + except (AttributeError, TypeError) as e: + logger.warning("Could not save dictation pad setting: %s", e) + logger.info("Dictation pad capture %s", "enabled" if enabled else "disabled") + + def _on_copy_all_clicked(self, *_args: Any) -> None: + """Copy the entire buffer to the clipboard and flash a confirmation.""" + # The widget's text may include the user's own in-pad edits; fall back + # to the dictated buffer when there is no widget (headless). + text = self.controller.text + if self._buffer is not None: + # Land queued appends first so Copy All never misses the latest + # dictated segment still sitting in the idle queue. + self._flush_idle_ops() + try: + text = self._buffer.get_text( + self._buffer.get_start_iter(), self._buffer.get_end_iter(), False + ) + except _ui_errors(): + pass + try: + clipboard = self._clipboard() + clipboard.set_text(text, -1) + clipboard.store() + except _ui_errors() as e: + logger.warning("Could not copy dictation pad contents: %s", e) + return + if self._copy_button is None or self._copied_feedback_id is not None: + return + self._copy_button.set_label("Copied") + self._copied_feedback_id = self._GLib.timeout_add( + _COPIED_FEEDBACK_MS, self._reset_copy_button + ) + + def _reset_copy_button(self) -> bool: + """Restore the Copy All label after the brief confirmation flash.""" + if self._copy_button is not None: + self._copy_button.set_label("Copy All") + self._copied_feedback_id = None + return False + + def _on_clear_clicked(self, *_args: Any) -> None: + """Erase the buffer and drop appends still queued for the widget.""" + # Bump the generation first: any append or delete already sitting in + # the GTK idle queue becomes stale and cannot resurrect cleared text. + self._bump_generation() + self.controller.clear() + if self._buffer is not None: + try: + self._syncing_widget = True + self._buffer.set_text("") + except _ui_errors() as e: + logger.debug("Could not clear dictation pad view: %s", e) + finally: + self._syncing_widget = False + + def handle_action(self, action: str) -> bool: + """ + Perform an editing voice command on the pad (any thread). + + While capture mode is on, the pad owns every editing action — they + must never reach the application that happens to hold keyboard + focus. History commands run on the controller; selection and + clipboard commands are marshalled onto the GTK main loop. + """ + if action in _HISTORY_ACTIONS: + changed = self.controller.undo() if action == "undo" else self.controller.redo() + if changed: + self._sync_widget_from_controller() + return changed + if action in _WIDGET_ACTIONS: + if self._buffer is None: + logger.debug("Dictation pad cannot %s without a text view", action) + return False + self._idle_add(self._apply_action, action, self._generation) + return True + logger.warning("Dictation pad does not handle action: %s", action) + return False diff --git a/src/vocalinux/ui/settings_dialog.py b/src/vocalinux/ui/settings_dialog.py index 6d12f90..e4ddddd 100644 --- a/src/vocalinux/ui/settings_dialog.py +++ b/src/vocalinux/ui/settings_dialog.py @@ -102,6 +102,7 @@ from .config_manager import ( # noqa: E402 DEFAULT_SOUND_EFFECT_TONE, PASTE_SHORTCUTS, SOUND_EFFECT_TONES, + normalize_language_shortcuts, resolve_whispercpp_variant, ) from .keyboard_backends import ( # noqa: E402 @@ -140,6 +141,11 @@ logger = logging.getLogger(__name__) # must agree on whose snapshot is newest. _apply_settings_lock = threading.Lock() _apply_settings_generation = 0 +# Generation of the apply that last wrote each advanced key. A deferred +# persist uses it to tell "a newer apply rewrote this key" apart from "the +# config still holds the baseline" — a newer user choice can legally equal +# that baseline, so comparing values alone is not enough. +_apply_settings_written: dict[str, int] = {} def _raw_audio_device_name(device_name: Optional[str]) -> Optional[str]: @@ -3735,6 +3741,18 @@ class SettingsDialog(Gtk.Dialog): logger.info(f"Copy to clipboard {'enabled' if enabled else 'disabled'}") return False + def _on_dictation_pad_toggled(self, widget: Gtk.Switch, state: bool) -> bool: + """Handle toggle of the in-app dictation pad switch.""" + if _handlers_suppressed(self): + return False + + enabled = bool(state) + logger.info(f"Dictate to pad toggled: {enabled}") + self.config_manager.set("text_injection", "dictate_to_pad", enabled) + self.config_manager.save_settings() + logger.info(f"Dictate to pad {'enabled' if enabled else 'disabled'}") + return False + def _on_auto_capitalize_toggled(self, widget, state): """Handle toggle of the auto-capitalize switch.""" if _handlers_suppressed(self): @@ -4289,6 +4307,19 @@ class SettingsDialog(Gtk.Dialog): ), ) + self.dictation_pad_switch = _add_switch_row( + output_group, + title="Dictation Pad", + subtitle="Capture dictation in an in-app text box instead of other apps", + keywords=("wayland", "pad", "fallback", "text box"), + tooltip=( + "Type dictation into Vocalinux's own Dictation Pad window instead " + "of injecting it into other apps. The pad opens from the tray menu; " + "copy text out of it by hand. Useful on Wayland, where injecting " + "keystrokes into other windows is restricted." + ), + ) + self.append_trailing_space_switch = _add_switch_row( output_group, title="Trailing Space After Dictation", @@ -4323,6 +4354,7 @@ class SettingsDialog(Gtk.Dialog): self.recognition_settings_tab.pack_start(output_group, False, False, 0) self.copy_to_clipboard_switch.connect("state-set", self._on_copy_to_clipboard_toggled) + self.dictation_pad_switch.connect("state-set", self._on_dictation_pad_toggled) self.auto_capitalize_switch.connect("state-set", self._on_auto_capitalize_toggled) self.append_trailing_space_switch.connect( "state-set", self._on_append_trailing_space_toggled @@ -4755,17 +4787,33 @@ class SettingsDialog(Gtk.Dialog): """ if self._initializing: return + # Each row's remembered binding belongs to that row first, so a + # shortcut typed into one row can never displace a binding another + # row already holds — the stealing edit is rejected instead. + owners: dict[str, int] = {} + for index, refs in enumerate(self._language_shortcut_rows): + owned = (refs.get("last_valid_shortcut") or "").strip().lower() + if owned and owned not in owners: + owners[owned] = index entries = [] rejected = [] claimed = set() - for refs in self._language_shortcut_rows: + for index, refs in enumerate(self._language_shortcut_rows): language = refs["language_picker"].get_active_id() or "auto" shortcut = refs["shortcut_entry"].get_text().strip().lower() last_valid = refs.get("last_valid_shortcut", "") - if shortcut and is_valid_shortcut(shortcut) and shortcut not in claimed: + owner = owners.get(shortcut) + if ( + shortcut + and is_valid_shortcut(shortcut) + and shortcut not in claimed + and (owner is None or owner == index) + ): refs["last_valid_shortcut"] = shortcut claimed.add(shortcut) - elif last_valid and last_valid not in claimed: + elif ( + last_valid and last_valid not in claimed and owners.get(last_valid, index) == index + ): if shortcut and is_valid_shortcut(shortcut): rejected.append(shortcut) shortcut = last_valid @@ -4775,6 +4823,12 @@ class SettingsDialog(Gtk.Dialog): rejected.append(shortcut) continue entries.append({"shortcut": shortcut, "language": language}) + if normalize_language_shortcuts(entries) == self.config_manager.get_language_shortcuts(): + # Nothing effective changed — a mid-edit keystroke, a rejected + # duplicate, or a reverted field — so the live listeners do not + # need another rebuild (and dictation never pauses for one). + self._report_language_shortcut_rejections(rejected) + return self.config_manager.set_language_shortcuts(entries) self.config_manager.save_settings() if self.language_shortcuts_update_callback: @@ -6475,11 +6529,16 @@ class SettingsDialog(Gtk.Dialog): # A newer apply already reconfigured the shared engine # and saved newer values. Keep only the keys it left # untouched — the rest of the snapshot is stale and - # must not reach the engine or the config again. + # must not reach the engine or the config again. A key + # counts as touched only when the newer apply wrote + # it: its choice can legally equal the baseline, and + # the older edit must not overwrite it. surviving = { key: value for key, value in pending.items() - if self.config_manager.get("advanced", key) == baseline.get(key) + if _apply_settings_written.get(key, self._pending_apply_generation) + <= self._pending_apply_generation + and self.config_manager.get("advanced", key) == baseline.get(key) } else: surviving = dict(pending) @@ -6494,6 +6553,7 @@ class SettingsDialog(Gtk.Dialog): self.speech_engine.reconfigure(**surviving) for key, value in surviving.items(): self.config_manager.set("advanced", key, value) + _apply_settings_written[key] = _apply_settings_generation self.config_manager.save_settings() except (OSError, ValueError, TypeError, RuntimeError) as e: logger.warning( @@ -6573,6 +6633,7 @@ class SettingsDialog(Gtk.Dialog): show_missing_tray_warning = ui_settings.get("show_missing_tray_warning", True) show_overlay = ui_settings.get("show_overlay", True) copy_to_clipboard = text_injection_settings.get("copy_to_clipboard", False) + dictate_to_pad = text_injection_settings.get("dictate_to_pad", False) 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() @@ -6586,6 +6647,7 @@ class SettingsDialog(Gtk.Dialog): self.missing_tray_warning_switch.set_active(show_missing_tray_warning) self.show_overlay_switch.set_active(show_overlay) self.copy_to_clipboard_switch.set_active(copy_to_clipboard) + self.dictation_pad_switch.set_active(dictate_to_pad) self.auto_capitalize_switch.set_active(auto_capitalize) self.append_trailing_space_switch.set_active(append_trailing_space) if not self.paste_shortcut_combo.set_active_id(paste_shortcut): @@ -8101,6 +8163,7 @@ class SettingsDialog(Gtk.Dialog): # generation that already includes this in-flight apply. global _apply_settings_generation _apply_settings_generation += 1 + apply_generation = _apply_settings_generation self._applying_settings = True # Already-downloaded apply is handed to a worker that clears this flag # via GLib.idle_add. Download still holds it for the modal run() below. @@ -8203,7 +8266,9 @@ class SettingsDialog(Gtk.Dialog): def apply_already_downloaded() -> None: try: - self._apply_settings_internal(settings, raise_errors=True) + self._apply_settings_internal( + settings, raise_errors=True, apply_generation=apply_generation + ) logger.info("Settings auto-applied successfully") except Exception as e: logger.error(f"Failed to auto-apply settings: {e}") @@ -8438,13 +8503,15 @@ 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 + # stamp the window so test speech stays out of Recent Snippets while # leftover segments from a dictation still file normally. + self.speech_engine.test_capture_ceiling = None 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 + self.speech_engine.test_capture_ceiling = None del self._saved_text_callbacks self.test_output_revealer.set_reveal_child(True) if getattr(self.speech_engine, "is_auto_paused", False): @@ -8650,7 +8717,10 @@ 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 + # Close the test-capture window rather than reopening history for + # everything after the floor: segments captured during the test but + # still decoding must keep failing the window check. + self.speech_engine.test_capture_ceiling = time.monotonic() # Check result after giving time for final callbacks to complete GLib.timeout_add(300, self._check_test_result) @@ -8802,7 +8872,11 @@ For now, the engine has been reverted to VOSK.""" dialog.destroy() def _apply_settings_internal( - self, settings: dict, raise_errors: bool = False, force_reinit: bool = False + self, + settings: dict, + raise_errors: bool = False, + force_reinit: bool = False, + apply_generation: Optional[int] = None, ) -> bool: """Internal method to apply settings. @@ -8816,20 +8890,37 @@ For now, the engine has been reverted to VOSK.""" matches its live state. The download threads pass True: they only run because the model is missing on disk, and a no-op reconfigure would report success for a download that never ran. + apply_generation: The apply generation captured when this snapshot + was collected. When a newer apply began since, this snapshot is + stale and its write is skipped — otherwise lock order, not + collection order, would decide which snapshot wins. """ try: - was_running = self.speech_engine.state != RecognitionState.IDLE - if was_running: - self.speech_engine.stop_recognition() - time.sleep(0.5) - # Persist only once the engine really runs these settings: this call # downloads missing models, and a config saved up front would keep # pointing at a model that never made it to disk. The lock orders - # this snapshot against _persist_pending_text_edits' deferred one. + # this snapshot against _persist_pending_text_edits' deferred one — + # and the staleness check must precede stop_recognition, or a + # superseded apply would still interrupt a live dictation session. with _apply_settings_lock: + if apply_generation is not None and _apply_settings_generation != apply_generation: + logger.info( + "Settings apply superseded by a newer one; " "skipping the stale snapshot" + ) + return True + was_running = self.speech_engine.state != RecognitionState.IDLE + if was_running: + self.speech_engine.stop_recognition() + time.sleep(0.5) + self.speech_engine.reconfigure(force_reinit=force_reinit, **settings) self._save_selected_settings(settings) + written_gen = ( + apply_generation if apply_generation is not None else _apply_settings_generation + ) + for key in settings: + if key.startswith("whispercpp_"): + _apply_settings_written[key] = written_gen logger.info("Settings applied successfully.") return True diff --git a/src/vocalinux/ui/transcription_history.py b/src/vocalinux/ui/transcription_history.py index ac996aa..7891646 100644 --- a/src/vocalinux/ui/transcription_history.py +++ b/src/vocalinux/ui/transcription_history.py @@ -58,9 +58,12 @@ class TranscriptionHistory: 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._entries: deque[tuple[int, str]] = deque(maxlen=self._max_items) self._lock = threading.Lock() self._change_callback: Optional[Callable[[], None]] = None + # Monotonic id stamped on each snippet so late segments can extend + # the session they belong to even after newer snippets arrive. + self._next_id = 0 # Bumped every time the entries are wiped; lets callers refuse text # produced before a clear so it cannot reappear afterwards. self._epoch = 0 @@ -130,8 +133,8 @@ class TranscriptionHistory: 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. + def add(self, text: str, *, expected_epoch: Optional[int] = None) -> Optional[int]: + """Add a snippet, returning its id when recorded (``None`` on refusal). 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 @@ -139,16 +142,42 @@ class TranscriptionHistory: produced before the clear cannot reappear as a new entry. """ if not text: - return False + return None text = text.strip() if not text: - return False + return None with self._lock: if not self._enabled: + return None + if expected_epoch is not None and expected_epoch != self._epoch: + return None + self._next_id += 1 + self._entries.append((self._next_id, text)) + self._notify() + return self._next_id + + def extend_entry( + self, snippet_id: int, text: str, *, expected_epoch: Optional[int] = None + ) -> bool: + """Append a late-arriving segment to the snippet with ``snippet_id``. + + Unlike ``extend_latest`` the target is the entry's own id, so a + straggler still lands in the session that produced it even when a + newer session has since committed its snippet on top. + """ + 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.append(text) + for index, (entry_id, existing) in enumerate(self._entries): + if entry_id == snippet_id: + self._entries[index] = (entry_id, f"{existing} {text.strip()}") + break + else: + return False self._notify() return True @@ -173,14 +202,15 @@ class TranscriptionHistory: return False if expected_epoch is not None and expected_epoch != self._epoch: return False - self._entries[-1] = f"{self._entries[-1]} {text.strip()}" + entry_id, existing = self._entries[-1] + self._entries[-1] = (entry_id, f"{existing} {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)) + return [text for _, text in reversed(self._entries)] def clear(self) -> None: """Remove all snippets. diff --git a/src/vocalinux/ui/tray_indicator.py b/src/vocalinux/ui/tray_indicator.py index 9e367ee..ed9e2ba 100644 --- a/src/vocalinux/ui/tray_indicator.py +++ b/src/vocalinux/ui/tray_indicator.py @@ -149,6 +149,7 @@ class TrayIndicator: speech_engine: SpeechRecognitionManagerProtocol, text_injector: TextInjectorProtocol, transcription_history: Optional[TranscriptionHistory] = None, + dictation_pad: Optional[Any] = None, ) -> None: """ Initialize the system tray indicator. @@ -158,10 +159,13 @@ class TrayIndicator: text_injector: The text injector instance transcription_history: Optional in-memory store of recent dictation snippets. When provided, a "Recent Snippets" submenu is shown. + dictation_pad: Optional in-app Dictation Pad window the tray menu + can open (the Wayland-safe dictation fallback, #726) """ self.speech_engine = speech_engine self.text_injector = text_injector self.transcription_history = transcription_history + self.dictation_pad = dictation_pad # 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() @@ -190,6 +194,9 @@ class TrayIndicator: # One listener per bound language shortcut (#805); rebuilt by # _setup_language_shortcuts alongside the main one. self._language_shortcut_managers: list[KeyboardShortcutManager] = [] + # A settings edit landed while a dictation session was live; the + # listener rebuild is deferred to the next IDLE transition (#805). + self._language_shortcuts_refresh_pending = False # The binding whose press started the live push-to-talk session, so an # unrelated key's release cannot end it (#805). self._ptt_owner: Optional[KeyboardShortcutManager] = None @@ -466,12 +473,20 @@ class TrayIndicator: self._language_shortcut_managers = [] def refresh_language_shortcuts(self) -> None: - """Rebuild the per-language listeners after a settings change (#805).""" + """Rebuild the per-language listeners after a settings change (#805). + + While a dictation session is live the rebuild is deferred to the + next IDLE transition instead: rebuilding stops recognition first, + and a keystroke-level settings edit must never cut an utterance off. + """ if self._external_activation_active(): # The internal listener family is off; a binding edit must not # arm language listeners behind the external-activation setting. self._stop_language_shortcut_managers() return + if self.speech_engine.state != RecognitionState.IDLE: + self._language_shortcuts_refresh_pending = True + return self._setup_language_shortcuts() def _init_icons(self): @@ -570,6 +585,8 @@ class TrayIndicator: self._update_autostart_checkbox() self._add_menu_separator() + if self.dictation_pad is not None: + self._add_menu_item("Dictation Pad", self._on_dictation_pad_clicked) self._add_menu_item("Settings", self._on_settings_clicked) self._add_menu_item("View Logs", self._on_logs_clicked) self._gateway_stop_menu_item = self._add_menu_item( @@ -1198,13 +1215,20 @@ class TrayIndicator: state wins: GLib.idle_add can deliver a captured PROCESSING argument after stop already went IDLE (#739). """ - if not hasattr(self, "indicator"): - return False - engine_state = getattr(self.speech_engine, "state", None) if isinstance(engine_state, RecognitionState): state = engine_state + if state == RecognitionState.IDLE and getattr( + self, "_language_shortcuts_refresh_pending", False + ): + # The session the listener rebuild was deferred for has ended. + self._language_shortcuts_refresh_pending = False + self.refresh_language_shortcuts() + + if not hasattr(self, "indicator"): + return False + if state == RecognitionState.IDLE: self._set_indicator_icon(self._icon_keys["default"], "Microphone off") self._set_menu_item_enabled("Start Voice Typing", True) @@ -1325,6 +1349,12 @@ class TrayIndicator: if self.transcription_history is not None: self.transcription_history.clear() + def _on_dictation_pad_clicked(self, widget: Gtk.MenuItem) -> None: + """Handle click on the Dictation Pad menu item.""" + logger.debug("Dictation Pad clicked") + if self.dictation_pad is not None: + self.dictation_pad.show_pad() + def _on_settings_clicked(self, widget): """Handle click on the Settings menu item.""" logger.debug("Settings clicked") diff --git a/tests/test_dictation_pad.py b/tests/test_dictation_pad.py new file mode 100644 index 0000000..f3b4340 --- /dev/null +++ b/tests/test_dictation_pad.py @@ -0,0 +1,806 @@ +""" +Tests for the in-app dictation pad (issue #726). + +Drives the shipped DictationPadController (and config helpers) so capture +mode routes text into the pad's buffer instead of cross-app injection. + +Important: these tests must not import real GTK / tray_indicator into +sys.modules — that breaks later tests that mock gi (see CI isolation). +""" + +import os +import tempfile +import unittest +from unittest.mock import MagicMock, patch + +from vocalinux.ui.config_manager import DEFAULT_CONFIG, ConfigManager +from vocalinux.ui.dictation_pad import DictationPad, DictationPadController + + +def _ensure_test_config_dir(path: str) -> None: + 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) + + +def _pad_without_gtk(enabled: bool = False) -> DictationPad: + """ + Build DictationPad without initializing a real GTK window. + + Patches window construction so tests never load gi.repository (which + would pollute the process for later gi-mocked tests). + """ + with patch.object( + DictationPad, + "_init_gtk_window", + side_effect=RuntimeError("gtk disabled in unit tests"), + ): + return DictationPad(enabled=enabled) + + +class TestDictationPadController(unittest.TestCase): + """Unit tests for the pure buffer controller.""" + + def test_default_disabled_and_empty(self) -> None: + ctrl = DictationPadController() + self.assertFalse(ctrl.enabled) + self.assertEqual(ctrl.text, "") + + def test_append_accumulates_segments(self) -> None: + ctrl = DictationPadController(enabled=True) + ctrl.append("Hello ") + ctrl.append("world.") + self.assertEqual(ctrl.text, "Hello world.") + + def test_append_newlines_preserved(self) -> None: + ctrl = DictationPadController() + ctrl.append("first line") + ctrl.append("\n") + ctrl.append("second") + self.assertEqual(ctrl.text, "first line\nsecond") + + def test_delete_last_removes_tail_chars(self) -> None: + ctrl = DictationPadController() + ctrl.append("Hello world") + self.assertEqual(ctrl.delete_last(6), 6) + self.assertEqual(ctrl.text, "Hello") + + def test_delete_last_clamps_to_buffer_length(self) -> None: + ctrl = DictationPadController() + ctrl.append("hi") + self.assertEqual(ctrl.delete_last(10), 2) + self.assertEqual(ctrl.text, "") + + def test_delete_last_on_empty_returns_zero(self) -> None: + ctrl = DictationPadController() + self.assertEqual(ctrl.delete_last(5), 0) + ctrl.append("x") + self.assertEqual(ctrl.delete_last(0), 0) + self.assertEqual(ctrl.delete_last(-3), 0) + self.assertEqual(ctrl.text, "x") + + def test_clear_empties_buffer(self) -> None: + ctrl = DictationPadController() + ctrl.append("some dictation") + ctrl.clear() + self.assertEqual(ctrl.text, "") + + def test_set_text_replaces_buffer(self) -> None: + ctrl = DictationPadController() + ctrl.append("dictated") + ctrl.set_text("dictated plus edits") + self.assertEqual(ctrl.text, "dictated plus edits") + # Identical content is a no-op, not a mutation: the first undo goes + # straight back to the pre-edit text. + ctrl.set_text("dictated plus edits") + self.assertTrue(ctrl.undo()) + self.assertEqual(ctrl.text, "dictated") + + def test_undo_restores_last_mutation_and_redo_replays(self) -> None: + ctrl = DictationPadController() + ctrl.append("one ") + ctrl.append("two ") + self.assertTrue(ctrl.undo()) + self.assertEqual(ctrl.text, "one ") + self.assertTrue(ctrl.undo()) + self.assertEqual(ctrl.text, "") + self.assertFalse(ctrl.undo()) + self.assertTrue(ctrl.redo()) + self.assertEqual(ctrl.text, "one ") + self.assertTrue(ctrl.redo()) + self.assertEqual(ctrl.text, "one two ") + self.assertFalse(ctrl.redo()) + + def test_new_edit_clears_redo_lane(self) -> None: + ctrl = DictationPadController() + ctrl.append("a") + ctrl.undo() + ctrl.append("b") + self.assertFalse(ctrl.redo()) + + def test_delete_last_and_clear_are_undoable(self) -> None: + ctrl = DictationPadController() + ctrl.append("hello world") + ctrl.delete_last(6) + self.assertEqual(ctrl.text, "hello") + self.assertTrue(ctrl.undo()) + self.assertEqual(ctrl.text, "hello world") + ctrl.clear() + self.assertEqual(ctrl.text, "") + self.assertTrue(ctrl.undo()) + self.assertEqual(ctrl.text, "hello world") + + def test_set_enabled_does_not_touch_buffer(self) -> None: + ctrl = DictationPadController() + ctrl.append("keep me") + ctrl.set_enabled(True) + ctrl.set_enabled(False) + self.assertEqual(ctrl.text, "keep me") + + def test_last_segment_tracks_appends(self) -> None: + ctrl = DictationPadController() + self.assertIsNone(ctrl.last_segment) + ctrl.append("hello ") + ctrl.append("world ") + self.assertEqual(ctrl.last_segment, "world ") + + def test_last_segment_tracks_undo_and_redo(self) -> None: + """Undo pops the deletion target with the text it removed.""" + ctrl = DictationPadController() + ctrl.append("hello ") + ctrl.append("world ") + self.assertTrue(ctrl.undo()) + self.assertEqual(ctrl.last_segment, "hello ") + self.assertTrue(ctrl.undo()) + self.assertIsNone(ctrl.last_segment) + self.assertTrue(ctrl.redo()) + self.assertEqual(ctrl.last_segment, "hello ") + self.assertTrue(ctrl.redo()) + self.assertEqual(ctrl.last_segment, "world ") + + def test_last_segment_survives_delete_and_drops_on_set_text(self) -> None: + ctrl = DictationPadController() + ctrl.append("hello ") + ctrl.append("world ") + ctrl.delete_last(2) + self.assertEqual(ctrl.last_segment, "worl") + ctrl.delete_last(10) + self.assertIsNone(ctrl.last_segment) + ctrl.append("again ") + ctrl.set_text("manual rewrite") + self.assertIsNone(ctrl.last_segment) + # Undo restores the boundaries the manual rewrite blurred. + self.assertTrue(ctrl.undo()) + self.assertEqual(ctrl.last_segment, "again ") + + def test_clear_drops_segment_tracking(self) -> None: + ctrl = DictationPadController() + ctrl.append("gone ") + ctrl.clear() + self.assertIsNone(ctrl.last_segment) + + +class TestDictationPadConfig(unittest.TestCase): + """ConfigManager helpers for the dictate_to_pad preference.""" + + def setUp(self) -> None: + self.temp_dir = tempfile.TemporaryDirectory() + self.temp_config_dir = os.path.join(self.temp_dir.name, ".config/vocalinux") + _ensure_test_config_dir(self.temp_config_dir) + self.temp_config_file = os.path.join(self.temp_config_dir, "config.json") + + self.config_dir_patcher = patch( + "vocalinux.ui.config_manager.CONFIG_DIR", self.temp_config_dir + ) + 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() + _ensure_test_config_dir(self.temp_config_dir) + + def tearDown(self) -> None: + self.config_dir_patcher.stop() + self.config_file_patcher.stop() + self.makedirs_patcher.stop() + self.temp_dir.cleanup() + + def test_default_dictate_to_pad_disabled(self) -> None: + self.assertFalse(DEFAULT_CONFIG["text_injection"]["dictate_to_pad"]) + cm = ConfigManager() + self.assertFalse(cm.is_dictate_to_pad_enabled()) + + def test_set_dictate_to_pad_persists(self) -> None: + cm = ConfigManager() + cm.set_dictate_to_pad(True) + self.assertTrue(cm.is_dictate_to_pad_enabled()) + cm.save_config() + + cm2 = ConfigManager() + self.assertTrue(cm2.is_dictate_to_pad_enabled()) + + def test_set_dictate_to_pad_false(self) -> None: + cm = ConfigManager() + cm.set_dictate_to_pad(True) + cm.set_dictate_to_pad(False) + self.assertFalse(cm.is_dictate_to_pad_enabled()) + + def test_corrupt_config_file_keeps_in_memory_capture_state(self) -> None: + """A torn config.json write must not flip a running session's routing. + + The live toggle is read from the shared manager's in-memory cache, + never re-parsed per segment: a partial save on disk cannot misroute + dictation back into whichever application holds focus. + """ + cm = ConfigManager() + cm.set_dictate_to_pad(True) + # Simulate a torn mid-write file. + with open(self.temp_config_file, "w") as f: + f.write('{"text_injection": {"dictate_to_pad": tr') + self.assertTrue(cm.is_dictate_to_pad_enabled()) + # A fresh manager loading the torn file fails closed to the default. + self.assertFalse(ConfigManager().is_dictate_to_pad_enabled()) + + +class TestDictationPadFacade(unittest.TestCase): + """ + Drive the shipped DictationPad API without a display. + + Window construction is patched out so these tests never load GTK; the + controller must still capture text so nothing is silently dropped. + """ + + def test_gtk_init_failure_keeps_buffer_usable(self) -> None: + pad = _pad_without_gtk(enabled=True) + self.assertFalse(pad._gtk_ready) + pad.append_text("dictated words ") + pad.append_text("more") + self.assertEqual(pad.controller.text, "dictated words more") + pad.destroy() + + def test_delete_last_chars_headless(self) -> None: + pad = _pad_without_gtk() + try: + pad.append_text("hello ") + self.assertEqual(pad.delete_last_chars(6), 6) + self.assertEqual(pad.controller.text, "") + finally: + pad.destroy() + + def test_show_pad_is_safe_headless(self) -> None: + pad = _pad_without_gtk() + try: + pad.show_pad() # must not raise + finally: + pad.destroy() + + def test_set_capture_enabled_updates_controller(self) -> None: + pad = _pad_without_gtk() + try: + pad.set_capture_enabled(True) + self.assertTrue(pad.controller.enabled) + pad.set_capture_enabled(False) + self.assertFalse(pad.controller.enabled) + finally: + pad.destroy() + + def test_capture_checkbox_persists_config(self) -> None: + """The pad checkbox writes dictate_to_pad via the config manager.""" + config_manager = MagicMock() + with patch.object( + DictationPad, + "_init_gtk_window", + side_effect=RuntimeError("gtk disabled in unit tests"), + ): + pad = DictationPad(enabled=False, config_manager=config_manager) + try: + widget = MagicMock() + widget.get_active.return_value = True + pad._on_capture_toggled(widget) + config_manager.set.assert_called_once_with("text_injection", "dictate_to_pad", True) + config_manager.save_settings.assert_called_once() + self.assertTrue(pad.controller.enabled) + finally: + pad.destroy() + + def test_apply_append_inserts_at_end_and_shows(self) -> None: + """Widget path: append inserts at buffer end; enabled capture shows.""" + pad = _pad_without_gtk(enabled=True) + try: + pad._gtk_ready = True + pad._window = MagicMock() + pad._window.get_visible.return_value = False + pad._textview = MagicMock() + pad._buffer = MagicMock() + pad._Gtk = MagicMock() + + pad.controller.append("segment ") + pad._apply_append("segment ", pad._generation) + + pad._buffer.insert.assert_called_once() + args = pad._buffer.insert.call_args[0] + self.assertEqual(args[1], "segment ") + pad._window.show_all.assert_called_once() + finally: + pad._window = None + pad.destroy() + + def test_apply_append_does_not_show_when_disabled(self) -> None: + pad = _pad_without_gtk(enabled=False) + try: + pad._gtk_ready = True + pad._window = MagicMock() + pad._window.get_visible.return_value = False + pad._textview = MagicMock() + pad._buffer = MagicMock() + + pad.controller.append("segment ") + pad._apply_append("segment ", pad._generation) + + pad._window.show_all.assert_not_called() + finally: + pad._window = None + pad.destroy() + + def test_apply_append_reveals_when_enabled_via_settings(self) -> None: + """Regression: capture toggled in Settings must reveal the pad too. + + The reveal gate consults live config, not just the controller flag, + so a dictate_to_pad=True written by the Settings switch while the + pad is closed still surfaces the window on the first segment. + """ + config_manager = MagicMock() + config_manager.get_bool.return_value = True + with patch.object( + DictationPad, + "_init_gtk_window", + side_effect=RuntimeError("gtk disabled in unit tests"), + ): + pad = DictationPad(enabled=False, config_manager=config_manager) + try: + pad._gtk_ready = True + pad._window = MagicMock() + pad._window.get_visible.return_value = False + pad._textview = MagicMock() + pad._buffer = MagicMock() + pad._Gtk = MagicMock() + + pad.controller.append("segment ") + pad._apply_append("segment ", pad._generation) + + config_manager.get_bool.assert_called_with("text_injection", "dictate_to_pad", False) + self.assertTrue(pad.controller.enabled) + pad._window.show_all.assert_called_once() + finally: + pad._window = None + pad.destroy() + + def test_apply_append_stays_hidden_when_settings_off(self) -> None: + """A stale controller flag must not override live config.""" + config_manager = MagicMock() + config_manager.get_bool.return_value = False + with patch.object( + DictationPad, + "_init_gtk_window", + side_effect=RuntimeError("gtk disabled in unit tests"), + ): + pad = DictationPad(enabled=True, config_manager=config_manager) + try: + pad._gtk_ready = True + pad._window = MagicMock() + pad._window.get_visible.return_value = False + pad._textview = MagicMock() + pad._buffer = MagicMock() + + pad.controller.append("segment ") + pad._apply_append("segment ", pad._generation) + + pad._window.show_all.assert_not_called() + self.assertFalse(pad.controller.enabled) + finally: + pad._window = None + pad.destroy() + + def test_copy_all_prefers_widget_text(self) -> None: + """Copy All copies the widget contents (dictation + in-pad edits).""" + pad = _pad_without_gtk() + try: + pad._gtk_ready = True + pad._Gtk = MagicMock() + pad._Gdk = MagicMock() + pad._GLib = MagicMock() + pad._copy_button = MagicMock() + pad._buffer = MagicMock() + pad._buffer.get_text.return_value = "dictated plus edits" + pad.controller.append("dictated") + + pad._on_copy_all_clicked() + + clipboard = pad._Gtk.Clipboard.get.return_value + clipboard.set_text.assert_called_once_with("dictated plus edits", -1) + finally: + pad.destroy() + + def test_show_pad_does_not_clobber_widget_edits(self) -> None: + """Re-showing the pad must not overwrite in-pad manual edits.""" + pad = _pad_without_gtk() + try: + pad._gtk_ready = True + pad._window = MagicMock() + pad._window.get_visible.return_value = True + pad._buffer = MagicMock() + pad._Gtk = MagicMock() + + pad.show_pad() + + # Never a wholesale set_text on (re)show: widget content is kept + # incrementally in sync by _apply_append/_apply_delete. + pad._buffer.set_text.assert_not_called() + finally: + pad._window = None + pad.destroy() + + def test_apply_delete_removes_tail_from_widget(self) -> None: + pad = _pad_without_gtk() + try: + pad._gtk_ready = True + pad._textview = MagicMock() + pad._buffer = MagicMock() + end_iter = MagicMock() + pad._buffer.get_end_iter.return_value = end_iter + + pad.controller.append("hello") + pad._apply_delete(5, pad._generation) + + pad._buffer.delete.assert_called_once() + start_iter = pad._buffer.delete.call_args[0][0] + start_iter.backward_chars.assert_called_once_with(5) + finally: + pad.destroy() + + def test_clear_drops_appends_still_queued_for_widget(self) -> None: + """A Clear between append_text and its idle callback must win.""" + pad = _pad_without_gtk(enabled=True) + try: + pad._gtk_ready = True + pad._GLib = MagicMock() + pad._buffer = MagicMock() + pad._textview = MagicMock() + pad._window = MagicMock() + pad._window.get_visible.return_value = True + + pad.append_text("stale segment ") + pad._on_clear_clicked() + self.assertEqual(pad.controller.text, "") + + # Run the queued idle callback — the stale generation skips it. + queued = pad._GLib.idle_add.call_args.args[0] + queued() + pad._buffer.insert.assert_not_called() + pad._buffer.set_text.assert_called_once_with("") + finally: + pad._window = None + pad.destroy() + + def test_queued_append_before_clear_still_applies(self) -> None: + """Appends queued before the bump run normally; only stale ones die.""" + pad = _pad_without_gtk(enabled=True) + try: + pad._gtk_ready = True + pad._GLib = MagicMock() + pad._buffer = MagicMock() + pad._textview = MagicMock() + pad._window = MagicMock() + pad._window.get_visible.return_value = True + + pad.append_text("kept ") + queued = pad._GLib.idle_add.call_args.args[0] + queued() + pad._buffer.insert.assert_called_once() + + pad.append_text("post clear ") + pad._on_clear_clicked() + stale = pad._GLib.idle_add.call_args.args[0] + stale() + self.assertEqual(pad._buffer.insert.call_count, 1) + finally: + pad._window = None + pad.destroy() + + def test_widget_edits_sync_back_to_controller(self) -> None: + """Manual edits in the text view update the controller's buffer.""" + pad = _pad_without_gtk() + try: + pad._buffer = MagicMock() + pad._buffer.get_text.return_value = "user edited text" + pad._on_buffer_changed(pad._buffer) + self.assertEqual(pad.controller.text, "user edited text") + # delete_last now computes against the edited content. + self.assertEqual(pad.delete_last_chars(4), 4) + self.assertEqual(pad.controller.text, "user edited ") + finally: + pad.destroy() + + def test_buffer_changed_skips_programmatic_sync(self) -> None: + """The pad's own widget writes must not feed back into the controller.""" + pad = _pad_without_gtk() + try: + pad.controller.append("dictated") + pad._syncing_widget = True + pad._on_buffer_changed(MagicMock()) + self.assertEqual(pad.controller.text, "dictated") + finally: + pad.destroy() + + def test_widget_edit_flushes_queued_ops_before_sync(self) -> None: + """A manual edit must not overwrite queued dictation in the controller. + + Input events outrank idle callbacks, so a keystroke can land before a + queued append. The changed handler replays the queue first, keeping + the view — and the controller sync — in delivery order. + """ + pad = _pad_without_gtk(enabled=True) + try: + pad._gtk_ready = True + pad._GLib = MagicMock() + pad._buffer = MagicMock() + pad._textview = MagicMock() + pad._window = MagicMock() + pad._window.get_visible.return_value = True + + pad.append_text("queued dictation ") + # The user edits before the idle callback runs; the queued op must + # be flushed into the widget before the view is read back. + pad._buffer.get_text.return_value = "user queued dictation " + pad._on_buffer_changed(pad._buffer) + + calls = [c[0] for c in pad._buffer.mock_calls] + self.assertLess(calls.index("insert"), calls.index("get_text")) + self.assertEqual(pad.controller.text, "user queued dictation ") + self.assertEqual(pad._pending_idle, []) + finally: + pad._window = None + pad.destroy() + + def test_flushed_ops_do_not_run_again_on_glib_dispatch(self) -> None: + """An op a flush already ran must no-op when GLib still dispatches it.""" + pad = _pad_without_gtk(enabled=True) + try: + pad._gtk_ready = True + pad._GLib = MagicMock() + pad._buffer = MagicMock() + pad._textview = MagicMock() + pad._window = MagicMock() + pad._window.get_visible.return_value = True + + pad.append_text("queued ") + queued = pad._GLib.idle_add.call_args.args[0] + pad._flush_idle_ops() + pad._buffer.insert.assert_called_once() + + # GLib still fires the callback later; without the pending-entry + # guard the insert would land a second time. + queued() + pad._buffer.insert.assert_called_once() + finally: + pad._window = None + pad.destroy() + + def test_queued_append_lands_before_user_edit(self) -> None: + """Dictation queued before a keystroke must precede it in the buffer.""" + pad = _pad_without_gtk(enabled=True) + try: + pad._gtk_ready = True + pad._GLib = MagicMock() + pad._buffer = MagicMock() + pad._textview = MagicMock() + pad._window = MagicMock() + pad._window.get_visible.return_value = True + + pad.append_text("dictated ") + # insert-text/delete-range fire before the edit is applied, so + # the queued append lands first and cannot leapfrog new text. + pad._on_buffer_user_edit() + pad._buffer.insert.assert_called_once() + self.assertEqual(pad._pending_idle, []) + finally: + pad._window = None + pad.destroy() + + def test_same_text_appends_do_not_cancel_each_other(self) -> None: + """A stale dispatch must not consume a later entry with the same text.""" + pad = _pad_without_gtk(enabled=True) + try: + pad._gtk_ready = True + pad._GLib = MagicMock() + pad._buffer = MagicMock() + pad._textview = MagicMock() + pad._window = MagicMock() + pad._window.get_visible.return_value = True + + pad.append_text("y") + pad._flush_idle_ops() + pad.append_text("z") + pad.append_text("y") + calls = pad._GLib.idle_add.call_args_list + calls[0].args[0]() # late dispatch of the already-flushed "y" + calls[1].args[0]() # "z" + calls[2].args[0]() # "y" + + inserted = [c.args[1] for c in pad._buffer.insert.call_args_list] + self.assertEqual(inserted, ["y", "z", "y"]) + finally: + pad._window = None + pad.destroy() + + def test_widget_edit_drops_stale_full_refresh(self) -> None: + """A queued full refresh must not erase a manual edit. + + After a voice undo the post-undo refresh waits in the idle queue; a + keystroke landing first makes that refresh stale — running it would + wipe the edit before it ever reaches the controller. + """ + pad = _pad_without_gtk(enabled=True) + try: + pad._gtk_ready = True + pad._GLib = MagicMock() + pad._buffer = MagicMock() + pad._textview = MagicMock() + pad._window = MagicMock() + pad._window.get_visible.return_value = True + + pad.append_text("one ") + pad.append_text("two ") + pad.handle_action("undo") + # The undo refresh is still queued when the user types. + pad._buffer.get_text.return_value = "one two edited" + pad._on_buffer_changed(pad._buffer) + + pad._buffer.set_text.assert_not_called() + self.assertEqual(pad.controller.text, "one two edited") + self.assertEqual(pad._pending_idle, []) + + # GLib still dispatches the dropped refresh later — it must no-op. + queued = pad._GLib.idle_add.call_args.args[0] + queued() + pad._buffer.set_text.assert_not_called() + finally: + pad._window = None + pad.destroy() + + def test_copy_all_flushes_queued_appends(self) -> None: + """Copy All sees segments still waiting in the idle queue.""" + pad = _pad_without_gtk() + try: + pad._gtk_ready = True + pad._Gtk = MagicMock() + pad._Gdk = MagicMock() + pad._GLib = MagicMock() + pad._copy_button = MagicMock() + pad._buffer = MagicMock() + pad._buffer.get_text.return_value = "whole pad " + + pad.append_text("queued ") + pad._on_copy_all_clicked() + + calls = [c[0] for c in pad._buffer.mock_calls] + self.assertLess(calls.index("insert"), calls.index("get_text")) + clipboard = pad._Gtk.Clipboard.get.return_value + clipboard.set_text.assert_called_once_with("whole pad ", -1) + finally: + pad.destroy() + + def test_last_segment_passthrough_to_controller(self) -> None: + pad = _pad_without_gtk() + try: + self.assertIsNone(pad.last_segment) + pad.append_text("dictated ") + self.assertEqual(pad.last_segment, "dictated ") + finally: + pad.destroy() + + def test_apply_append_stale_generation_is_dropped(self) -> None: + pad = _pad_without_gtk() + try: + pad._buffer = MagicMock() + pad._apply_append("old", pad._generation - 1) + pad._buffer.insert.assert_not_called() + finally: + pad.destroy() + + def test_handle_action_undo_redo_sync_widget(self) -> None: + pad = _pad_without_gtk() + try: + pad._GLib = MagicMock() + pad._buffer = MagicMock() + pad.controller.append("segment ") + + self.assertTrue(pad.handle_action("undo")) + self.assertEqual(pad.controller.text, "") + pad._GLib.idle_add.call_args.args[0]() + pad._buffer.set_text.assert_called_once_with("") + + self.assertTrue(pad.handle_action("redo")) + self.assertEqual(pad.controller.text, "segment ") + finally: + pad.destroy() + + def test_handle_action_with_empty_history_returns_false(self) -> None: + pad = _pad_without_gtk() + try: + self.assertFalse(pad.handle_action("undo")) + self.assertFalse(pad.handle_action("redo")) + finally: + pad.destroy() + + def test_handle_action_selection_commands(self) -> None: + pad = _pad_without_gtk() + try: + pad._GLib = MagicMock() + pad._buffer = MagicMock() + it = MagicMock() + it.get_offset.return_value = 0 + pad._buffer.get_iter_at_mark.return_value = it + pad._buffer.get_text.return_value = "para one\n\npara two" + + for action in ( + "select_all", + "select_line", + "select_word", + "select_paragraph", + ): + self.assertTrue(pad.handle_action(action)) + self.assertEqual(pad._GLib.idle_add.call_count, 4) + + # Run the queued select_all and select_line applies. + pad._GLib.idle_add.call_args_list[0].args[0]() + pad._buffer.select_range.assert_called_once_with( + pad._buffer.get_start_iter.return_value, + pad._buffer.get_end_iter.return_value, + ) + pad._GLib.idle_add.call_args_list[1].args[0]() + it.copy.return_value.set_line_offset.assert_called_once_with(0) + pad._GLib.idle_add.call_args_list[3].args[0]() + # "para one\n\npara two" — next blank line after offset 0 is at 8. + pad._buffer.get_iter_at_offset.assert_called_with(8) + finally: + pad.destroy() + + def test_handle_action_clipboard_commands(self) -> None: + pad = _pad_without_gtk() + try: + pad._GLib = MagicMock() + pad._buffer = MagicMock() + pad._Gtk = MagicMock() + pad._Gdk = MagicMock() + + for action in ("cut", "copy", "paste"): + self.assertTrue(pad.handle_action(action)) + pad._GLib.idle_add.call_args.args[0]() + + clipboard = pad._Gtk.Clipboard.get.return_value + pad._buffer.cut_clipboard.assert_called_once_with(clipboard, True) + pad._buffer.copy_clipboard.assert_called_once_with(clipboard) + pad._buffer.paste_clipboard.assert_called_once_with(clipboard, None, True) + finally: + pad.destroy() + + def test_handle_action_rejects_unknown_and_headless(self) -> None: + """Unknown actions and widget commands without a view are refused.""" + pad = _pad_without_gtk() + try: + self.assertFalse(pad.handle_action("select_all")) + self.assertFalse(pad.handle_action("nonexistent")) + finally: + pad.destroy() + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_language_shortcuts.py b/tests/test_language_shortcuts.py index b210682..0def881 100644 --- a/tests/test_language_shortcuts.py +++ b/tests/test_language_shortcuts.py @@ -574,7 +574,12 @@ def test_refresh_does_not_rearm_under_external_activation() -> None: def test_refresh_stops_a_held_push_to_talk_first() -> None: - """Rebuilding listeners cannot strand a session waiting on its release.""" + """Rebuilding listeners cannot strand a session waiting on its release. + + While a push-to-talk hold is live the rebuild is deferred instead of + stopping the session: the manager whose release callback will end the + hold is kept alive until the IDLE transition drains the refresh. + """ from vocalinux.ui.tray_indicator import TrayIndicator tray = _tray_stub( @@ -582,11 +587,17 @@ def test_refresh_stops_a_held_push_to_talk_first() -> None: entries=[{"shortcut": "alt+d", "language": "de"}], ) tray.speech_engine.state = RecognitionState.LISTENING + tray._language_shortcuts_refresh_pending = False + armed = MagicMock() + tray._language_shortcut_managers = [armed] with patch("vocalinux.ui.tray_indicator.KeyboardShortcutManager"): TrayIndicator.refresh_language_shortcuts(tray) - tray.speech_engine.stop_recognition.assert_called_once() + tray.speech_engine.stop_recognition.assert_not_called() + assert tray._language_shortcuts_refresh_pending is True + assert tray._language_shortcut_managers == [armed] # release callback survives + armed.stop.assert_not_called() def test_refresh_rebuilds_the_language_listeners() -> None: @@ -909,3 +920,87 @@ def test_begin_recording_disarms_the_previous_target(dialog_class: type[Any]) -> ) dialog._stop_recording_shortcut.assert_called_once() + + +def test_persist_rejects_an_edit_that_claims_another_rows_binding( + dialog_class: type[Any], +) -> None: + """Editing a row onto another row's key must not displace that row.""" + dialog = _dialog_stub() + first_picker, second_picker = MagicMock(), MagicMock() + first_entry, second_entry = MagicMock(), MagicMock() + first_picker.get_active_id.return_value = "de" + second_picker.get_active_id.return_value = "fr" + first_entry.get_text.return_value = "alt+f" # typed onto row two's key + second_entry.get_text.return_value = "alt+f" # row two still holds it + dialog._language_shortcut_rows = [ + { + "language_picker": first_picker, + "shortcut_entry": first_entry, + "last_valid_shortcut": "alt+d", + }, + { + "language_picker": second_picker, + "shortcut_entry": second_entry, + "last_valid_shortcut": "alt+f", + }, + ] + dialog._report_language_shortcut_rejections = ( + lambda rejected: dialog_class._report_language_shortcut_rejections(dialog, rejected) + ) + dialog.config_manager.get_language_shortcuts.return_value = [ + {"shortcut": "alt+d", "language": "de"}, + {"shortcut": "alt+f", "language": "fr"}, + ] + + dialog_class._persist_language_shortcuts(dialog) + + # The resolved rows match the saved bindings — the stealing edit lost — + # so nothing is rewritten and no listener rebuild is triggered. + dialog.config_manager.set_language_shortcuts.assert_not_called() + dialog.language_shortcuts_update_callback.assert_not_called() + dialog.language_shortcuts_info_label.set_markup.assert_called_once() + assert dialog._language_shortcut_rows[0]["last_valid_shortcut"] == "alt+d" + + +def test_refresh_defers_listener_rebuild_while_dictating() -> None: + """A binding edit during a session waits for IDLE, it must not stop it.""" + from vocalinux.ui.tray_indicator import TrayIndicator + + tray = _tray_stub(entries=[{"shortcut": "alt+d", "language": "de"}]) + tray.speech_engine.state = RecognitionState.LISTENING + tray._language_shortcuts_refresh_pending = False + + TrayIndicator.refresh_language_shortcuts(tray) + + assert tray._language_shortcuts_refresh_pending is True + tray.speech_engine.stop_recognition.assert_not_called() + assert tray._language_shortcut_managers == [] + + # The next IDLE transition drains the deferred rebuild. + tray.speech_engine.state = RecognitionState.IDLE + with patch("vocalinux.ui.tray_indicator.KeyboardShortcutManager") as manager_class: + manager = MagicMock() + manager_class.return_value = manager + TrayIndicator._update_ui(tray, RecognitionState.IDLE) + + assert tray._language_shortcuts_refresh_pending is False + manager_class.assert_called_once_with(shortcut="alt+d", mode="toggle") + manager.start.assert_called_once() + + +def test_refresh_rebuilds_immediately_when_idle() -> None: + """The normal path still rebuilds the listeners right away.""" + from vocalinux.ui.tray_indicator import TrayIndicator + + tray = _tray_stub(entries=[{"shortcut": "alt+d", "language": "de"}]) + tray._language_shortcuts_refresh_pending = False + + with patch("vocalinux.ui.tray_indicator.KeyboardShortcutManager") as manager_class: + manager = MagicMock() + manager_class.return_value = manager + TrayIndicator.refresh_language_shortcuts(tray) + + assert tray._language_shortcuts_refresh_pending is False + manager_class.assert_called_once_with(shortcut="alt+d", mode="toggle") + manager.start.assert_called_once() diff --git a/tests/test_main.py b/tests/test_main.py index e073bc0..bbdc5c2 100644 --- a/tests/test_main.py +++ b/tests/test_main.py @@ -8,6 +8,7 @@ import threading import time import unittest from contextlib import ExitStack +from types import SimpleNamespace from typing import Any, Callable, Dict, Optional, Tuple from unittest.mock import ANY, MagicMock, patch @@ -162,6 +163,7 @@ class TestMainModule(unittest.TestCase): mock_exit.assert_called_once_with(1) @patch("vocalinux.main.check_dependencies") + @patch("vocalinux.ui.dictation_pad.DictationPad") @patch("vocalinux.ui.action_handler.ActionHandler") @patch("vocalinux.speech_recognition.recognition_manager.SpeechRecognitionManager") @patch("vocalinux.text_injection.text_injector.TextInjector") @@ -178,6 +180,7 @@ class TestMainModule(unittest.TestCase): mock_text, mock_speech, mock_action_handler, + mock_dictation_pad, mock_check_deps, ): """Test that main initializes all the required components.""" @@ -192,6 +195,7 @@ class TestMainModule(unittest.TestCase): "general": {"first_run": False}, } mock_config_instance.get_model_size_for_engine.return_value = "medium" + mock_config_instance.is_dictate_to_pad_enabled.return_value = False mock_config_manager.return_value = mock_config_instance # Mock objects @@ -199,12 +203,14 @@ class TestMainModule(unittest.TestCase): mock_text_instance = MagicMock() mock_tray_instance = MagicMock() mock_action_instance = MagicMock() + mock_pad_instance = MagicMock() # Setup return values mock_speech.return_value = mock_speech_instance mock_text.return_value = mock_text_instance mock_tray.return_value = mock_tray_instance mock_action_handler.return_value = mock_action_instance + mock_dictation_pad.return_value = mock_pad_instance # Mock the arguments with patch("vocalinux.main.parse_arguments") as mock_parse: @@ -253,14 +259,21 @@ class TestMainModule(unittest.TestCase): speech_engine=mock_speech_instance, text_injector=mock_text_instance, transcription_history=ANY, + dictation_pad=mock_pad_instance, + ) + # The pad starts from the shared manager's live capture flag. + mock_dictation_pad.assert_called_once_with( + enabled=False, config_manager=mock_config_instance ) # 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 - ) + mock_speech_instance.register_action_callback.assert_called_once() + # The registered wrapper delegates non-pad actions to ActionHandler. + action_callback = mock_speech_instance.register_action_callback.call_args.args[0] + action_callback("undo") + mock_action_instance.handle_action.assert_called_once_with("undo") mock_speech_instance.register_state_callback.assert_called_once() # Verify the tray indicator was started @@ -289,6 +302,7 @@ class TestMainModule(unittest.TestCase): "speech_recognition": {}, "general": {"first_run": False}, } + mock_config_instance.is_dictate_to_pad_enabled.return_value = False mock_config_manager.return_value = mock_config_instance mock_speech_instance = MagicMock() @@ -352,6 +366,7 @@ class TestMainModule(unittest.TestCase): mock_config_instance.get.side_effect = lambda section, key, default=None: ( auto_capitalize if section == "text_injection" and key == "auto_capitalize" else default ) + mock_config_instance.is_dictate_to_pad_enabled.return_value = False mock_config_manager.return_value = mock_config_instance mock_speech_instance = MagicMock() @@ -1255,111 +1270,301 @@ class TestShouldAppendTrailingSpace(unittest.TestCase): self.assertTrue(_should_append_trailing_space()) +def _boot_main_callbacks( + *, + append_trailing_space: bool = True, + inject_ok: bool = True, + dictate_to_pad: bool = False, +) -> SimpleNamespace: + """Boot main() with mocked deps; return the registered callbacks and mocks. + + The patch stack stays open — the caller owns ``stack`` and must close it + in a finally block so no patch leaks into later tests. + """ + from contextlib import ExitStack + + 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() + mock_config.get_settings.return_value = { + "speech_recognition": {}, + "general": {"first_run": False}, + } + mock_config.get.return_value = False # auto_capitalize off + mock_config.is_dictate_to_pad_enabled.return_value = dictate_to_pad + 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 = inject_ok + mock_text_cls.return_value = mock_text + + mock_pad_cls = stack.enter_context(patch("vocalinux.ui.dictation_pad.DictationPad")) + mock_pad = mock_pad_cls.return_value + + 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")) + stack.enter_context( + patch( + "vocalinux.main._should_append_trailing_space", + return_value=append_trailing_space, + ) + ) + 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] + action_cb = mock_speech.register_action_callback.call_args.args[0] + state_cb = mock_speech.register_state_callback.call_args.args[0] + except BaseException: + # The caller only closes the stack once it gets one back, so a + # failure here would leak these patches into every later test. + stack.close() + raise + + return SimpleNamespace( + stack=stack, + text_cb=text_cb, + action_cb=action_cb, + state_cb=state_cb, + text_system=mock_text, + pad=mock_pad, + pad_cls=mock_pad_cls, + config=mock_config, + ) + + class TestMainCallbackTrailingSpaceEdges(unittest.TestCase): """Exercise trailing-space edge paths through the real main() callback.""" - def _boot_under_patches(self, *, append_trailing_space: bool = True, inject_ok: bool = True): - """Return (exit_stack, text_cb, mock_text) with patches still active.""" - from contextlib import ExitStack - - 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() - mock_config.get_settings.return_value = { - "speech_recognition": {}, - "general": {"first_run": False}, - } - 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 = inject_ok - 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")) - stack.enter_context( - patch( - "vocalinux.main._should_append_trailing_space", - return_value=append_trailing_space, - ) - ) - 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 + def test_whitespace_only_is_skipped_through_main(self) -> None: + boot = _boot_main_callbacks() try: - main() - text_cb = mock_speech.register_text_callback.call_args.args[0] - except BaseException: - # The caller only closes the stack once it gets one back, so a - # failure here would leak these patches into every later test. - stack.close() - raise - - return stack, text_cb, mock_text - - def test_whitespace_only_is_skipped_through_main(self): - stack, text_cb, mock_text = self._boot_under_patches() - try: - text_cb(" \t ") - mock_text.inject_text.assert_not_called() + boot.text_cb(" \t ") + boot.text_system.inject_text.assert_not_called() finally: - stack.close() + boot.stack.close() - def test_newline_segment_skips_trailing_space_through_main(self): - stack, text_cb, mock_text = self._boot_under_patches() + def test_newline_segment_skips_trailing_space_through_main(self) -> None: + boot = _boot_main_callbacks() try: - text_cb("Hello.\n") - mock_text.inject_text.assert_called_once_with("Hello.\n") + boot.text_cb("Hello.\n") + boot.text_system.inject_text.assert_called_once_with("Hello.\n") finally: - stack.close() + boot.stack.close() - def test_legacy_mode_adds_leading_space_in_session(self): - stack, text_cb, mock_text = self._boot_under_patches(append_trailing_space=False) + def test_legacy_mode_adds_leading_space_in_session(self) -> None: + boot = _boot_main_callbacks(append_trailing_space=False) try: - text_cb("Hello.") - text_cb("World") - calls = [c.args[0] for c in mock_text.inject_text.call_args_list] + boot.text_cb("Hello.") + boot.text_cb("World") + calls = [c.args[0] for c in boot.text_system.inject_text.call_args_list] self.assertEqual(calls, ["Hello.", " World"]) finally: - stack.close() + boot.stack.close() - def test_failed_inject_does_not_remember_text(self): - stack, text_cb, mock_text = self._boot_under_patches( - append_trailing_space=False, inject_ok=False - ) + def test_failed_inject_does_not_remember_text(self) -> None: + boot = _boot_main_callbacks(append_trailing_space=False, inject_ok=False) try: - text_cb("Hello.") - mock_text.inject_text.reset_mock() - mock_text.inject_text.return_value = True - text_cb("World") + boot.text_cb("Hello.") + boot.text_system.inject_text.reset_mock() + boot.text_system.inject_text.return_value = True + boot.text_cb("World") # Failure means last_injected stays empty; next segment has no leading space - mock_text.inject_text.assert_called_once_with("World") + boot.text_system.inject_text.assert_called_once_with("World") finally: - stack.close() + boot.stack.close() + + +class TestPadRoutingCallbacks(unittest.TestCase): + """Pad capture routing through the real main() callback wrappers (#726). + + The pad path is the feature's main routing: these exercise the callbacks + actually registered on the speech engine rather than re-implementations. + """ + + def test_capture_routes_transcription_to_pad_not_injector(self) -> None: + boot = _boot_main_callbacks(dictate_to_pad=True) + try: + boot.text_cb("hello") + boot.pad.append_text.assert_called_once_with("hello ") + boot.text_system.inject_text.assert_not_called() + finally: + boot.stack.close() + + def test_capture_routes_delete_that_to_pad(self) -> None: + boot = _boot_main_callbacks(dictate_to_pad=True) + try: + boot.pad.last_segment = "hello " + boot.text_cb("hello") + boot.action_cb("delete_last") + boot.pad.delete_last_chars.assert_called_once_with(len("hello ")) + boot.text_system.press_backspace.assert_not_called() + finally: + boot.stack.close() + + def test_capture_routes_editing_commands_to_pad(self) -> None: + """undo/select/cut/copy/paste must never reach the focused app.""" + boot = _boot_main_callbacks(dictate_to_pad=True) + try: + for action in ( + "undo", + "redo", + "select_all", + "select_line", + "select_word", + "select_paragraph", + "cut", + "copy", + "paste", + ): + boot.pad.handle_action.reset_mock() + self.assertTrue(boot.action_cb(action)) + boot.pad.handle_action.assert_called_once_with(action) + boot.text_system._inject_keyboard_shortcut.assert_not_called() + boot.text_system.press_backspace.assert_not_called() + finally: + boot.stack.close() + + def test_delete_that_follows_destination_across_mode_change(self) -> None: + """Deletion targets where the last segment went, not the live toggle.""" + boot = _boot_main_callbacks(dictate_to_pad=False) + try: + # Dictated into the app; capture switched on afterwards: the + # backspaces must still go to the app that received the text. + boot.text_cb("into app") + boot.config.is_dictate_to_pad_enabled.return_value = True + boot.action_cb("delete_last") + boot.text_system.press_backspace.assert_called_once_with(len("into app ")) + boot.pad.delete_last_chars.assert_not_called() + + # Dictated into the pad; capture switched off afterwards: the + # pad's copy is still the one removed. + boot.config.is_dictate_to_pad_enabled.return_value = True + boot.pad.last_segment = "into pad " + boot.pad.delete_last_chars.return_value = len("into pad ") + boot.text_cb("into pad") + boot.config.is_dictate_to_pad_enabled.return_value = False + boot.action_cb("delete_last") + boot.pad.delete_last_chars.assert_called_once_with(len("into pad ")) + self.assertEqual(boot.text_system.press_backspace.call_count, 1) + finally: + boot.stack.close() + + def test_capture_toggle_flips_routing_without_restart(self) -> None: + """Routing follows the shared manager's live value, not a snapshot.""" + boot = _boot_main_callbacks(dictate_to_pad=False) + try: + boot.text_cb("to app") + boot.text_system.inject_text.assert_called_once_with("to app ") + + boot.config.is_dictate_to_pad_enabled.return_value = True + boot.text_cb("to pad") + boot.pad.append_text.assert_called_once_with("to pad ") + self.assertEqual(boot.text_system.inject_text.call_count, 1) + finally: + boot.stack.close() + + def test_delete_that_with_empty_history_is_swallowed(self) -> None: + boot = _boot_main_callbacks(dictate_to_pad=True) + try: + self.assertTrue(boot.action_cb("delete_last")) + boot.pad.delete_last_chars.assert_not_called() + boot.text_system.press_backspace.assert_not_called() + finally: + boot.stack.close() + + def test_idle_state_resets_delete_that_target(self) -> None: + boot = _boot_main_callbacks(dictate_to_pad=True) + try: + boot.text_cb("pad text") + boot.state_cb(RecognitionState.IDLE) + boot.action_cb("delete_last") + boot.pad.delete_last_chars.assert_not_called() + boot.text_system.press_backspace.assert_not_called() + finally: + boot.stack.close() + + def test_pad_undo_retargets_delete_that(self) -> None: + """Undoing the last pad segment must not leave a stale delete length.""" + boot = _boot_main_callbacks(dictate_to_pad=True) + try: + boot.text_cb("hello") + boot.text_cb("world") + # The pad reports the surviving segment after its own undo pops + # "world "; the next delete targets it, not the stale segment. + boot.pad.last_segment = "hello " + boot.pad.handle_action.return_value = True + self.assertTrue(boot.action_cb("undo")) + boot.pad.handle_action.assert_called_once_with("undo") + + boot.action_cb("delete_last") + boot.pad.delete_last_chars.assert_called_once_with(len("hello ")) + boot.text_system.press_backspace.assert_not_called() + finally: + boot.stack.close() + + def test_pad_undo_to_empty_clears_delete_target(self) -> None: + """When pad undo removes the last segment, "delete that" is a no-op.""" + boot = _boot_main_callbacks(dictate_to_pad=True) + try: + boot.text_cb("only") + boot.pad.last_segment = None + boot.pad.handle_action.return_value = True + self.assertTrue(boot.action_cb("undo")) + + self.assertTrue(boot.action_cb("delete_last")) + boot.pad.delete_last_chars.assert_not_called() + boot.text_system.press_backspace.assert_not_called() + finally: + boot.stack.close() + + def test_pad_redo_restores_delete_target(self) -> None: + """A redo that brings the segment back must re-arm "delete that".""" + boot = _boot_main_callbacks(dictate_to_pad=True) + try: + boot.text_cb("only") + boot.pad.handle_action.return_value = True + + # Undo to an empty pad clears the target but keeps the pad as the + # last dictation destination. + boot.pad.last_segment = None + self.assertTrue(boot.action_cb("undo")) + self.assertTrue(boot.action_cb("delete_last")) + boot.pad.delete_last_chars.assert_not_called() + + # Redo restores the segment; the next delete reaches the pad. + boot.pad.last_segment = "only" + self.assertTrue(boot.action_cb("redo")) + self.assertTrue(boot.action_cb("delete_last")) + boot.pad.delete_last_chars.assert_called_once_with(len("only")) + boot.text_system.press_backspace.assert_not_called() + finally: + boot.stack.close() class TestSessionHistoryRecording(unittest.TestCase): @@ -1402,6 +1607,10 @@ class TestSessionHistoryRecording(unittest.TestCase): mock_speech = MagicMock() mock_speech.engine = "whisper_cpp" mock_speech_cls.return_value = mock_speech + # Stashed so tests can drive engine-side attributes (capture floors, + # the live recognition thread) through the same instance the + # callbacks close over. + self._engine = mock_speech mock_text_cls = stack.enter_context( patch("vocalinux.text_injection.text_injector.TextInjector") @@ -1613,6 +1822,64 @@ class TestSessionHistoryRecording(unittest.TestCase): finally: stack.close() + def test_straggler_extends_its_own_snippet_during_next_session(self) -> None: + """A late segment lands in its session's entry, not the newest one.""" + stack, _, segment_cb, state_cb, history = self._boot() + try: + state_cb(RecognitionState.LISTENING) + w1 = threading.Thread(target=segment_cb, args=("one", time.monotonic())) + self._engine.recognition_thread = w1 + w1.start() + w1.join() + state_cb(RecognitionState.IDLE) + + state_cb(RecognitionState.LISTENING) + # Session one's leftover worker finally delivers while session + # two is open — it must grow "one", never the pending "two". + straggler = threading.Thread(target=segment_cb, args=("tail", time.monotonic())) + straggler.start() + straggler.join() + # Session two's own segment arrives on its live worker. + w2 = threading.Thread(target=segment_cb, args=("two", time.monotonic())) + self._engine.recognition_thread = w2 + w2.start() + w2.join() + state_cb(RecognitionState.IDLE) + self.assertEqual(history.get_all(), ["two", "one tail"]) + finally: + stack.close() + + def test_pre_session_capture_is_not_attributed_to_open_session(self) -> None: + """Audio captured before a session opened cannot join its snippet.""" + stack, _, segment_cb, state_cb, history = self._boot() + try: + state_cb(RecognitionState.LISTENING) + session_start = time.monotonic() + segment_cb("new", time.monotonic()) + # A queued leftover captured before this session began, delivered + # on the session's worker, still belongs to what came before. + segment_cb("old", session_start - 10.0) + state_cb(RecognitionState.IDLE) + self.assertEqual(history.get_all(), ["new", "old"]) + finally: + stack.close() + + def test_mic_test_window_drops_segments_after_test_ends(self) -> None: + """Test speech still decoding after restore stays out of history.""" + stack, _, segment_cb, state_cb, history = self._boot() + try: + floor = time.monotonic() + self._engine.test_capture_floor = floor + self._engine.test_capture_ceiling = floor + 60.0 + + state_cb(RecognitionState.LISTENING) + segment_cb("test speech", floor + 5.0) + segment_cb("real dictation", floor + 120.0) + state_cb(RecognitionState.IDLE) + self.assertEqual(history.get_all(), ["real dictation"]) + finally: + stack.close() + if __name__ == "__main__": unittest.main() diff --git a/tests/test_settings_model_state.py b/tests/test_settings_model_state.py index 7e79286..6268bf2 100644 --- a/tests/test_settings_model_state.py +++ b/tests/test_settings_model_state.py @@ -213,7 +213,9 @@ def test_already_downloaded_auto_apply_runs_on_a_worker(settings_dialog, dialog_ ): dialog_class._auto_apply_settings(dialog) - dialog._apply_settings_internal.assert_called_once_with(settings, raise_errors=True) + dialog._apply_settings_internal.assert_called_once_with( + settings, raise_errors=True, apply_generation=settings_dialog._apply_settings_generation + ) dialog.speech_engine.reconfigure.assert_not_called() assert dialog._applying_settings is True _run_finish_idle(dialog, dialog_class, idle_calls) @@ -304,7 +306,11 @@ def test_second_pick_during_apply_resyncs_ui_when_the_worker_finishes( assert len(workers) == 1 workers[0].target() - dialog._apply_settings_internal.assert_called_once_with(first, raise_errors=True) + dialog._apply_settings_internal.assert_called_once_with( + first, + raise_errors=True, + apply_generation=settings_dialog._apply_settings_generation, + ) assert dialog._applying_settings is True _run_finish_idle(dialog, dialog_class, idle_calls) @@ -1385,13 +1391,73 @@ def test_deferred_text_edits_dropped_when_a_newer_apply_ran( """A newer apply owns the engine and config; the stashed snapshot stays stale.""" dialog = _dialog_stub() dialog._pending_text_edits = {"whispercpp_language_candidates": "en,es"} + dialog._pending_text_edit_baseline = {"whispercpp_language_candidates": "en"} dialog._pending_apply_generation = settings_dialog._apply_settings_generation - # A newer apply began after the edits were stashed. + # A newer apply began after the edits were stashed and rewrote the key. + settings_dialog._apply_settings_generation += 1 + settings_dialog._apply_settings_written["whispercpp_language_candidates"] = ( + settings_dialog._apply_settings_generation + ) + dialog.config_manager.get.side_effect = lambda section, key, default=None: "en" + + try: + with patch.object(settings_dialog.threading, "Thread", _InlineThread): + dialog_class._persist_pending_text_edits(dialog) + finally: + settings_dialog._apply_settings_written.pop("whispercpp_language_candidates", None) + + dialog.speech_engine.reconfigure.assert_not_called() + dialog.config_manager.set.assert_not_called() + dialog.config_manager.save_settings.assert_not_called() + + +def test_deferred_text_edits_dropped_when_newer_apply_wrote_baseline_value( + settings_dialog: Any, dialog_class: type[Any] +) -> None: + """A newer choice equal to the baseline is still a choice the edit must lose to.""" + dialog = _dialog_stub() + dialog._pending_text_edits = {"whispercpp_language_candidates": "en,es"} + dialog._pending_text_edit_baseline = {"whispercpp_language_candidates": "en"} + dialog._pending_apply_generation = settings_dialog._apply_settings_generation + # The reopened dialog applied the key back to its baseline value. The + # config is indistinguishable from untouched, yet the write counts as + # touched — the closed dialog's older edit must not overwrite it. + settings_dialog._apply_settings_generation += 1 + settings_dialog._apply_settings_written["whispercpp_language_candidates"] = ( + settings_dialog._apply_settings_generation + ) + dialog.config_manager.get.side_effect = lambda section, key, default=None: "en" + + try: + with patch.object(settings_dialog.threading, "Thread", _InlineThread): + dialog_class._persist_pending_text_edits(dialog) + finally: + settings_dialog._apply_settings_written.pop("whispercpp_language_candidates", None) + + dialog.speech_engine.reconfigure.assert_not_called() + dialog.config_manager.set.assert_not_called() + dialog.config_manager.save_settings.assert_not_called() + + +def test_superseded_apply_skips_the_stale_snapshot( + settings_dialog: Any, dialog_class: type[Any] +) -> None: + """A worker whose snapshot a newer apply predates must not write at all.""" + dialog = _dialog_stub() + dialog.speech_engine.state = RecognitionState.LISTENING + stale_generation = settings_dialog._apply_settings_generation + # A newer apply began after this snapshot was collected. settings_dialog._apply_settings_generation += 1 - with patch.object(settings_dialog.threading, "Thread", _InlineThread): - dialog_class._persist_pending_text_edits(dialog) + result = dialog_class._apply_settings_internal( + dialog, + {"whispercpp_language_candidates": "en,es"}, + apply_generation=stale_generation, + ) + assert result is True + # Dictation must keep running: the staleness check precedes the stop. + dialog.speech_engine.stop_recognition.assert_not_called() dialog.speech_engine.reconfigure.assert_not_called() dialog.config_manager.set.assert_not_called() dialog.config_manager.save_settings.assert_not_called() diff --git a/tests/test_transcription_history.py b/tests/test_transcription_history.py index 4c20919..d5a475c 100644 --- a/tests/test_transcription_history.py +++ b/tests/test_transcription_history.py @@ -111,6 +111,50 @@ class TestTranscriptionHistory(unittest.TestCase): history.clear() self.assertFalse(history.extend_latest("x")) + def test_add_returns_snippet_id(self) -> None: + history = TranscriptionHistory() + first = history.add("one") + second = history.add("two") + self.assertIsInstance(first, int) + self.assertIsInstance(second, int) + self.assertNotEqual(first, second) + + def test_add_returns_none_when_refused(self) -> None: + history = TranscriptionHistory(enabled=False) + self.assertIsNone(history.add("ignored")) + + def test_extend_entry_targets_the_named_snippet(self) -> None: + """A straggler extends its own session even after newer commits.""" + history = TranscriptionHistory() + first = history.add("one") + assert first is not None + history.add("two") + self.assertTrue(history.extend_entry(first, "tail")) + self.assertEqual(history.get_all(), ["two", "one tail"]) + + def test_extend_entry_unknown_id_returns_false(self) -> None: + history = TranscriptionHistory() + history.add("one") + self.assertFalse(history.extend_entry(999, "tail")) + self.assertEqual(history.get_all(), ["one"]) + + def test_extend_entry_after_clear_returns_false(self) -> None: + history = TranscriptionHistory() + first = history.add("one") + assert first is not None + history.clear() + self.assertFalse(history.extend_entry(first, "tail")) + + def test_extend_entry_refused_from_stale_epoch(self) -> None: + history = TranscriptionHistory() + first = history.add("one") + assert first is not None + epoch = history.epoch + history.clear() + history.add("two") + self.assertFalse(history.extend_entry(first, "tail", expected_epoch=epoch)) + self.assertEqual(history.get_all(), ["two"]) + def test_extend_latest_fires_change_callback(self) -> None: history = TranscriptionHistory() history.add("a")