fix(tts): preserve Telegram Opus output for native providers

tests/tools/test_tts_speed.py

@@ -136,12 +136,12 @@ def _hex_response(payload_audio: bytes = b"\x00\x01\x02\x03"):
 class TestMinimaxTtsT2aV2:
     """Default path: base_url contains 't2a_v2'."""
 
-    def _run(self, tts_config, tmp_path, monkeypatch, response=None):
+    def _run(self, tts_config, tmp_path, monkeypatch, response=None, suffix=".mp3"):
         monkeypatch.setenv("MINIMAX_API_KEY", "test-key")
         resp = response if response is not None else _hex_response()
         with patch("requests.post", return_value=resp) as mock_post:
             from tools.tts_tool import _generate_minimax_tts
-            output = _generate_minimax_tts("Hello", str(tmp_path / "out.mp3"), tts_config)
+            output = _generate_minimax_tts("Hello", str(tmp_path / f"out{suffix}"), tts_config)
         return mock_post, output
 
     def test_nested_payload(self, tmp_path, monkeypatch):
@@ -157,6 +157,12 @@ def test_nested_payload(self, tmp_path, monkeypatch):
         # Don't send flat top-level voice_id alongside nested voice_setting.
         assert "voice_id" not in payload
 
+    def test_ogg_output_requests_opus(self, tmp_path, monkeypatch):
+        """Native Opus is requested when the output path is an OGG file."""
+        mock_post, _ = self._run({}, tmp_path, monkeypatch, suffix=".ogg")
+        payload = mock_post.call_args[1]["json"]
+        assert payload["audio_setting"]["format"] == "opus"
+
     def test_decodes_hex_audio(self, tmp_path, monkeypatch):
         """t2a_v2 hex-encoded audio is decoded and written verbatim."""
         _, output = self._run({}, tmp_path, monkeypatch)
@@ -207,6 +213,74 @@ def test_api_error_raises(self, tmp_path, monkeypatch):
             self._run({}, tmp_path, monkeypatch, response=resp)
 
 
+class TestMinimaxTelegramOutput:
+    def test_explicit_mp3_path_is_rewritten_to_native_ogg(self, tmp_path):
+        """Gateway temp paths must not force Telegram MiniMax replies to MP3."""
+        import json
+
+        from tools.tts_tool import text_to_speech_tool
+
+        requested_paths = []
+
+        def fake_generate(_text, output_path, _config):
+            requested_paths.append(output_path)
+            with open(output_path, "wb") as f:
+                f.write(b"opus")
+
+        with patch(
+            "tools.tts_tool._load_tts_config",
+            return_value={"provider": "minimax"},
+        ), patch(
+            "gateway.session_context.get_session_env",
+            return_value="telegram",
+        ), patch(
+            "tools.tts_tool._generate_minimax_tts",
+            side_effect=fake_generate,
+        ):
+            result = json.loads(text_to_speech_tool(
+                "Hello", output_path=str(tmp_path / "reply.mp3")
+            ))
+
+        assert requested_paths == [str(tmp_path / "reply.ogg")]
+        assert result["file_path"] == str(tmp_path / "reply.ogg")
+        assert result["voice_compatible"] is True
+
+
+class TestElevenLabsTelegramOutput:
+    def test_explicit_mp3_path_is_rewritten_to_native_ogg(self, tmp_path):
+        """Gateway temp paths must not force Telegram ElevenLabs replies to MP3."""
+        import json
+
+        from tools.tts_tool import text_to_speech_tool
+
+        requested_paths = []
+
+        def fake_generate(_text, output_path, _config):
+            requested_paths.append(output_path)
+            with open(output_path, "wb") as f:
+                f.write(b"opus")
+
+        with patch(
+            "tools.tts_tool._load_tts_config",
+            return_value={"provider": "elevenlabs"},
+        ), patch(
+            "gateway.session_context.get_session_env",
+            return_value="telegram",
+        ), patch(
+            "tools.tts_tool._import_elevenlabs",
+        ), patch(
+            "tools.tts_tool._generate_elevenlabs",
+            side_effect=fake_generate,
+        ):
+            result = json.loads(text_to_speech_tool(
+                "Hello", output_path=str(tmp_path / "reply.mp3")
+            ))
+
+        assert requested_paths == [str(tmp_path / "reply.ogg")]
+        assert result["file_path"] == str(tmp_path / "reply.ogg")
+        assert result["voice_compatible"] is True
+
+
 class TestMinimaxTtsLegacyTextToSpeech:
     """Legacy path: caller pins base_url to the old text_to_speech endpoint."""
 

