From ffc3ce27d53a80dff9e8ed53877270dcd90275b2 Mon Sep 17 00:00:00 2001 From: Charles Cha <92324143+ypwcharles@users.noreply.github.com> Date: Fri, 17 Jul 2026 00:46:40 +0800 Subject: [PATCH] fix(stt): scope upload size limits to remote providers --- .../test_transcription_command_providers.py | 28 ++++- .../test_transcription_plugin_dispatch.py | 95 +++++++++------ tests/tools/test_transcription_tools.py | 45 +++++++ tests/tools/test_voice_mode.py | 112 ++++++++++++++++++ tools/transcription_tools.py | 49 ++++++-- tools/voice_mode.py | 9 +- 6 files changed, 280 insertions(+), 58 deletions(-) diff --git a/tests/tools/test_transcription_command_providers.py b/tests/tools/test_transcription_command_providers.py index 749ab5e839c3..10d6e9b332a8 100644 --- a/tests/tools/test_transcription_command_providers.py +++ b/tests/tools/test_transcription_command_providers.py @@ -396,7 +396,7 @@ class TestTranscribeCommandSTT: audio = _make_silent_wav(tmp_path / "input.wav") # Write the model into the transcript so we can assert it propagated. interpreter = sys.executable - payload = "import sys; open(sys.argv[2], 'w').write(sys.argv[1])" + payload = "import sys; open(sys.argv[2], 'w', encoding='utf-8').write(sys.argv[1])" cfg = { "command": f'"{interpreter}" -c "{payload}" {{model}} {{output_path}}', "model": "config-model", @@ -410,7 +410,7 @@ class TestTranscribeCommandSTT: def test_config_model_used_when_no_override(self, tmp_path): audio = _make_silent_wav(tmp_path / "input.wav") interpreter = sys.executable - payload = "import sys; open(sys.argv[2], 'w').write(sys.argv[1])" + payload = "import sys; open(sys.argv[2], 'w', encoding='utf-8').write(sys.argv[1])" cfg = { "command": f'"{interpreter}" -c "{payload}" {{model}} {{output_path}}', "model": "config-model", @@ -421,7 +421,7 @@ class TestTranscribeCommandSTT: def test_language_from_provider_config_wins(self, tmp_path): audio = _make_silent_wav(tmp_path / "input.wav") interpreter = sys.executable - payload = "import sys; open(sys.argv[2], 'w').write(sys.argv[1])" + payload = "import sys; open(sys.argv[2], 'w', encoding='utf-8').write(sys.argv[1])" cfg = { "command": f'"{interpreter}" -c "{payload}" {{language}} {{output_path}}', "language": "fr", @@ -435,7 +435,7 @@ class TestTranscribeCommandSTT: def test_language_falls_back_to_stt_section(self, tmp_path): audio = _make_silent_wav(tmp_path / "input.wav") interpreter = sys.executable - payload = "import sys; open(sys.argv[2], 'w').write(sys.argv[1])" + payload = "import sys; open(sys.argv[2], 'w', encoding='utf-8').write(sys.argv[1])" cfg = { "command": f'"{interpreter}" -c "{payload}" {{language}} {{output_path}}', } @@ -447,7 +447,7 @@ class TestTranscribeCommandSTT: def test_language_defaults_to_en(self, tmp_path): audio = _make_silent_wav(tmp_path / "input.wav") interpreter = sys.executable - payload = "import sys; open(sys.argv[2], 'w').write(sys.argv[1])" + payload = "import sys; open(sys.argv[2], 'w', encoding='utf-8').write(sys.argv[1])" cfg = { "command": f'"{interpreter}" -c "{payload}" {{language}} {{output_path}}', } @@ -486,6 +486,24 @@ class TestTranscribeAudioDispatchToCommandProvider: assert result["transcript"] == "dispatched via command" assert result["provider"] == "fake-cli" + def test_oversized_command_provider_file_is_rejected(self, tmp_path): + from tools.transcription_tools import MAX_FILE_SIZE + + audio = tmp_path / "oversized.wav" + with audio.open("wb") as audio_file: + audio_file.seek(MAX_FILE_SIZE) + audio_file.write(b"\0") + cfg = self._config_with_command_provider("fake-cli", "unused {input_path}") + + with patch("tools.transcription_tools._load_stt_config", return_value=cfg), \ + patch("tools.transcription_tools._transcribe_command_stt", + return_value={"success": True, "transcript": "hi"}) as mock_command: + result = transcribe_audio(str(audio)) + + assert result["success"] is False + assert "File too large" in result["error"] + mock_command.assert_not_called() + def test_builtin_name_shadow_does_not_route_to_command(self, tmp_path): # User mis-configures stt.providers.openai as a command — must NOT # hijack the real OpenAI built-in. The built-in elif chain owns diff --git a/tests/tools/test_transcription_plugin_dispatch.py b/tests/tools/test_transcription_plugin_dispatch.py index 834246769527..d7ea8dac4ee3 100644 --- a/tests/tools/test_transcription_plugin_dispatch.py +++ b/tests/tools/test_transcription_plugin_dispatch.py @@ -66,6 +66,14 @@ def _reset_registry(): transcription_registry._reset_for_tests() +@pytest.fixture +def sample_audio_file(tmp_path): + """Return a real, small input so E2E tests exercise validation too.""" + audio_path = tmp_path / "audio.mp3" + audio_path.write_bytes(b"fake audio data") + return str(audio_path) + + # --------------------------------------------------------------------------- # Built-in always wins # --------------------------------------------------------------------------- @@ -209,42 +217,39 @@ class TestTranscribeAudioE2E: """transcribe_audio() routes plugin dispatch correctly when the configured name is unknown to the built-in branches. - Note: we mock _validate_audio_file and _get_provider so the real - file-validation and provider-resolution don't fire — we're testing - the plugin-dispatch wiring, not those helpers. + Provider resolution is mocked to isolate routing, while a real small + input keeps the validation chain active. """ - def test_unknown_name_with_plugin_dispatches(self): + def test_unknown_name_with_plugin_dispatches(self, sample_audio_file): from unittest.mock import patch provider = _FakeProvider(name="openrouter") transcription_registry.register_provider(provider) - with patch("tools.transcription_tools._validate_audio_file", return_value=None), \ - patch("tools.transcription_tools._load_stt_config", return_value={"provider": "openrouter"}), \ + with patch("tools.transcription_tools._load_stt_config", return_value={"provider": "openrouter"}), \ patch("tools.transcription_tools.is_stt_enabled", return_value=True), \ patch("tools.transcription_tools._get_provider", return_value="openrouter"): - result = transcription_tools.transcribe_audio("/tmp/audio.mp3") + result = transcription_tools.transcribe_audio(sample_audio_file) assert result["success"] is True assert result["transcript"] == "fake transcript" assert result["provider"] == "openrouter" - def test_unknown_name_without_plugin_falls_to_legacy_error(self): + def test_unknown_name_without_plugin_falls_to_legacy_error(self, sample_audio_file): """When no plugin is registered for the unknown name, the dispatcher returns None and transcribe_audio falls through to the legacy 'No STT provider available' error message.""" from unittest.mock import patch - with patch("tools.transcription_tools._validate_audio_file", return_value=None), \ - patch("tools.transcription_tools._load_stt_config", return_value={"provider": "openrouter"}), \ + with patch("tools.transcription_tools._load_stt_config", return_value={"provider": "openrouter"}), \ patch("tools.transcription_tools.is_stt_enabled", return_value=True), \ patch("tools.transcription_tools._get_provider", return_value="openrouter"): - result = transcription_tools.transcribe_audio("/tmp/audio.mp3") + result = transcription_tools.transcribe_audio(sample_audio_file) assert result["success"] is False assert "No STT provider" in result["error"] - def test_builtin_name_does_not_consult_plugin_registry(self): + def test_builtin_name_does_not_consult_plugin_registry(self, sample_audio_file): """Even if a plugin's name collides with a built-in (which the registry blocks, but defense in depth matters), transcribe_audio with provider='groq' goes through the legacy elif chain, never @@ -255,12 +260,11 @@ class TestTranscribeAudioE2E: provider = _FakeProvider(name="openrouter") transcription_registry.register_provider(provider) - with patch("tools.transcription_tools._validate_audio_file", return_value=None), \ - patch("tools.transcription_tools._load_stt_config", return_value={"provider": "groq"}), \ + with patch("tools.transcription_tools._load_stt_config", return_value={"provider": "groq"}), \ patch("tools.transcription_tools._get_provider", return_value="groq"), \ patch("tools.transcription_tools._transcribe_groq", return_value={"success": True, "transcript": "from groq", "provider": "groq"}) as mock_groq: - result = transcription_tools.transcribe_audio("/tmp/audio.mp3") + result = transcription_tools.transcribe_audio(sample_audio_file) assert result["provider"] == "groq" assert result["transcript"] == "from groq" @@ -268,6 +272,25 @@ class TestTranscribeAudioE2E: # Plugin was never called assert provider.last_call is None + def test_oversized_plugin_file_is_rejected_before_dispatch(self, tmp_path): + from unittest.mock import patch + + provider = _FakeProvider(name="openrouter") + transcription_registry.register_provider(provider) + audio_path = tmp_path / "oversized.mp3" + with audio_path.open("wb") as audio_file: + audio_file.seek(transcription_tools.MAX_FILE_SIZE) + audio_file.write(b"\0") + + with patch("tools.transcription_tools._load_stt_config", return_value={"provider": "openrouter"}), \ + patch("tools.transcription_tools.is_stt_enabled", return_value=True), \ + patch("tools.transcription_tools._get_provider", return_value="openrouter"): + result = transcription_tools.transcribe_audio(str(audio_path)) + + assert result["success"] is False + assert "File too large" in result["error"] + assert provider.last_call is None + # --------------------------------------------------------------------------- # Availability gating (codex review feedback on PR #30493) @@ -330,7 +353,7 @@ class TestAvailabilityGate: assert "not available" in result["error"] assert provider.last_call is None - def test_unavailable_plugin_at_transcribe_audio_level(self): + def test_unavailable_plugin_at_transcribe_audio_level(self, sample_audio_file): """End-to-end: ``stt.provider: openrouter`` + plugin reports unavailable → ``transcribe_audio`` returns the unavailability envelope, NOT the generic "No STT provider available" message. @@ -339,11 +362,10 @@ class TestAvailabilityGate: provider = _FakeProvider(name="openrouter", available=False) transcription_registry.register_provider(provider) - with patch("tools.transcription_tools._validate_audio_file", return_value=None), \ - patch("tools.transcription_tools._load_stt_config", return_value={"provider": "openrouter"}), \ + with patch("tools.transcription_tools._load_stt_config", return_value={"provider": "openrouter"}), \ patch("tools.transcription_tools.is_stt_enabled", return_value=True), \ patch("tools.transcription_tools._get_provider", return_value="openrouter"): - result = transcription_tools.transcribe_audio("/tmp/audio.mp3") + result = transcription_tools.transcribe_audio(sample_audio_file) assert result["success"] is False # Must surface the plugin's unavailability — NOT the generic @@ -364,7 +386,7 @@ class TestLanguageForwardingFromConfig: ``stt.local.language``). """ - def test_language_read_from_provider_namespaced_config(self): + def test_language_read_from_provider_namespaced_config(self, sample_audio_file): """``stt.openrouter.language: ja`` reaches the plugin's transcribe() call as language='ja'.""" from unittest.mock import patch @@ -375,16 +397,15 @@ class TestLanguageForwardingFromConfig: "provider": "openrouter", "openrouter": {"language": "ja"}, } - with patch("tools.transcription_tools._validate_audio_file", return_value=None), \ - patch("tools.transcription_tools._load_stt_config", return_value=stt_config), \ + with patch("tools.transcription_tools._load_stt_config", return_value=stt_config), \ patch("tools.transcription_tools.is_stt_enabled", return_value=True), \ patch("tools.transcription_tools._get_provider", return_value="openrouter"): - transcription_tools.transcribe_audio("/tmp/audio.mp3") + transcription_tools.transcribe_audio(sample_audio_file) assert provider.last_call is not None assert provider.last_call["kwargs"]["language"] == "ja" - def test_model_from_provider_namespaced_config(self): + def test_model_from_provider_namespaced_config(self, sample_audio_file): """``stt.openrouter.model: whisper-large-v3`` reaches the plugin as model='whisper-large-v3' when caller doesn't override.""" @@ -396,15 +417,14 @@ class TestLanguageForwardingFromConfig: "provider": "openrouter", "openrouter": {"model": "whisper-large-v3"}, } - with patch("tools.transcription_tools._validate_audio_file", return_value=None), \ - patch("tools.transcription_tools._load_stt_config", return_value=stt_config), \ + with patch("tools.transcription_tools._load_stt_config", return_value=stt_config), \ patch("tools.transcription_tools.is_stt_enabled", return_value=True), \ patch("tools.transcription_tools._get_provider", return_value="openrouter"): - transcription_tools.transcribe_audio("/tmp/audio.mp3") + transcription_tools.transcribe_audio(sample_audio_file) assert provider.last_call["kwargs"]["model"] == "whisper-large-v3" - def test_caller_model_overrides_config_model(self): + def test_caller_model_overrides_config_model(self, sample_audio_file): """An explicit ``model`` arg to transcribe_audio wins over ``stt..model`` in config.""" from unittest.mock import patch @@ -415,33 +435,31 @@ class TestLanguageForwardingFromConfig: "provider": "openrouter", "openrouter": {"model": "config-model"}, } - with patch("tools.transcription_tools._validate_audio_file", return_value=None), \ - patch("tools.transcription_tools._load_stt_config", return_value=stt_config), \ + with patch("tools.transcription_tools._load_stt_config", return_value=stt_config), \ patch("tools.transcription_tools.is_stt_enabled", return_value=True), \ patch("tools.transcription_tools._get_provider", return_value="openrouter"): transcription_tools.transcribe_audio( - "/tmp/audio.mp3", model="explicit-arg-model", + sample_audio_file, model="explicit-arg-model", ) assert provider.last_call["kwargs"]["model"] == "explicit-arg-model" - def test_missing_provider_namespace_passes_none(self): + def test_missing_provider_namespace_passes_none(self, sample_audio_file): """No ``stt.`` subsection → language is None, model falls back to caller arg or None. No crash.""" from unittest.mock import patch provider = _FakeProvider(name="openrouter") transcription_registry.register_provider(provider) - with patch("tools.transcription_tools._validate_audio_file", return_value=None), \ - patch("tools.transcription_tools._load_stt_config", return_value={"provider": "openrouter"}), \ + with patch("tools.transcription_tools._load_stt_config", return_value={"provider": "openrouter"}), \ patch("tools.transcription_tools.is_stt_enabled", return_value=True), \ patch("tools.transcription_tools._get_provider", return_value="openrouter"): - transcription_tools.transcribe_audio("/tmp/audio.mp3") + transcription_tools.transcribe_audio(sample_audio_file) assert provider.last_call["kwargs"]["language"] is None assert provider.last_call["kwargs"]["model"] is None - def test_non_dict_provider_namespace_does_not_crash(self): + def test_non_dict_provider_namespace_does_not_crash(self, sample_audio_file): """If someone accidentally writes ``stt.openrouter: "foo"`` (a string instead of a dict), we should not crash — treat as empty config.""" @@ -450,11 +468,10 @@ class TestLanguageForwardingFromConfig: transcription_registry.register_provider(provider) stt_config = {"provider": "openrouter", "openrouter": "garbage"} - with patch("tools.transcription_tools._validate_audio_file", return_value=None), \ - patch("tools.transcription_tools._load_stt_config", return_value=stt_config), \ + with patch("tools.transcription_tools._load_stt_config", return_value=stt_config), \ patch("tools.transcription_tools.is_stt_enabled", return_value=True), \ patch("tools.transcription_tools._get_provider", return_value="openrouter"): - result = transcription_tools.transcribe_audio("/tmp/audio.mp3") + result = transcription_tools.transcribe_audio(sample_audio_file) # Should still dispatch successfully (config is just ignored) assert result["success"] is True diff --git a/tests/tools/test_transcription_tools.py b/tests/tools/test_transcription_tools.py index c3d8eefc3215..c0186513a0eb 100644 --- a/tests/tools/test_transcription_tools.py +++ b/tests/tools/test_transcription_tools.py @@ -54,6 +54,18 @@ def sample_ogg(tmp_path): return str(ogg_path) +@pytest.fixture +def oversized_wav(tmp_path): + """Create a sparse WAV-shaped file just above the remote upload cap.""" + from tools.transcription_tools import MAX_FILE_SIZE + + wav_path = tmp_path / "oversized.wav" + with wav_path.open("wb") as audio_file: + audio_file.seek(MAX_FILE_SIZE) + audio_file.write(b"\0") + return str(wav_path) + + pytestmark = pytest.mark.usefixtures("disable_lazy_stt_install") @@ -1110,6 +1122,39 @@ class TestTranscribeAudioDispatch: assert result["success"] is True mock_local.assert_called_once() + def test_oversized_local_file_reaches_dispatcher(self, oversized_wav): + with patch("tools.transcription_tools._load_stt_config", return_value={"provider": "local"}), \ + patch("tools.transcription_tools._get_provider", return_value="local"), \ + patch("tools.transcription_tools._transcribe_local", + return_value={"success": True, "transcript": "hi"}) as mock_local: + from tools.transcription_tools import transcribe_audio + result = transcribe_audio(oversized_wav) + + assert result["success"] is True + mock_local.assert_called_once() + + def test_oversized_local_command_file_reaches_dispatcher(self, oversized_wav): + with patch("tools.transcription_tools._load_stt_config", return_value={"provider": "local_command"}), \ + patch("tools.transcription_tools._get_provider", return_value="local_command"), \ + patch("tools.transcription_tools._transcribe_local_command", + return_value={"success": True, "transcript": "hi"}) as mock_command: + from tools.transcription_tools import transcribe_audio + result = transcribe_audio(oversized_wav) + + assert result["success"] is True + mock_command.assert_called_once() + + def test_oversized_remote_file_is_rejected_before_dispatch(self, oversized_wav): + with patch("tools.transcription_tools._load_stt_config", return_value={"provider": "openai"}), \ + patch("tools.transcription_tools._get_provider", return_value="openai"), \ + patch("tools.transcription_tools._transcribe_openai") as mock_openai: + from tools.transcription_tools import transcribe_audio + result = transcribe_audio(oversized_wav) + + assert result["success"] is False + assert "File too large" in result["error"] + mock_openai.assert_not_called() + def test_dispatches_to_openai(self, sample_ogg): with patch("tools.transcription_tools._load_stt_config", return_value={"provider": "openai"}), \ patch("tools.transcription_tools._get_provider", return_value="openai"), \ diff --git a/tests/tools/test_voice_mode.py b/tests/tools/test_voice_mode.py index 4611be901531..87ccdb297057 100644 --- a/tests/tools/test_voice_mode.py +++ b/tests/tools/test_voice_mode.py @@ -837,9 +837,20 @@ class TestTranscribeRecording: monkeypatch.setattr("tools.voice_mode._TEMP_DIR", str(temp_dir)) monkeypatch.setattr("tools.transcription_tools.MAX_FILE_SIZE", 70 * 1024) + call_count = 0 seen_paths = [] def fake_transcribe(path, model=None): + nonlocal call_count + call_count += 1 + # First call is on the original file — simulate remote provider + # rejecting it as too large so chunking kicks in. + if call_count == 1: + return { + "success": False, + "transcript": "", + "error": "File too large: 0.1MB (max 0.1MB)", + } seen_paths.append(path) assert model == "base" assert path != str(wav_path) @@ -877,7 +888,17 @@ class TestTranscribeRecording: monkeypatch.setattr("tools.voice_mode._TEMP_DIR", str(temp_dir)) monkeypatch.setattr("tools.transcription_tools.MAX_FILE_SIZE", 70 * 1024) + call_count = 0 + def fake_transcribe(path, model=None): + nonlocal call_count + call_count += 1 + if call_count == 1: + return { + "success": False, + "transcript": "", + "error": "File too large: 0.1MB (max 0.1MB)", + } return {"success": False, "transcript": "", "error": "provider rejected audio"} with patch("tools.transcription_tools.transcribe_audio", side_effect=fake_transcribe): @@ -889,6 +910,97 @@ class TestTranscribeRecording: assert "provider rejected audio" in result["error"] assert list(temp_dir.iterdir()) == [] + def test_trusts_transcribe_audio_skip_chunk_for_local(self, tmp_path, monkeypatch): + """Local providers never return 'File too large' — no chunking.""" + wav_path = tmp_path / "record.wav" + n_frames = 50000 + audio = struct.pack(f"<{n_frames}h", *([1000] * n_frames)) + with wave.open(str(wav_path), "wb") as wf: + wf.setnchannels(1) + wf.setsampwidth(2) + wf.setframerate(16000) + wf.writeframes(audio) + + mock_transcribe = MagicMock(return_value={ + "success": True, + "transcript": "local whisper result", + "provider": "local", + }) + + with patch("tools.transcription_tools.transcribe_audio", mock_transcribe): + from tools.voice_mode import transcribe_recording + result = transcribe_recording(str(wav_path), model="base") + + assert result["success"] is True + assert result["transcript"] == "local whisper result" + assert "chunks" not in result + mock_transcribe.assert_called_once_with(str(wav_path), model="base") + + def test_chunks_when_transcribe_audio_returns_file_too_large(self, tmp_path, monkeypatch): + """Remote provider rejects large file → chunking fallback.""" + wav_path = tmp_path / "record.wav" + n_frames = 50000 + audio = struct.pack(f"<{n_frames}h", *([1000] * n_frames)) + with wave.open(str(wav_path), "wb") as wf: + wf.setnchannels(1) + wf.setsampwidth(2) + wf.setframerate(16000) + wf.writeframes(audio) + + temp_dir = tmp_path / "chunks" + temp_dir.mkdir() + monkeypatch.setattr("tools.voice_mode._TEMP_DIR", str(temp_dir)) + monkeypatch.setattr("tools.transcription_tools.MAX_FILE_SIZE", 70 * 1024) + + call_count = 0 + + def fake_transcribe(path, model=None): + nonlocal call_count + call_count += 1 + if call_count == 1: + return { + "success": False, + "transcript": "", + "error": "File too large: 30.0MB (max 25MB)", + } + return { + "success": True, + "transcript": f"chunk {call_count - 1}", + "provider": "openai", + } + + with patch("tools.transcription_tools.transcribe_audio", side_effect=fake_transcribe): + from tools.voice_mode import transcribe_recording + result = transcribe_recording(str(wav_path), model="whisper-1") + + assert result["success"] is True + assert result.get("chunks", 0) > 1 + + def test_other_error_does_not_trigger_chunk(self, tmp_path, monkeypatch): + """Non-size errors from transcribe_audio are returned as-is.""" + wav_path = tmp_path / "record.wav" + n_frames = 50000 + audio = struct.pack(f"<{n_frames}h", *([1000] * n_frames)) + with wave.open(str(wav_path), "wb") as wf: + wf.setnchannels(1) + wf.setsampwidth(2) + wf.setframerate(16000) + wf.writeframes(audio) + + mock_transcribe = MagicMock(return_value={ + "success": False, + "transcript": "", + "error": "STT is disabled in config.yaml", + }) + + with patch("tools.transcription_tools.transcribe_audio", mock_transcribe): + from tools.voice_mode import transcribe_recording + result = transcribe_recording(str(wav_path), model="base") + + assert result["success"] is False + assert "STT is disabled" in result["error"] + mock_transcribe.assert_called_once() + class TestWhisperHallucinationFilter: def test_known_hallucinations(self): diff --git a/tools/transcription_tools.py b/tools/transcription_tools.py index 69124a2808cb..50a2ee89085f 100644 --- a/tools/transcription_tools.py +++ b/tools/transcription_tools.py @@ -393,6 +393,14 @@ def _resolve_command_stt_provider_config( return None +def _is_local_stt_provider(provider: str, stt_config: Dict[str, Any]) -> bool: + """Return whether *provider* is exempt from Hermes's remote upload cap.""" + key = (provider or "").lower().strip() + if key in {"local", "local_command"}: + return True + return False + + def _iter_command_stt_providers(stt_config: Dict[str, Any]): """Yield (name, config) pairs for every declared command-type STT provider.""" if not isinstance(stt_config, dict): @@ -1094,7 +1102,26 @@ def _dispatch_to_plugin_provider( # --------------------------------------------------------------------------- -def _validate_audio_file(file_path: str) -> Optional[Dict[str, Any]]: +def _validate_audio_file_size(audio_path: Path) -> Optional[Dict[str, Any]]: + """Return an error when *audio_path* exceeds the remote upload cap.""" + try: + file_size = audio_path.stat().st_size + except OSError as e: + return {"success": False, "transcript": "", "error": f"Failed to access file: {e}"} + if file_size > MAX_FILE_SIZE: + return { + "success": False, + "transcript": "", + "error": f"File too large: {file_size / (1024*1024):.1f}MB (max {MAX_FILE_SIZE / (1024*1024):.0f}MB)", + } + return None + + +def _validate_audio_file( + file_path: str, + *, + enforce_size_limit: bool = True, +) -> Optional[Dict[str, Any]]: """Validate the audio file. Returns an error dict or None if OK.""" audio_path = Path(file_path) @@ -1110,17 +1137,12 @@ def _validate_audio_file(file_path: str) -> Optional[Dict[str, Any]]: "transcript": "", "error": f"Unsupported format: {audio_path.suffix}. Supported: {', '.join(sorted(SUPPORTED_FORMATS))}", } + if enforce_size_limit: + return _validate_audio_file_size(audio_path) try: - file_size = audio_path.stat().st_size - if file_size > MAX_FILE_SIZE: - return { - "success": False, - "transcript": "", - "error": f"File too large: {file_size / (1024*1024):.1f}MB (max {MAX_FILE_SIZE / (1024*1024):.0f}MB)", - } + audio_path.stat() except OSError as e: return {"success": False, "transcript": "", "error": f"Failed to access file: {e}"} - return None # --------------------------------------------------------------------------- @@ -1874,8 +1896,9 @@ def transcribe_audio(file_path: str, model: Optional[str] = None) -> Dict[str, A - "error" (str, optional): Error message if success is False - "provider" (str, optional): Which provider was used """ - # Validate input - error = _validate_audio_file(file_path) + # Apply common path validation before provider resolution so invalid files + # cannot trigger provider setup or lazy installation. + error = _validate_audio_file(file_path, enforce_size_limit=False) if error: return error @@ -1889,6 +1912,10 @@ def transcribe_audio(file_path: str, model: Optional[str] = None) -> Dict[str, A } provider = _get_provider(stt_config) + if not _is_local_stt_provider(provider, stt_config): + error = _validate_audio_file_size(Path(file_path)) + if error: + return error if provider == "local": local_cfg = stt_config.get("local") or {} diff --git a/tools/voice_mode.py b/tools/voice_mode.py index 093534c3be52..6f8b22bd520f 100644 --- a/tools/voice_mode.py +++ b/tools/voice_mode.py @@ -941,10 +941,13 @@ def transcribe_recording(wav_path: str, model: Optional[str] = None) -> Dict[str """ from tools.transcription_tools import MAX_FILE_SIZE, transcribe_audio - if _should_chunk_for_transcription(wav_path, MAX_FILE_SIZE): + result = transcribe_audio(wav_path, model=model) + + # Only chunk when the provider itself reports "File too large" — + # local providers (faster-whisper, whisper.cpp, etc.) have no upload + # cap so ``transcribe_audio`` will never return this error for them. + if not result.get("success") and "File too large" in result.get("error", ""): result = _transcribe_wav_in_chunks(wav_path, model=model, max_file_size=MAX_FILE_SIZE) - else: - result = transcribe_audio(wav_path, model=model) # Filter out Whisper hallucinations (common on silent/near-silent audio) if result.get("success") and is_whisper_hallucination(result.get("transcript", "")):