From 28ce674430dab497ecd17fea3c7b6fb6f1c4b342 Mon Sep 17 00:00:00 2001 From: DavidHruby1 Date: Wed, 14 Oct 2026 21:24:34 +0200 Subject: [PATCH 1/2] fix: keep end of speech, scale STT timeout, widen Groq cap - Keep samples drained during stream.stop() and capture a 0.25 s release tail - Scale STT read timeout with audio length; 10 s connect timeout - Size the Groq completion cap for dense (e.g. Czech) tokenization - Warn once about plain-http provider URLs on non-loopback hosts - Load history records missing defaulted fields - Raise MIC_UNAVAILABLE from calibrate() instead of swallowing errors - Send only the primary language subtag to STT (pt-br -> pt) Co-Authored-By: Claude Opus 5.5 --- docs/IMPLEMENTATION.md | 24 ++++++++++++++++------- src/audio.py | 37 ++++++++++++++++++++++++++---------- src/config.py | 32 ++++++++++++++++++++++++++++++- src/main.py | 16 +++++++++++----- src/results.py | 31 ++++++++++++++++++++++-------- src/rewrite.py | 6 ++++-- src/settings_dialog.py | 24 ++++++++++++++++++++++- src/stt.py | 13 ++++++++++++- tests/test_audio.py | 40 ++++++++++++++++++++++++++++++++++----- tests/test_config.py | 14 ++++++++++++++ tests/test_results.py | 21 ++++++++++++++++++++ tests/test_stt_rewrite.py | 32 ++++++++++++++++++++++++++++--- 12 files changed, 247 insertions(+), 43 deletions(-) diff --git a/docs/IMPLEMENTATION.md b/docs/IMPLEMENTATION.md index 5ac3d12..d00e399 100644 --- a/docs/IMPLEMENTATION.md +++ b/docs/IMPLEMENTATION.md @@ -283,6 +283,10 @@ def parse_custom_headers(custom_headers: str) -> dict[str, str]: ... def validate_config(cfg: AppConfig) -> list[ConfigValidationIssue]: ... """Return all startup/settings validation issues for the current config.""" +def insecure_base_urls(cfg: AppConfig) -> list[tuple[str, str]]: ... + """(label, url) for http:// provider URLs on a non-loopback host. Settings warns + once when such a URL is newly entered; it never blocks saving.""" + def stt_provider_is_configured(provider: ProviderConfig) -> bool: ... """STT requires URL + model; key is optional. LLM still requires key + URL + model.""" @@ -377,16 +381,19 @@ class AudioRecorder: @property def rms_threshold(self) -> float: ... def calibrate(self, duration: float = 2.0) -> float: ... - """Return max(noise_floor * 2.0, DEFAULT_RMS_THRESHOLD); measurement failure falls back to 5.0.""" + """Return max(noise_floor * 2.0, DEFAULT_RMS_THRESHOLD); measurement failure raises + MIC_UNAVAILABLE and leaves the current threshold unchanged.""" def start(self) -> None: ... """Open and start the microphone; on failure close the stream and raise MIC_UNAVAILABLE.""" def snapshot(self) -> CaptureSnapshot: ... """Immutable evidence under the recorder lock; no enumeration or audio copy. Before callbacks: level 0, has_callback_data False. Captured zeros: True, level 0.""" - def stop(self) -> bytes: ... - """Return mono int16 WAV; short/quiet capture returns b"". A valid-length - attempt without samples, unexpected stream finish, or stop/close failure - raises MIC_DISCONNECTED and discards frames. Late callbacks cannot append.""" + def stop(self, tail: float = 0.0) -> bytes: ... + """Capture `tail` more seconds (hotkey stops pass RELEASE_TAIL = 0.25), then + return mono int16 WAV; short/quiet capture returns b"". Samples the backend + drains during stream.stop() are kept. A valid-length attempt without samples, + unexpected stream finish, or stop/close failure raises MIC_DISCONNECTED and + discards frames. Callbacks after stop() returns cannot append.""" def resolve_device( preferred_id: int | None, @@ -601,7 +608,8 @@ ruff format --check src/ tests/ .github/scripts/ - The tray Language radio submenu switches the persisted active STT code immediately. Auto omits the STT language field and rewrite hint; a selected code is sent to both primary and fallback STT and added to the optional rewrite prompt. Provider support - for a language code is not guaranteed. Settings manages the ordered favorites; + for a language code is not guaranteed. STT receives only the primary subtag + (`pt-br` is sent as `pt`), since Whisper-compatible endpoints expect ISO 639-1. Settings manages the ordered favorites; removing an active additional favorite selects Auto. A failed tray save restores the previous visible and live selection. - Tray icon: grey (idle) → red (recording) → yellow (processing) → grey. @@ -666,7 +674,9 @@ ruff format --check src/ tests/ .github/scripts/ - Latest usable text remains in RAM even with history disabled. Persistent history is opt-in (default off) and stores validated version-1 JSON as DPAPI-protected ciphertext at `APP_DIR/history.enc`; writes use ciphertext-only temporary files, fsync and atomic - replacement. Retention is 1..100 on commit. Read/decrypt/schema/write failures fail + replacement. Records missing a defaulted field load with its default and unknown + fields are ignored, so adding a field does not invalidate older files. + Retention is 1..100 on commit. Read/decrypt/schema/write failures fail closed while RAM recovery remains available. Turning history off stops ordinary reads/writes but leaves prior saved entries until explicit Clear. Delete is not a secure-erasure guarantee for backups or other copies. diff --git a/src/audio.py b/src/audio.py index 1bd4fea..e30d472 100644 --- a/src/audio.py +++ b/src/audio.py @@ -26,6 +26,8 @@ CHANNELS = 1 DTYPE = "int16" MIN_DURATION = 0.3 +# Speech often ends a moment after the hotkey is released; keep capturing briefly. +RELEASE_TAIL = 0.25 @dataclass @@ -116,6 +118,7 @@ def __init__(self, device_id: int | None = None, sample_rate: int = SAMPLE_RATE) self._start_time: float = 0.0 self._rms_threshold: float = DEFAULT_RMS_THRESHOLD self._accepting_frames = False + self._stop_requested = False self._level_rms = 0.0 self._has_callback_data = False self._capture_error: str | None = None @@ -147,7 +150,11 @@ def snapshot(self) -> CaptureSnapshot: ) def calibrate(self, duration: float = 2.0) -> float: - """Record ambient noise and return a usable silence-gate threshold.""" + """Record ambient noise and return a usable silence-gate threshold. + + Raise ``ScreamerError(AppError.MIC_UNAVAILABLE)`` if recording fails; the + current threshold is left unchanged. + """ _require_sd() try: with log_duration(log, f"Calibration for {duration:.1f}s"): @@ -169,9 +176,8 @@ def calibrate(self, duration: float = 2.0) -> float: ) return threshold except Exception as e: - log.warning("Calibration failed: %s; using fallback %.1f", e, DEFAULT_RMS_THRESHOLD) - self._rms_threshold = DEFAULT_RMS_THRESHOLD - return DEFAULT_RMS_THRESHOLD + log.warning("Calibration failed: %s", e) + raise ScreamerError(AppError.MIC_UNAVAILABLE, str(e)) from e def _callback(self, indata: np.ndarray, frames: int, time_info, status) -> None: # type: ignore[no-untyped-def] level_rms = float(np.sqrt(np.mean(indata.astype(np.float64) ** 2))) @@ -185,7 +191,7 @@ def _callback(self, indata: np.ndarray, frames: int, time_info, status) -> None: def _on_stream_finished(self) -> None: with self._lock: - if self._accepting_frames: + if self._accepting_frames and not self._stop_requested: self._capture_error = "Input stream stopped unexpectedly." self._accepting_frames = False @@ -199,6 +205,7 @@ def start(self) -> None: self._capture_error = None self._device = None self._input_status = "" + self._stop_requested = False self._accepting_frames = True self._start_time = time.monotonic() try: @@ -233,9 +240,12 @@ def start(self) -> None: self._stream = stream log.info("Recording started (device=%s)", self._device_id) - def stop(self) -> bytes: + def stop(self, tail: float = 0.0) -> bytes: """Stop recording and return 16kHz mono int16 WAV bytes. + ``tail`` keeps capturing for that many seconds first. Samples the backend + drains while stopping are kept, so the end of speech is not cut off. + Raise ``ScreamerError(AppError.MIC_DISCONNECTED)`` on stream failure or a valid-length attempt without callback data. Return empty bytes if the recording is too short or silent. @@ -245,13 +255,16 @@ def stop(self) -> bytes: stream = self._stream self._stream = None + if tail > 0: + time.sleep(tail) with self._lock: - self._accepting_frames = False - capture_error = self._capture_error + self._stop_requested = True try: try: stream.stop(ignore_errors=False) finally: + with self._lock: + self._accepting_frames = False stream.close(ignore_errors=False) except Exception as e: with self._lock: @@ -261,6 +274,7 @@ def stop(self) -> bytes: with self._lock: frames = self._frames self._frames = [] + capture_error = self._capture_error if capture_error is not None: raise ScreamerError(AppError.MIC_DISCONNECTED, capture_error) @@ -349,8 +363,11 @@ def resolve_device( recorder = AudioRecorder() print("Calibrating (2s ambient noise)...") - threshold = recorder.calibrate(2.0) - print(f"RMS threshold: {threshold:.1f}") + try: + threshold = recorder.calibrate(2.0) + print(f"RMS threshold: {threshold:.1f}") + except ScreamerError as e: + print(f" Calibration failed: {e}; using {recorder.rms_threshold:.1f}") print() print("Recording 3 seconds... speak now!") diff --git a/src/config.py b/src/config.py index f330633..c7143d4 100644 --- a/src/config.py +++ b/src/config.py @@ -2,6 +2,7 @@ from __future__ import annotations +import ipaddress import json import logging import ntpath @@ -328,7 +329,7 @@ def normalize_language_code(value: str) -> str: raise ValueError("Language code must be text") code = value.strip().lower() if len(code) > _MAX_LANGUAGE_CODE_LENGTH or not _LANGUAGE_CODE.fullmatch(code): - raise ValueError("Use a language code such as cs, en, or pt-br (up to 32 characters)") + raise ValueError("Use a language code such as cs, en, or de (up to 32 characters)") return code @@ -1201,6 +1202,35 @@ def validate_config(cfg: AppConfig) -> list[ConfigValidationIssue]: return issues +def insecure_base_urls(cfg: AppConfig) -> list[tuple[str, str]]: + """Return (label, url) for plain-http provider URLs that leave this machine. + + Such URLs send API keys, custom headers and transcripts unencrypted. This is + a warning, not a validation issue: LAN-hosted servers may use http on purpose. + """ + result: list[tuple[str, str]] = [] + for url, label in ( + (cfg.stt_base_url, "Primary STT"), + (cfg.stt_fallback_base_url, "Fallback STT"), + (cfg.llm_base_url, "Primary LLM"), + (cfg.llm_fallback_base_url, "Fallback LLM"), + ): + try: + parsed = urlsplit(url) + host = parsed.hostname + except ValueError: + continue # Invalid URLs are reported by validate_config. + if parsed.scheme != "http" or not host or host == "localhost": + continue + try: + if ipaddress.ip_address(host).is_loopback: + continue + except ValueError: + pass # A hostname other than localhost. + result.append((label, url)) + return result + + def _env_path() -> str: """Locate .env: next to the executable when frozen, else at cwd (dev runs).""" if getattr(sys, "frozen", False): diff --git a/src/main.py b/src/main.py index 3974945..11ed8ac 100644 --- a/src/main.py +++ b/src/main.py @@ -31,7 +31,13 @@ QWidgetAction, ) -from src.audio import AudioRecorder, default_input_device_id, list_devices, resolve_device +from src.audio import ( + RELEASE_TAIL, + AudioRecorder, + default_input_device_id, + list_devices, + resolve_device, +) from src.config import ( DEFAULT_LLM_SYSTEM_PROMPT, HOTKEY_OPTIONS, @@ -539,7 +545,7 @@ def _poll_recording_level(self) -> None: label, min(capture.level_rms / 32767.0, 1.0), capture.has_callback_data ) - def _finalize_recording(self) -> None: + def _finalize_recording(self, tail: float = 0.0) -> None: """Stop recording and start the processing worker.""" self._recording_timer.stop() self._level_timer.stop() @@ -547,7 +553,7 @@ def _finalize_recording(self) -> None: self._apply_state(TrayState.PROCESSING) try: - audio_wav = self._recorder.stop() + audio_wav = self._recorder.stop(tail=tail) except ScreamerError as e: self._session_config = None self._session = None @@ -599,7 +605,7 @@ def _on_hotkey_pressed(self) -> None: if self._recording: # Toggle mode: second press → finalize and process. - self._finalize_recording() + self._finalize_recording(tail=RELEASE_TAIL) else: self._start_recording() @@ -608,7 +614,7 @@ def _on_hotkey_released(self) -> None: if self._is_hotkey_capture_active() and not self._recording: return if self._recording: - self._finalize_recording() + self._finalize_recording(tail=RELEASE_TAIL) def _is_hotkey_capture_active(self) -> bool: return self._settings_dlg is not None and self._settings_dlg.is_hotkey_capture_active() diff --git a/src/results.py b/src/results.py index 53f86af..d899608 100644 --- a/src/results.py +++ b/src/results.py @@ -5,7 +5,7 @@ import json import os import tempfile -from dataclasses import asdict, dataclass +from dataclasses import MISSING, asdict, dataclass, fields from datetime import datetime, timezone from pathlib import Path @@ -77,10 +77,28 @@ def _timestamp(value: object) -> None: raise ValueError("Expected UTC timestamp") -def _decode_record(value: object) -> DictationRecord: - if not isinstance(value, dict) or set(value) != set(DictationRecord.__dataclass_fields__): +def _known_fields(cls: type, value: object) -> dict: + """Require fields without defaults and fill missing defaulted ones. + + Records written before a defaulted field was added must still load instead of + invalidating the whole history file. Unknown fields (e.g. audio) stay rejected. + """ + if not isinstance(value, dict) or not set(value) <= {f.name for f in fields(cls)}: raise ValueError("Invalid record fields") - data = dict(value) + data = {} + for f in fields(cls): + if f.name in value: + data[f.name] = value[f.name] + elif f.default is MISSING: + raise ValueError("Invalid record fields") + else: + # Defaults in their serialized form (tuples are stored as JSON lists). + data[f.name] = list(f.default) if isinstance(f.default, tuple) else f.default + return data + + +def _decode_record(value: object) -> DictationRecord: + data = _known_fields(DictationRecord, value) _string(data["id"], nonempty=True) _timestamp(data["created_at"]) _string(data["raw_text"], nonempty=True) @@ -104,10 +122,7 @@ def _decode_record(value: object) -> DictationRecord: data["warnings"] = tuple(AppError[name] for name in warnings) attempt = data["last_delivery"] if attempt is not None: - if not isinstance(attempt, dict) or set(attempt) != set( - DeliveryAttempt.__dataclass_fields__ - ): - raise ValueError("Invalid attempt fields") + attempt = _known_fields(DeliveryAttempt, attempt) _timestamp(attempt["attempted_at"]) if attempt["mode"] not in {"type", "copy", "copy_and_type", "manual"}: raise ValueError("Invalid output mode") diff --git a/src/rewrite.py b/src/rewrite.py index f099188..402a658 100644 --- a/src/rewrite.py +++ b/src/rewrite.py @@ -163,8 +163,10 @@ def _call_llm( def _groq_completion_cap(user_text: str) -> int: - estimated_input_tokens = max(1, len(user_text) // 4) - cap = int(estimated_input_tokens * 1.5) + 32 + # Assume ~2 chars/token: accented languages such as Czech tokenize far + # denser than English, and a too-tight cap discards the whole rewrite. + estimated_input_tokens = max(1, len(user_text) // 2) + cap = estimated_input_tokens * 2 + 64 return max(128, min(4096, cap)) diff --git a/src/settings_dialog.py b/src/settings_dialog.py index 00a9420..21bff57 100644 --- a/src/settings_dialog.py +++ b/src/settings_dialog.py @@ -63,6 +63,7 @@ normalize_vocabulary_entries, reset_config, save_config, + insecure_base_urls, validate_config, ) from src.audio import resolve_device @@ -168,6 +169,8 @@ def __init__( # Edit a deep copy so the original is untouched until accept. self._working = copy.deepcopy(config) + # Warn about unencrypted provider URLs once, and only when newly entered. + self._acknowledged_insecure_urls = {url for _label, url in insecure_base_urls(config)} self._profile_loading = False self._rewrite_error: str | None = None @@ -407,7 +410,7 @@ def _build_stt_tab(self) -> None: form.addRow("Favorite languages:", self._stt_favorites) self._stt_favorite_input = QLineEdit() - self._stt_favorite_input.setPlaceholderText("Language code, e.g. de or pt-br") + self._stt_favorite_input.setPlaceholderText("ISO 639-1 language code, e.g. de or pt") form.addRow("Favorite code:", self._stt_favorite_input) btn_add = QPushButton("Add") @@ -1519,6 +1522,7 @@ def _show_validation_issue(self) -> bool: None, ) if issue is None: + self._warn_insecure_urls() return True QMessageBox.warning(self, "Missing Configuration", issue.message) @@ -1527,6 +1531,24 @@ def _show_validation_issue(self) -> bool: self._llm_group.setVisible(True) return False + def _warn_insecure_urls(self) -> None: + new = [ + (label, url) + for label, url in insecure_base_urls(self._working) + if url not in self._acknowledged_insecure_urls + ] + if not new: + return + self._acknowledged_insecure_urls.update(url for _label, url in new) + lines = "\n".join(f"{label}: {url}" for label, url in new) + QMessageBox.warning( + self, + "Unencrypted Connection", + "These providers use http:// on a non-local host, so the API key, custom " + f"headers and transcripts are sent unencrypted:\n\n{lines}\n\n" + "Use https:// unless the server is on a network you trust.", + ) + def _save_or_warn(self) -> bool: from src.utils import ScreamerError diff --git a/src/stt.py b/src/stt.py index 8bdae07..13c5004 100644 --- a/src/stt.py +++ b/src/stt.py @@ -115,6 +115,8 @@ def _call_stt( data: dict[str, str] = {"model": provider.model, "response_format": response_format} if language: + # Whisper-compatible endpoints accept ISO 639-1 codes only; "pt-br" -> "pt". + language = language.split("-", 1)[0] data["language"] = language if prompt: data["prompt"] = prompt @@ -129,7 +131,9 @@ def _call_stt( len(audio_wav), ) request_started = perf_counter() - resp = http_client.post(url, headers=headers, data=data, files=files, timeout=60.0) + resp = http_client.post( + url, headers=headers, data=data, files=files, timeout=_stt_timeout(audio_wav) + ) request_elapsed = perf_counter() - request_started log.info( @@ -163,6 +167,13 @@ def _call_stt( return text +def _stt_timeout(audio_wav: bytes) -> httpx.Timeout: + """Scale the read budget with audio length; local CPU models can be slower than real time.""" + # 16 kHz mono int16 = 32000 bytes per second, after the 44-byte WAV header. + audio_seconds = max(0, len(audio_wav) - 44) / 32000 + return httpx.Timeout(60.0 + 3.0 * audio_seconds, connect=10.0) + + # --------------------------------------------------------------------------- # CLI smoke test # --------------------------------------------------------------------------- diff --git a/tests/test_audio.py b/tests/test_audio.py index 09b3e68..7cc45be 100644 --- a/tests/test_audio.py +++ b/tests/test_audio.py @@ -69,13 +69,16 @@ def test_calibration_scales_audible_noise_floor(self) -> None: self.assertEqual(threshold, 20.0) - def test_calibration_failure_uses_low_default(self) -> None: + def test_calibration_failure_raises_and_keeps_threshold(self) -> None: fake_sd = FakeSoundDevice(error=RuntimeError("no mic")) + recorder = AudioRecorder() + recorder.rms_threshold = 42.0 - with patch("src.audio.sd", fake_sd): - threshold = AudioRecorder().calibrate(1.0) + with patch("src.audio.sd", fake_sd), self.assertRaises(ScreamerError) as error: + recorder.calibrate(1.0) - self.assertEqual(threshold, DEFAULT_RMS_THRESHOLD) + self.assertEqual(error.exception.code, AppError.MIC_UNAVAILABLE) + self.assertEqual(recorder.rms_threshold, 42.0) class AudioDeviceDefaultTests(unittest.TestCase): @@ -201,6 +204,7 @@ def __init__(self, callback, finished_callback) -> None: self.stop_error = False self.close_error = False self.closed = False + self.drain: np.ndarray | None = None def feed(self, samples, status="") -> None: self.callback(samples, len(samples), None, status) @@ -213,7 +217,8 @@ def start(self) -> None: def stop(self, *, ignore_errors) -> None: assert ignore_errors is False # Backends may finish an in-flight callback while stop is in progress. - self.feed(np.full((160, 1), 5000, dtype=np.int16)) + if self.drain is not None: + self.feed(self.drain) self.finished_callback() if self.stop_error: raise OSError("stop failed") @@ -291,6 +296,31 @@ def test_latest_level_and_actual_identity_with_unchanged_wav_samples(self) -> No self.assertEqual(self.recorder.snapshot().level_rms, 600.0) self.assertEqual(self.recorder._frames, []) + def test_samples_drained_while_stopping_are_kept(self) -> None: + spoken = np.full((160, 1), 1200, dtype=np.int16) + drained = np.full((160, 1), 5000, dtype=np.int16) + self.backend.stream.feed(spoken) + self.backend.stream.drain = drained + self.now = 11.0 + wav = self.recorder.stop() + with wave.open(io.BytesIO(wav), "rb") as audio: + self.assertEqual(audio.getnframes(), 320) + self.assertEqual(audio.readframes(320), spoken.tobytes() + drained.tobytes()) + + def test_tail_keeps_capturing_before_stopping(self) -> None: + tail_samples = np.full((160, 1), 2400, dtype=np.int16) + self.backend.stream.feed(np.full((160, 1), 1200, dtype=np.int16)) + self.now = 11.0 + with patch( + "src.audio.time.sleep", side_effect=lambda _s: self.backend.stream.feed(tail_samples) + ) as sleep: + wav = self.recorder.stop(tail=0.25) + sleep.assert_called_once_with(0.25) + with wave.open(io.BytesIO(wav), "rb") as audio: + self.assertEqual(audio.getnframes(), 320) + audio.readframes(160) + self.assertEqual(audio.readframes(160), tail_samples.tobytes()) + def test_captured_zero_and_quiet_input_are_not_missing_data(self) -> None: for value in (0, 2): with self.subTest(value=value): diff --git a/tests/test_config.py b/tests/test_config.py index a8b9ced..08ab5f1 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -4,6 +4,7 @@ import sys import tempfile import unittest +from dataclasses import replace from pathlib import Path from unittest.mock import patch @@ -13,6 +14,7 @@ ProviderConfig, _env_path, import_from_env, + insecure_base_urls, language_choices, load_config, normalize_language_code, @@ -121,6 +123,18 @@ def test_provider_config_detects_groq_by_host(self) -> None: self.assertTrue(ProviderConfig(base_url="https://api.groq.com/openai/v1").is_groq) self.assertFalse(ProviderConfig(base_url="https://api.openai.com/v1").is_groq) + def test_plain_http_is_flagged_only_off_loopback(self) -> None: + cfg = AppConfig( + stt_base_url="http://192.168.1.5:8000/v1", + stt_fallback_base_url="http://localhost:8000/v1", + llm_base_url="http://127.0.0.1:1234/v1", + llm_fallback_base_url="https://api.example.test/v1", + ) + self.assertEqual( + insecure_base_urls(cfg), [("Primary STT", "http://192.168.1.5:8000/v1")] + ) + self.assertEqual(insecure_base_urls(replace(cfg, stt_base_url="http://[::1]/v1")), []) + def test_invalid_http_headers_and_urls_are_rejected(self) -> None: cfg = AppConfig( stt_api_key="key", diff --git a/tests/test_results.py b/tests/test_results.py index 8df71b0..dddac42 100644 --- a/tests/test_results.py +++ b/tests/test_results.py @@ -53,6 +53,27 @@ def test_raw_and_final_with_attempt_roundtrip_without_audio_or_credentials(self) self.assertNotIn("audio", payload["entries"][0]) self.assertNotIn("config", payload["entries"][0]) + def test_records_from_before_a_field_was_added_still_load(self): + attempt = DeliveryAttempt("2026-09-30T12:01:00+00:00", "copy") + stored = asdict(record(last_delivery=attempt)) + for key in ("profile_id", "language", "warnings"): + del stored[key] + del stored["last_delivery"]["reason"] + self.path.write_bytes( + base64.b64encode(json.dumps({"version": 1, "entries": [stored]}).encode()) + ) + self.assertEqual(self.store.load(), [record(last_delivery=attempt)]) + + def test_record_missing_required_field_fails_closed(self): + stored = asdict(record()) + stored["warnings"] = [] + del stored["raw_text"] + self.path.write_bytes( + base64.b64encode(json.dumps({"version": 1, "entries": [stored]}).encode()) + ) + with self.assertRaises(ScreamerError): + self.store.load() + def test_upsert_checkpoint_updates_one_id_and_retention_evicts_oldest(self): self.store.upsert(record(), 2) final = record(rewrite_status="succeeded", rewritten_text="Final") diff --git a/tests/test_stt_rewrite.py b/tests/test_stt_rewrite.py index f56b5b6..1a6429e 100644 --- a/tests/test_stt_rewrite.py +++ b/tests/test_stt_rewrite.py @@ -375,6 +375,32 @@ def fake_post(_url, **kwargs): self.assertEqual(ctx.exception.code, AppError.NO_SPEECH) + def test_groq_cap_fits_dense_czech_output(self) -> None: + from src.rewrite import _groq_completion_cap + + # ~1 minute of Czech dictation; Llama tokenizers average ~2.5-3 chars/token. + text = "Příliš žluťoučký kůň úpěl ďábelské ódy. " * 25 + self.assertGreaterEqual(_groq_completion_cap(text), int(len(text) / 2.5 * 1.5)) + + def test_stt_timeout_scales_with_audio_length(self) -> None: + from src.stt import _stt_timeout + + short = _stt_timeout(b"\0" * (44 + 32000)) + five_minutes = _stt_timeout(b"\0" * (44 + 300 * 32000)) + self.assertEqual(short.connect, 10.0) + self.assertGreaterEqual(short.read, 60.0) + self.assertGreaterEqual(five_minutes.read, 900.0) + + def test_stt_sends_primary_language_subtag(self) -> None: + cfg = AppConfig( + stt_base_url="https://stt.test/v1", stt_model="whisper", stt_language="pt-br" + ) + with patch( + "src.http_client.post", return_value=FakeResponse({"text": "olá"}) + ) as post: + transcribe(b"wav", cfg) + self.assertEqual(post.call_args.kwargs["data"]["language"], "pt") + def test_rewrite_groq_sets_dynamic_completion_cap(self) -> None: cfg = AppConfig( llm_enabled=True, @@ -415,10 +441,10 @@ def fake_post(_url, **kwargs): ) with patch("src.http_client.post", side_effect=fake_post): - rewrite("a" * 10_000, cfg) + rewrite("a" * 3_000, cfg) - # 10_000 chars -> ~2500 estimated input tokens -> cap int(2500 * 1.5) + 32 = 3782 - self.assertEqual(captured["max_completion_tokens"], 3782) + # 3_000 chars -> ~1500 estimated input tokens -> cap 1500 * 2 + 64 = 3064 + self.assertEqual(captured["max_completion_tokens"], 3064) def test_rewrite_length_finish_keeps_original_text(self) -> None: cfg = AppConfig( From feb6d4609e73a7ef2fe645a61cd9b90a6a44fd1a Mon Sep 17 00:00:00 2001 From: DavidHruby1 Date: Wed, 14 Oct 2026 21:25:42 +0200 Subject: [PATCH 2/2] style: ruff format tests Co-Authored-By: Claude Opus 5.5 --- tests/test_config.py | 4 +--- tests/test_stt_rewrite.py | 4 +--- 2 files changed, 2 insertions(+), 6 deletions(-) diff --git a/tests/test_config.py b/tests/test_config.py index 08ab5f1..64a9a1c 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -130,9 +130,7 @@ def test_plain_http_is_flagged_only_off_loopback(self) -> None: llm_base_url="http://127.0.0.1:1234/v1", llm_fallback_base_url="https://api.example.test/v1", ) - self.assertEqual( - insecure_base_urls(cfg), [("Primary STT", "http://192.168.1.5:8000/v1")] - ) + self.assertEqual(insecure_base_urls(cfg), [("Primary STT", "http://192.168.1.5:8000/v1")]) self.assertEqual(insecure_base_urls(replace(cfg, stt_base_url="http://[::1]/v1")), []) def test_invalid_http_headers_and_urls_are_rejected(self) -> None: diff --git a/tests/test_stt_rewrite.py b/tests/test_stt_rewrite.py index 1a6429e..cc0382b 100644 --- a/tests/test_stt_rewrite.py +++ b/tests/test_stt_rewrite.py @@ -395,9 +395,7 @@ def test_stt_sends_primary_language_subtag(self) -> None: cfg = AppConfig( stt_base_url="https://stt.test/v1", stt_model="whisper", stt_language="pt-br" ) - with patch( - "src.http_client.post", return_value=FakeResponse({"text": "olá"}) - ) as post: + with patch("src.http_client.post", return_value=FakeResponse({"text": "olá"})) as post: transcribe(b"wav", cfg) self.assertEqual(post.call_args.kwargs["data"]["language"], "pt")