tools/tts_tool.py

@@ -1245,7 +1245,7 @@ def _generate_minimax_tts(text: str, output_path: str, tts_config: Dict[str, Any
             "audio_setting": {
                 "sample_rate": sample_rate,
                 "bitrate": bitrate,
-                "format": "mp3",
+                "format": "opus" if output_path.endswith(".ogg") else "mp3",
                 "channel": 1,
             },
         }
@@ -1834,6 +1834,18 @@ def _generate_kittentts(text: str, output_path: str, tts_config: Dict[str, Any])
 # ===========================================================================
 # Main tool function
 # ===========================================================================
+def _supports_opus_output(provider: str, tts_config: Dict[str, Any]) -> bool:
+    """Return whether *provider* can write an Opus OGG output path."""
+    if provider in {"elevenlabs", "openai", "mistral", "gemini"}:
+        return True
+    if provider == "minimax":
+        base_url = tts_config.get("minimax", {}).get(
+            "base_url", DEFAULT_MINIMAX_BASE_URL
+        )
+        return "t2a_v2" in base_url
+    return False
+
+
 def text_to_speech_tool(
     text: str,
     output_path: Optional[str] = None,
@@ -1884,6 +1896,7 @@ def text_to_speech_tool(
     from gateway.session_context import get_session_env
     platform = get_session_env("HERMES_SESSION_PLATFORM", "").lower()
     want_opus = (platform == "telegram")
+    supports_opus_output = _supports_opus_output(provider, tts_config)
 
     # Determine output path
     if output_path:
@@ -1913,6 +1926,8 @@ def text_to_speech_tool(
             file_path = _configured_command_tts_output_path(
                 file_path, command_provider_config
             )
+        elif want_opus and supports_opus_output and file_path.suffix.lower() == ".mp3":
+            file_path = file_path.with_suffix(".ogg")
     else:
         timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
         out_dir = Path(DEFAULT_OUTPUT_DIR)
@@ -1922,7 +1937,7 @@ def text_to_speech_tool(
             file_path = out_dir / f"tts_{timestamp}.{fmt}"
         # Use .ogg for Telegram with providers that support native Opus output,
         # otherwise fall back to .mp3 (Edge TTS will attempt ffmpeg conversion later).
-        elif want_opus and provider in {"openai", "elevenlabs", "mistral", "gemini"}:
+        elif want_opus and supports_opus_output:
             file_path = out_dir / f"tts_{timestamp}.ogg"
         else:
             file_path = out_dir / f"tts_{timestamp}.mp3"
@@ -2110,7 +2125,7 @@ def text_to_speech_tool(
             if opus_path:
                 file_str = opus_path
                 voice_compatible = True
-        elif provider in {"elevenlabs", "openai", "mistral", "gemini"}:
+        elif supports_opus_output:
             voice_compatible = want_opus and file_str.endswith(".ogg")
 
         file_size = os.path.getsize(file_str)

website/docs/user-guide/features/tts.md

@@ -141,7 +141,7 @@ Telegram voice bubbles require Opus/OGG audio format:
 
 - **OpenAI, ElevenLabs, and Mistral** produce Opus natively — no extra setup
 - **Edge TTS** (default) outputs MP3 and needs **ffmpeg** to convert:
-- **MiniMax TTS** outputs MP3 and needs **ffmpeg** to convert for Telegram voice bubbles
+- **MiniMax TTS** produces Opus natively with the default `t2a_v2` endpoint; legacy endpoints output MP3 and need **ffmpeg**
 - **Google Gemini TTS** outputs raw PCM and uses **ffmpeg** to encode Opus directly for Telegram voice bubbles
 - **xAI TTS** outputs MP3 and needs **ffmpeg** to convert for Telegram voice bubbles
 - **NeuTTS** outputs WAV and also needs **ffmpeg** to convert for Telegram voice bubbles