fix(gateway): make Telegram auto-TTS provider-aware

gateway/platforms/base.py

@@ -15,6 +15,7 @@
 import socket as _socket
 import subprocess
 import sys
+import tempfile
 import time
 import uuid
 from abc import ABC, abstractmethod
@@ -130,6 +131,22 @@ def should_send_media_as_audio(platform, ext: str, is_voice: bool = False) -> bo
     return True
 
 
+def build_auto_tts_output_path(platform) -> str:
+    """Return a unique temp path for gateway auto-TTS output.
+
+    The TTS tool may rewrite the suffix for explicit Telegram auto-TTS
+    depending on provider capability. Start from MP3 so conversion-based
+    providers do not skip their Opus conversion path.
+    """
+    audio_path = os.path.join(
+        tempfile.gettempdir(),
+        "hermes_voice",
+        f"tts_reply_{uuid.uuid4().hex[:12]}.mp3",
+    )
+    os.makedirs(os.path.dirname(audio_path), exist_ok=True)
+    return audio_path
+
+
 def utf16_len(s: str) -> int:
     """Count UTF-16 code units in *s*.
 
@@ -4954,6 +4971,8 @@ async def _stop_typing_task() -> None:
                 # an explicit ``/voice on|tts`` opt-in OR when ``voice.auto_tts`` is
                 # True globally and no ``/voice off`` has been issued.
                 _tts_path = None
+                _tts_requested_path = None
+                _tts_attempted_path = None
                 if (self._should_auto_tts_for_chat(event.source.chat_id)
                         and event.message_type == MessageType.VOICE
                         and text_content
@@ -4965,16 +4984,24 @@ async def _stop_typing_task() -> None:
                             speech_text = self.prepare_tts_text(text_content)
                             if not speech_text:
                                 raise ValueError("Empty text after markdown cleanup")
+                            _tts_requested_path = build_auto_tts_output_path(self.platform)
                             tts_result_str = await asyncio.to_thread(
-                                text_to_speech_tool, text=speech_text
+                                text_to_speech_tool,
+                                text=speech_text,
+                                output_path=_tts_requested_path,
+                                target_platform=self.platform,
+                                prefer_voice=True,
                             )
                             tts_data = _json.loads(tts_result_str)
-                            _tts_path = tts_data.get("file_path")
+                            _tts_attempted_path = tts_data.get("attempted_file_path")
+                            if tts_data.get("success", True):
+                                _tts_path = tts_data.get("file_path") or _tts_requested_path
                     except Exception as tts_err:
                         logger.warning("[%s] Auto-TTS failed: %s", self.name, tts_err)
 
                 # Play TTS audio before text (voice-first experience)
                 _tts_caption_delivered = False
+                _tts_cleanup_paths = {_tts_requested_path, _tts_path, _tts_attempted_path} - {None}
                 if _tts_path and Path(_tts_path).exists():
                     try:
                         telegram_tts_caption = None
@@ -4994,8 +5021,15 @@ async def _stop_typing_task() -> None:
                             telegram_tts_caption and getattr(tts_result, "success", False)
                         )
                     finally:
+                        for _cleanup_path in _tts_cleanup_paths:
+                            try:
+                                os.remove(_cleanup_path)
+                            except OSError:
+                                pass
+                elif _tts_cleanup_paths:
+                    for _cleanup_path in _tts_cleanup_paths:
                         try:
-                            os.remove(_tts_path)
+                            os.remove(_cleanup_path)
                         except OSError:
                             pass
 

gateway/run.py

@@ -36,7 +36,6 @@
 import site
 import sys
 import signal
-import tempfile
 import threading
 import time
 import sqlite3
@@ -1764,6 +1763,7 @@ def _profile_runtime_scope(profile_home: "Path"):
     MessageType,
     _prefix_within_utf16_limit,
     _reply_anchor_for_event,
+    build_auto_tts_output_path,
     merge_pending_message_event,
     utf16_len,
 )
@@ -13069,9 +13069,9 @@ def _should_echo_stt_transcripts(self) -> bool:
 
     async def _send_voice_reply(self, event: MessageEvent, text: str) -> None:
         """Generate TTS audio and send as a voice message before the text reply."""
-        import uuid as _uuid
         audio_path = None
         actual_path = None
+        attempted_path = None
         try:
             from tools.tts_tool import text_to_speech_tool, _strip_markdown_for_tts
 
@@ -13081,15 +13081,15 @@ async def _send_voice_reply(self, event: MessageEvent, text: str) -> None:
 
             # Telegram's adapter only sends native voice bubbles for OGG/Opus.
             # Other platforms keep the existing MP3 default.
-            audio_ext = "ogg" if event.source.platform == Platform.TELEGRAM else "mp3"
-            audio_path = os.path.join(
-                tempfile.gettempdir(), "hermes_voice",
-                f"tts_reply_{_uuid.uuid4().hex[:12]}.{audio_ext}",
-            )
-            os.makedirs(os.path.dirname(audio_path), exist_ok=True)
+            # build_auto_tts_output_path uses uuid-backed names to avoid collisions.
+            audio_path = build_auto_tts_output_path(event.source.platform)
 
             result_json = await asyncio.to_thread(
-                text_to_speech_tool, text=tts_text, output_path=audio_path
+                text_to_speech_tool,
+                text=tts_text,
+                output_path=audio_path,
+                target_platform=event.source.platform,
+                prefer_voice=True,
             )
             try:
                 result = json.loads(result_json)
@@ -13098,6 +13098,7 @@ async def _send_voice_reply(self, event: MessageEvent, text: str) -> None:
                 return
 
             # Use the actual file path from result (may differ after opus conversion)
+            attempted_path = result.get("attempted_file_path")
             actual_path = result.get("file_path", audio_path)
             if not result.get("success") or not os.path.isfile(actual_path):
                 logger.warning("Auto voice reply TTS failed: %s", result.get("error"))
@@ -13137,7 +13138,7 @@ async def _send_voice_reply(self, event: MessageEvent, text: str) -> None:
         except Exception as e:
             logger.warning("Auto voice reply failed: %s", e, exc_info=True)
         finally:
-            for p in {audio_path, actual_path} - {None}:
+            for p in {audio_path, actual_path, attempted_path} - {None}:
                 try:
                     os.unlink(p)
                 except OSError: