From 1b4a0d5eb455e58ae547aefb6c2ec6deb60c90c2 Mon Sep 17 00:00:00 2001 From: Gilad Bauman Date: Fri, 10 Jul 2026 11:35:30 +0000 Subject: [PATCH] fix(gateway): make Telegram auto-TTS provider-aware --- gateway/platforms/base.py | 40 +++- gateway/run.py | 21 +- tests/gateway/test_auto_voice_reply_format.py | 219 +++++++++++++++++- tests/gateway/test_voice_command.py | 6 +- tests/tools/test_tts_opus_routing.py | 187 +++++++++++++++ tools/tts_tool.py | 140 +++++++---- 6 files changed, 538 insertions(+), 75 deletions(-) diff --git a/gateway/platforms/base.py b/gateway/platforms/base.py index 21ca4af6bb4d..16db4fd1aa43 100644 --- a/gateway/platforms/base.py +++ b/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 diff --git a/gateway/run.py b/gateway/run.py index ccfa8e92c143..e39f67e3b3e7 100644 --- a/gateway/run.py +++ b/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: diff --git a/tests/gateway/test_auto_voice_reply_format.py b/tests/gateway/test_auto_voice_reply_format.py index eeb39ab60e78..4b379a706a92 100644 --- a/tests/gateway/test_auto_voice_reply_format.py +++ b/tests/gateway/test_auto_voice_reply_format.py @@ -1,35 +1,40 @@ """Tests for gateway auto-TTS voice reply audio format selection.""" +import asyncio import json from pathlib import Path from unittest.mock import AsyncMock, MagicMock, patch import pytest -from gateway.config import Platform -from gateway.platforms.base import MessageEvent +from gateway.config import Platform, PlatformConfig +from gateway.platforms.base import BasePlatformAdapter, MessageEvent, MessageType, SendResult from gateway.run import GatewayRunner -from gateway.session import SessionSource +from gateway.session import SessionSource, build_session_key class TestAutoVoiceReplyFormat: @pytest.mark.asyncio async def test_telegram_auto_voice_reply_requests_ogg_for_native_voice_bubble(self): - """Telegram auto-TTS should request OGG/Opus so send_voice sends a voice bubble.""" + """Telegram auto-TTS should target Telegram and send returned OGG path.""" runner = _make_runner() adapter = _make_adapter(Platform.TELEGRAM) runner.adapters[Platform.TELEGRAM] = adapter event = _make_event(Platform.TELEGRAM) requested_paths = [] - def fake_tts(*, text, output_path): + def fake_tts(*, text, output_path, target_platform, prefer_voice): requested_paths.append(output_path) - assert output_path.endswith(".ogg") + assert output_path.endswith(".mp3") + assert target_platform == Platform.TELEGRAM + assert prefer_voice is True Path(output_path).parent.mkdir(parents=True, exist_ok=True) - Path(output_path).write_bytes(b"fake ogg opus") + Path(output_path).write_bytes(b"fake intermediate mp3") + actual_path = Path(output_path).with_suffix(".ogg") + actual_path.write_bytes(b"fake ogg opus") return json.dumps({ "success": True, - "file_path": output_path, + "file_path": str(actual_path), "provider": "gemini", "voice_compatible": True, }) @@ -38,7 +43,7 @@ def fake_tts(*, text, output_path): await runner._send_voice_reply(event, "hello from auto tts") assert requested_paths - assert requested_paths[0].endswith(".ogg") + assert requested_paths[0].endswith(".mp3") adapter.send_voice.assert_awaited_once() assert adapter.send_voice.await_args.kwargs["audio_path"].endswith(".ogg") @@ -51,9 +56,11 @@ async def test_non_telegram_auto_voice_reply_keeps_mp3_default(self): event = _make_event(Platform.SLACK) requested_paths = [] - def fake_tts(*, text, output_path): + def fake_tts(*, text, output_path, target_platform, prefer_voice): requested_paths.append(output_path) assert output_path.endswith(".mp3") + assert target_platform == Platform.SLACK + assert prefer_voice is True Path(output_path).parent.mkdir(parents=True, exist_ok=True) Path(output_path).write_bytes(b"fake mp3") return json.dumps({ @@ -71,6 +78,156 @@ def fake_tts(*, text, output_path): adapter.send_voice.assert_awaited_once() assert adapter.send_voice.await_args.kwargs["audio_path"].endswith(".mp3") + @pytest.mark.asyncio + async def test_telegram_auto_voice_reply_cleans_attempted_path_on_tts_failure(self, tmp_path, monkeypatch): + """Failed native-provider auto-TTS must clean the rewritten attempted path.""" + monkeypatch.setattr("gateway.platforms.base.tempfile.gettempdir", lambda: str(tmp_path)) + runner = _make_runner() + adapter = _make_adapter(Platform.TELEGRAM) + runner.adapters[Platform.TELEGRAM] = adapter + event = _make_event(Platform.TELEGRAM) + attempted_paths = [] + + def fake_tts(*, text, output_path, target_platform, prefer_voice): + requested_path = Path(output_path) + attempted_path = requested_path.with_suffix(".ogg") + requested_path.write_bytes(b"requested mp3") + attempted_path.write_bytes(b"partial ogg") + attempted_paths.append((requested_path, attempted_path)) + return json.dumps({ + "success": False, + "error": "provider failed", + "attempted_file_path": str(attempted_path), + }) + + with patch("tools.tts_tool.text_to_speech_tool", side_effect=fake_tts): + await runner._send_voice_reply(event, "hello from auto tts") + + requested_path, attempted_path = attempted_paths[0] + adapter.send_voice.assert_not_awaited() + assert not requested_path.exists() + assert not attempted_path.exists() + + +class TestBaseAdapterAutoVoiceReplyFormat: + @pytest.mark.asyncio + async def test_telegram_voice_input_auto_tts_targets_platform_and_sends_returned_path(self): + """Base adapter voice-input auto-TTS should send provider-returned OGG.""" + adapter = _AutoTtsAdapter(Platform.TELEGRAM) + adapter._keep_typing = _hold_typing + adapter._should_auto_tts_for_chat = lambda _chat_id: True + adapter.play_tts = AsyncMock(return_value=SendResult(success=True, message_id="voice-1")) + adapter.set_message_handler(lambda _event: asyncio.sleep(0, result="short reply")) + event = _make_voice_event(Platform.TELEGRAM) + requested_paths = [] + actual_paths = [] + + def fake_tts(*, text, output_path, target_platform, prefer_voice): + requested_path = Path(output_path) + requested_paths.append(requested_path) + assert requested_path.suffix == ".mp3" + assert target_platform == Platform.TELEGRAM + assert prefer_voice is True + assert requested_path.parent.is_dir() + requested_path.write_bytes(b"intermediate mp3") + + actual_path = requested_path.with_name(f"{requested_path.stem}_actual.ogg") + actual_paths.append(actual_path) + actual_path.write_bytes(b"actual ogg") + return json.dumps({ + "success": True, + "file_path": str(actual_path), + "provider": "gemini", + "voice_compatible": True, + }) + + with patch("tools.tts_tool.check_tts_requirements", return_value=True), patch( + "tools.tts_tool.text_to_speech_tool", + side_effect=fake_tts, + ): + await adapter._process_message_background(event, build_session_key(event.source)) + + assert len(requested_paths) == 1 + assert requested_paths[0].suffix == ".mp3" + assert actual_paths[0].suffix == ".ogg" + adapter.play_tts.assert_awaited_once() + assert adapter.play_tts.await_args.kwargs["audio_path"] == str(actual_paths[0]) + assert not requested_paths[0].exists() + assert not actual_paths[0].exists() + + @pytest.mark.asyncio + async def test_non_telegram_voice_input_auto_tts_keeps_mp3_default(self): + """Base adapter voice-input auto-TTS should preserve MP3 for other platforms.""" + adapter = _AutoTtsAdapter(Platform.SLACK) + adapter._keep_typing = _hold_typing + adapter._should_auto_tts_for_chat = lambda _chat_id: True + adapter.play_tts = AsyncMock(return_value=SendResult(success=True, message_id="voice-1")) + adapter.set_message_handler(lambda _event: asyncio.sleep(0, result="short reply")) + event = _make_voice_event(Platform.SLACK) + requested_paths = [] + + def fake_tts(*, text, output_path, target_platform, prefer_voice): + requested_path = Path(output_path) + requested_paths.append(requested_path) + assert requested_path.suffix == ".mp3" + assert target_platform == Platform.SLACK + assert prefer_voice is True + assert requested_path.parent.is_dir() + requested_path.write_bytes(b"actual mp3") + return json.dumps({ + "success": True, + "file_path": str(requested_path), + "provider": "gemini", + "voice_compatible": False, + }) + + with patch("tools.tts_tool.check_tts_requirements", return_value=True), patch( + "tools.tts_tool.text_to_speech_tool", + side_effect=fake_tts, + ): + await adapter._process_message_background(event, build_session_key(event.source)) + + assert len(requested_paths) == 1 + assert requested_paths[0].suffix == ".mp3" + adapter.play_tts.assert_awaited_once() + assert adapter.play_tts.await_args.kwargs["audio_path"] == str(requested_paths[0]) + assert not requested_paths[0].exists() + + @pytest.mark.asyncio + async def test_telegram_voice_input_auto_tts_cleans_attempted_path_on_tts_failure(self, tmp_path, monkeypatch): + """Base adapter auto-TTS should clean attempted paths from failed TTS results.""" + monkeypatch.setattr("gateway.platforms.base.tempfile.gettempdir", lambda: str(tmp_path)) + adapter = _AutoTtsAdapter(Platform.TELEGRAM) + adapter._keep_typing = _hold_typing + adapter._should_auto_tts_for_chat = lambda _chat_id: True + adapter.play_tts = AsyncMock(return_value=SendResult(success=True, message_id="voice-1")) + adapter.set_message_handler(lambda _event: asyncio.sleep(0, result="short reply")) + event = _make_voice_event(Platform.TELEGRAM) + attempted_paths = [] + + def fake_tts(*, text, output_path, target_platform, prefer_voice): + requested_path = Path(output_path) + attempted_path = requested_path.with_suffix(".ogg") + requested_path.write_bytes(b"requested mp3") + attempted_path.write_bytes(b"partial ogg") + attempted_paths.append((requested_path, attempted_path)) + return json.dumps({ + "success": False, + "error": "provider failed", + "attempted_file_path": str(attempted_path), + }) + + with patch("tools.tts_tool.check_tts_requirements", return_value=True), patch( + "tools.tts_tool.text_to_speech_tool", + side_effect=fake_tts, + ): + await adapter._process_message_background(event, build_session_key(event.source)) + + requested_path, attempted_path = attempted_paths[0] + adapter.play_tts.assert_not_awaited() + assert not requested_path.exists() + assert not attempted_path.exists() + def _make_runner() -> GatewayRunner: with patch("gateway.run.GatewayRunner._load_voice_modes", return_value={}): @@ -98,3 +255,45 @@ def _make_event(platform: Platform) -> MessageEvent: ), message_id="456", ) + + +class _AutoTtsAdapter(BasePlatformAdapter): + def __init__(self, platform: Platform): + super().__init__(PlatformConfig(enabled=True, token="fake-token"), platform) + self.sent = [] + + async def connect(self, *, is_reconnect: bool = False) -> bool: + return True + + async def disconnect(self) -> None: + return None + + async def send(self, chat_id, content, reply_to=None, metadata=None) -> SendResult: + self.sent.append({ + "chat_id": chat_id, + "content": content, + "reply_to": reply_to, + "metadata": metadata, + }) + return SendResult(success=True, message_id="text-1") + + async def get_chat_info(self, chat_id: str): + return {"id": chat_id} + + +async def _hold_typing(_chat_id, interval=2.0, metadata=None): + await asyncio.Event().wait() + + +def _make_voice_event(platform: Platform) -> MessageEvent: + return MessageEvent( + text="voice input", + message_type=MessageType.VOICE, + source=SessionSource( + platform=platform, + chat_id="123", + user_id="u1", + user_name="User", + ), + message_id="voice-456", + ) diff --git a/tests/gateway/test_voice_command.py b/tests/gateway/test_voice_command.py index 539648886daf..46e42b98af6f 100644 --- a/tests/gateway/test_voice_command.py +++ b/tests/gateway/test_voice_command.py @@ -433,7 +433,9 @@ async def test_calls_tts_and_send_voice(self, runner): await runner._send_voice_reply(event, "Hello world") mock_adapter.send_voice.assert_called_once() - assert mock_tts.call_args.kwargs["output_path"].endswith(".ogg") + assert mock_tts.call_args.kwargs["output_path"].endswith(".mp3") + assert mock_tts.call_args.kwargs["target_platform"] == Platform.TELEGRAM + assert mock_tts.call_args.kwargs["prefer_voice"] is True call_args = mock_adapter.send_voice.call_args assert call_args.kwargs.get("chat_id") == "123" @@ -458,6 +460,8 @@ async def test_non_telegram_auto_voice_reply_uses_mp3(self, runner): mock_adapter.send_voice.assert_called_once() assert mock_tts.call_args.kwargs["output_path"].endswith(".mp3") + assert mock_tts.call_args.kwargs["target_platform"] == Platform.SLACK + assert mock_tts.call_args.kwargs["prefer_voice"] is True @pytest.mark.asyncio async def test_auto_voice_reply_uses_thread_metadata_helper(self, runner): diff --git a/tests/tools/test_tts_opus_routing.py b/tests/tools/test_tts_opus_routing.py index 0073146c3045..e4347a9caa81 100644 --- a/tests/tools/test_tts_opus_routing.py +++ b/tests/tools/test_tts_opus_routing.py @@ -68,3 +68,190 @@ def fake_convert(path: str) -> str: assert result["voice_compatible"] is True assert result["media_tag"] == f"[[audio_as_voice]]\nMEDIA:{opus}" convert.assert_called_once_with(str(out)) + + +def test_telegram_target_native_provider_aligns_explicit_path_to_ogg(tmp_path, monkeypatch): + requested = tmp_path / "speech.mp3" + generated_paths = [] + + def fake_gemini(text: str, output_path: str, tts_config: dict) -> str: + generated_paths.append(output_path) + assert output_path.endswith(".ogg") + Path(output_path).write_bytes(b"ogg") + return output_path + + monkeypatch.setattr(tts_tool, "_load_tts_config", lambda: {"provider": "gemini"}) + monkeypatch.setattr(tts_tool, "_generate_gemini_tts", fake_gemini) + monkeypatch.setattr(tts_tool, "_convert_to_opus", Mock()) + + result = json.loads( + tts_tool.text_to_speech_tool( + "hello", + output_path=str(requested), + target_platform="telegram", + ) + ) + + assert generated_paths == [str(requested.with_suffix(".ogg"))] + assert result["success"] is True + assert result["file_path"] == str(requested.with_suffix(".ogg")) + assert result["voice_compatible"] is True + assert result["media_tag"] == f"[[audio_as_voice]]\nMEDIA:{requested.with_suffix('.ogg')}" + tts_tool._convert_to_opus.assert_not_called() + + +def test_telegram_target_native_provider_failure_reports_attempted_path(tmp_path, monkeypatch): + requested = tmp_path / "speech.mp3" + attempted = requested.with_suffix(".ogg") + + def fake_gemini(text: str, output_path: str, tts_config: dict) -> str: + assert output_path == str(attempted) + Path(output_path).write_bytes(b"partial ogg") + raise RuntimeError("provider failed") + + monkeypatch.setattr(tts_tool, "_load_tts_config", lambda: {"provider": "gemini"}) + monkeypatch.setattr(tts_tool, "_generate_gemini_tts", fake_gemini) + + result = json.loads( + tts_tool.text_to_speech_tool( + "hello", + output_path=str(requested), + target_platform="telegram", + ) + ) + + assert result["success"] is False + assert result["attempted_file_path"] == str(attempted) + assert "file_path" not in result + assert "media_tag" not in result + + +def test_telegram_target_conversion_provider_starts_mp3_and_returns_ogg(tmp_path, monkeypatch): + requested = tmp_path / "speech.ogg" + generation_path = requested.with_suffix(".mp3") + opus = requested.with_suffix(".ogg") + generated_paths = [] + + async def fake_edge(text: str, output_path: str, tts_config: dict) -> str: + generated_paths.append(output_path) + assert output_path == str(generation_path) + Path(output_path).write_bytes(b"mp3") + return output_path + + def fake_convert(path: str) -> str: + assert path == str(generation_path) + opus.write_bytes(b"ogg") + return str(opus) + + convert = Mock(side_effect=fake_convert) + + monkeypatch.setattr(tts_tool, "_load_tts_config", lambda: {"provider": "edge"}) + monkeypatch.setattr(tts_tool, "_import_edge_tts", lambda: object()) + monkeypatch.setattr(tts_tool, "_generate_edge_tts", fake_edge) + monkeypatch.setattr(tts_tool, "_convert_to_opus", convert) + + result = json.loads( + tts_tool.text_to_speech_tool( + "hello", + output_path=str(requested), + target_platform="telegram", + ) + ) + + assert generated_paths == [str(generation_path)] + assert result["success"] is True + assert result["file_path"] == str(opus) + assert result["voice_compatible"] is True + assert result["media_tag"] == f"[[audio_as_voice]]\nMEDIA:{opus}" + convert.assert_called_once_with(str(generation_path)) + + +def test_target_platform_absent_preserves_manual_native_output_path(tmp_path, monkeypatch): + requested = tmp_path / "speech.mp3" + generated_paths = [] + + def fake_gemini(text: str, output_path: str, tts_config: dict) -> str: + generated_paths.append(output_path) + assert output_path.endswith(".mp3") + Path(output_path).write_bytes(b"mp3") + return output_path + + monkeypatch.setenv("HERMES_SESSION_PLATFORM", "telegram") + monkeypatch.setattr(tts_tool, "_load_tts_config", lambda: {"provider": "gemini"}) + monkeypatch.setattr(tts_tool, "_generate_gemini_tts", fake_gemini) + + result = json.loads(tts_tool.text_to_speech_tool("hello", output_path=str(requested))) + + assert generated_paths == [str(requested)] + assert result["success"] is True + assert result["file_path"] == str(requested) + assert result["voice_compatible"] is False + + +def test_telegram_target_command_provider_keeps_configured_format(tmp_path, monkeypatch): + requested = tmp_path / "speech.mp3" + convert = Mock() + + def fake_command(text, output_path, provider_name, config, tts_config): + Path(output_path).write_bytes(b"mp3") + return output_path + + monkeypatch.setattr( + tts_tool, + "_load_tts_config", + lambda: { + "provider": "cmd", + "providers": { + "cmd": { + "type": "command", + "command": "fake", + "output_format": "mp3", + "voice_compatible": False, + }, + }, + }, + ) + monkeypatch.setattr(tts_tool, "_generate_command_tts", fake_command) + monkeypatch.setattr(tts_tool, "_convert_to_opus", convert) + + result = json.loads( + tts_tool.text_to_speech_tool( + "hello", + output_path=str(requested), + target_platform="telegram", + ) + ) + + assert result["success"] is True + assert result["file_path"] == str(requested) + assert result["voice_compatible"] is False + assert result["media_tag"] == f"MEDIA:{requested}" + convert.assert_not_called() + + +def test_telegram_target_plugin_provider_keeps_non_voice_output(tmp_path, monkeypatch): + requested = tmp_path / "speech.mp3" + convert = Mock() + + def fake_plugin(text, output_path, provider, tts_config): + Path(output_path).write_bytes(b"mp3") + return output_path + + monkeypatch.setattr(tts_tool, "_load_tts_config", lambda: {"provider": "cartesia"}) + monkeypatch.setattr(tts_tool, "_dispatch_to_plugin_provider", fake_plugin) + monkeypatch.setattr(tts_tool, "_plugin_provider_is_voice_compatible", lambda provider: False) + monkeypatch.setattr(tts_tool, "_convert_to_opus", convert) + + result = json.loads( + tts_tool.text_to_speech_tool( + "hello", + output_path=str(requested), + target_platform="telegram", + ) + ) + + assert result["success"] is True + assert result["file_path"] == str(requested) + assert result["voice_compatible"] is False + assert result["media_tag"] == f"MEDIA:{requested}" + convert.assert_not_called() diff --git a/tools/tts_tool.py b/tools/tts_tool.py index ec7b361b8e14..81a586afb1f3 100644 --- a/tools/tts_tool.py +++ b/tools/tts_tool.py @@ -398,6 +398,8 @@ def _get_provider(tts_config: Dict[str, Any]) -> str: "kittentts", "piper", }) +NATIVE_OPUS_TTS_PROVIDERS = frozenset({"elevenlabs", "openai", "mistral", "gemini"}) +CONVERSION_OPUS_TTS_PROVIDERS = frozenset({"edge", "neutts", "minimax", "xai", "kittentts", "piper"}) DEFAULT_COMMAND_TTS_TIMEOUT_SECONDS = 120 DEFAULT_COMMAND_TTS_OUTPUT_FORMAT = "mp3" @@ -405,6 +407,27 @@ def _get_provider(tts_config: Dict[str, Any]) -> str: DEFAULT_COMMAND_TTS_MAX_TEXT_LENGTH = 5000 +def _normalize_platform_name(platform: Optional[str]) -> str: + value = getattr(platform, "value", platform) + return str(value or "").strip().lower() + + +def _auto_tts_effective_output_path(path: Path, provider: str) -> Path: + """Align an explicit auto-TTS path with the provider's Opus strategy.""" + if provider in NATIVE_OPUS_TTS_PROVIDERS: + return path.with_suffix(".ogg") + if provider in CONVERSION_OPUS_TTS_PROVIDERS: + return path.with_suffix(".mp3") + return path + + +def _tts_error_json(error: str, *, attempted_file_path: Optional[str] = None) -> str: + payload: Dict[str, Any] = {"success": False, "error": error} + if attempted_file_path: + payload["attempted_file_path"] = str(attempted_file_path) + return json.dumps(payload, ensure_ascii=False) + + def _get_provider_section(tts_config: Dict[str, Any], name: str) -> Dict[str, Any]: """Return a provider config block if it's a dict, else an empty dict.""" if not isinstance(tts_config, dict): @@ -2153,6 +2176,9 @@ def _generate_kittentts(text: str, output_path: str, tts_config: Dict[str, Any]) def text_to_speech_tool( text: str, output_path: Optional[str] = None, + *, + target_platform: Optional[str] = None, + prefer_voice: Optional[bool] = None, ) -> str: """ Convert text to speech audio. @@ -2167,6 +2193,10 @@ def text_to_speech_tool( Args: text: The text to convert to speech. output_path: Optional custom save path. Defaults to ~/voice-memos/.mp3 + target_platform: Internal gateway hint for automatic TTS delivery. + Not exposed in the model tool schema. + prefer_voice: Internal gateway hint to prefer native voice-bubble + output when the target platform supports it. Returns: str: JSON result with success, file_path, and optionally MEDIA tag. @@ -2193,13 +2223,19 @@ def text_to_speech_tool( ) text = text[:max_len] - # Detect platform from gateway env var to choose the best output format. - # Telegram voice bubbles require Opus (.ogg); OpenAI and ElevenLabs can - # produce Opus natively (no ffmpeg needed). Edge TTS always outputs MP3 - # and needs ffmpeg for conversion. + # Detect platform from an explicit gateway hint first, falling back to the + # historical ambient session context. Manual custom output paths are only + # rewritten when target_platform is explicit, preserving existing CLI/tool + # semantics when the hint is absent. from gateway.session_context import get_session_env - platform = get_session_env("HERMES_SESSION_PLATFORM", "").lower() - want_opus = (platform == "telegram") + explicit_target_platform = target_platform is not None + platform = ( + _normalize_platform_name(target_platform) + if explicit_target_platform + else get_session_env("HERMES_SESSION_PLATFORM", "").lower() + ) + prefer_voice = True if prefer_voice is None else bool(prefer_voice) + want_opus = platform == "telegram" and prefer_voice # Determine output path if output_path: @@ -2229,6 +2265,8 @@ def text_to_speech_tool( file_path = _configured_command_tts_output_path( file_path, command_provider_config ) + elif explicit_target_platform and want_opus: + file_path = _auto_tts_effective_output_path(file_path, provider) else: timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S") out_dir = Path(DEFAULT_OUTPUT_DIR) @@ -2238,7 +2276,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 provider in NATIVE_OPUS_TTS_PROVIDERS: file_path = out_dir / f"tts_{timestamp}.ogg" else: file_path = out_dir / f"tts_{timestamp}.mp3" @@ -2276,10 +2314,10 @@ def text_to_speech_tool( try: _import_elevenlabs() except ImportError: - return json.dumps({ - "success": False, - "error": "ElevenLabs provider selected but 'elevenlabs' package not installed. Run: pip install elevenlabs" - }, ensure_ascii=False) + return _tts_error_json( + "ElevenLabs provider selected but 'elevenlabs' package not installed. Run: pip install elevenlabs", + attempted_file_path=file_str, + ) logger.info("Generating speech with ElevenLabs...") _generate_elevenlabs(text, file_str, tts_config) @@ -2287,10 +2325,10 @@ def text_to_speech_tool( try: _import_openai_client() except ImportError: - return json.dumps({ - "success": False, - "error": "OpenAI provider selected but 'openai' package not installed." - }, ensure_ascii=False) + return _tts_error_json( + "OpenAI provider selected but 'openai' package not installed.", + attempted_file_path=file_str, + ) logger.info("Generating speech with OpenAI TTS...") _generate_openai_tts(text, file_str, tts_config) @@ -2306,11 +2344,11 @@ def text_to_speech_tool( try: _import_mistral_client() except ImportError: - return json.dumps({ - "success": False, - "error": "Mistral provider selected but 'mistralai' package not installed. " - "Run: pip install 'hermes-agent[mistral]'" - }, ensure_ascii=False) + return _tts_error_json( + "Mistral provider selected but 'mistralai' package not installed. " + "Run: pip install 'hermes-agent[mistral]'", + attempted_file_path=file_str, + ) logger.info("Generating speech with Mistral Voxtral TTS...") _generate_mistral_tts(text, file_str, tts_config) @@ -2320,11 +2358,11 @@ def text_to_speech_tool( elif provider == "neutts": if not _check_neutts_available(): - return json.dumps({ - "success": False, - "error": "NeuTTS provider selected but neutts is not installed. " - "Run hermes setup and choose NeuTTS, or install espeak-ng and run python -m pip install -U neutts[all]." - }, ensure_ascii=False) + return _tts_error_json( + "NeuTTS provider selected but neutts is not installed. " + "Run hermes setup and choose NeuTTS, or install espeak-ng and run python -m pip install -U neutts[all].", + attempted_file_path=file_str, + ) logger.info("Generating speech with NeuTTS (local)...") _generate_neutts(text, file_str, tts_config) @@ -2332,12 +2370,12 @@ def text_to_speech_tool( try: _import_kittentts() except ImportError: - return json.dumps({ - "success": False, - "error": "KittenTTS provider selected but 'kittentts' package not installed. " - "Run 'hermes setup tts' and choose KittenTTS, or install manually: " - "pip install https://github.com/KittenML/KittenTTS/releases/download/0.8.1/kittentts-0.8.1-py3-none-any.whl" - }, ensure_ascii=False) + return _tts_error_json( + "KittenTTS provider selected but 'kittentts' package not installed. " + "Run 'hermes setup tts' and choose KittenTTS, or install manually: " + "pip install https://github.com/KittenML/KittenTTS/releases/download/0.8.1/kittentts-0.8.1-py3-none-any.whl", + attempted_file_path=file_str, + ) logger.info("Generating speech with KittenTTS (local, ~25MB)...") _generate_kittentts(text, file_str, tts_config) @@ -2345,12 +2383,12 @@ def text_to_speech_tool( try: _import_piper() except ImportError: - return json.dumps({ - "success": False, - "error": "Piper provider selected but 'piper-tts' package not installed. " - "Run 'hermes tools' and select Piper under TTS, or install manually: " - "pip install piper-tts", - }, ensure_ascii=False) + return _tts_error_json( + "Piper provider selected but 'piper-tts' package not installed. " + "Run 'hermes tools' and select Piper under TTS, or install manually: " + "pip install piper-tts", + attempted_file_path=file_str, + ) logger.info("Generating speech with Piper (local)...") _generate_piper_tts(text, file_str, tts_config) @@ -2377,18 +2415,18 @@ def text_to_speech_tool( provider = "neutts" _generate_neutts(text, file_str, tts_config) else: - return json.dumps({ - "success": False, - "error": "No TTS provider available. Install edge-tts (pip install edge-tts) " - "or set up NeuTTS for local synthesis." - }, ensure_ascii=False) + return _tts_error_json( + "No TTS provider available. Install edge-tts (pip install edge-tts) " + "or set up NeuTTS for local synthesis.", + attempted_file_path=file_str, + ) # Check the file was actually created if not os.path.exists(file_str) or os.path.getsize(file_str) == 0: - return json.dumps({ - "success": False, - "error": f"TTS generation produced no output (provider: {provider})" - }, ensure_ascii=False) + return _tts_error_json( + f"TTS generation produced no output (provider: {provider})", + attempted_file_path=file_str, + ) # Try Opus conversion for Telegram compatibility. # Edge TTS outputs MP3, NeuTTS/KittenTTS output WAV. Keep those native @@ -2419,14 +2457,14 @@ def text_to_speech_tool( voice_compatible = file_str.endswith(".ogg") elif ( want_opus - and provider in {"edge", "neutts", "minimax", "xai", "kittentts", "piper"} + and provider in CONVERSION_OPUS_TTS_PROVIDERS and not file_str.endswith(".ogg") ): opus_path = _convert_to_opus(file_str) if opus_path: file_str = opus_path voice_compatible = True - elif provider in {"elevenlabs", "openai", "mistral", "gemini"}: + elif provider in NATIVE_OPUS_TTS_PROVIDERS: voice_compatible = want_opus and file_str.endswith(".ogg") file_size = os.path.getsize(file_str) @@ -2449,17 +2487,17 @@ def text_to_speech_tool( # Configuration errors (missing API keys, etc.) error_msg = f"TTS configuration error ({provider}): {e}" logger.error("%s", error_msg) - return tool_error(error_msg, success=False) + return _tts_error_json(error_msg, attempted_file_path=file_str) except FileNotFoundError as e: # Missing dependencies or files error_msg = f"TTS dependency missing ({provider}): {e}" logger.error("%s", error_msg, exc_info=True) - return tool_error(error_msg, success=False) + return _tts_error_json(error_msg, attempted_file_path=file_str) except Exception as e: # Unexpected errors error_msg = f"TTS generation failed ({provider}): {e}" logger.error("%s", error_msg, exc_info=True) - return tool_error(error_msg, success=False) + return _tts_error_json(error_msg, attempted_file_path=file_str) # ===========================================================================