fix(stt): provider robustness — device config, CUDA fallbacks, container normalization (SILK/m4a), lazy-install recovery

contributors/emails/anthony.ai.assistant@gmail.com

@@ -0,0 +1 @@
+AnthonyFrancis

contributors/emails/carl@sempervirens.no

@@ -0,0 +1 @@
+carljborg

contributors/emails/damian.kluk.92@gmail.com

@@ -0,0 +1 @@
+damiankluk

contributors/emails/richardhojunjang@gmail.com

@@ -0,0 +1 @@
+RichardHojunJang

contributors/emails/tusharanshu18@gmail.com

@@ -0,0 +1 @@
+tusharui

contributors/emails/zehuaw@mit.edu

@@ -0,0 +1 @@
+zehuaw1

gateway/run.py

@@ -17971,7 +17971,17 @@ async def _enrich_message_with_transcription(
                 return f"{prefix}\n\n{user_text}", []
             return prefix, []
 
-        from tools.transcription_tools import transcribe_audio
+        try:
+            from tools.transcription_tools import transcribe_audio
+        except ModuleNotFoundError as e:
+            logger.error("Transcription module unavailable: %s", e)
+            unavailable_note = "[voice message could not be transcribed]"
+            _placeholder = "(The user sent a message with no text content)"
+            if user_text and user_text.strip() == _placeholder:
+                return unavailable_note, []
+            if user_text:
+                return f"{unavailable_note}\n\n{user_text}", []
+            return unavailable_note, []
 
         enriched_parts = []
         successful_transcripts: List[str] = []

tests/gateway/test_stt_config.py

@@ -143,6 +143,44 @@ async def test_enrich_message_with_transcription_returns_tuple_for_empty_content
     assert transcripts == ["hello from a captionless voice note"]
 
 
+@pytest.mark.parametrize(
+    ("user_text", "expected_text"),
+    [
+        ("caption", "[voice message could not be transcribed]\n\ncaption"),
+        ("", "[voice message could not be transcribed]"),
+        (
+            "(The user sent a message with no text content)",
+            "[voice message could not be transcribed]",
+        ),
+    ],
+)
+@pytest.mark.asyncio
+async def test_enrich_message_with_transcription_handles_missing_transcription_module_gracefully(
+    user_text,
+    expected_text,
+):
+    from gateway.run import GatewayRunner
+
+    runner = GatewayRunner.__new__(GatewayRunner)
+    runner.config = GatewayConfig(stt_enabled=True)
+
+    real_import = __import__
+
+    def fake_import(name, globals=None, locals=None, fromlist=(), level=0):
+        if name == "tools.transcription_tools":
+            raise ModuleNotFoundError("No module named 'tools.transcription_tools'")
+        return real_import(name, globals, locals, fromlist, level)
+
+    with patch("builtins.__import__", side_effect=fake_import):
+        result, transcripts = await runner._enrich_message_with_transcription(
+            user_text,
+            ["/tmp/voice.ogg"],
+        )
+
+    assert result == expected_text
+    assert transcripts == []
+
+
 @pytest.mark.asyncio
 async def test_prepare_inbound_message_text_transcribes_queued_voice_event():
     from gateway.run import GatewayRunner

tests/tools/test_managed_media_gateways.py

@@ -169,6 +169,7 @@ def close(self):
         APIError=Exception,
         APIConnectionError=Exception,
         APITimeoutError=Exception,
+        BadRequestError=type("BadRequestError", (Exception,), {}),
     )
     sys.modules["openai"] = fake_module
 
@@ -347,6 +348,33 @@ def test_transcription_uses_model_specific_response_formats(monkeypatch, tmp_pat
     assert json_capture["close_calls"] == 1
 
 
+@pytest.mark.parametrize(
+    ("transcription", "expected"),
+    [
+        ("language English<asr_text>Hello from Qwen.", "Hello from Qwen."),
+        (
+            types.SimpleNamespace(text="language Chinese<asr_text>Object response."),
+            "Object response.",
+        ),
+        (
+            {"text": "language English<asr_text>Dictionary response."},
+            "Dictionary response.",
+        ),
+    ],
+)
+def test_extract_transcript_text_strips_qwen3_asr_prefix(
+    transcription,
+    expected,
+):
+    _install_fake_tools_package()
+    transcription_tools = _load_tool_module(
+        "tools.transcription_tools",
+        "transcription_tools.py",
+    )
+
+    assert transcription_tools._extract_transcript_text(transcription) == expected
+
+
 PLUGINS_DIR = Path(__file__).resolve().parents[2] / "plugins"
 
 

tests/tools/test_transcription_command_providers.py

@@ -396,7 +396,7 @@ def test_model_override_passed_to_template(self, tmp_path):
         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 @@ def test_model_override_passed_to_template(self, tmp_path):
     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 @@ def test_config_model_used_when_no_override(self, tmp_path):
     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 @@ def test_language_from_provider_config_wins(self, tmp_path):
     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 @@ def test_language_falls_back_to_stt_section(self, tmp_path):
     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 @@ def test_command_provider_dispatches_via_transcribe_audio(self, tmp_path):
         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
@@ -510,7 +528,10 @@ def test_unknown_provider_no_command_falls_through_to_error(self, tmp_path):
         with patch("tools.transcription_tools._load_stt_config", return_value=cfg):
             result = transcribe_audio(str(audio))
         assert result["success"] is False
-        assert "No STT provider available" in result["error"]
+        # Explicitly-configured unknown providers now get a named
+        # registration error instead of the generic legacy message.
+        assert result["error_type"] == "provider_not_registered"
+        assert "unknown-cli" in result["error"]
 
 
 # ---------------------------------------------------------------------------