From dca937fc2105d835b7e14e8bfb61fad67e3f530f Mon Sep 17 00:00:00 2001 From: Jatin K Malik Date: Wed, 30 Sep 2026 02:33:49 -0700 Subject: [PATCH] feat(dictionary): combined custom dictionary with corrections (#890) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * feat(dictionary): combined custom dictionary with corrections Rebased onto latest main (conflict-resolution squash for PR #818). * fix(dictionary): stabilize the terms and corrections file contracts - corrections_path() stays at the fixed config_dir() location promised by the file contract instead of following a custom or --dictionary-file terms path, which could strand or duplicate corrections; a leftover file beside the terms file is still read once and copied over. - _atomic_write() preserves the replaced file's permission mode instead of resetting it to the process umask. - build_initial_prompt() streams terms and stops at the configured cap instead of reading and processing the whole file for every segment. * fix(settings): restore the selected dictionary pane after a search Searching can auto-switch the Custom Dictionary management pane to reveal a match, but clearing the query only restored widget visibility. Record the pane selection in the search baseline snapshot and restore it alongside the previous page so users return to the pane they chose. * fix(dictionary): close the remaining custom dictionary file-contract gaps - A --dictionary-file session override no longer imports a corrections file found beside the override into the standard config directory; the override scopes to terms only. - Terms reads validate the whole file before yielding: corrupt UTF-8 mid-file now yields no terms at all, matching the invalid status Settings reports instead of leaking a partial dictionary to recognition. - The terms read is bounded by MAX_TERMS_FILE_BYTES (1 MiB); a larger file is invalid everywhere with a dedicated status instead of being read on every segment. * fix(dictionary): keep usable terms prefix on oversized files and legacy corrections under overrides * fix(dictionary): cap yielded terms and keep added terms visible on oversized files * fix(dictionary): keep added terms visible without dropping scanner-owned lines * fix(dictionary): keep oversized terms manageable and honor mid-search pane switches - terms list gains a Show all / Show first 500 toggle; adding a term beyond the cap auto-expands so the new row renders with a Remove button instead of being invisible until the file is hand-edited - a manual dictionary pane switch during an active search drops the saved pre-search pane so clearing the query no longer rolls back the user's newer selection - annotate the new settings tests and type model_info so the parakeet branch stops adding fresh mypy errors * fix(settings): do not auto-expand terms on add past display cap Rebuilding up to thousands of term rows on the UI thread stalls the dialog; the overflow button already shows the true count, so the added term stays one click away. * test(dictionary): stub segment_callbacks on bare manager for history dispatch * fix(dictionary): stream oversized terms files when building the prompt build_initial_prompt decoded the whole 1 MiB read window before every transcription segment even though it only needs the first max_words terms. _iter_terms_for_prompt now reads oversized files line by line and stops at the prompt limit, keeping the same contract: complete leading lines only, invalid UTF-8 inside the usable window yields nothing. * fix(history): only the ended session's worker may extend its snippet A late segment could extend the newest history entry regardless of which session produced it, so a worker finishing after the next dictation session leaked its text into that session's entry. Track the worker that owned the ended session and require late segments to arrive on the same thread before merging; anything else forms its own snippet. * fix(speech): transcribe queued segments the reload worker never drained The recognition worker can exit during stop_recognition's bounded audio-thread join, so the released key's final buffer lands in the queue after the worker is gone and was silently dropped when the reload teardown replaced the queue. Drain whatever remains before swapping it — except on capture failure or an explicit cancel, which still discard the session's leftovers. * fix(settings): build term rows in idle slices and pin hidden additions Two problems shared one cause. Rebuilding up to 4096 rows synchronously stalled the dialog, and a term appended past the 500-row cap reported success while staying invisible. Rows now build in bounded GLib idle slices guarded by a refresh token, and a saved term outside the visible window is pinned to the top of the capped list until it disappears or becomes visible on its own. Adding under "show all" scrolls to the new term instead. * test(engine): annotate new test method return types * fix(ci): match a candidate row that leads the latest track block snapcraft status normally prints candidate as a continuation line under the latest track (channel in column 1). If candidate ever leads the track block it rides the track line instead (channel in column 3), and the resolver missed it and refused to promote. Match both layouts in the resolve and the pre-release re-verify. * fix(dictionary): bound prompt reads and scroll to landed terms - _iter_terms_for_prompt now reads each line via readline() capped by the remaining byte window, so an overlong line can never pull more than MAX_TERMS_FILE_BYTES into memory - the prompt path keeps decoding every complete line inside the usable window after the term limit is collected, so invalid UTF-8 anywhere in the window still yields no terms and matches the reader's invalid status - under "show all" Settings scrolls to the index where an added term landed (prepended or appended), verified against the row's label, instead of always scrolling to the end * fix(history,reload): keep late-segment ownership and drain stamped tuples - Track session_workers_seen so a worker from an earlier session can never merge into a newer snippet while a session is open - Unpack (segment, language) queue tuples in _transcribe_queued_segments; passing the tuple as audio raised TypeError and lost final speech - Refuse 'delete that' when a manual pad edit blurred segment boundaries - Translate term-row coordinates into the scrolled card's space so the scroll actually reaches rows below the fold - Type _on_dictionary_pane_changed's varargs * style: drop unneeded nonlocal for mutated worker set --------- Co-authored-by: Graham --- .github/workflows/snap-promote.yml | 12 +- AGENTS.md | 1 + docs/CUSTOM_DICTIONARY_DESIGN.md | 125 ++ docs/USER_GUIDE.md | 33 + src/vocalinux/custom_dictionary.py | 734 +++++++++++ src/vocalinux/main.py | 82 +- .../engines/faster_whisper_engine.py | 10 +- .../speech_recognition/recognition_manager.py | 96 +- src/vocalinux/ui/config_manager.py | 6 + src/vocalinux/ui/settings_dialog.py | 630 +++++++++- tests/test_custom_dictionary.py | 1106 +++++++++++++++++ tests/test_faster_whisper_engine.py | 15 + tests/test_main.py | 77 +- tests/test_record_during_reload.py | 41 + tests/test_settings_dialog.py | 105 +- 15 files changed, 3048 insertions(+), 25 deletions(-) create mode 100644 docs/CUSTOM_DICTIONARY_DESIGN.md create mode 100644 src/vocalinux/custom_dictionary.py create mode 100644 tests/test_custom_dictionary.py diff --git a/.github/workflows/snap-promote.yml b/.github/workflows/snap-promote.yml index 9e0c52c..45b2eff 100644 --- a/.github/workflows/snap-promote.yml +++ b/.github/workflows/snap-promote.yml @@ -65,8 +65,10 @@ jobs: - name: Install snapcraft run: sudo snap install snapcraft --classic - # `snapcraft status` prints the channel map grouped by track and arch; - # the candidate row is a continuation line, so column 1 is the channel. + # `snapcraft status` prints the channel map grouped by track and arch. + # The candidate row is normally a continuation line (column 1 is the + # channel), but if candidate ever leads the latest track block it rides + # the track line itself — column 3 is the channel then — so match both. - name: Resolve the candidate revision id: candidate env: @@ -78,7 +80,8 @@ jobs: /snap/bin/snapcraft status vocalinux | tee /tmp/snap-status.txt # POSIX: mawk has no \S — use a literal non-space class. row=$(awk '/^[^ \t]/ { track = $1 } - track == "latest" && $1 == "candidate" { print $2, $3; exit }' \ + track == "latest" && $1 == "candidate" { print $2, $3; exit } + track == "latest" && $3 == "candidate" { print $4, $5; exit }' \ /tmp/snap-status.txt) read -r seen_version revision <<<"$row" echo "latest/candidate holds version=${seen_version:-none} revision=${revision:-none}" @@ -105,7 +108,8 @@ jobs: set -euo pipefail VERSION="${TAG#v}" row=$(awk '/^[^ \t]/ { track = $1 } - track == "latest" && $1 == "candidate" { print $2, $3; exit }' \ + track == "latest" && $1 == "candidate" { print $2, $3; exit } + track == "latest" && $3 == "candidate" { print $4, $5; exit }' \ <(/snap/bin/snapcraft status vocalinux)) read -r seen_version revision <<<"$row" if [ "${revision:-}" != "$REVISION" ] || [ "$seen_version" != "$VERSION" ]; then diff --git a/AGENTS.md b/AGENTS.md index c3fb3c6..b74b92a 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -144,6 +144,7 @@ src/vocalinux/ ├── auto_pause_monitor.py # unload model while configured apps run ├── model_keepalive.py # idle unload ├── suspend_handler.py # logind PrepareForSleep +├── custom_dictionary.py # terms bias + transcript correction file contracts ├── speech_recognition/ │ ├── recognition_manager.py # whisper.cpp / Whisper / Vosk / remote │ ├── command_processor.py # voice commands diff --git a/docs/CUSTOM_DICTIONARY_DESIGN.md b/docs/CUSTOM_DICTIONARY_DESIGN.md new file mode 100644 index 0000000..1a421b3 --- /dev/null +++ b/docs/CUSTOM_DICTIONARY_DESIGN.md @@ -0,0 +1,125 @@ +# Custom Dictionary Support Design + +## Scope + +Custom dictionary support has two separate jobs: + +1. **Custom terms** bias local Whisper, whisper.cpp, and Faster Whisper recognition + toward names and jargon. +2. **Transcript corrections** deterministically replace a known misheard phrase + after transcription. + +They appear together on the **Custom Dictionary** Settings page, but are not +interchangeable. A correction replacement does not implicitly become a prompt +term. That coupling would make recognition bias depend on correction order and +would make the external terms contract surprising. + +## File contracts + +Both files live in VocaLinux's XDG-aware configuration directory: +`$XDG_CONFIG_HOME/vocalinux` (or `~/.config/vocalinux` when unset). + +| File | Contract | Consumer | +| --- | --- | --- | +| `dictionary.txt` | UTF-8 (BOM accepted), one term per line. Blank lines and lines beginning with `#` are ignored. Terms are de-duplicated case-insensitively while retaining the first spelling. | VocaLinux and an external accessibility scanner. | +| `custom-dictionary-corrections.json` | UTF-8 JSON object: `{"version": 1, "corrections": [{"heard": "super base", "replacement": "Supabase"}]}`. | VocaLinux only. | + +The terms file remains deliberately simple and scanner-friendly. Corrections +need two unambiguous fields, a version, and exact replacement casing, so JSON is +used rather than overloading the line file with an ambiguous delimiter grammar. +The UI writes either file by atomic replacement; a failed UI save preserves the +old file and reports an error. + +The application reads both files for each completed transcript. An externally +edited file therefore takes effect on the next dictation segment without an app +restart. Missing, unreadable, malformed, or unsupported correction files are +ignored safely; dictation continues without their entries. + +The terms file may contain any number of lines, but VocaLinux limits a single +prompt to the first 200 valid terms and 2,000 characters. This bounds decoding +work while leaving the scanner's complete line-file contract intact. The Custom +Dictionary page keeps both an add/remove editor and a validated file chooser +for the terms file. + +## Processing and engine behavior + +For every completed audio segment, VocaLinux applies this order: + +1. The selected engine transcribes audio. +2. Live phrase corrections run against the raw transcript. +3. Voice-command interpretation runs on that corrected text. +4. Remaining text and actions are dispatched. + +Corrections run before commands so a correction can prevent a command-like +misrecognition from being acted on. This also means users must not create a +replacement that deliberately creates a voice command unless that is intended. + +Corrections match literal phrases case-insensitively, only where the adjacent +characters are not Unicode word characters (or common combining marks). Longer +phrases win over shorter overlaps; equal-length entries retain JSON order. +Replacement text is inserted exactly as written. Inputs are normalized to NFC +for matching, so composed/decomposed accents are handled consistently. + +| Engine | Custom terms | Corrections | +| --- | --- | --- | +| OpenAI Whisper | `initial_prompt` | Yes | +| whisper.cpp | Advanced initial prompt followed by custom terms; explicit empty prompt clears reused native state | Yes | +| Faster Whisper | `initial_prompt` | Yes | +| VOSK | No prompt-bias API | Yes | +| Parakeet | No prompt-bias API | Yes | +| Remote API | No prompt is added to current request formats | Yes | + +`--dictionary-file PATH` preserves the #767 session-only behavior: `PATH` is +used as the custom terms file, it enables terms for that session, and Settings +does not modify the saved terms file or enablement. Corrections remain loaded +from the standard JSON file. The previous experimental #768 +`text_injection.custom_dictionary` config list is read only as a fallback until +the JSON corrections file exists; it is never silently deleted or migrated. + +The #767 persisted configuration keys are retained unchanged: +`dictionary.enabled`, `dictionary.file_path` (default +`~/.config/vocalinux/dictionary.txt`), and `dictionary.max_words`. Existing +Peony settings therefore continue to select the scanner's `dictionary.txt` +file. New configured paths are accepted only when they can be expanded and are +either absent (so the UI can create them) or readable regular files; a failed +save leaves the prior setting in place. Invalid configured and `--dictionary-file` +paths are safely ignored and shown as unavailable rather than raising. + +## Alternatives rejected + +- **One mixed line file:** cannot express arbitrary phrases and replacements + without reserving/escaping delimiters, which weakens the scanner contract. +- **Corrections in `config.json`:** makes a scanner compete with application + settings writes and offers no durable external-file schema. +- **Post-command correction:** allows command-like transcription errors to be + acted on before they can be fixed. +- **Automatically adding correction replacements to prompts:** rejected because + replacements are not automatically included in the recognition prompt; + corrections work post-transcription on every engine. + +## Human test checklist + +1. Add `VocaLinux`, `PyGObject`, and a non-ASCII term in **Custom terms**; + confirm `dictionary.txt` is UTF-8, one line per term, and an external + edit changes the next Whisper, whisper.cpp, and Faster Whisper dictation segment. +2. Add `super base` → `Supabase`; test lowercase, uppercase, punctuation, an + overlapping short correction, `C++`, and non-ASCII text. Confirm replacement + casing is exact and embedded text (for example `supersize`) is not replaced. +3. With voice commands enabled, add a correction that changes a command-like + misrecognition into ordinary text. Confirm it is injected rather than acted + on. Also verify an intentionally command-like replacement is avoided. +4. Edit each file externally while VocaLinux is running. Test an invalid JSON + correction file and invalid UTF-8 terms file; confirm dictation continues and + the Settings status/save feedback is clear. +5. Test the UI with keyboard only: Add buttons via Enter, empty validation, + update/remove feedback, narrow window, long entries, and high-contrast theme. +6. Test all engines: Whisper, whisper.cpp, and Faster Whisper should use terms and + corrections; VOSK, Parakeet, and remote API should use corrections only, with + the VOSK terms warning. +7. Start with `--dictionary-file /path/to/terms.txt`; confirm the session uses + that file, Settings disables terms editing, saved settings remain unchanged, + and corrections still apply. +8. Start with an unresolved path such as `--dictionary-file ~missing-user/terms.txt` + and configure an unreadable path through the file chooser. Confirm Settings + reports the path safely, dictation continues without terms, and the prior + saved `dictionary.file_path` remains intact. diff --git a/docs/USER_GUIDE.md b/docs/USER_GUIDE.md index fed1f1f..5654121 100644 --- a/docs/USER_GUIDE.md +++ b/docs/USER_GUIDE.md @@ -212,6 +212,39 @@ vocalinux --start-minimized vocalinux --transcribe-file meeting.wav # diarized transcript on stdout, then exits ``` +## Custom Dictionary Support + +Open **Settings → Custom Dictionary** to configure two separate capabilities: + +- **Custom terms** bias recognition toward product names, people, and jargon. + They are read from `~/.config/vocalinux/dictionary.txt` by default, as UTF-8 + with one term per line. Blank lines and `#` comments are allowed, so the same + file remains friendly to an accessibility scanner. Use the add/remove editor + or choose a different readable terms file with the file picker. +- **Transcript corrections** replace a known misheard word or phrase after + transcription, for example `super base` with `Supabase`. Corrections are + stored separately in `~/.config/vocalinux/custom-dictionary-corrections.json`. + +VocaLinux re-reads both files before each completed dictation segment, so an +external edit applies to the next segment without restarting the app. Vocabulary +bias works with Whisper, whisper.cpp, and Faster Whisper. Corrections work with +every engine, including VOSK, Parakeet, and the configured remote API. + +Corrections run before voice-command interpretation. This can prevent a +command-like misrecognition from acting, but avoid replacements that create a +voice command unless that is intentional. Correction replacements are not +automatically added to recognition bias. + +For a one-session terms override, start Vocalinux with: + +```bash +vocalinux --dictionary-file /path/to/dictionary.txt +``` + +This temporarily enables terms from that file without changing saved settings; +the Custom terms controls are disabled for the session. It does not disable +transcript corrections. + `--transcribe-file` uses the TinyDiarize model (`small.en-tdrz`) and needs it downloaded first — the tray "Transcribe Audio File…" entry fetches it on demand. ## Troubleshooting diff --git a/src/vocalinux/custom_dictionary.py b/src/vocalinux/custom_dictionary.py new file mode 100644 index 0000000..199ac41 --- /dev/null +++ b/src/vocalinux/custom_dictionary.py @@ -0,0 +1,734 @@ +"""File-backed custom dictionary support for recognition bias and transcript fixes.""" + +import itertools +import json +import logging +import os +import re +import shutil +import unicodedata +import uuid +from pathlib import Path +from typing import TYPE_CHECKING, Any, Iterator, Optional + +from .utils.paths import config_dir + +if TYPE_CHECKING: + from .ui.config_manager import ConfigManager + +logger = logging.getLogger(__name__) + +TERMS_FILENAME = "dictionary.txt" +DEFAULT_TERMS_PATH = str(Path(config_dir()) / TERMS_FILENAME) +# Pre-XDG default persisted by older builds; keep as a configured path. +LEGACY_DEFAULT_TERMS_PATH = "~/.config/vocalinux/dictionary.txt" +CORRECTIONS_FILENAME = "custom-dictionary-corrections.json" +CORRECTIONS_VERSION = 1 +CORRECTIONS_TOP_LEVEL_KEYS = frozenset({"version", "corrections"}) +CORRECTIONS_ENTRY_KEYS = frozenset({"heard", "replacement"}) +DEFAULT_MAX_TERMS = 200 +MAX_TERM_CHARACTERS = 200 +MAX_PROMPT_CHARACTERS = 2_000 +MAX_CORRECTIONS = 500 +MAX_CORRECTION_CHARACTERS = 500 +# A terms file beyond this bound is invalid everywhere rather than read in +# part: the read itself stays bounded and no partial dictionary can reach +# recognition while Settings reports the file unusable. +MAX_TERMS_FILE_BYTES = 1_048_576 +# Even within the byte bound a scanner-managed file can hold ~130k terms; the +# dictionary is a prompt vocabulary, so yields stop here — Settings would +# otherwise build a row per term. +MAX_TERMS_YIELDED = 4_096 + + +class _DuplicateJsonKeyError(ValueError): + """Raised when a JSON object contains the same key more than once.""" + + +def _object_pairs_without_duplicates(pairs: list[tuple[Any, Any]]) -> dict[Any, Any]: + """Return a dict, raising if any key repeats at this object level.""" + payload: dict[Any, Any] = {} + for key, value in pairs: + if key in payload: + raise _DuplicateJsonKeyError(key) + payload[key] = value + return payload + + +def normalize_corrections(raw_entries: Any) -> list[dict[str, str]]: + """Validate correction entries, retaining the first case-insensitive source. + + Input order is meaningful for equal-length overlapping entries. Keeping the + first duplicate makes hand-edited JSON deterministic and agrees with the + order used when building the matching expression. + """ + if not isinstance(raw_entries, list): + return [] + + entries: list[dict[str, str]] = [] + seen: set[str] = set() + for index, entry in enumerate(raw_entries): + if len(entries) >= MAX_CORRECTIONS: + logger.warning("Ignoring corrections after the supported limit of %d", MAX_CORRECTIONS) + break + if not isinstance(entry, dict): + logger.warning("Ignoring correction %d: expected an object", index) + continue + heard = entry.get("heard") + replacement = entry.get("replacement") + if not isinstance(heard, str) or not isinstance(replacement, str): + logger.warning("Ignoring correction %d: both fields must be strings", index) + continue + heard = unicodedata.normalize("NFC", heard.strip()) + replacement = replacement.strip() + if not heard or not replacement: + logger.warning("Ignoring correction %d: both fields must be non-empty", index) + continue + if len(heard) > MAX_CORRECTION_CHARACTERS or len(replacement) > MAX_CORRECTION_CHARACTERS: + logger.warning("Ignoring correction %d: phrase is too long", index) + continue + normalized_heard = heard.casefold() + if normalized_heard in seen: + logger.warning("Ignoring duplicate correction source %r", heard) + continue + seen.add(normalized_heard) + entries.append({"heard": heard, "replacement": replacement}) + return entries + + +def apply_corrections(text: str, entries: list[dict[str, str]]) -> str: + """Replace configured phrases in *text* using the custom-dictionary policy. + + Sources are matched case-insensitively only when neither adjacent character + is a Unicode word character. Replacements retain the exact user-provided + spelling. Longer source phrases take precedence; ties retain JSON order. + """ + if not text or not entries: + return text + + normalized_text = unicodedata.normalize("NFC", text) + ordered = sorted( + entries, + key=lambda entry: (len(entry["heard"].split()), len(entry["heard"])), + reverse=True, + ) + alternatives = "|".join( + f"(?P{re.escape(entry['heard'])})" + for index, entry in enumerate(ordered) + ) + pattern = re.compile( + r"(? str: + """Return the replacement associated with the matched alternative.""" + for index, entry in enumerate(ordered): + if match.group(f"correction_{index}") is not None: + return entry["replacement"] + return match.group(0) + + corrected = pattern.sub(replace, normalized_text) + if corrected != text: + logger.debug("Applied custom dictionary corrections to final transcript") + return corrected + + +class CustomDictionaryManager: + """Read, write, and live-reload the two custom dictionary file contracts.""" + + def __init__( + self, config_manager: "ConfigManager", transient_terms_path: Optional[str] = None + ) -> None: + self.config = config_manager + self._transient_terms_path = transient_terms_path + + @property + def is_transient_terms(self) -> bool: + """Return whether a CLI session-only terms file supersedes the saved file.""" + return self._transient_terms_path is not None + + def terms_enabled(self) -> bool: + """Return whether custom terms should be offered to Whisper-family engines.""" + return self.is_transient_terms or bool(self.config.get("dictionary", "enabled", False)) + + def set_terms_enabled(self, enabled: bool) -> bool: + """Persist terms enablement, rolling back the in-memory setting on failure.""" + if self.is_transient_terms: + logger.info("Ignoring saved terms enablement while a CLI override is active") + return False + old_value = self.config.get("dictionary", "enabled", False) + if not self.config.set("dictionary", "enabled", bool(enabled)): + return False + if self.config.save_config(): + return True + self.config.set("dictionary", "enabled", old_value) + logger.warning("Could not save custom terms enablement; keeping previous setting") + return False + + def terms_path_text(self) -> str: + """Return the configured or session-only terms path without expansion. + + A missing or blank saved path uses the live XDG default. Historical + leftover pre-XDG default strings are kept, not rewritten to XDG. + """ + if self._transient_terms_path is not None: + return self._transient_terms_path + default_path = str(Path(config_dir()) / TERMS_FILENAME) + configured = self.config.get("dictionary", "file_path", default_path) + if not isinstance(configured, str) or not configured.strip(): + return default_path + configured = configured.strip() + if ( + self._is_legacy_default_terms_path(configured, default_path) + and not self._file_path_is_explicit() + ): + self._stamp_legacy_default_terms_path_explicit() + return configured + + def _file_path_is_explicit(self) -> bool: + """Return whether Settings persisted the current terms path on purpose.""" + return bool(self.config.get("dictionary", "file_path_explicit", False)) + + @staticmethod + def _is_legacy_default_terms_path(configured: str, default_path: str) -> bool: + """Return whether a persisted path matches the pre-XDG default string.""" + if configured == LEGACY_DEFAULT_TERMS_PATH: + return True + try: + expanded_legacy = str(Path.home() / ".config" / "vocalinux" / TERMS_FILENAME) + except RuntimeError: + return False + return configured == expanded_legacy and expanded_legacy != default_path + + def _stamp_legacy_default_terms_path_explicit(self) -> None: + """Best-effort persist that a leftover pre-XDG path is an explicit choice. + + Historical configs cannot distinguish leftover defaults from Settings + picks. Keep the stored path and mark it explicit so later reads stay + stable. Failures leave both the path and the marker unchanged. + """ + if self.is_transient_terms: + return + old_explicit = self.config.get("dictionary", "file_path_explicit", False) + if not self.config.set("dictionary", "file_path_explicit", True): + return + if self.config.save_config(): + return + self.config.set("dictionary", "file_path_explicit", old_explicit) + logger.warning( + "Could not persist custom terms path explicit marker; keeping the configured path" + ) + + def terms_path(self) -> Optional[Path]: + """Return the active expanded terms path, or None when it is invalid.""" + configured = self.terms_path_text() + try: + return Path(configured).expanduser() + except RuntimeError as error: + logger.warning("Could not expand custom terms path %r: %s", configured, error) + return None + + def _saved_terms_path(self) -> Optional[Path]: + """Return the persisted terms path, ignoring any CLI override.""" + default_path = str(Path(config_dir()) / TERMS_FILENAME) + configured = self.config.get("dictionary", "file_path", default_path) + if not isinstance(configured, str) or not configured.strip(): + configured = default_path + try: + return Path(configured.strip()).expanduser() + except RuntimeError as error: + logger.warning("Could not expand saved terms path %r: %s", configured, error) + return None + + def set_terms_path(self, path: str) -> bool: + """Persist a usable terms path, retaining the prior setting on save failure.""" + if self.is_transient_terms: + logger.info("Ignoring terms path change while a CLI override is active") + return False + if not isinstance(path, str) or not path.strip(): + logger.warning("Ignoring empty custom terms path") + return False + configured = path.strip() + try: + candidate = Path(configured).expanduser() + if candidate.exists() and (not candidate.is_file() or not self._is_readable(candidate)): + logger.warning("Ignoring unusable custom terms path %s", candidate) + return False + except (OSError, RuntimeError) as error: + logger.warning("Ignoring invalid custom terms path %r: %s", configured, error) + return False + + old_value = self.config.get( + "dictionary", "file_path", str(Path(config_dir()) / TERMS_FILENAME) + ) + old_explicit = self.config.get("dictionary", "file_path_explicit", False) + if not self.config.set("dictionary", "file_path", configured): + return False + if not self.config.set("dictionary", "file_path_explicit", True): + self.config.set("dictionary", "file_path", old_value) + return False + if self.config.save_config(): + return True + self.config.set("dictionary", "file_path", old_value) + self.config.set("dictionary", "file_path_explicit", old_explicit) + logger.warning("Could not save custom terms path; keeping previous setting") + return False + + def corrections_path(self) -> Path: + """Return the fixed structured corrections path. + + Corrections always live under config_dir(), the location promised by + the file contract, so a custom or session-only terms path cannot + strand or duplicate them. + """ + return Path(config_dir()) / CORRECTIONS_FILENAME + + def get_terms(self) -> list[str]: + """Read the current UTF-8 line file; external edits apply next segment.""" + path = self.terms_path() + if path is None: + return [] + return list(self._iter_terms(path)) + + @staticmethod + def _iter_terms(path: Path) -> Iterator[str]: + """Yield normalized, de-duplicated terms from a bounded read. + + The bounded read validates the whole file when it fits the limit, so + a partly invalid file yields nothing and recognition never consumes + terms a status check would call invalid. Files past the limit still + supply their complete leading lines — a scanner-managed file can grow + past the bound without losing its usable prefix. Iterating over the + decoded lines still lets callers stop processing early. + """ + try: + with path.open("rb") as terms_file: + raw = terms_file.read(MAX_TERMS_FILE_BYTES + 1) + except FileNotFoundError: + return + except OSError as error: + logger.warning("Could not read custom terms file: %s", error) + return + if len(raw) > MAX_TERMS_FILE_BYTES: + logger.warning( + "Custom terms file exceeds %d bytes; reading complete leading lines only", + MAX_TERMS_FILE_BYTES, + ) + # Keep the valid leading lines: a scanner-managed file can grow + # past the bound, but its usable prefix still supplies terms. A + # prefix with no complete line supplies nothing. + last_newline = raw.rfind(b"\n") + if last_newline == -1: + return + raw = raw[:last_newline] + try: + text = raw.decode("utf-8-sig") + except UnicodeError as error: + logger.warning("Could not read custom terms file: %s", error) + return + seen: set[str] = set() + for line in text.splitlines(): + term = unicodedata.normalize("NFC", line.strip()) + normalized_term = term.casefold() + if not term or term.startswith("#") or normalized_term in seen: + continue + seen.add(normalized_term) + yield term + if len(seen) >= MAX_TERMS_YIELDED: + logger.warning( + "Custom terms file supplies more than %d terms; trailing entries ignored", + MAX_TERMS_YIELDED, + ) + return + + def _iter_terms_for_prompt(self, path: Path, limit: int) -> Iterator[str]: + """Yield up to *limit* terms, streaming oversized files line by line. + + Files inside the read bound keep the same whole-file-validated + iteration as ``_iter_terms``. Past it — where the usable-prefix + policy already accepts complete leading lines — the prompt only + collects the first *limit* terms, but every complete line in the + bounded window is still decoded so an invalid file supplies no + terms at all. Each read is capped by the remaining window, so an + overlong line can never pull more than the byte bound into memory. + """ + if limit <= 0: + return + try: + oversized = path.stat().st_size > MAX_TERMS_FILE_BYTES + except OSError: + oversized = False + if not oversized: + yield from itertools.islice(self._iter_terms(path), limit) + return + terms: list[str] = [] + seen: set[str] = set() + consumed = 0 + try: + with path.open("rb") as terms_file: + index = 0 + while True: + raw_line = terms_file.readline(MAX_TERMS_FILE_BYTES + 2 - consumed) + consumed += len(raw_line) + if ( + not raw_line + or consumed > MAX_TERMS_FILE_BYTES + 1 + or not raw_line.endswith(b"\n") + ): + # Lines complete inside the byte bound are usable; a + # line straddling it is not a complete leading line. + break + try: + line = raw_line.decode("utf-8-sig" if index == 0 else "utf-8") + except UnicodeError as error: + logger.warning("Could not read custom terms file: %s", error) + return + index += 1 + # Collection stops at the prompt limit, but every + # complete leading line is still decoded: the prompt + # fails closed on the same invalid files the terms + # reader rejects. + if len(terms) >= limit: + continue + term = unicodedata.normalize("NFC", line.strip()) + normalized_term = term.casefold() + if not term or term.startswith("#") or normalized_term in seen: + continue + seen.add(normalized_term) + terms.append(term) + except FileNotFoundError: + return + except OSError as error: + logger.warning("Could not read custom terms file: %s", error) + return + yield from terms + + def save_terms(self, terms: list[str]) -> bool: + """Safely replace the standard terms file with normalized line entries.""" + if self.is_transient_terms: + logger.info("Ignoring terms edit while a CLI override is active") + return False + path = self.terms_path() + if path is None: + logger.warning("Cannot save custom terms because the configured path is invalid") + return False + normalized_terms = self._normalize_terms(terms) + contents = "\n".join(normalized_terms) + if contents: + contents += "\n" + return self._atomic_write(path, contents) + + def add_term(self, term: str) -> bool: + """Add one valid term without dropping or rewriting unrelated lines.""" + if self.is_transient_terms: + logger.info("Ignoring terms edit while a CLI override is active") + return False + cleaned = self._single_term(term) + if cleaned is None: + return False + path = self.terms_path() + if path is None: + logger.warning("Cannot add custom term because the configured path is invalid") + return False + contents = self._read_terms_contents(path, missing_value="") + if contents is None: + return False + existing_terms = self.get_terms() + if any(existing.casefold() == cleaned.casefold() for existing in existing_terms): + logger.warning("Ignoring duplicate custom term %r", cleaned) + return False + if ( + len(contents.encode("utf-8")) > MAX_TERMS_FILE_BYTES + or len(existing_terms) >= MAX_TERMS_YIELDED + ): + # A tail append would land past the read window where the new term + # stays invisible, while rewriting from a capped read would drop + # scanner-owned lines. Prepending keeps the addition readable and + # leaves every existing line, comment, and blank line intact. + return self._atomic_write(path, f"{cleaned}\n{contents}") + separator = "" if not contents or contents.endswith(("\n", "\r")) else "\n" + return self._atomic_write(path, f"{contents}{separator}{cleaned}\n") + + def remove_term(self, term: str) -> bool: + """Remove matching term lines without changing unrelated file content.""" + if self.is_transient_terms: + logger.info("Ignoring terms edit while a CLI override is active") + return False + cleaned = self._single_term(term) + if cleaned is None: + return False + path = self.terms_path() + if path is None: + logger.warning("Cannot remove custom term because the configured path is invalid") + return False + contents = self._read_terms_contents(path) + if contents is None: + return False + + remaining_lines: list[str] = [] + removed = False + for line in contents.splitlines(keepends=True): + line_term = unicodedata.normalize("NFC", line.rstrip("\r\n").strip()) + if not line_term.startswith("#") and line_term.casefold() == cleaned.casefold(): + removed = True + continue + remaining_lines.append(line) + if not removed: + logger.warning("Custom term %r was not present in the terms file", cleaned) + return False + return self._atomic_write(path, "".join(remaining_lines)) + + def build_initial_prompt(self) -> Optional[str]: + """Build the current Whisper prompt from enabled terms, if any.""" + if not self.terms_enabled(): + return None + max_terms = self.config.get("dictionary", "max_words", DEFAULT_MAX_TERMS) + try: + max_terms = max(0, int(max_terms)) + except (TypeError, ValueError): + logger.warning("Invalid custom terms limit %r; using %d", max_terms, DEFAULT_MAX_TERMS) + max_terms = DEFAULT_MAX_TERMS + path = self.terms_path() + if path is None: + return None + prompt_terms: list[str] = [] + prompt_characters = 0 + for term in self._iter_terms_for_prompt(path, max_terms): + additional_characters = len(term) + (1 if prompt_terms else 0) + if prompt_characters + additional_characters > MAX_PROMPT_CHARACTERS: + logger.warning("Custom terms prompt reached %d characters", MAX_PROMPT_CHARACTERS) + break + prompt_terms.append(term) + prompt_characters += additional_characters + return " ".join(prompt_terms) or None + + def get_corrections(self) -> list[dict[str, str]]: + """Read corrections from JSON for every segment, failing closed on errors.""" + entries = self._read_corrections(for_edit=False) + return entries if entries is not None else [] + + def get_corrections_for_edit(self) -> Optional[list[dict[str, str]]]: + """Return losslessly editable entries, or None for an unsafe source file. + + Runtime correction reads may safely ignore malformed individual entries, + but a UI edit must never rewrite the file from that filtered view. + """ + return self._read_corrections(for_edit=True) + + def _read_corrections_text(self, path: Path) -> Optional[str]: + """Return corrections JSON text, or None when no file exists. + + If *path* is missing, copy once from beside the terms file when a + leftover file from the earlier co-located contract exists there so + those entries survive the move to the fixed location. Copy failure + still returns the old text so entries are not lost. + """ + try: + return path.read_text(encoding="utf-8") + except FileNotFoundError: + pass + # Legacy corrections lived beside the persisted terms file. Resolve + # that path from the saved setting — never the CLI override — so an + # override's sibling is not adopted while a saved file's corrections + # still apply. + terms = self._saved_terms_path() + old_path = terms.parent / CORRECTIONS_FILENAME if terms is not None else None + if old_path is None or old_path == path or not old_path.is_file(): + return None + try: + contents = old_path.read_text(encoding="utf-8") + except FileNotFoundError: + return None + self._atomic_write(path, contents) + return contents + + def _read_corrections(self, *, for_edit: bool) -> Optional[list[dict[str, str]]]: + """Read corrections with stricter validation for write-back workflows.""" + path = self.corrections_path() + try: + contents = self._read_corrections_text(path) + if contents is None: + return self._legacy_corrections() + payload = json.loads( + contents, + object_pairs_hook=_object_pairs_without_duplicates if for_edit else None, + ) + except _DuplicateJsonKeyError: + logger.warning( + "Refusing to edit custom corrections because the source has duplicate JSON keys" + ) + return None + except (OSError, UnicodeError, json.JSONDecodeError) as error: + logger.warning("Could not read custom corrections file %s: %s", path, error) + return None if for_edit else [] + + version = payload.get("version") if isinstance(payload, dict) else None + if ( + not isinstance(payload, dict) + or not isinstance(version, int) + or isinstance(version, bool) + or version != CORRECTIONS_VERSION + ): + logger.warning("Ignoring custom corrections file with an unsupported schema") + return None if for_edit else [] + + raw_entries = payload.get("corrections") + entries = normalize_corrections(raw_entries) + if for_edit: + if set(payload) - CORRECTIONS_TOP_LEVEL_KEYS: + logger.warning( + "Refusing to edit custom corrections because the source has extra top-level fields" + ) + return None + if not isinstance(raw_entries, list) or len(entries) != len(raw_entries): + logger.warning( + "Refusing to edit custom corrections because some source entries are invalid" + ) + return None + if any( + not isinstance(entry, dict) or set(entry) - CORRECTIONS_ENTRY_KEYS + for entry in raw_entries + ): + logger.warning( + "Refusing to edit custom corrections because some source entries have extra fields" + ) + return None + if any( + entry.get("heard") != normalized["heard"] + or entry.get("replacement") != normalized["replacement"] + for entry, normalized in zip(raw_entries, entries) + ): + logger.warning( + "Refusing to edit custom corrections because normalization would change values" + ) + return None + return entries + + def save_corrections(self, entries: list[dict[str, str]]) -> bool: + """Safely write validated corrections in the versioned JSON contract.""" + payload = { + "version": CORRECTIONS_VERSION, + "corrections": normalize_corrections(entries), + } + return self._atomic_write( + self.corrections_path(), json.dumps(payload, ensure_ascii=False, indent=2) + "\n" + ) + + def apply_corrections(self, text: str) -> str: + """Live-reload and apply corrections to a completed transcript.""" + return apply_corrections(text, self.get_corrections()) + + def terms_status(self) -> str: + """Return a concise, user-facing description of the custom terms file.""" + path = self.terms_path() + if path is None: + return "Configured terms path is invalid or cannot be expanded." + try: + if not path.exists(): + return "Terms file does not exist yet; add a term to create it." + if not path.is_file(): + return "Terms path is not a regular file." + if path.stat().st_size > MAX_TERMS_FILE_BYTES: + count = len(self.get_terms()) + return f"{count} term(s) available from the file's leading lines only." + if len(list(self._iter_terms(path))) >= MAX_TERMS_YIELDED: + return f"Only the first {MAX_TERMS_YIELDED} terms are used." + path.read_text(encoding="utf-8-sig") + except UnicodeError: + return "Terms file is not valid UTF-8." + except OSError: + return "Terms file cannot be inspected." + return f"{len(self.get_terms())} term(s) available from the live file." + + @staticmethod + def _is_readable(path: Path) -> bool: + """Return whether a path can be opened for reading.""" + try: + with path.open("r", encoding="utf-8"): + return True + except OSError: + return False + + @staticmethod + def _normalize_terms(terms: list[str]) -> list[str]: + """Return trimmed, de-duplicated terms suitable for the line-file contract.""" + normalized: list[str] = [] + seen: set[str] = set() + for term in terms: + if not isinstance(term, str): + continue + cleaned = unicodedata.normalize("NFC", term.strip()) + if len(cleaned) > MAX_TERM_CHARACTERS: + logger.warning( + "Ignoring custom term longer than %d characters", MAX_TERM_CHARACTERS + ) + continue + key = cleaned.casefold() + if not cleaned or cleaned.startswith("#") or key in seen: + continue + seen.add(key) + normalized.append(cleaned) + return normalized + + @staticmethod + def _single_term(term: str) -> Optional[str]: + """Validate a single line-file term for an incremental edit.""" + if not isinstance(term, str) or "\n" in term or "\r" in term: + logger.warning("Ignoring invalid custom term") + return None + normalized = CustomDictionaryManager._normalize_terms([term]) + return normalized[0] if normalized else None + + @staticmethod + def _read_terms_contents(path: Path, missing_value: Optional[str] = None) -> Optional[str]: + """Read terms text for an edit while reporting invalid source files safely.""" + try: + return path.read_text(encoding="utf-8-sig") + except FileNotFoundError: + return missing_value + except (OSError, UnicodeError) as error: + logger.warning("Could not read custom terms file %s: %s", path, error) + return None + + def _legacy_corrections(self) -> list[dict[str, str]]: + """Read the never-released #768 config shape without migrating it silently.""" + raw_entries = self.config.get("text_injection", "custom_dictionary", []) + entries = normalize_corrections( + [ + {"heard": entry.get("spoken"), "replacement": entry.get("replacement")} + for entry in raw_entries + if isinstance(entry, dict) + ] + ) + if entries: + logger.warning("Using legacy config entries until %s is saved", CORRECTIONS_FILENAME) + return entries + + @staticmethod + def _atomic_write(path: Path, contents: str) -> bool: + """Atomically write UTF-8 user data and leave the old file intact on failure.""" + temporary_path: Optional[Path] = None + try: + path.parent.mkdir(parents=True, exist_ok=True) + temporary_path = path.with_name(f".{path.name}.{uuid.uuid4().hex}") + with temporary_path.open("x", encoding="utf-8") as temporary_file: + temporary_file.write(contents) + temporary_file.flush() + os.fsync(temporary_file.fileno()) + try: + # A replacement keeps the existing file's permissions; a new + # file keeps the umask-derived mode of the temporary file. + shutil.copymode(path, temporary_path) + except OSError: + pass + os.replace(temporary_path, path) + return True + except OSError as error: + logger.error("Could not save custom dictionary file %s: %s", path, error) + if temporary_path is not None: + try: + temporary_path.unlink(missing_ok=True) + except OSError as cleanup_error: + logger.warning("Could not remove temporary dictionary file: %s", cleanup_error) + return False diff --git a/src/vocalinux/main.py b/src/vocalinux/main.py index 632e813..a01ba6f 100644 --- a/src/vocalinux/main.py +++ b/src/vocalinux/main.py @@ -111,6 +111,14 @@ def parse_arguments(): action="store_true", help="Stop voice typing on a running instance (via D-Bus) and exit", ) + parser.add_argument( + "--dictionary-file", + type=str, + help=( + "Use this custom terms file for this session only (UTF-8, one term per line); " + "it does not change saved settings" + ), + ) parser.add_argument( "--transcribe-file", type=str, @@ -414,6 +422,7 @@ def main(): # Now it's safe to import GTK-dependent modules from .common_types import RecognitionState + from .custom_dictionary import CustomDictionaryManager from .speech_recognition import recognition_manager from .text_injection import text_injector from .ui import tray_indicator @@ -526,6 +535,16 @@ def main(): # sanitize on load so a bad preference cannot abort startup. history_max_items = sanitize_max_items(history_settings.get("max_items", DEFAULT_MAX_ITEMS)) + dictionary_file = getattr(args, "dictionary_file", None) + transient_terms_path = ( + dictionary_file.strip() + if isinstance(dictionary_file, str) and dictionary_file.strip() + else None + ) + if transient_terms_path is not None: + logger.info("Using session-only custom terms file: %s", transient_terms_path) + dictionary_manager = CustomDictionaryManager(config_manager, transient_terms_path) + logger.info(f"Final settings: engine={engine}, language={language}, model={model_size}") if audio_device_index is not None: logger.info( @@ -564,6 +583,7 @@ def main(): whispercpp_no_speech_thold=advanced_settings.get("whispercpp_no_speech_thold", 0.6), whispercpp_n_threads=advanced_settings.get("whispercpp_n_threads", 0), whispercpp_gpu_device=advanced_settings.get("whispercpp_gpu_device", None), + dictionary_manager=dictionary_manager, remote_api_url=saved_settings.get("remote_api_url", ""), remote_api_key=saved_settings.get("remote_api_key", ""), remote_api_endpoint=saved_settings.get("remote_api_endpoint", "/inference"), @@ -610,10 +630,19 @@ def main(): # TranscriptionHistory.add), so late segments extend that entry # rather than whichever snippet happens to be newest. latest_snippet_id: Optional[int] = None + # Worker thread that produced the most-recently-ended session's + # segments; deliveries on that same thread may extend its committed + # snippet once the next session has closed, so an older worker + # finishing two sessions later cannot leak into a newer entry. + ended_session_worker: Optional[threading.Thread] = None + # Every worker that has delivered in-session segments; a delivery on + # a thread never associated with a session is treated as the + # just-ended session's trailing decode, while a worker seen producing + # an earlier session can never merge into a newer entry. + session_workers_seen: set[threading.Thread] = set() # 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. @@ -695,7 +724,7 @@ def main(): captured after the clear are kept, so dictation that continues across a clear is still recoverable. """ - nonlocal session_worker, latest_snippet_id + nonlocal session_worker, latest_snippet_id, ended_session_worker if not transcription_history.enabled: return segment = _normalize_segment_text(segment) @@ -736,23 +765,40 @@ def main(): ) ): session_worker = worker + session_workers_seen.add(worker) session_segments.append((segment, started_at)) return # Late segment from a session that already ended: merge into # 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 + # Two kinds are admitted: trickles on the ended session's own + # worker thread (the engine's decode finishing after IDLE), + # and any straggler while a newer session is still open — + # the open session has not committed yet, so the tracked id + # still points to the ended session's own snippet. An older + # worker delivering after that newer session committed forms + # its own entry rather than growing the wrong snippet. 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 + if ( + latest_snippet_id is not None + and ( + worker is ended_session_worker + or (session_open and worker not in session_workers_seen) + ) + 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. + # and form their own snippet, which its worker keeps owning. snippet_id = transcription_history.add(segment, expected_epoch=ended_session_epoch) if snippet_id is not None: latest_snippet_id = snippet_id + ended_session_worker = worker + session_workers_seen.add(worker) def text_callback_wrapper(text: str) -> None: """Bridge between speech engine text events and the text injector. @@ -807,13 +853,14 @@ def main(): transcription history as a single snippet. """ nonlocal session_open, session_worker, latest_snippet_id - nonlocal ended_session_epoch, session_started_floor + nonlocal ended_session_epoch, ended_session_worker, 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 + closing_worker = session_worker session_worker = None ended_session_epoch = transcription_history.epoch # Only segments captured after the last clear() join the @@ -832,10 +879,12 @@ def main(): latest_snippet_id = transcription_history.add( joined, expected_epoch=ended_session_epoch ) + ended_session_worker = closing_worker else: with session_lock: if not session_open: session_open = True + leftover_worker = session_worker session_worker = None session_started_floor = time.monotonic() # Segments left over by a session that ended without a @@ -850,8 +899,9 @@ def main(): for text, started_at in session_segments if started_at > cleared_at ), - expected_epoch=transcription_history.epoch, + expected_epoch=ended_session_epoch, ) + ended_session_worker = leftover_worker session_segments.clear() # Connect speech recognition to text injection and action handling. @@ -874,10 +924,18 @@ def main(): 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 + # Only the pad's own segment bookkeeping may size the + # deletion: a manual edit or a pad "undo" blurs the + # boundaries (last_segment is None), and falling back to + # the recorded text's length could erase characters the + # user typed after dictating. Refuse rather than + # misdelete — the tracking is still cleared so a repeated + # command cannot retry the stale length. + target = dictation_pad.last_segment + if target is None: + action_handler.set_last_injected_text("") + last_injected["to_pad"] = False + return True deleted = dictation_pad.delete_last_chars(len(target)) if deleted: action_handler.set_last_injected_text("") diff --git a/src/vocalinux/speech_recognition/engines/faster_whisper_engine.py b/src/vocalinux/speech_recognition/engines/faster_whisper_engine.py index 478a5c7..9941d4a 100644 --- a/src/vocalinux/speech_recognition/engines/faster_whisper_engine.py +++ b/src/vocalinux/speech_recognition/engines/faster_whisper_engine.py @@ -39,6 +39,7 @@ class _WhisperModel(Protocol): beam_size: int, best_of: int, condition_on_previous_text: bool, + initial_prompt: Optional[str], ) -> tuple[Iterable[_Segment], object]: """Transcribe audio and return segments with metadata.""" ... @@ -130,7 +131,12 @@ class FasterWhisperEngine: return "en" return language - def transcribe(self, audio_buffer: list[bytes], language: Optional[str] = None) -> str: + def transcribe( + self, + audio_buffer: list[bytes], + language: Optional[str] = None, + initial_prompt: Optional[str] = None, + ) -> str: """Transcribe the provided audio buffer. Args: @@ -139,6 +145,7 @@ class FasterWhisperEngine: (#805): a one-shot dictation binds its session language here so a later restore cannot rewrite it mid-flight. None uses ``self.language``. + initial_prompt: Optional vocabulary prompt used to bias recognition. Returns: Recognized text. @@ -162,6 +169,7 @@ class FasterWhisperEngine: beam_size=5, best_of=5, condition_on_previous_text=False, + initial_prompt=initial_prompt, ) text_parts = [segment.text.strip() for segment in segments if segment.text] diff --git a/src/vocalinux/speech_recognition/recognition_manager.py b/src/vocalinux/speech_recognition/recognition_manager.py index f31c329..9fe5fbb 100644 --- a/src/vocalinux/speech_recognition/recognition_manager.py +++ b/src/vocalinux/speech_recognition/recognition_manager.py @@ -81,6 +81,9 @@ from ..version import __version__ from .command_processor import CommandProcessor from .silero_vad import SILERO_CHUNK_SIZE, load_silero_vad +if TYPE_CHECKING: + from ..custom_dictionary import CustomDictionaryManager + class _AudioSegment(list): """A queued audio segment that knows when its capture began. @@ -400,6 +403,10 @@ class SpeechRecognitionManager: # change a queued segment's language (#805). self._session_language: Optional[str] = None self.stop_sound_guard_ms = kwargs.get("stop_sound_guard_ms", 200) + self.dictionary_manager: Optional["CustomDictionaryManager"] = kwargs.get( + "dictionary_manager" + ) + self._vosk_dictionary_warned = False self.state = RecognitionState.IDLE self.audio_thread = None self.recognition_thread = None @@ -837,6 +844,36 @@ class SpeechRecognitionManager: self._faster_whisper_engine.language = previous logger.debug(f"Restored dictation language {previous} after one-shot") + def _get_dictionary_prompt(self) -> Optional[str]: + """Return a live custom-terms prompt without interrupting dictation on errors.""" + if self.dictionary_manager is None: + return None + try: + return self.dictionary_manager.build_initial_prompt() + except (OSError, TypeError, ValueError) as error: + logger.warning("Could not build custom terms prompt: %s", error) + return None + + def _get_whispercpp_prompt(self) -> Optional[str]: + """Combine the Advanced prompt with the live custom terms prompt. + + The explicit Advanced prompt remains first and custom terms are appended, + so enabling custom dictionary support never discards user configuration. + """ + parts = [self.whispercpp_initial_prompt.strip(), self._get_dictionary_prompt() or ""] + prompt = " ".join(part for part in parts if part) + return prompt or None + + def _apply_dictionary_corrections(self, text: str) -> str: + """Apply live corrections before voice-command interpretation.""" + if self.dictionary_manager is None: + return text + try: + return self.dictionary_manager.apply_corrections(text) + except (OSError, TypeError, ValueError) as error: + logger.warning("Could not apply custom dictionary corrections: %s", error) + return text + def _init_vosk(self): """Initialize the VOSK speech recognition engine.""" # VOSK doesn't support auto-detect, so fall back to en-us for "auto" @@ -884,6 +921,15 @@ class SpeechRecognitionManager: self.recognizer = KaldiRecognizer(self.model, 16000) self._model_initialized = True logger.info("VOSK engine initialized successfully.") + if ( + self.dictionary_manager is not None + and self.dictionary_manager.terms_enabled() + and not self._vosk_dictionary_warned + ): + self._vosk_dictionary_warned = True + logger.warning( + "Custom terms are ignored by VOSK; transcript corrections still apply." + ) except ImportError: logger.error("Failed to import VOSK. Please install it with 'pip install vosk'") @@ -1067,6 +1113,7 @@ class SpeechRecognitionManager: temperature=0.0, # Greedy decoding for consistency no_speech_threshold=0.6, fp16=use_fp16, # Explicitly set to avoid warning on CPU + initial_prompt=self._get_dictionary_prompt(), ) text = result.get("text", "").strip() @@ -1288,7 +1335,9 @@ class SpeechRecognitionManager: return "" return self._faster_whisper_engine.transcribe( - audio_buffer, language=self._dictation_language(language) + audio_buffer, + language=self._dictation_language(language), + initial_prompt=self._get_dictionary_prompt(), ) except (RuntimeError, OSError, ValueError) as e: logger.error(f"Error in faster-whisper transcription: {e}", exc_info=True) @@ -1787,7 +1836,11 @@ class SpeechRecognitionManager: # Transcribe with whisper.cpp # pywhispercpp expects audio as numpy array transcribe_start = time.time() - segments = self.model.transcribe(audio_float, language=lang) + transcribe_kwargs = {"language": lang} + # pywhispercpp reuses native parameter state. Passing an empty + # value explicitly clears a prompt that was active last segment. + transcribe_kwargs["initial_prompt"] = self._get_whispercpp_prompt() or "" + segments = self.model.transcribe(audio_float, **transcribe_kwargs) transcribe_duration = time.time() - transcribe_start # Extract text from segments, filtering non-speech tokens @@ -3221,6 +3274,18 @@ class SpeechRecognitionManager: # stop_recognition owns the final buffer. Do not finish or allow a # new session until key release has completed that handoff. self._capture_finished.wait() + # The worker can exit during the bounded audio-thread join in + # stop_recognition, leaving the released key's final buffer — and + # any stragglers it never consumed — queued behind the stop + # sentinel. Transcribe what is still here before replacing the + # queue so the tail of the recording is not silently dropped. + # A failed capture or an explicit cancel still discards it. + if ( + self.model_ready + and not self._buffered_capture_failed + and not self._cancel_buffered_session.is_set() + ): + self._transcribe_queued_segments() with self._buffer_lock: self.audio_buffer = [] self._segment_queue = queue.Queue(maxsize=32) @@ -3656,6 +3721,9 @@ class SpeechRecognitionManager: # Process text - either with voice commands or pass through directly logger.debug(f"_process_audio_buffer got text='{text[:50] if text else '(empty)'}...'") if text: + # Corrections must precede command parsing: a safe replacement can + # stop a model mishearing from triggering a destructive action. + text = self._apply_dictionary_corrections(text) if self._voice_commands_enabled: # Process with voice commands (original behavior) processed_text, actions = self.command_processor.process_text(text) @@ -3782,6 +3850,30 @@ class SpeechRecognitionManager: except queue.Empty: logger.debug("Recognition queue emptied before stop signal") + def _transcribe_queued_segments(self) -> None: + """Transcribe audio segments still queued after the worker exited. + + Runs on the reload thread once key release has finished its handoff, + so segments that missed the worker's final drain — the released + key's tail buffer queued during its bounded join — are still + transcribed instead of being thrown away with the queue. + """ + while True: + try: + queued = self._segment_queue.get_nowait() + except queue.Empty: + return + if queued is None: + continue + # Queue items are (segment, language) tuples stamped at enqueue + # time — the language must reach the decoder or the buffer is + # transcribed under the wrong language. + segment, segment_language = queued + try: + self._process_audio_buffer(segment, segment_language) + except (ChecksumError, ImportError, OSError, RuntimeError, ValueError): + logger.exception("Failed to transcribe a leftover buffered segment") + def reconfigure( self, engine: Optional[str] = None, diff --git a/src/vocalinux/ui/config_manager.py b/src/vocalinux/ui/config_manager.py index 976a767..be6fe38 100644 --- a/src/vocalinux/ui/config_manager.py +++ b/src/vocalinux/ui/config_manager.py @@ -207,6 +207,12 @@ DEFAULT_CONFIG = { "enabled": True, # Keep recent dictation snippets in the tray menu "max_items": 10, # How many snippets to retain (in-memory only, cleared on quit) }, + "dictionary": { + # Preserve the custom-dictionary contract used by the accessibility scanner. + "enabled": False, + "file_path": os.path.join(CONFIG_DIR, "dictionary.txt"), + "max_words": 200, + }, "advanced": { "power_user_mode": False, "debug_logging": False, diff --git a/src/vocalinux/ui/settings_dialog.py b/src/vocalinux/ui/settings_dialog.py index e4ddddd..c264b5a 100644 --- a/src/vocalinux/ui/settings_dialog.py +++ b/src/vocalinux/ui/settings_dialog.py @@ -17,6 +17,7 @@ UX Design Notes: - Modal dialog for model downloads (explicit confirmation for large downloads) """ +import itertools import logging import math import os @@ -35,6 +36,7 @@ gi.require_version("Gdk", "3.0") from gi.repository import Gdk, GLib, GObject, Gtk, Pango # noqa: E402 from ..common_types import RecognitionState, _EvdevCaptureDevice # noqa: E402 +from ..custom_dictionary import normalize_corrections # noqa: E402 from ..gateway_embed import GatewayStatus, get_gateway_embed_manager # noqa: E402 from ..speech_recognition.silero_vad import is_silero_available # noqa: E402 from ..utils import parakeet_model_info as parakeet # noqa: E402 @@ -480,6 +482,12 @@ _ACTION_WIDTH = 96 _SPIN_WIDTH = 88 _ICON_BUTTON_WIDTH = 36 _PAIRED_COMBO_WIDTH = _CONTROL_WIDTH - _ICON_BUTTON_WIDTH - 8 +# Term rows are built on the UI thread, so a scanner-managed file with +# thousands of entries is capped and the remainder summarized in one row. +_MAX_TERMS_DISPLAYED = 500 +# Rows appended per GLib idle slice: building thousands at once stalls the +# UI the same way an uncapped list would. +_TERMS_ROWS_PER_IDLE = 150 def _style_combo(combo: Gtk.ComboBox, width: int = _CONTROL_WIDTH) -> Gtk.ComboBox: @@ -947,6 +955,21 @@ SETTINGS_CSS = """ border: 1px solid alpha(@borders, 0.5); } +/* Keep child list backgrounds inside the card's rounded lower corners. */ +.preferences-group-list { + background-color: transparent; + border-radius: 0 0 11px 11px; +} + +.preferences-group-list > row:last-child, +.preferences-group-list > row:last-child:hover { + border-radius: 0 0 11px 11px; +} + +.dictionary-entry-list { + background-color: transparent; +} + .preferences-group-title { font-weight: bold; font-size: 0.9em; @@ -2121,6 +2144,7 @@ class PreferencesGroup(Gtk.Box): # Content area with listbox for rows self.listbox = Gtk.ListBox() + self.listbox.get_style_context().add_class("preferences-group-list") self.listbox.set_selection_mode(Gtk.SelectionMode.NONE) self.listbox.set_activate_on_single_click(False) self.pack_start(self.listbox, False, False, 0) @@ -2587,6 +2611,7 @@ class SettingsDialog(Gtk.Dialog): self.config_manager = config_manager self.speech_engine = speech_engine + self.dictionary_manager = getattr(speech_engine, "dictionary_manager", None) self.shortcut_update_callback = shortcut_update_callback self.update_status_callback = update_status_callback self.overlay_enabled_callback = overlay_enabled_callback @@ -2675,6 +2700,7 @@ class SettingsDialog(Gtk.Dialog): # every distro (some system-monitor icons only ship with Yaru). self._pages = [ SettingsPage("dictation", "Dictation", "input-keyboard-symbolic"), + SettingsPage("dictionary", "Custom Dictionary", "accessories-dictionary-symbolic"), SettingsPage("model", "Speech Model", "audio-input-microphone-symbolic"), SettingsPage("audio", "Audio", "audio-speakers-symbolic"), SettingsPage("performance", "Performance", "power-profile-performance-symbolic"), @@ -2688,6 +2714,7 @@ class SettingsDialog(Gtk.Dialog): self.dictation_page = pages_by_name["dictation"] self.shortcuts_tab = self.dictation_page.box self.recognition_settings_tab = self.dictation_page.box + self.dictionary_tab = pages_by_name["dictionary"].box self.speech_engine_tab = pages_by_name["model"].box self.audio_tab = pages_by_name["audio"].box self.power_tab = pages_by_name["performance"].box @@ -2711,7 +2738,10 @@ class SettingsDialog(Gtk.Dialog): self.settings_stack.set_transition_duration(120) self.settings_stack.set_hexpand(True) for page in self._pages: - self.settings_stack.add_titled(_scrollable(page.box), page.name, page.title) + # The dictionary page owns two independently scrollable management + # panes so its pane chooser stays visible above long term lists. + content = page.box if page.name == "dictionary" else _scrollable(page.box) + self.settings_stack.add_titled(content, page.name, page.title) self.settings_stack.add_named(self._build_search_empty_page(), "search-empty") self.settings_stack.connect("notify::visible-child", self._on_settings_page_changed) @@ -2748,6 +2778,23 @@ class SettingsDialog(Gtk.Dialog): # Search state: baseline visibility snapshot while a query is active. self._search_baseline = None self._search_previous_page = None + self._search_previous_dictionary_pane = None + # Marks dictionary pane switches made by the search filter itself so a + # user-initiated switch can drop the saved pre-search pane. + self._search_pane_programmatic = False + # Custom-terms list stays capped for GTK row count until the user + # asks to render the whole file. + self._show_all_terms = False + # Bumps on every refresh so idle-sliced row builders left over from + # an older pass stop instead of re-adding cleared rows. + self._terms_build_token = 0 + # Terms added in this dialog that land past the display cap; they are + # pinned to the top so a successful save is never invisible. + self._pinned_terms: list[str] = [] + # A term saved under "show all" scrolls into view once the idle-sliced + # rebuild finishes: (row index, displayed term) so a stale target left + # by a superseded refresh can never jump to the wrong row. + self._scroll_terms_to_row: Optional[tuple[int, str]] = None # Set content_box to speech_engine_tab for backward compatibility self.content_box = self.speech_engine_tab @@ -2757,6 +2804,7 @@ class SettingsDialog(Gtk.Dialog): self._build_language_shortcuts_section() self._build_recognition_section() self._build_simple_model_section() + self._build_dictionary_section() self._build_engine_section() self._build_remote_server_section() self._build_gateway_embed_section() @@ -2964,6 +3012,9 @@ class SettingsDialog(Gtk.Dialog): visible_page = self.settings_stack.get_visible_child_name() page_names = {page.name for page in self._pages} self._search_previous_page = visible_page if visible_page in page_names else "dictation" + self._search_previous_dictionary_pane = ( + self.dictionary_management_stack.get_visible_child_name() + ) baseline = {"rows": {}, "groups": {}, "extras": {}, "islands": {}} for page in self._pages: @@ -3018,6 +3069,27 @@ class SettingsDialog(Gtk.Dialog): self.settings_stack.set_visible_child_name(page.name) self._search_previous_page = None + if self._search_previous_dictionary_pane is not None: + # A search may have auto-switched the dictionary pane to reveal a + # match; clearing the query returns to the pane the user selected. + self.dictionary_management_stack.set_visible_child_name( + self._search_previous_dictionary_pane + ) + self._search_previous_dictionary_pane = None + + def _on_dictionary_pane_changed(self, *_args: Any) -> None: + """Keep a manual dictionary pane switch made during an active search. + + The filter auto-switches panes to reveal a match and restores the + pre-search pane when the query clears. If the user clicks the pane + switcher themselves mid-search, that newer choice wins and clearing + the search must not roll it back. + """ + if self._search_pane_programmatic: + return + if self._search_baseline is not None: + self._search_previous_dictionary_pane = None + def _on_search_changed(self, entry): """Live-filter settings rows across all pages.""" query = entry.get_text().strip() @@ -3057,6 +3129,21 @@ class SettingsDialog(Gtk.Dialog): for extra in page.extras: extra.hide() + if page.name == "dictionary": + # A search can hide the selected management pane while the + # other pane contains the match. Keep matching controls visible. + current = self.dictionary_management_stack.get_visible_child_name() + terms_visible = self.dictionary_terms_group.get_visible() + corrections_visible = self.dictionary_corrections_group.get_visible() + self._search_pane_programmatic = True + try: + if current == "terms" and not terms_visible and corrections_visible: + self.dictionary_management_stack.set_visible_child_name("corrections") + elif current == "corrections" and not corrections_visible and terms_visible: + self.dictionary_management_stack.set_visible_child_name("terms") + finally: + self._search_pane_programmatic = False + if page_matches > 0: if page.update_badge_label is not None: page.update_badge_label.hide() @@ -3292,6 +3379,546 @@ class SettingsDialog(Gtk.Dialog): self.history_enabled_switch.connect("state-set", self._on_history_enabled_toggled) self.history_max_items_spin.connect("value-changed", self._on_history_max_items_changed) + def _build_dictionary_section(self) -> None: + """Build the Custom Dictionary page for terms and transcript corrections.""" + self.dictionary_management_stack = Gtk.Stack() + self.dictionary_management_stack.set_transition_type(Gtk.StackTransitionType.CROSSFADE) + self.dictionary_management_stack.set_transition_duration(120) + self.dictionary_management_stack.set_homogeneous(False) + self.dictionary_management_stack.set_hexpand(True) + self.dictionary_management_stack.set_vexpand(True) + self.dictionary_management_stack.connect( + "notify::visible-child-name", self._on_dictionary_pane_changed + ) + + self.dictionary_management_switcher = Gtk.StackSwitcher() + self.dictionary_management_switcher.set_stack(self.dictionary_management_stack) + self.dictionary_management_switcher.set_halign(Gtk.Align.CENTER) + self.dictionary_management_switcher.get_accessible().set_name("Custom dictionary section") + self.dictionary_tab.pack_start(self.dictionary_management_switcher, False, False, 0) + + self.dictionary_feedback_label = Gtk.Label(xalign=0) + self.dictionary_feedback_label.set_line_wrap(True) + self.dictionary_feedback_label.get_accessible().set_name("Custom dictionary status") + self.dictionary_feedback_label.get_style_context().add_class("tip-label") + self.dictionary_tab.pack_start(self.dictionary_feedback_label, False, False, 0) + + self.dictionary_tab.pack_start(self.dictionary_management_stack, True, True, 0) + + terms_group = PreferencesGroup( + title="Custom terms", + description=( + "Terms bias Whisper, whisper.cpp, and Faster Whisper recognition. The UTF-8 " + "terms file is live-reloaded and has one term per line, so accessibility tools " + "can edit it." + ), + keywords=("dictionary", "terms", "vocabulary", "whisper", "scanner"), + ) + self.dictionary_terms_group = terms_group + self.dictionary_terms_enabled_switch = Gtk.Switch() + terms_group.add_row( + PreferenceRow( + title="Use custom terms", + subtitle=( + "Vocabulary bias works with Whisper, whisper.cpp, and Faster Whisper; " + "corrections work with every engine." + ), + widget=self.dictionary_terms_enabled_switch, + ) + ) + terms_path_box = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=6) + self.dictionary_terms_path_entry = Gtk.Entry() + self.dictionary_terms_path_entry.set_hexpand(True) + self.dictionary_terms_path_entry.set_placeholder_text("~/.config/vocalinux/dictionary.txt") + self.dictionary_terms_path_entry.set_tooltip_text( + "UTF-8 line file used for custom terms and the accessibility scanner" + ) + self.dictionary_terms_path_entry.get_accessible().set_name("Custom terms file") + terms_path_box.pack_start(self.dictionary_terms_path_entry, True, True, 0) + self.dictionary_terms_file_button = Gtk.FileChooserButton(title="Choose Terms File") + terms_path_box.pack_start(self.dictionary_terms_file_button, False, False, 0) + terms_group.add_row( + PreferenceRow( + title="Terms file", + subtitle="UTF-8, one term per line; # starts a comment", + widget=terms_path_box, + ) + ) + self.dictionary_terms_status_label = Gtk.Label(xalign=0) + self.dictionary_terms_status_label.set_line_wrap(True) + self.dictionary_terms_status_label.get_style_context().add_class("tip-label") + terms_group.add_row( + PreferenceRow( + title="Live status", + subtitle="The terms file is re-read before every transcription.", + widget=self.dictionary_terms_status_label, + ) + ) + + terms_add_box = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=8) + terms_add_box.set_margin_top(8) + terms_add_box.set_margin_bottom(4) + terms_add_box.set_margin_start(16) + terms_add_box.set_margin_end(16) + self.dictionary_term_entry = Gtk.Entry() + self.dictionary_term_entry.set_placeholder_text("Add a term") + self.dictionary_term_entry.set_tooltip_text( + "Term to add to the scanner-friendly terms file" + ) + self.dictionary_term_entry.get_accessible().set_name("Custom term") + self.dictionary_term_entry.set_hexpand(True) + terms_add_box.pack_start(self.dictionary_term_entry, True, True, 0) + self.dictionary_add_term_button = Gtk.Button(label="Add term") + self.dictionary_add_term_button.connect("clicked", self._on_dictionary_add_term) + self.dictionary_term_entry.connect("activate", self._on_dictionary_add_term) + terms_add_box.pack_start(self.dictionary_add_term_button, False, False, 0) + terms_add_row = Gtk.ListBoxRow() + terms_add_row.set_activatable(False) + terms_add_row.add(terms_add_box) + terms_group.add_row(terms_add_row) + + self.dictionary_terms_listbox = Gtk.ListBox() + self.dictionary_terms_listbox.get_style_context().add_class("dictionary-entry-list") + self.dictionary_terms_listbox.set_selection_mode(Gtk.SelectionMode.NONE) + self.dictionary_terms_listbox.set_placeholder( + Gtk.Label(label="No custom terms yet.", xalign=0.5) + ) + terms_list_row = Gtk.ListBoxRow() + terms_list_row.set_activatable(False) + terms_list_row.add(self.dictionary_terms_listbox) + terms_group.add_row(terms_list_row) + + self.dictionary_terms_scroller = Gtk.ScrolledWindow() + self.dictionary_terms_scroller.set_policy(Gtk.PolicyType.NEVER, Gtk.PolicyType.AUTOMATIC) + self.dictionary_terms_scroller.set_shadow_type(Gtk.ShadowType.NONE) + self.dictionary_terms_scroller.add(terms_group) + self.dictionary_management_stack.add_titled( + self.dictionary_terms_scroller, "terms", "Custom terms" + ) + + corrections_group = PreferencesGroup( + title="Transcript corrections", + description=( + "Corrections replace a misheard whole word or phrase in every completed " + "transcript. They run before voice commands, so avoid replacement text that " + "is itself a voice command." + ), + keywords=("dictionary", "correction", "replacement", "misheard", "transcript"), + ) + self.dictionary_corrections_group = corrections_group + corrections_box = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=8) + corrections_box.set_margin_top(8) + corrections_box.set_margin_bottom(4) + corrections_box.set_margin_start(16) + corrections_box.set_margin_end(16) + self.dictionary_heard_entry = Gtk.Entry() + self.dictionary_heard_entry.set_placeholder_text("Heard as, e.g. super base") + self.dictionary_heard_entry.set_tooltip_text("Misheard phrase, matched case-insensitively") + self.dictionary_heard_entry.get_accessible().set_name("Heard phrase") + self.dictionary_heard_entry.set_hexpand(True) + corrections_box.pack_start(self.dictionary_heard_entry, True, True, 0) + self.dictionary_replacement_entry = Gtk.Entry() + self.dictionary_replacement_entry.set_placeholder_text("Replace with, e.g. Supabase") + self.dictionary_replacement_entry.set_tooltip_text( + "Replacement inserted with exactly this spelling" + ) + self.dictionary_replacement_entry.get_accessible().set_name("Replacement text") + self.dictionary_replacement_entry.set_hexpand(True) + corrections_box.pack_start(self.dictionary_replacement_entry, True, True, 0) + self.dictionary_add_correction_button = Gtk.Button(label="Add correction") + self.dictionary_add_correction_button.connect("clicked", self._on_dictionary_add_correction) + self.dictionary_heard_entry.connect("activate", self._on_dictionary_add_correction) + self.dictionary_replacement_entry.connect("activate", self._on_dictionary_add_correction) + corrections_box.pack_start(self.dictionary_add_correction_button, False, False, 0) + corrections_add_row = Gtk.ListBoxRow() + corrections_add_row.set_activatable(False) + corrections_add_row.add(corrections_box) + corrections_group.add_row(corrections_add_row) + + self.dictionary_corrections_listbox = Gtk.ListBox() + self.dictionary_corrections_listbox.get_style_context().add_class("dictionary-entry-list") + self.dictionary_corrections_listbox.set_selection_mode(Gtk.SelectionMode.NONE) + self.dictionary_corrections_listbox.set_placeholder( + Gtk.Label(label="No transcript corrections yet.", xalign=0.5) + ) + corrections_list_row = Gtk.ListBoxRow() + corrections_list_row.set_activatable(False) + corrections_list_row.add(self.dictionary_corrections_listbox) + corrections_group.add_row(corrections_list_row) + + corrections_scroller = Gtk.ScrolledWindow() + corrections_scroller.set_policy(Gtk.PolicyType.NEVER, Gtk.PolicyType.AUTOMATIC) + corrections_scroller.set_shadow_type(Gtk.ShadowType.NONE) + corrections_scroller.add(corrections_group) + self.dictionary_management_stack.add_titled( + corrections_scroller, "corrections", "Corrections" + ) + + self.dictionary_terms_enabled_switch.connect("state-set", self._on_dictionary_terms_enabled) + self.dictionary_terms_path_entry.connect("activate", self._on_dictionary_terms_path_changed) + self.dictionary_terms_path_entry.connect( + "focus-out-event", self._on_dictionary_terms_path_changed + ) + self.dictionary_terms_file_button.connect("file-set", self._on_dictionary_terms_file_chosen) + + def _on_dictionary_terms_enabled(self, widget: Any, state: bool) -> bool: + """Persist terms enablement and restore the control when persistence fails.""" + if self._initializing or self._applying_settings or self.dictionary_manager is None: + return False + if not self.dictionary_manager.set_terms_enabled(bool(state)): + self.dictionary_feedback_label.set_text("Could not save custom terms setting.") + self._refresh_dictionary_ui() + return False + + def _on_dictionary_terms_path_changed(self, widget: Any, *args: Any) -> bool: + """Persist a validated custom terms path and restore it on failure.""" + if self._initializing or self._applying_settings or self.dictionary_manager is None: + return False + if self.dictionary_manager.set_terms_path(self.dictionary_terms_path_entry.get_text()): + self.dictionary_feedback_label.set_text("Custom terms path saved.") + else: + self.dictionary_feedback_label.set_text( + "Could not save that custom terms path; keeping the previous path." + ) + self._refresh_dictionary_ui() + return False + + def _on_dictionary_terms_file_chosen(self, widget: Any) -> None: + """Apply a terms path selected through the GTK file chooser.""" + path = widget.get_filename() + if path: + self.dictionary_terms_path_entry.set_text(path) + self._on_dictionary_terms_path_changed(self.dictionary_terms_path_entry) + + def _on_dictionary_add_term(self, widget: Any) -> None: + """Add a term to the fixed line file, reporting an observable result.""" + if self._initializing or self._applying_settings or self.dictionary_manager is None: + return + term = self.dictionary_term_entry.get_text().strip() + if not term: + self.dictionary_feedback_label.set_text("Enter a term before adding it.") + return + terms = self.dictionary_manager.get_terms() + if any(existing.casefold() == term.casefold() for existing in terms): + self.dictionary_feedback_label.set_text("That term is already in the terms file.") + return + if not self.dictionary_manager.add_term(term): + self.dictionary_feedback_label.set_text( + "Could not save the terms file; no term was added." + ) + return + self.dictionary_term_entry.set_text("") + terms_now = self.dictionary_manager.get_terms() + if any(existing.casefold() == term.casefold() for existing in terms_now): + self.dictionary_feedback_label.set_text("Term saved to the live terms file.") + if self._show_all_terms: + # Oversized or yield-capped files prepend the term, other + # files append it — scroll to wherever it landed once the + # idle-sliced rebuild finishes. Under "show all" the rebuilt + # rows match terms_now one to one. + new_index = next( + index + for index, existing in enumerate(terms_now) + if existing.casefold() == term.casefold() + ) + self._scroll_terms_to_row = (new_index, terms_now[new_index]) + else: + # A term that lands past the display cap is pinned to the top + # of the capped list — a saved term is never invisible. The + # next refresh drops the pin if the term is already visible. + self._pinned_terms.append(term) + else: + self.dictionary_feedback_label.set_text("That term is not valid for the terms file.") + self._refresh_dictionary_ui() + + def _on_terms_show_all(self, widget: Any) -> None: + """Render every term, including those past the display cap.""" + self._show_all_terms = True + self._refresh_dictionary_ui() + + def _on_terms_show_fewer(self, widget: Any) -> None: + """Collapse the terms list back to the display cap.""" + self._show_all_terms = False + self._refresh_dictionary_ui() + + def _on_dictionary_remove_term(self, widget: Any, term: str) -> None: + """Remove one term from the fixed line file.""" + if self._initializing or self._applying_settings or self.dictionary_manager is None: + return + if self.dictionary_manager.remove_term(term): + self.dictionary_feedback_label.set_text("Term removed from the live terms file.") + else: + self.dictionary_feedback_label.set_text( + "Could not save the terms file; the term was not removed." + ) + self._refresh_dictionary_ui() + + def _on_dictionary_add_correction(self, widget: Any) -> None: + """Add or update a phrase correction in the structured corrections file.""" + if self._initializing or self._applying_settings or self.dictionary_manager is None: + return + heard = self.dictionary_heard_entry.get_text().strip() + replacement = self.dictionary_replacement_entry.get_text().strip() + if not heard or not replacement: + self.dictionary_feedback_label.set_text( + "Enter both the heard phrase and its replacement." + ) + return + entries = self.dictionary_manager.get_corrections_for_edit() + if entries is None: + self.dictionary_feedback_label.set_text( + "Could not edit corrections: fix or replace the malformed corrections file first." + ) + return + candidate = {"heard": heard, "replacement": replacement} + normalized_entries = normalize_corrections([candidate]) + if not normalized_entries: + self.dictionary_feedback_label.set_text("That correction is not valid for the file.") + return + candidate = normalized_entries[0] + heard_key = candidate["heard"].casefold() + existing = [entry for entry in entries if entry["heard"].casefold() == heard_key] + entries = [entry for entry in entries if entry["heard"].casefold() != heard_key] + entries.append(candidate) + if not self.dictionary_manager.save_corrections(entries): + self.dictionary_feedback_label.set_text( + "Could not save corrections; no correction was changed." + ) + return + self.dictionary_heard_entry.set_text("") + self.dictionary_replacement_entry.set_text("") + if any( + entry["heard"].casefold() == heard_key + for entry in self.dictionary_manager.get_corrections() + ): + self.dictionary_feedback_label.set_text( + "Correction updated." + if existing + else "Correction saved to the live corrections file." + ) + else: + self.dictionary_feedback_label.set_text("That correction is not valid for the file.") + self._refresh_dictionary_ui() + + def _on_dictionary_remove_correction(self, widget: Any, heard: str) -> None: + """Remove one correction from the structured corrections file.""" + if self._initializing or self._applying_settings or self.dictionary_manager is None: + return + editable_entries = self.dictionary_manager.get_corrections_for_edit() + if editable_entries is None: + self.dictionary_feedback_label.set_text( + "Could not edit corrections: fix or replace the malformed corrections file first." + ) + return + entries = [ + entry for entry in editable_entries if entry["heard"].casefold() != heard.casefold() + ] + if self.dictionary_manager.save_corrections(entries): + self.dictionary_feedback_label.set_text("Correction removed.") + else: + self.dictionary_feedback_label.set_text( + "Could not save corrections; nothing was removed." + ) + self._refresh_dictionary_ui() + + def _make_term_row(self, term: str, transient: bool) -> Gtk.ListBoxRow: + """Build one removable term row for the custom terms list.""" + row = Gtk.ListBoxRow() + row.set_activatable(False) + row_box = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=8) + row_box.set_margin_top(6) + row_box.set_margin_bottom(6) + row_box.set_margin_start(16) + row_box.set_margin_end(16) + label = Gtk.Label(label=term, xalign=0) + label.set_hexpand(True) + row_box.pack_start(label, True, True, 0) + remove_button = Gtk.Button(label="Remove") + remove_button.set_tooltip_text(f"Remove term {term}") + remove_button.get_accessible().set_name(f"Remove term {term}") + remove_button.set_sensitive(not transient) + remove_button.connect("clicked", self._on_dictionary_remove_term, term) + row_box.pack_start(remove_button, False, False, 0) + row.add(row_box) + return row + + def _build_term_rows( + self, terms: list[str], total_terms: int, transient: bool, token: int + ) -> None: + """Populate the terms listbox in idle slices of bounded size. + + Building thousands of GTK rows synchronously stalls the whole + dialog, so each slice appends at most ``_TERMS_ROWS_PER_IDLE`` rows + and reschedules itself. Every refresh bumps the build token, which + retires pending slices from an older pass instead of letting them + re-add rows the refresh already cleared. + """ + pending = iter(terms) + + def build_chunk() -> bool: + if token != self._terms_build_token: + return False + batch = list(itertools.islice(pending, _TERMS_ROWS_PER_IDLE)) + for term in batch: + self.dictionary_terms_listbox.add(self._make_term_row(term, transient)) + self.dictionary_terms_listbox.show_all() + if len(batch) == _TERMS_ROWS_PER_IDLE: + return True + self._finish_term_rows(total_terms) + return False + + # The first slice runs inline so small lists appear immediately; + # a slice never exceeds _TERMS_ROWS_PER_IDLE rows of work. + if build_chunk(): + GLib.idle_add(build_chunk) + + def _finish_term_rows(self, total_terms: int) -> None: + """Append the overflow summary row after the last term slice.""" + if total_terms > _MAX_TERMS_DISPLAYED: + overflow_row = Gtk.ListBoxRow() + overflow_row.set_activatable(False) + overflow_box = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=8) + overflow_box.set_margin_top(6) + overflow_box.set_margin_bottom(6) + overflow_box.set_margin_start(16) + overflow_box.set_margin_end(16) + if self._show_all_terms: + overflow_text = f"Showing all {total_terms} terms." + toggle_button = Gtk.Button(label=f"Show first {_MAX_TERMS_DISPLAYED}") + toggle_button.connect("clicked", self._on_terms_show_fewer) + else: + overflow_text = f"Showing the first {_MAX_TERMS_DISPLAYED} of {total_terms} terms." + toggle_button = Gtk.Button(label=f"Show all {total_terms}") + toggle_button.connect("clicked", self._on_terms_show_all) + overflow_label = Gtk.Label(label=overflow_text, xalign=0) + overflow_label.set_line_wrap(True) + overflow_label.set_hexpand(True) + overflow_label.get_style_context().add_class("tip-label") + overflow_box.pack_start(overflow_label, True, True, 0) + overflow_box.pack_start(toggle_button, False, False, 0) + overflow_row.add(overflow_box) + self.dictionary_terms_listbox.add(overflow_row) + self.dictionary_terms_listbox.show_all() + if self._scroll_terms_to_row is not None: + index, expected_term = self._scroll_terms_to_row + self._scroll_terms_to_row = None + row = self.dictionary_terms_listbox.get_row_at_index(index) + row_text = self._term_row_text(row) if row is not None else None + if row_text is not None and row_text.casefold() == expected_term.casefold(): + self._scroll_terms_row_into_view(row) + + @staticmethod + def _term_row_text(row: Gtk.ListBoxRow) -> Optional[str]: + """Return the term displayed by a row built by ``_make_term_row``.""" + row_box = row.get_child() + if not isinstance(row_box, Gtk.Box) or not row_box.get_children(): + return None + label = row_box.get_children()[0] + return label.get_label() if isinstance(label, Gtk.Label) else None + + def _scroll_terms_row_into_view(self, row: Gtk.ListBoxRow) -> None: + """Bring a term row inside the terms scroller's viewport once laid out.""" + retries_left = 20 + + def scroll_once() -> bool: + nonlocal retries_left + retries_left -= 1 + if row.get_parent() is not self.dictionary_terms_listbox: + return False + allocation = row.get_allocation() + if allocation.height <= 0: + # Not laid out yet — retry briefly instead of never scrolling. + return retries_left > 0 + # The row's y is relative to the terms listbox, but the scroller's + # adjustment addresses the whole card (header and controls above + # the list) — translate into that space or the scroll stops short. + viewport = self.dictionary_terms_scroller.get_child() + content = viewport.get_child() if isinstance(viewport, Gtk.Viewport) else None + position = row.translate_coordinates(content, 0, 0) if content else None + if position is None: + return False + _, row_y = position + adjustment = self.dictionary_terms_scroller.get_vadjustment() + page_size = adjustment.get_page_size() + value = adjustment.get_value() + if row_y < value: + adjustment.set_value(max(adjustment.get_lower(), row_y)) + elif row_y + allocation.height > value + page_size: + adjustment.set_value( + min( + adjustment.get_upper() - page_size, + row_y + allocation.height - page_size, + ) + ) + return False + + GLib.idle_add(scroll_once) + + def _refresh_dictionary_ui(self) -> None: + """Rebuild custom dictionary controls from the live, file-backed state.""" + if not hasattr(self, "dictionary_terms_enabled_switch"): + return + if self.dictionary_manager is None: + self.dictionary_terms_enabled_switch.set_sensitive(False) + self.dictionary_terms_status_label.set_text("Custom dictionary support is unavailable.") + return + transient = self.dictionary_manager.is_transient_terms + enabled = self.dictionary_manager.terms_enabled() + self.dictionary_terms_enabled_switch.set_active(enabled) + self.dictionary_terms_enabled_switch.set_sensitive(not transient) + self.dictionary_terms_path_entry.set_text(self.dictionary_manager.terms_path_text()) + self.dictionary_terms_path_entry.set_sensitive(not transient) + self.dictionary_terms_file_button.set_sensitive(not transient) + self.dictionary_terms_status_label.set_text( + ("Session-only --dictionary-file override is active. " if transient else "") + + self.dictionary_manager.terms_status() + ) + self.dictionary_term_entry.set_sensitive(not transient) + self.dictionary_add_term_button.set_sensitive(not transient) + + for child in list(self.dictionary_terms_listbox.get_children()): + self.dictionary_terms_listbox.remove(child) + terms = self.dictionary_manager.get_terms() + visible_terms = terms if self._show_all_terms else terms[:_MAX_TERMS_DISPLAYED] + # Recently added terms that would fall outside the visible window + # stay pinned to the top while they remain unseen; a pin expires as + # soon as the term disappears or becomes visible on its own. + visible_keys = {term.casefold() for term in visible_terms} + term_keys = {term.casefold() for term in terms} + self._pinned_terms = [ + pinned + for pinned in self._pinned_terms + if pinned.casefold() in term_keys and pinned.casefold() not in visible_keys + ] + self._terms_build_token += 1 + self._build_term_rows( + [*self._pinned_terms, *visible_terms], len(terms), transient, self._terms_build_token + ) + + for child in list(self.dictionary_corrections_listbox.get_children()): + self.dictionary_corrections_listbox.remove(child) + + for entry in self.dictionary_manager.get_corrections(): + row = Gtk.ListBoxRow() + row.set_activatable(False) + row_box = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=8) + row_box.set_margin_top(6) + row_box.set_margin_bottom(6) + row_box.set_margin_start(16) + row_box.set_margin_end(16) + label = Gtk.Label(label=f"{entry['heard']} → {entry['replacement']}", xalign=0) + label.set_hexpand(True) + row_box.pack_start(label, True, True, 0) + remove_button = Gtk.Button(label="Remove") + remove_button.set_tooltip_text(f"Remove correction for {entry['heard']}") + remove_button.get_accessible().set_name(f"Remove correction for {entry['heard']}") + remove_button.connect("clicked", self._on_dictionary_remove_correction, entry["heard"]) + row_box.pack_start(remove_button, False, False, 0) + row.add(row_box) + self.dictionary_corrections_listbox.add(row) + self.dictionary_terms_listbox.show_all() + self.dictionary_corrections_listbox.show_all() + def _build_auto_pause_section(self): """Build Auto-Pause settings: enable toggle + process name list.""" group = PreferencesGroup( @@ -6670,6 +7297,7 @@ class SettingsDialog(Gtk.Dialog): self.auto_pause_switch.set_active(auto_pause_enabled) self._update_auto_pause_sensitivity(auto_pause_enabled) self._refresh_auto_pause_list() + self._refresh_dictionary_ui() keepalive_settings = self.config_manager.get_settings().get("model_keepalive", {}) keepalive_enabled = bool(keepalive_settings.get("enabled", False)) diff --git a/tests/test_custom_dictionary.py b/tests/test_custom_dictionary.py new file mode 100644 index 0000000..6cba739 --- /dev/null +++ b/tests/test_custom_dictionary.py @@ -0,0 +1,1106 @@ +"""Tests for the file-backed custom dictionary contract.""" + +import json +import re +import stat +import threading +from pathlib import Path +from typing import Any +from unittest.mock import MagicMock, Mock, patch + +from vocalinux.custom_dictionary import ( + CORRECTIONS_FILENAME, + DEFAULT_TERMS_PATH, + LEGACY_DEFAULT_TERMS_PATH, + MAX_CORRECTION_CHARACTERS, + MAX_TERMS_FILE_BYTES, + MAX_TERMS_YIELDED, + TERMS_FILENAME, + CustomDictionaryManager, + apply_corrections, + normalize_corrections, +) +from vocalinux.main import parse_arguments +from vocalinux.speech_recognition.recognition_manager import SpeechRecognitionManager + + +class FakeConfig: + """Minimal ConfigManager substitute with controllable persistence.""" + + def __init__(self, values: dict[str, dict[str, Any]] | None = None, save_result: bool = True): + self.values = values or { + "dictionary": { + "enabled": False, + "file_path": DEFAULT_TERMS_PATH, + "max_words": 200, + } + } + self.save_result = save_result + + def get(self, section: str, key: str, default: Any = None) -> Any: + """Return one configured value.""" + return self.values.get(section, {}).get(key, default) + + def set(self, section: str, key: str, value: Any) -> bool: + """Set one configured value.""" + self.values.setdefault(section, {})[key] = value + return True + + def save_config(self) -> bool: + """Return the requested persistence result.""" + return self.save_result + + +def manager_at( + tmp_path: Path, monkeypatch, config: FakeConfig | None = None +) -> CustomDictionaryManager: + """Create a manager whose configured files live in pytest's temp directory.""" + monkeypatch.setattr("vocalinux.custom_dictionary.config_dir", lambda: str(tmp_path)) + config = config or FakeConfig() + dictionary = config.values.setdefault("dictionary", {}) + if dictionary.get("file_path") in (None, DEFAULT_TERMS_PATH): + dictionary["file_path"] = str(tmp_path / TERMS_FILENAME) + return CustomDictionaryManager(config) + + +def recognition_manager( + engine: str, dictionary: CustomDictionaryManager, advanced_prompt: str = "" +) -> SpeechRecognitionManager: + """Build a recognition manager without loading a real speech model.""" + with patch.object(SpeechRecognitionManager, "_init_vosk"): + with patch.object(SpeechRecognitionManager, "_init_whisper"): + with patch.object(SpeechRecognitionManager, "_init_whispercpp"): + return SpeechRecognitionManager( + engine=engine, + dictionary_manager=dictionary, + whispercpp_initial_prompt=advanced_prompt, + ) + + +def mock_numpy() -> MagicMock: + """Return the minimum NumPy surface used by transcription helpers.""" + numpy = MagicMock() + numpy.frombuffer.return_value = MagicMock(__len__=lambda _: 16_000) + numpy.frombuffer.return_value.astype.return_value = numpy.frombuffer.return_value + numpy.int16 = "int16" + numpy.float32 = "float32" + return numpy + + +def test_terms_are_live_reloaded_and_scanner_friendly(tmp_path: Path, monkeypatch) -> None: + """The UTF-8 line file accepts comments and applies an external replacement.""" + manager = manager_at(tmp_path, monkeypatch, FakeConfig({"dictionary": {"enabled": True}})) + terms_file = tmp_path / TERMS_FILENAME + terms_file.write_text("# scanner comment\nVocaLinux\nvocalinux\nSupabase\n", encoding="utf-8") + + assert manager.get_terms() == ["VocaLinux", "Supabase"] + assert manager.build_initial_prompt() == "VocaLinux Supabase" + + terms_file.write_text("PyGObject\n", encoding="utf-8") + assert manager.build_initial_prompt() == "PyGObject" + + +def test_terms_save_is_atomic_shape_and_transient_override_is_not_written( + tmp_path: Path, monkeypatch +) -> None: + """UI saves standard terms safely while CLI terms remain session-only.""" + manager = manager_at(tmp_path, monkeypatch) + assert manager.save_terms(["VocaLinux", "vocalinux", "# ignored", " PyGObject "]) + assert (tmp_path / TERMS_FILENAME).read_text(encoding="utf-8") == "VocaLinux\nPyGObject\n" + + override = tmp_path / "override.txt" + override.write_text("Override\n", encoding="utf-8") + transient = CustomDictionaryManager(FakeConfig(), str(override)) + assert transient.terms_enabled() + assert transient.get_terms() == ["Override"] + assert not transient.save_terms(["No write"]) + + +def test_incremental_term_edits_preserve_comments_and_blank_lines( + tmp_path: Path, monkeypatch +) -> None: + """GTK add/remove operations retain unrelated scanner-managed source lines.""" + manager = manager_at(tmp_path, monkeypatch) + terms_file = tmp_path / TERMS_FILENAME + terms_file.write_text("# scanner note\n\nVocaLinux\n# retain me\nPyGObject\n", encoding="utf-8") + + assert manager.add_term("Supabase") + assert terms_file.read_text(encoding="utf-8") == ( + "# scanner note\n\nVocaLinux\n# retain me\nPyGObject\nSupabase\n" + ) + assert manager.remove_term("PyGObject") + assert terms_file.read_text(encoding="utf-8") == ( + "# scanner note\n\nVocaLinux\n# retain me\nSupabase\n" + ) + + +def test_enable_rollback_when_config_persistence_fails(tmp_path: Path, monkeypatch) -> None: + """A failed config save preserves the previous in-memory setting.""" + config = FakeConfig(save_result=False) + manager = manager_at(tmp_path, monkeypatch, config) + + assert not manager.set_terms_enabled(True) + assert not manager.terms_enabled() + + +def test_prompt_read_stops_at_the_terms_limit(tmp_path: Path, monkeypatch) -> None: + """Prompt assembly stops processing terms after the configured term cap.""" + config = FakeConfig({"dictionary": {"enabled": True, "max_words": 2}}) + manager = manager_at(tmp_path, monkeypatch, config) + (tmp_path / TERMS_FILENAME).write_text("one\ntwo\nthree\n", encoding="utf-8") + + assert manager.build_initial_prompt() == "one two" + + +def test_partially_invalid_terms_file_fails_closed(tmp_path: Path, monkeypatch) -> None: + """Valid lines before corrupt UTF-8 must not reach recognition.""" + manager = manager_at(tmp_path, monkeypatch, FakeConfig({"dictionary": {"enabled": True}})) + payload = b"one\ntwo\n" + b"x" * 9000 + b"\xff\nthree\n" + (tmp_path / TERMS_FILENAME).write_bytes(payload) + + assert manager.get_terms() == [] + assert manager.build_initial_prompt() is None + assert manager.terms_status() == "Terms file is not valid UTF-8." + + +def test_oversized_terms_file_supplies_its_leading_lines(tmp_path: Path, monkeypatch) -> None: + """A file beyond the read bound still supplies its complete leading lines.""" + manager = manager_at(tmp_path, monkeypatch, FakeConfig({"dictionary": {"enabled": True}})) + line_count = MAX_TERMS_FILE_BYTES // 8 + 2 # 8-byte lines plus a partial tail + (tmp_path / TERMS_FILENAME).write_bytes( + b"".join(f"t{i:06d}\n".encode() for i in range(line_count)) + ) + + terms = manager.get_terms() + assert terms + assert all(re.fullmatch(r"t\d{6}", term) for term in terms) + assert len(terms) == MAX_TERMS_YIELDED + assert manager.build_initial_prompt() is not None + assert "leading lines only" in manager.terms_status() + + +def test_prompt_on_oversized_file_streams_only_the_needed_prefix( + tmp_path: Path, monkeypatch +) -> None: + """Prompt assembly stops after max_terms instead of decoding the bound.""" + config = FakeConfig({"dictionary": {"enabled": True, "max_words": 3}}) + manager = manager_at(tmp_path, monkeypatch, config) + line_count = MAX_TERMS_FILE_BYTES // 8 + 2 + (tmp_path / TERMS_FILENAME).write_bytes( + b"".join(f"t{i:06d}\n".encode() for i in range(line_count)) + ) + + assert manager.build_initial_prompt() == "t000000 t000001 t000002" + + +def test_prompt_on_oversized_file_with_corrupt_prefix_still_fails_closed( + tmp_path: Path, monkeypatch +) -> None: + """Invalid UTF-8 inside the usable prefix supplies no prompt at all.""" + manager = manager_at(tmp_path, monkeypatch, FakeConfig({"dictionary": {"enabled": True}})) + line_count = MAX_TERMS_FILE_BYTES // 8 + 2 + (tmp_path / TERMS_FILENAME).write_bytes( + b"one\ntwo\n" + b"\xff\n" + b"".join(f"t{i:06d}\n".encode() for i in range(line_count)) + ) + + assert manager.build_initial_prompt() is None + + +def test_prompt_on_oversized_file_validates_past_the_term_limit( + tmp_path: Path, monkeypatch +) -> None: + """Invalid UTF-8 after the collected prompt terms still supplies none.""" + config = FakeConfig({"dictionary": {"enabled": True, "max_words": 2}}) + manager = manager_at(tmp_path, monkeypatch, config) + line_count = MAX_TERMS_FILE_BYTES // 8 + 2 + (tmp_path / TERMS_FILENAME).write_bytes( + b"one\ntwo\nthree\n" + + b"\xff\n" + + b"".join(f"t{i:06d}\n".encode() for i in range(line_count)) + ) + + assert manager.build_initial_prompt() is None + assert manager.get_terms() == [] + + +def test_prompt_on_oversized_file_with_single_huge_line_stays_bounded( + tmp_path: Path, monkeypatch +) -> None: + """A first line longer than the read bound supplies no prompt terms.""" + manager = manager_at(tmp_path, monkeypatch, FakeConfig({"dictionary": {"enabled": True}})) + (tmp_path / TERMS_FILENAME).write_bytes(b"x" * (MAX_TERMS_FILE_BYTES + 10) + b"\nvalid\n") + + assert manager.build_initial_prompt() is None + + +def test_add_term_on_oversized_file_stays_visible(tmp_path: Path, monkeypatch) -> None: + """Prepending past the read window keeps the new term visible while + preserving every scanner-owned line, comment, and blank line.""" + manager = manager_at(tmp_path, monkeypatch, FakeConfig({"dictionary": {"enabled": True}})) + line_count = MAX_TERMS_FILE_BYTES // 8 + 2 + original = b"# scanner tail\n" + b"".join(f"t{i:06d}\n".encode() for i in range(line_count)) + (tmp_path / TERMS_FILENAME).write_bytes(original) + + assert manager.add_term("brandnew") + assert (tmp_path / TERMS_FILENAME).read_bytes() == b"brandnew\n" + original + terms = manager.get_terms() + assert terms[0] == "brandnew" + assert len(terms) == MAX_TERMS_YIELDED + + +def test_add_term_on_yield_capped_file_stays_visible(tmp_path: Path, monkeypatch) -> None: + """A file at the yield cap but under the byte bound also needs the new + term prepended so Settings and the prompt can see it.""" + manager = manager_at(tmp_path, monkeypatch, FakeConfig({"dictionary": {"enabled": True}})) + original = "".join(f"t{i:05d}\n" for i in range(MAX_TERMS_YIELDED)) + terms_file = tmp_path / TERMS_FILENAME + terms_file.write_text(original, encoding="utf-8") + + assert manager.add_term("brandnew") + assert terms_file.read_text(encoding="utf-8") == f"brandnew\n{original}" + terms = manager.get_terms() + assert terms[0] == "brandnew" + assert len(terms) == MAX_TERMS_YIELDED + + +def test_oversized_terms_file_without_complete_line_yields_nothing( + tmp_path: Path, monkeypatch +) -> None: + """A huge file whose first line never ends supplies no terms.""" + manager = manager_at(tmp_path, monkeypatch, FakeConfig({"dictionary": {"enabled": True}})) + (tmp_path / TERMS_FILENAME).write_bytes(b"x" * (MAX_TERMS_FILE_BYTES + 10)) + + assert manager.get_terms() == [] + + +def test_pr_767_dictionary_configuration_keys_and_contract_are_preserved( + tmp_path: Path, monkeypatch +) -> None: + """Existing enabled, file_path, and max_words settings remain effective.""" + configured_path = tmp_path / TERMS_FILENAME + configured_path.write_text("VocaLinux\nPyGObject\n", encoding="utf-8") + config = FakeConfig( + { + "dictionary": { + "enabled": True, + "file_path": str(configured_path), + "max_words": 1, + } + } + ) + manager = manager_at(tmp_path, monkeypatch, config) + + assert TERMS_FILENAME == "dictionary.txt" + assert Path(DEFAULT_TERMS_PATH).name == TERMS_FILENAME + assert manager.terms_enabled() + assert manager.terms_path() == configured_path + assert manager.build_initial_prompt() == "VocaLinux" + + +def test_default_terms_path_follows_xdg_config_home(tmp_path: Path, monkeypatch) -> None: + """A missing saved path uses the same XDG-aware root as corrections.""" + monkeypatch.setattr("vocalinux.custom_dictionary.config_dir", lambda: str(tmp_path)) + manager = CustomDictionaryManager(FakeConfig({"dictionary": {}})) + + assert manager.terms_path() == tmp_path / TERMS_FILENAME + assert manager.corrections_path() == tmp_path / CORRECTIONS_FILENAME + + +def test_legacy_vocalinux_terms_root_keeps_corrections_at_config_dir( + tmp_path: Path, monkeypatch +) -> None: + """Persisted ~/.config/vocalinux terms do not move the corrections file.""" + monkeypatch.setattr("vocalinux.custom_dictionary.config_dir", lambda: str(tmp_path)) + legacy_corrections = Path.home() / ".config" / "vocalinux" / CORRECTIONS_FILENAME + expanded_legacy = str(Path.home() / ".config" / "vocalinux" / TERMS_FILENAME) + assert legacy_corrections != tmp_path / CORRECTIONS_FILENAME + + tilde_manager = CustomDictionaryManager( + FakeConfig({"dictionary": {"file_path": LEGACY_DEFAULT_TERMS_PATH}}) + ) + assert tilde_manager.corrections_path() == tmp_path / CORRECTIONS_FILENAME + + expanded_manager = CustomDictionaryManager( + FakeConfig({"dictionary": {"file_path": expanded_legacy}}) + ) + assert expanded_manager.corrections_path() == tmp_path / CORRECTIONS_FILENAME + + +def test_explicit_legacy_terms_path_keeps_corrections_at_config_dir( + tmp_path: Path, monkeypatch +) -> None: + """An explicit leftover pre-XDG terms path cannot relocate corrections.""" + monkeypatch.setattr("vocalinux.custom_dictionary.config_dir", lambda: str(tmp_path)) + legacy_root = Path.home() / ".config" / "vocalinux" + expanded_legacy = str(legacy_root / TERMS_FILENAME) + assert legacy_root / CORRECTIONS_FILENAME != tmp_path / CORRECTIONS_FILENAME + + tilde_manager = CustomDictionaryManager( + FakeConfig( + { + "dictionary": { + "file_path": LEGACY_DEFAULT_TERMS_PATH, + "file_path_explicit": True, + } + } + ) + ) + assert tilde_manager.corrections_path() == tmp_path / CORRECTIONS_FILENAME + + stamped = FakeConfig({"dictionary": {"file_path": expanded_legacy}}) + stamped_manager = CustomDictionaryManager(stamped) + stamped_manager.terms_path_text() + assert stamped.get("dictionary", "file_path_explicit") is True + assert stamped_manager.corrections_path() == tmp_path / CORRECTIONS_FILENAME + + +def test_corrections_fall_back_from_terms_dir_when_primary_missing( + tmp_path: Path, monkeypatch +) -> None: + """Leftover terms-adjacent corrections are read and copied to the standard file.""" + fake_home = tmp_path / "home" + xdg_dir = tmp_path / "xdg-config" + legacy_root = fake_home / ".config" / "vocalinux" + legacy_root.mkdir(parents=True) + xdg_dir.mkdir() + monkeypatch.setenv("HOME", str(fake_home)) + monkeypatch.setattr("vocalinux.custom_dictionary.config_dir", lambda: str(xdg_dir)) + + old_path = Path(LEGACY_DEFAULT_TERMS_PATH).expanduser().parent / CORRECTIONS_FILENAME + payload = { + "version": 1, + "corrections": [{"heard": "super base", "replacement": "Supabase"}], + } + old_path.write_text(json.dumps(payload) + "\n", encoding="utf-8") + + manager = CustomDictionaryManager( + FakeConfig( + { + "dictionary": { + "file_path": LEGACY_DEFAULT_TERMS_PATH, + "file_path_explicit": True, + } + } + ) + ) + primary = xdg_dir / CORRECTIONS_FILENAME + assert manager.corrections_path() == primary + assert primary != old_path + assert not primary.exists() + assert manager.get_corrections() == [{"heard": "super base", "replacement": "Supabase"}] + assert primary.is_file() + old_path.unlink() + assert manager.get_corrections() == [{"heard": "super base", "replacement": "Supabase"}] + + +def test_corrections_fallback_survives_copy_failure(tmp_path: Path, monkeypatch) -> None: + """A failed copy-once still returns leftover terms-adjacent corrections.""" + terms_root = tmp_path / "legacy-root" + xdg_dir = tmp_path / "xdg-config" + terms_root.mkdir() + xdg_dir.mkdir() + monkeypatch.setattr("vocalinux.custom_dictionary.config_dir", lambda: str(xdg_dir)) + old_path = terms_root / CORRECTIONS_FILENAME + payload = { + "version": 1, + "corrections": [{"heard": "super base", "replacement": "Supabase"}], + } + old_path.write_text(json.dumps(payload) + "\n", encoding="utf-8") + + manager = CustomDictionaryManager( + FakeConfig( + { + "dictionary": { + "file_path": str(terms_root / TERMS_FILENAME), + "file_path_explicit": True, + } + } + ) + ) + primary = xdg_dir / CORRECTIONS_FILENAME + + def fail_replace(source: Path, destination: Path) -> None: + """Simulate a filesystem failure during atomic replacement.""" + raise OSError("disk full") + + monkeypatch.setattr("vocalinux.custom_dictionary.os.replace", fail_replace) + + assert manager.corrections_path() == primary + assert not primary.exists() + assert manager.get_corrections() == [{"heard": "super base", "replacement": "Supabase"}] + assert not primary.exists() + + +def test_corrections_stay_at_config_dir_with_arbitrary_terms_parent( + tmp_path: Path, monkeypatch +) -> None: + """Corrections ignore the terms file location outside a vocalinux directory.""" + xdg_dir = tmp_path / "xdg-config" + xdg_dir.mkdir() + monkeypatch.setattr("vocalinux.custom_dictionary.config_dir", lambda: str(xdg_dir)) + terms = tmp_path / "my-terms.txt" + manager = CustomDictionaryManager( + FakeConfig( + { + "dictionary": { + "file_path": str(terms), + "file_path_explicit": True, + } + } + ) + ) + assert manager.corrections_path() == xdg_dir / CORRECTIONS_FILENAME + assert manager.corrections_path() != tmp_path / CORRECTIONS_FILENAME + + +def test_transient_terms_override_keeps_corrections_at_config_dir( + tmp_path: Path, monkeypatch +) -> None: + """A --dictionary-file session override must not move the corrections file.""" + xdg_dir = tmp_path / "xdg-config" + xdg_dir.mkdir() + monkeypatch.setattr("vocalinux.custom_dictionary.config_dir", lambda: str(xdg_dir)) + override = tmp_path / "override.txt" + override.write_text("Override\n", encoding="utf-8") + + manager = CustomDictionaryManager(FakeConfig(), str(override)) + + assert manager.corrections_path() == xdg_dir / CORRECTIONS_FILENAME + assert manager.corrections_path() != tmp_path / CORRECTIONS_FILENAME + + +def test_transient_terms_override_does_not_import_beside_corrections( + tmp_path: Path, monkeypatch +) -> None: + """A --dictionary-file session must not import the override's neighbor file.""" + xdg_dir = tmp_path / "xdg-config" + override_dir = tmp_path / "elsewhere" + xdg_dir.mkdir() + override_dir.mkdir() + monkeypatch.setattr("vocalinux.custom_dictionary.config_dir", lambda: str(xdg_dir)) + override = override_dir / "terms.txt" + override.write_text("Override\n", encoding="utf-8") + payload = { + "version": 1, + "corrections": [{"heard": "super base", "replacement": "Supabase"}], + } + (override_dir / CORRECTIONS_FILENAME).write_text(json.dumps(payload) + "\n", encoding="utf-8") + + manager = CustomDictionaryManager(FakeConfig(), str(override)) + + assert manager.get_corrections() == [] + assert manager.get_corrections_for_edit() == [] + assert not (xdg_dir / CORRECTIONS_FILENAME).exists() + + +def test_historical_legacy_looking_path_without_explicit_marker_is_not_migrated( + tmp_path: Path, monkeypatch +) -> None: + """A leftover pre-XDG path stays put when XDG_CONFIG_HOME points elsewhere.""" + monkeypatch.setattr("vocalinux.custom_dictionary.config_dir", lambda: str(tmp_path)) + xdg_path = str(tmp_path / TERMS_FILENAME) + tilde_path = LEGACY_DEFAULT_TERMS_PATH + expanded_legacy = str(Path.home() / ".config" / "vocalinux" / TERMS_FILENAME) + assert expanded_legacy != xdg_path + + config = FakeConfig({"dictionary": {"file_path": tilde_path}}) + manager = CustomDictionaryManager(config) + + assert manager.terms_path_text() == tilde_path + assert manager.terms_path() == Path(tilde_path).expanduser() + assert config.get("dictionary", "file_path") == tilde_path + assert config.get("dictionary", "file_path") != xdg_path + assert config.get("dictionary", "file_path_explicit") is True + + failing = FakeConfig( + {"dictionary": {"file_path": tilde_path}}, + save_result=False, + ) + failing_manager = CustomDictionaryManager(failing) + assert failing_manager.terms_path_text() == tilde_path + assert failing.get("dictionary", "file_path") == tilde_path + assert failing.get("dictionary", "file_path") != xdg_path + assert not failing.get("dictionary", "file_path_explicit", False) + + expanded_config = FakeConfig({"dictionary": {"file_path": expanded_legacy}}) + expanded_manager = CustomDictionaryManager(expanded_config) + assert expanded_manager.terms_path_text() == expanded_legacy + assert expanded_config.get("dictionary", "file_path") == expanded_legacy + assert expanded_config.get("dictionary", "file_path") != xdg_path + assert expanded_config.get("dictionary", "file_path_explicit") is True + + custom = str(tmp_path / "custom-terms.txt") + custom_config = FakeConfig({"dictionary": {"file_path": custom}}) + custom_manager = CustomDictionaryManager(custom_config) + assert custom_manager.terms_path_text() == custom + assert custom_config.get("dictionary", "file_path") == custom + + +def test_explicit_legacy_terms_path_is_not_migrated(tmp_path: Path, monkeypatch) -> None: + """set_terms_path of the pre-XDG string is kept when XDG points elsewhere.""" + monkeypatch.setattr("vocalinux.custom_dictionary.config_dir", lambda: str(tmp_path)) + xdg_path = str(tmp_path / TERMS_FILENAME) + expanded_legacy = str(Path.home() / ".config" / "vocalinux" / TERMS_FILENAME) + assert expanded_legacy != xdg_path + + tilde_config = FakeConfig({"dictionary": {"file_path": xdg_path}}) + tilde_manager = CustomDictionaryManager(tilde_config) + assert tilde_manager.set_terms_path(LEGACY_DEFAULT_TERMS_PATH) + assert tilde_config.get("dictionary", "file_path") == LEGACY_DEFAULT_TERMS_PATH + assert tilde_config.get("dictionary", "file_path_explicit") is True + assert tilde_manager.terms_path_text() == LEGACY_DEFAULT_TERMS_PATH + assert tilde_config.get("dictionary", "file_path") == LEGACY_DEFAULT_TERMS_PATH + + expanded_config = FakeConfig({"dictionary": {"file_path": xdg_path}}) + expanded_manager = CustomDictionaryManager(expanded_config) + assert expanded_manager.set_terms_path(expanded_legacy) + assert expanded_config.get("dictionary", "file_path") == expanded_legacy + assert expanded_config.get("dictionary", "file_path_explicit") is True + assert expanded_manager.terms_path_text() == expanded_legacy + assert expanded_config.get("dictionary", "file_path") == expanded_legacy + + +def test_invalid_or_unreadable_configured_terms_path_is_not_persisted( + tmp_path: Path, monkeypatch +) -> None: + """Invalid paths are harmless and leave the previous configured path intact.""" + previous_path = str(tmp_path / TERMS_FILENAME) + config = FakeConfig({"dictionary": {"file_path": previous_path}}) + manager = manager_at(tmp_path, monkeypatch, config) + + assert not manager.set_terms_path("~vocalinux-user-does-not-exist/dictionary.txt") + assert config.get("dictionary", "file_path") == previous_path + + assert not manager.set_terms_path(str(tmp_path)) + assert config.get("dictionary", "file_path") == previous_path + + +def test_invalid_saved_terms_path_is_safe_to_read_and_report(tmp_path: Path, monkeypatch) -> None: + """A pre-existing unresolved path cannot crash recognition or Settings.""" + config = FakeConfig( + { + "dictionary": { + "enabled": True, + "file_path": "~vocalinux-user-does-not-exist/dictionary.txt", + } + } + ) + manager = manager_at(tmp_path, monkeypatch, config) + + assert manager.terms_path() is None + assert manager.get_terms() == [] + assert manager.build_initial_prompt() is None + assert manager.terms_status() == "Configured terms path is invalid or cannot be expanded." + + +def test_terms_path_save_failure_preserves_the_previous_configured_path( + tmp_path: Path, monkeypatch +) -> None: + """A failed config save never claims a new path persisted.""" + previous_path = str(tmp_path / TERMS_FILENAME) + config = FakeConfig({"dictionary": {"file_path": previous_path}}, save_result=False) + manager = manager_at(tmp_path, monkeypatch, config) + + assert not manager.set_terms_path(str(tmp_path / "new-terms.txt")) + assert config.get("dictionary", "file_path") == previous_path + assert not config.get("dictionary", "file_path_explicit", False) + + +def test_invalid_transient_terms_path_does_not_crash(tmp_path: Path, monkeypatch) -> None: + """An unresolved CLI tilde path disables only its session's terms safely.""" + manager_at(tmp_path, monkeypatch) + manager = CustomDictionaryManager(FakeConfig(), "~vocalinux-user-does-not-exist/dictionary.txt") + + assert manager.terms_enabled() + assert manager.terms_path() is None + assert manager.get_terms() == [] + assert manager.terms_status() == "Configured terms path is invalid or cannot be expanded." + + +def test_atomic_save_failure_preserves_existing_terms_file(tmp_path: Path, monkeypatch) -> None: + """A failed replacement leaves the scanner-visible file intact.""" + manager = manager_at(tmp_path, monkeypatch) + terms_file = tmp_path / TERMS_FILENAME + terms_file.write_text("Existing\n", encoding="utf-8") + + def fail_replace(source: Path, destination: Path) -> None: + """Simulate a filesystem failure during atomic replacement.""" + raise OSError("disk full") + + monkeypatch.setattr("vocalinux.custom_dictionary.os.replace", fail_replace) + + assert not manager.save_terms(["New"]) + assert terms_file.read_text(encoding="utf-8") == "Existing\n" + + +def test_file_edits_preserve_existing_permissions(tmp_path: Path, monkeypatch) -> None: + """Replacing either dictionary file keeps the mode the owner set on it.""" + manager = manager_at(tmp_path, monkeypatch) + + terms_file = tmp_path / TERMS_FILENAME + terms_file.write_text("VocaLinux\n", encoding="utf-8") + terms_file.chmod(0o600) + assert manager.save_terms(["Supabase"]) + assert stat.S_IMODE(terms_file.stat().st_mode) == 0o600 + + corrections_file = tmp_path / CORRECTIONS_FILENAME + corrections_file.write_text('{"version": 1, "corrections": []}\n', encoding="utf-8") + corrections_file.chmod(0o640) + assert manager.save_corrections([{"heard": "a", "replacement": "b"}]) + assert stat.S_IMODE(corrections_file.stat().st_mode) == 0o640 + + +def test_dictionary_file_argument_is_available_for_a_session_override(monkeypatch) -> None: + """The retained CLI option accepts a terms file without changing config.""" + monkeypatch.setattr("sys.argv", ["vocalinux", "--dictionary-file", "/tmp/terms.txt"]) + + assert parse_arguments().dictionary_file == "/tmp/terms.txt" + + +def test_corrections_use_versioned_json_and_live_reload(tmp_path: Path, monkeypatch) -> None: + """Structured corrections use exact replacement spelling on every read.""" + manager = manager_at(tmp_path, monkeypatch) + assert manager.save_corrections([{"heard": "super base", "replacement": "Supabase"}]) + stored = json.loads((tmp_path / CORRECTIONS_FILENAME).read_text(encoding="utf-8")) + assert stored == { + "version": 1, + "corrections": [{"heard": "super base", "replacement": "Supabase"}], + } + assert manager.get_corrections_for_edit() == [ + {"heard": "super base", "replacement": "Supabase"} + ] + assert manager.apply_corrections("super base is useful") == "Supabase is useful" + + assert manager.save_corrections([{"heard": "pie object", "replacement": "PyGObject"}]) + assert manager.apply_corrections("pie object") == "PyGObject" + + +def test_corrections_are_literal_longest_first_and_unicode_safe() -> None: + """Matching is whole-phrase, case-insensitive, and never maps through lower().""" + entries = normalize_corrections( + [ + {"heard": "super", "replacement": "short"}, + {"heard": "super base", "replacement": "Supabase"}, + {"heard": "C++", "replacement": "C plus plus"}, + {"heard": "i", "replacement": "letter"}, + ] + ) + + assert apply_corrections("SUPER BASE and super", entries) == "Supabase and short" + assert apply_corrections("C++ is not C++foo", entries) == "C plus plus is not C++foo" + assert apply_corrections("İ", entries) == "letter" + assert apply_corrections("e\u0301clair", [{"heard": "e", "replacement": "E"}]) == "éclair" + + +def test_invalid_corrections_file_fails_closed(tmp_path: Path, monkeypatch) -> None: + """Malformed JSON never aborts a segment or invents replacements.""" + manager = manager_at(tmp_path, monkeypatch) + (tmp_path / CORRECTIONS_FILENAME).write_text("{not json", encoding="utf-8") + + assert manager.get_corrections() == [] + assert manager.get_corrections_for_edit() is None + assert manager.apply_corrections("super base") == "super base" + + +def test_partially_invalid_corrections_cannot_be_destructively_edited( + tmp_path: Path, monkeypatch +) -> None: + """UI write-back is refused when runtime reads had to filter source entries.""" + manager = manager_at(tmp_path, monkeypatch) + path = tmp_path / CORRECTIONS_FILENAME + original = { + "version": 1, + "corrections": [ + {"heard": "super base", "replacement": "Supabase"}, + {"heard": "missing replacement"}, + ], + } + path.write_text(json.dumps(original), encoding="utf-8") + + assert manager.get_corrections() == [{"heard": "super base", "replacement": "Supabase"}] + assert manager.get_corrections_for_edit() is None + assert json.loads(path.read_text(encoding="utf-8")) == original + + +def test_corrections_with_extra_fields_cannot_be_destructively_edited( + tmp_path: Path, monkeypatch +) -> None: + """Unknown fields still apply at runtime but block a UI rewrite that would drop them.""" + manager = manager_at(tmp_path, monkeypatch) + path = tmp_path / CORRECTIONS_FILENAME + original = { + "version": 1, + "corrections": [ + { + "heard": "super base", + "replacement": "Supabase", + "note": "scanner annotation", + } + ], + } + path.write_text(json.dumps(original), encoding="utf-8") + + assert manager.get_corrections() == [{"heard": "super base", "replacement": "Supabase"}] + assert manager.apply_corrections("super base is useful") == "Supabase is useful" + assert manager.get_corrections_for_edit() is None + assert json.loads(path.read_text(encoding="utf-8")) == original + + +def test_corrections_with_top_level_metadata_cannot_be_destructively_edited( + tmp_path: Path, monkeypatch +) -> None: + """Unknown top-level keys apply at runtime but block a UI rewrite that would drop them.""" + manager = manager_at(tmp_path, monkeypatch) + path = tmp_path / CORRECTIONS_FILENAME + original = { + "version": 1, + "metadata": {"author": "scanner"}, + "corrections": [{"heard": "super base", "replacement": "Supabase"}], + } + path.write_text(json.dumps(original), encoding="utf-8") + + assert manager.get_corrections() == [{"heard": "super base", "replacement": "Supabase"}] + assert manager.apply_corrections("super base is useful") == "Supabase is useful" + assert manager.get_corrections_for_edit() is None + assert json.loads(path.read_text(encoding="utf-8")) == original + + +def test_corrections_with_whitespace_cannot_be_destructively_edited( + tmp_path: Path, monkeypatch +) -> None: + """Leading or trailing whitespace still applies at runtime but blocks a UI rewrite.""" + manager = manager_at(tmp_path, monkeypatch) + path = tmp_path / CORRECTIONS_FILENAME + original = { + "version": 1, + "corrections": [{"heard": " super base ", "replacement": " Supabase "}], + } + path.write_text(json.dumps(original), encoding="utf-8") + + assert manager.get_corrections() == [{"heard": "super base", "replacement": "Supabase"}] + assert manager.apply_corrections("super base is useful") == "Supabase is useful" + assert manager.get_corrections_for_edit() is None + assert json.loads(path.read_text(encoding="utf-8")) == original + + +def test_corrections_with_decomposed_unicode_cannot_be_destructively_edited( + tmp_path: Path, monkeypatch +) -> None: + """NFD heard values still apply at runtime but block a UI rewrite to NFC.""" + manager = manager_at(tmp_path, monkeypatch) + path = tmp_path / CORRECTIONS_FILENAME + original = { + "version": 1, + "corrections": [{"heard": "e\u0301clair", "replacement": "éclair"}], + } + path.write_text(json.dumps(original, ensure_ascii=False), encoding="utf-8") + + assert manager.get_corrections() == [{"heard": "éclair", "replacement": "éclair"}] + assert manager.apply_corrections("e\u0301clair") == "éclair" + assert manager.get_corrections_for_edit() is None + assert json.loads(path.read_text(encoding="utf-8")) == original + + +def test_corrections_with_duplicate_keys_cannot_be_destructively_edited( + tmp_path: Path, monkeypatch +) -> None: + """Duplicate JSON keys still apply at runtime but block a UI rewrite that would drop them.""" + manager = manager_at(tmp_path, monkeypatch) + path = tmp_path / CORRECTIONS_FILENAME + cases = [ + ( + "{\n" + ' "version": 1,\n' + ' "corrections": [{"heard": "pie object", "replacement": "PyGObject"}],\n' + ' "corrections": [{"heard": "super base", "replacement": "Supabase"}]\n' + "}\n", + [{"heard": "super base", "replacement": "Supabase"}], + ), + ( + "{\n" + ' "version": 1,\n' + ' "corrections": [' + '{"heard": "super base", "replacement": "Wrong", "replacement": "Supabase"}' + "]\n}\n", + [{"heard": "super base", "replacement": "Supabase"}], + ), + ] + for original, expected_runtime in cases: + path.write_text(original, encoding="utf-8") + before = path.read_bytes() + + assert manager.get_corrections() == expected_runtime + assert manager.apply_corrections("super base is useful") == "Supabase is useful" + assert manager.get_corrections_for_edit() is None + assert path.read_bytes() == before + + +def test_invalid_correction_update_does_not_wipe_existing_entry( + tmp_path: Path, monkeypatch +) -> None: + """A too-long replacement must not delete the previous same-heard correction.""" + import importlib + import sys + + import vocalinux.ui as ui_pkg + + manager = manager_at(tmp_path, monkeypatch) + original = [{"heard": "super base", "replacement": "Supabase"}] + assert manager.save_corrections(original) + corrections_path = tmp_path / CORRECTIONS_FILENAME + before = corrections_path.read_text(encoding="utf-8") + + repository = sys.modules["gi.repository"] + bases = {name: type(name, (), {}) for name in ("Box", "ListBoxRow", "Dialog")} + saved = sys.modules.pop("vocalinux.ui.settings_dialog", None) + try: + with patch.object(repository, "Gtk", MagicMock(**bases)): + settings_dialog = importlib.import_module("vocalinux.ui.settings_dialog") + finally: + sys.modules.pop("vocalinux.ui.settings_dialog", None) + if saved is not None: + sys.modules["vocalinux.ui.settings_dialog"] = saved + ui_pkg.settings_dialog = saved + else: + ui_pkg.__dict__.pop("settings_dialog", None) + + dialog = Mock() + dialog._initializing = False + dialog._applying_settings = False + dialog._dictionary_available.return_value = True + dialog.dictionary_manager = manager + dialog.dictionary_heard_entry.get_text.return_value = "super base" + dialog.dictionary_replacement_entry.get_text.return_value = "x" * ( + MAX_CORRECTION_CHARACTERS + 1 + ) + + settings_dialog.SettingsDialog._on_dictionary_add_correction(dialog, None) + + assert corrections_path.read_text(encoding="utf-8") == before + assert manager.get_corrections() == original + dialog.dictionary_heard_entry.set_text.assert_not_called() + dialog.dictionary_replacement_entry.set_text.assert_not_called() + dialog.dictionary_feedback_label.set_text.assert_called_once_with( + "That correction is not valid for the file." + ) + + +def test_nfd_correction_update_replaces_nfc_entry(tmp_path: Path, monkeypatch) -> None: + """A decomposed heard phrase must update the existing NFC correction.""" + import importlib + import sys + + import vocalinux.ui as ui_pkg + + manager = manager_at(tmp_path, monkeypatch) + original = [{"heard": "éclair", "replacement": "OLD"}] + assert manager.save_corrections(original) + corrections_path = tmp_path / CORRECTIONS_FILENAME + + repository = sys.modules["gi.repository"] + bases = {name: type(name, (), {}) for name in ("Box", "ListBoxRow", "Dialog")} + saved = sys.modules.pop("vocalinux.ui.settings_dialog", None) + try: + with patch.object(repository, "Gtk", MagicMock(**bases)): + settings_dialog = importlib.import_module("vocalinux.ui.settings_dialog") + finally: + sys.modules.pop("vocalinux.ui.settings_dialog", None) + if saved is not None: + sys.modules["vocalinux.ui.settings_dialog"] = saved + ui_pkg.settings_dialog = saved + else: + ui_pkg.__dict__.pop("settings_dialog", None) + + dialog = Mock() + dialog._initializing = False + dialog._applying_settings = False + dialog._dictionary_available.return_value = True + dialog.dictionary_manager = manager + dialog.dictionary_heard_entry.get_text.return_value = "e\u0301clair" + dialog.dictionary_replacement_entry.get_text.return_value = "NEW" + + settings_dialog.SettingsDialog._on_dictionary_add_correction(dialog, None) + + stored = json.loads(corrections_path.read_text(encoding="utf-8")) + assert stored == { + "version": 1, + "corrections": [{"heard": "éclair", "replacement": "NEW"}], + } + assert manager.get_corrections() == [{"heard": "éclair", "replacement": "NEW"}] + dialog.dictionary_feedback_label.set_text.assert_called_once_with("Correction updated.") + + +def test_non_integer_corrections_version_fails_closed(tmp_path: Path, monkeypatch) -> None: + """Boolean, float, and string versions must not be treated as schema 1.""" + manager = manager_at(tmp_path, monkeypatch) + path = tmp_path / CORRECTIONS_FILENAME + cases = ( + '{"version": true, "corrections": [{"heard": "super base", "replacement": "Supabase"}]}\n', + '{"version": 1.0, "corrections": [{"heard": "super base", "replacement": "Supabase"}]}\n', + '{"version": "1", "corrections": [{"heard": "super base", "replacement": "Supabase"}]}\n', + '{"version": 1e0, "corrections": [{"heard": "super base", "replacement": "Supabase"}]}\n', + ) + for original in cases: + path.write_text(original, encoding="utf-8") + before = path.read_bytes() + + assert manager.get_corrections() == [] + assert manager.get_corrections_for_edit() is None + assert path.read_bytes() == before + + +def test_invalid_terms_file_is_ignored_and_explained(tmp_path: Path, monkeypatch) -> None: + """An invalid scanner file leaves prompt bias empty with a clear status.""" + manager = manager_at(tmp_path, monkeypatch, FakeConfig({"dictionary": {"enabled": True}})) + (tmp_path / TERMS_FILENAME).write_bytes(b"\xff\xfe") + + assert manager.build_initial_prompt() is None + assert manager.terms_status() == "Terms file is not valid UTF-8." + + +def test_whisper_receives_the_live_custom_terms_prompt(tmp_path: Path, monkeypatch) -> None: + """Whisper gets the current file contents as initial_prompt at transcription time.""" + dictionary = manager_at(tmp_path, monkeypatch, FakeConfig({"dictionary": {"enabled": True}})) + (tmp_path / TERMS_FILENAME).write_text("VocaLinux\n", encoding="utf-8") + manager = recognition_manager("whisper", dictionary) + manager.model = MagicMock() + manager.model.transcribe.return_value = {"text": "ok"} + manager.model.device = MagicMock() + + with patch.dict("sys.modules", {"numpy": mock_numpy(), "torch": MagicMock()}): + assert manager._transcribe_with_whisper([b"\x00\x00"]) == "ok" + + assert manager.model.transcribe.call_args.kwargs["initial_prompt"] == "VocaLinux" + + +def test_faster_whisper_receives_the_live_custom_terms_prompt(tmp_path: Path, monkeypatch) -> None: + """Faster Whisper receives the same live vocabulary bias as OpenAI Whisper.""" + dictionary = manager_at(tmp_path, monkeypatch, FakeConfig({"dictionary": {"enabled": True}})) + (tmp_path / TERMS_FILENAME).write_text("VocaLinux\n", encoding="utf-8") + manager = object.__new__(SpeechRecognitionManager) + manager.dictionary_manager = dictionary + manager._faster_whisper_engine = MagicMock() + manager._faster_whisper_engine.is_ready.return_value = True + manager._faster_whisper_engine.transcribe.return_value = "ok" + + assert manager._transcribe_with_faster_whisper([b"\x00\x00"]) == "ok" + manager._faster_whisper_engine.transcribe.assert_called_once_with( + [b"\x00\x00"], language="auto", initial_prompt="VocaLinux" + ) + + +def test_whispercpp_composes_and_clears_live_terms_prompt(tmp_path: Path, monkeypatch) -> None: + """whisper.cpp keeps Advanced context and explicitly clears a stale prompt.""" + config = FakeConfig({"dictionary": {"enabled": True}}) + dictionary = manager_at(tmp_path, monkeypatch, config) + terms_file = tmp_path / TERMS_FILENAME + terms_file.write_text("VocaLinux\n", encoding="utf-8") + manager = recognition_manager("whisper_cpp", dictionary, advanced_prompt="Explicit context") + manager._model_lock = threading.Lock() + manager.model = MagicMock() + manager.model.transcribe.return_value = [MagicMock(text="ok")] + + with patch.dict("sys.modules", {"numpy": mock_numpy()}): + assert manager._transcribe_with_whispercpp([b"\x00\x00"]) == "ok" + config.values["dictionary"]["enabled"] = False + assert manager._transcribe_with_whispercpp([b"\x00\x00"]) == "ok" + manager.whispercpp_initial_prompt = "" + assert manager._transcribe_with_whispercpp([b"\x00\x00"]) == "ok" + + prompts = [call.kwargs["initial_prompt"] for call in manager.model.transcribe.call_args_list] + assert prompts == ["Explicit context VocaLinux", "Explicit context", ""] + + +def test_vosk_warns_once_for_bias_while_corrections_remain_available( + tmp_path: Path, monkeypatch +) -> None: + """VOSK rejects vocabulary bias only; its transcript can still be corrected.""" + dictionary = manager_at(tmp_path, monkeypatch, FakeConfig({"dictionary": {"enabled": True}})) + (tmp_path / TERMS_FILENAME).write_text("VocaLinux\n", encoding="utf-8") + manager = recognition_manager("vosk", dictionary) + manager._get_vosk_model_path = MagicMock(return_value="/models/vosk") + vosk = MagicMock() + + with patch.dict("sys.modules", {"vosk": vosk}): + with patch( + "vocalinux.speech_recognition.recognition_manager.os.path.exists", return_value=True + ): + with patch( + "vocalinux.speech_recognition.recognition_manager.logger.warning" + ) as warning: + manager._init_vosk() + manager._init_vosk() + + warning.assert_called_once_with( + "Custom terms are ignored by VOSK; transcript corrections still apply." + ) + + +def test_whispercpp_prompt_composes_advanced_and_terms() -> None: + """Advanced prompt remains a prefix and an emptied terms file clears the value.""" + recognition_manager = object.__new__(SpeechRecognitionManager) + recognition_manager.whispercpp_initial_prompt = "Write product names accurately." + recognition_manager.dictionary_manager = type( + "Dictionary", (), {"build_initial_prompt": lambda self: "VocaLinux"} + )() + + assert ( + recognition_manager._get_whispercpp_prompt() == "Write product names accurately. VocaLinux" + ) + recognition_manager.dictionary_manager = type( + "Dictionary", (), {"build_initial_prompt": lambda self: None} + )() + assert recognition_manager._get_whispercpp_prompt() == "Write product names accurately." + + +def test_corrections_run_before_voice_commands() -> None: + """A correction changes raw text before the command processor can act on it.""" + + class Recognizer: + """Minimal VOSK recognizer.""" + + def AcceptWaveform(self, data: bytes) -> None: + """Accept a fake audio chunk.""" + + def FinalResult(self) -> str: + """Return the raw model text.""" + return '{"text": "delete that"}' + + recognition_manager = object.__new__(SpeechRecognitionManager) + recognition_manager.engine = "vosk" + recognition_manager.recognizer = Recognizer() + recognition_manager._model_lock = threading.Lock() + recognition_manager._voice_commands_enabled = True + recognition_manager.dictionary_manager = type( + "Dictionary", (), {"apply_corrections": lambda self, text: "delete note"} + )() + received: list[str] = [] + recognition_manager.command_processor = type( + "Commands", (), {"process_text": lambda self, text: (text, [])} + )() + recognition_manager.text_callbacks = [received.append] + recognition_manager.action_callbacks = [] + recognition_manager.segment_callbacks = [] + + recognition_manager._process_audio_buffer([b"audio"]) + + assert received == ["delete note"] + + +def test_correction_stage_is_shared_by_every_engine() -> None: + """The common post-transcription stage is available to local and remote engines.""" + for engine in ( + "vosk", + "whisper", + "whisper_cpp", + "faster_whisper", + "parakeet", + "remote_api", + ): + manager = object.__new__(SpeechRecognitionManager) + manager.engine = engine + manager.dictionary_manager = type( + "Dictionary", (), {"apply_corrections": lambda self, text: "Supabase"} + )() + + assert manager._apply_dictionary_corrections("super base") == "Supabase" diff --git a/tests/test_faster_whisper_engine.py b/tests/test_faster_whisper_engine.py index 4f2d8c0..a21d6a4 100644 --- a/tests/test_faster_whisper_engine.py +++ b/tests/test_faster_whisper_engine.py @@ -323,6 +323,21 @@ class TestFasterWhisperEngine: assert text == "Hello, world." + def test_transcribe_passes_initial_prompt(self) -> None: + """Test that vocabulary bias is forwarded to faster-whisper.""" + whisper_mock = self._mock_whisper_model([MagicMock(text="VocaLinux")]) + with patch.dict(sys.modules, {"faster_whisper": whisper_mock}): + engine = FasterWhisperEngine(model_size="tiny", device="cpu") + engine.init() + + audio = np.array([0, 1000, -1000, 0], dtype=np.int16) + engine.transcribe([audio.tobytes()], initial_prompt="VocaLinux") + + assert ( + whisper_mock.WhisperModel.return_value.transcribe.call_args.kwargs["initial_prompt"] + == "VocaLinux" + ) + def test_transcribe_empty_audio(self): """Test that empty audio returns empty text.""" whisper_mock = self._mock_whisper_model([]) diff --git a/tests/test_main.py b/tests/test_main.py index bbdc5c2..863c51c 100644 --- a/tests/test_main.py +++ b/tests/test_main.py @@ -3,6 +3,7 @@ Tests for the main module functionality. """ import argparse +import queue import sys import threading import time @@ -34,6 +35,7 @@ class TestMainModule(unittest.TestCase): self.assertIsNone(args.model) # No default set, loaded from config instead self.assertIsNone(args.engine) self.assertIsNone(args.language) + self.assertIsNone(args.dictionary_file) self.assertFalse(args.wayland) self.assertFalse(args.start_minimized) @@ -248,6 +250,7 @@ class TestMainModule(unittest.TestCase): whispercpp_no_speech_thold=0.6, whispercpp_n_threads=0, whispercpp_gpu_device=None, + dictionary_manager=ANY, remote_api_url="", remote_api_key="", remote_api_endpoint="/inference", @@ -1694,25 +1697,91 @@ class TestSessionHistoryRecording(unittest.TestCase): finally: stack.close() + def _worker_deliverer( + self, segment_cb: Callable[[str, float], None] + ) -> Tuple[Callable[[str], None], Callable[[], None]]: + """Deliver segments from one persistent worker thread, like the real engine. + + Returns (deliver, close): ``deliver(text)`` runs the segment callback + on the worker synchronously; ``close()`` stops it. + """ + work: queue.Queue = queue.Queue() + errors: list[BaseException] = [] + + def worker() -> None: + while True: + job = work.get() + try: + if job is not None: + job() + except BaseException as error: + errors.append(error) + finally: + work.task_done() + if job is None: + return + + thread = threading.Thread(target=worker) + thread.start() + + def deliver(text: str) -> None: + work.put(lambda: segment_cb(text, time.monotonic())) + work.join() + if errors: + raise errors.pop(0) + + def close() -> None: + work.put(None) + work.join() + thread.join() + + return deliver, close + def test_late_segment_from_old_worker_during_next_session(self) -> None: """A leftover worker delivering on its own thread stays out of the open session.""" stack, _, segment_cb, state_cb, history = self._boot() + on_old_worker, close_old_worker = self._worker_deliverer(segment_cb) try: state_cb(RecognitionState.LISTENING) - segment_cb("one", time.monotonic()) + on_old_worker("one") state_cb(RecognitionState.IDLE) state_cb(RecognitionState.LISTENING) segment_cb("two", time.monotonic()) # The previous session's worker finally delivers on its own thread. - t = threading.Thread(target=segment_cb, args=("trailing", time.monotonic())) - t.start() - t.join() + on_old_worker("trailing") self.assertEqual(history.get_all(), ["one trailing"]) state_cb(RecognitionState.IDLE) self.assertEqual(history.get_all(), ["two", "one trailing"]) finally: + close_old_worker() + stack.close() + + def test_straggler_after_newer_session_forms_own_snippet(self) -> None: + """An old worker finishing after the next session must not grow its entry.""" + stack, _, segment_cb, state_cb, history = self._boot() + on_old_worker, close_old_worker = self._worker_deliverer(segment_cb) + try: + state_cb(RecognitionState.LISTENING) + on_old_worker("first") + state_cb(RecognitionState.IDLE) + + state_cb(RecognitionState.LISTENING) + segment_cb("second", time.monotonic()) + state_cb(RecognitionState.IDLE) + + # The older session's worker decodes last: its text belongs to + # the first session, not to the currently-newest entry. + on_old_worker("late tail") + self.assertEqual(history.get_all(), ["late tail", "second", "first"]) + + # A later straggler on that same worker still extends the + # snippet it owns. + on_old_worker("more tail") + self.assertEqual(history.get_all(), ["late tail more tail", "second", "first"]) + finally: + close_old_worker() stack.close() def test_session_without_segments_creates_no_snippet(self) -> None: diff --git a/tests/test_record_during_reload.py b/tests/test_record_during_reload.py index ab59325..868bff4 100644 --- a/tests/test_record_during_reload.py +++ b/tests/test_record_during_reload.py @@ -2,6 +2,7 @@ import sys import threading +import time from collections.abc import Generator from unittest.mock import Mock @@ -90,6 +91,46 @@ def test_release_before_reload_finishes_preserves_whole_recording(session: Sessi assert not manager.is_idle_unloaded +def test_worker_exit_during_release_join_still_transcribes_final_buffer( + session: Session, monkeypatch: pytest.MonkeyPatch +) -> None: + """A release whose final enqueue lands after the worker exits is kept. + + stop_recognition joins the audio thread with a bounded wait; while it + waits, the recognition worker can notice ``should_record`` is false and + exit. The final buffer it then enqueues must still be transcribed by + the reload teardown instead of being thrown away with the queue. + """ + manager, captured, loading, release_load, transcribed = session + capture_release = threading.Event() + + def slow_record() -> None: + with manager._buffer_lock: + manager.audio_buffer = [b"final speech"] + manager._recording_segment_has_speech = True + captured.set() + assert capture_release.wait(3) + + monkeypatch.setattr(manager, "_record_audio", slow_record) + assert manager.start_recognition(mode="push_to_talk") + assert captured.wait(1) + assert loading.wait(1) + release_load.set() # Model loads; the worker starts polling an empty queue. + + stopper = threading.Thread(target=manager.stop_recognition) + stopper.start() + # The worker's 0.5s second-chance window elapses while the stop path is + # still joining the audio thread, so the tail buffer is enqueued only + # after the worker has already exited. + time.sleep(1.2) + capture_release.set() + stopper.join(3) + manager.recognition_thread.join(3) + + assert transcribed == [[b"final speech"]] + assert manager.state == RecognitionState.IDLE + + def test_toggle_transcribes_queued_speech_after_reload(session: Session) -> None: manager, captured, loading, release_load, transcribed = session assert manager.start_recognition() diff --git a/tests/test_settings_dialog.py b/tests/test_settings_dialog.py index 9f5c88a..b1e4a62 100644 --- a/tests/test_settings_dialog.py +++ b/tests/test_settings_dialog.py @@ -1081,9 +1081,10 @@ class TestSettingsNavigation(unittest.TestCase): self.assertIn("self.sidebar_listbox = Gtk.ListBox()", self.source_code) self.assertNotIn("Gtk.Notebook()", self.source_code) - def test_topic_pages_exist(self): + def test_topic_pages_exist(self) -> None: for name, title in [ ("dictation", "Dictation"), + ("dictionary", "Custom Dictionary"), ("model", "Speech Model"), ("audio", "Audio"), ("performance", "Performance"), @@ -1092,6 +1093,108 @@ class TestSettingsNavigation(unittest.TestCase): ]: self.assertIn(f'SettingsPage("{name}", "{title}"', self.source_code) + def test_custom_dictionary_page_has_both_accessible_subsections(self) -> None: + """The dictionary page keeps terms and corrections distinct and accessible.""" + body = self.source_code.split("def _build_dictionary_section")[1].split("\n def ")[0] + self.assertIn('title="Custom terms"', body) + self.assertIn('title="Transcript corrections"', body) + self.assertIn("Vocabulary bias works with Whisper, whisper.cpp, and Faster Whisper", body) + self.assertIn("self.dictionary_management_switcher = Gtk.StackSwitcher()", body) + self.assertIn( + "self.dictionary_tab.pack_start(self.dictionary_management_stack, True, True, 0)", + body, + ) + self.assertIn('terms_scroller, "terms", "Custom terms"', body) + self.assertIn('corrections_scroller, "corrections", "Corrections"', body) + for accessible_name in [ + "Custom terms file", + "Custom term", + "Heard phrase", + "Replacement text", + "Custom dictionary status", + ]: + self.assertIn(accessible_name, body) + # Shared status lives on the page, not only in the Corrections stack child. + self.assertIn( + "self.dictionary_tab.pack_start(self.dictionary_feedback_label, False, False, 0)", + body, + ) + switcher_pack = body.find( + "self.dictionary_tab.pack_start(self.dictionary_management_switcher" + ) + feedback_pack = body.find("self.dictionary_tab.pack_start(self.dictionary_feedback_label") + stack_pack = body.find("self.dictionary_tab.pack_start(self.dictionary_management_stack") + self.assertLess(switcher_pack, feedback_pack) + self.assertLess(feedback_pack, stack_pack) + corrections_child = body[ + body.find("corrections_group = PreferencesGroup") : body.find( + 'corrections_scroller, "corrections"' + ) + ] + self.assertNotIn("dictionary_feedback_label", corrections_child) + + def test_custom_dictionary_cards_preserve_rounded_bottom_corners(self) -> None: + """Transparent list backgrounds do not cover the card's lower radius.""" + self.assertIn(".preferences-group-list", self.source_code) + self.assertIn('add_class("preferences-group-list")', self.source_code) + self.assertIn("border-radius: 0 0 11px 11px", self.source_code) + + def test_custom_dictionary_path_and_persistence_handlers_are_present(self) -> None: + """Path chooser and failure feedback keep existing settings safe.""" + self.assertIn('Gtk.FileChooserButton(title="Choose Terms File")', self.source_code) + self.assertIn("def _on_dictionary_terms_path_changed", self.source_code) + self.assertIn("set_terms_path", self.source_code) + self.assertIn("Could not save that custom terms path", self.source_code) + self.assertIn("add_term(term)", self.source_code) + self.assertIn("remove_term(term)", self.source_code) + + def test_correction_add_validates_candidate_before_save(self) -> None: + """Invalid replacements must be rejected before the list is rewritten.""" + body = self.source_code.split("def _on_dictionary_add_correction")[1].split("\n def ")[0] + self.assertIn("normalize_corrections", body) + self.assertLess(body.find("normalize_corrections"), body.find("save_corrections")) + self.assertLess(body.find("normalize_corrections"), body.find("entries.append")) + + def test_terms_list_builds_in_bounded_idle_slices_with_pinned_adds(self) -> None: + """Large term files never build all their GTK rows synchronously. + + The stall finding requires two guards to hold together: the row + build is sliced through GLib idle callbacks, and a saved term that + lands past the display cap is pinned so success is never reported + for an invisible term. + """ + self.assertIn("_TERMS_ROWS_PER_IDLE", self.source_code) + refresh_body = self.source_code.split("def _refresh_dictionary_ui")[1].split("\n def ")[ + 0 + ] + self.assertNotIn("for term in visible_terms:", refresh_body) + self.assertIn("self._build_term_rows(", refresh_body) + build_body = self.source_code.split("def _build_term_rows")[1].split("\n def ")[0] + self.assertIn("GLib.idle_add", build_body) + self.assertIn("_TERMS_ROWS_PER_IDLE", build_body) + self.assertIn("self._terms_build_token", build_body) + add_body = self.source_code.split("def _on_dictionary_add_term")[1].split("\n def ")[0] + self.assertIn("self._pinned_terms.append(term)", add_body) + self.assertIn("self._pinned_terms", self.source_code) + + def test_show_all_add_scrolls_to_the_landed_term(self) -> None: + """Under "show all" the rebuild scrolls to the new term's own row. + + Oversized or yield-capped files prepend the term, so scrolling to + the bottom would hide it; the target is the term's index in the + refreshed list and is verified against the row's label. + """ + add_body = self.source_code.split("def _on_dictionary_add_term")[1].split("\n def ")[0] + self.assertIn("self._show_all_terms", add_body) + self.assertIn("self._scroll_terms_to_row", add_body) + self.assertIn("terms_now[new_index]", add_body) + finish_body = self.source_code.split("def _finish_term_rows")[1].split("\n def ")[0] + self.assertIn("self._scroll_terms_to_row", finish_body) + self.assertIn("get_row_at_index", finish_body) + self.assertIn("_term_row_text", finish_body) + self.assertIn("_scroll_terms_row_into_view", finish_body) + self.assertNotIn("_scroll_terms_to_end", self.source_code) + def test_application_page_has_tray_warning_toggle(self): self.assertIn('PreferencesGroup(title="General")', self.source_code) self.assertIn("self.missing_tray_warning_switch = Gtk.Switch()", self.source_code)