test(gateway): regression coverage for platform-aware voice delivery

- tests/gateway/test_base_auto_tts_output_format.py (new): the base
  adapter auto-TTS block passes an explicit .ogg output path on every
  OPUS_VOICE_PLATFORMS member (parametrized from the tts_tool set — the
  single source of truth), keeps .mp3 on non-opus platforms, honors the
  tool's success flag, and stays unique/uuid-based.
- tests/gateway/test_auto_voice_reply_format.py: runner _send_voice_reply
  parametrized across Matrix/Feishu/WhatsApp/Signal (.ogg) alongside the
  existing Telegram/Slack cases; streamed+global-auto-TTS gate regression.
- tests/gateway/test_telegram_voice_caption_markdown.py (new): caption
  MarkdownV2 formatting, entity-rejection plain fallback, overflow skip,
  and no-caption passthrough (#32029).
- test_voice_command.py filename-uuid contract test updated to point at
  build_auto_tts_output_path (construction moved there).

tests/gateway/test_auto_voice_reply_format.py

@@ -70,6 +70,65 @@ class TestAutoVoiceReplyFormat:
         assert requested_paths[0].endswith(".mp3")
         adapter.send_voice.assert_awaited_once()
         assert adapter.send_voice.await_args.kwargs["audio_path"].endswith(".mp3")
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize(
+        "platform",
+        [Platform.MATRIX, Platform.FEISHU, Platform.WHATSAPP, Platform.SIGNAL],
+    )
+    async def test_opus_platform_auto_voice_reply_requests_ogg(self, platform):
+        """Every OPUS_VOICE_PLATFORMS member gets an explicit .ogg output path.
+
+        Regression for #14841 (Matrix) / #45557 (Feishu): _send_voice_reply
+        hardcoded .ogg for Telegram only, so Matrix/Feishu voice replies were
+        synthesized as MP3 and delivered as plain attachments instead of
+        native voice bubbles.
+        """
+        runner = _make_runner()
+        adapter = _make_adapter(platform)
+        runner.adapters[platform] = adapter
+        event = _make_event(platform)
+        requested_paths = []
+
+        def fake_tts(*, text, output_path):
+            requested_paths.append(output_path)
+            Path(output_path).parent.mkdir(parents=True, exist_ok=True)
+            Path(output_path).write_bytes(b"fake ogg opus")
+            return json.dumps({
+                "success": True,
+                "file_path": output_path,
+                "provider": "gemini",
+                "voice_compatible": True,
+            })
+
+        with patch("tools.tts_tool.text_to_speech_tool", side_effect=fake_tts):
+            await runner._send_voice_reply(event, "hello from auto tts")
+
+        assert requested_paths and requested_paths[0].endswith(".ogg")
+        adapter.send_voice.assert_awaited_once()
+        assert adapter.send_voice.await_args.kwargs["audio_path"].endswith(".ogg")
+
+    def test_should_send_voice_reply_streamed_global_auto_tts_fires(self):
+        """Streamed reply + global voice.auto_tts (no /voice opt-in) sends voice.
+
+        Regression for the #51867/#23983 remainder: when streaming consumed
+        the text, the base adapter's auto-TTS gets text_content=None, and the
+        runner path used to consult only self._voice_mode — so a chat relying
+        purely on the global voice.auto_tts default silently lost its voice
+        reply.
+        """
+        runner = _make_runner()
+        adapter = _make_adapter(Platform.TELEGRAM)
+        adapter._should_auto_tts_for_chat = MagicMock(return_value=True)
+        runner.adapters[Platform.TELEGRAM] = adapter
+        voice_event = _make_event(
+            Platform.TELEGRAM, chat_id="123", message_type=MessageType.VOICE
+        )
+
+        assert runner._should_send_voice_reply(
+            voice_event, "hello", [], already_sent=True
+        ) is True
+
     def test_should_send_voice_reply_uses_global_auto_tts_adapter_default(self):
         """voice.auto_tts=true should make normal text replies get voice too."""
         runner = _make_runner()

tests/gateway/test_base_auto_tts_output_format.py

@@ -0,0 +1,182 @@
+"""Base-adapter auto-TTS must pass a platform-aware explicit output path.
+
+Regression tests for the cleared-contextvar bug (#57049, #36685): the
+post-handler auto-TTS block in ``BasePlatformAdapter._process_message_background``
+runs AFTER ``_clear_session_env`` wiped ``HERMES_SESSION_PLATFORM``, so the
+TTS tool's contextvar-based ``want_opus`` detection always resolved False on
+that path and Opus platforms received MP3 (audio attachment, not a native
+voice bubble). The fix passes an explicit output path from
+``build_auto_tts_output_path(platform)``, which consults the TTS tool's
+``OPUS_VOICE_PLATFORMS`` set — the single source of truth.
+"""
+
+import asyncio
+import json
+from unittest.mock import AsyncMock, patch
+
+import pytest
+
+from gateway.config import Platform, PlatformConfig
+from gateway.platforms.base import (
+    BasePlatformAdapter,
+    MessageEvent,
+    MessageType,
+    SendResult,
+    build_auto_tts_output_path,
+)
+from gateway.session import SessionSource, build_session_key
+from tools.tts_tool import OPUS_VOICE_PLATFORMS
+
+
+class _DummyAdapter(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})
+        return SendResult(success=True, message_id="1")
+
+    async def send_typing(self, chat_id: str, metadata=None) -> None:
+        return None
+
+    async def stop_typing(self, chat_id: str, metadata=None) -> None:
+        return None
+
+    async def get_chat_info(self, chat_id: str):
+        return {"id": chat_id}
+
+
+def _make_voice_event(platform: Platform) -> MessageEvent:
+    return MessageEvent(
+        text="hello",
+        message_type=MessageType.VOICE,
+        source=SessionSource(
+            platform=platform,
+            chat_id="-1001",
+            chat_type="group",
+        ),
+        message_id="voice-1",
+    )
+
+
+def _hold_typing():
+    async def hold(*_args, **_kwargs):
+        await asyncio.Event().wait()
+
+    return hold
+
+
+# ---------------------------------------------------------------------------
+# build_auto_tts_output_path: OPUS_VOICE_PLATFORMS is the single source of truth
+# ---------------------------------------------------------------------------
+
+@pytest.mark.parametrize("platform_name", sorted(OPUS_VOICE_PLATFORMS))
+def test_output_path_is_ogg_for_every_opus_voice_platform(platform_name):
+    path = build_auto_tts_output_path(platform_name)
+    assert path.endswith(".ogg"), path
+
+
+@pytest.mark.parametrize(
+    "platform", [Platform.DISCORD, Platform.SLACK, "irc", None]
+)
+def test_output_path_is_mp3_for_non_opus_platforms(platform):
+    path = build_auto_tts_output_path(platform)
+    assert path.endswith(".mp3"), path
+
+
+def test_output_path_accepts_platform_enum():
+    assert build_auto_tts_output_path(Platform.TELEGRAM).endswith(".ogg")
+    assert build_auto_tts_output_path(Platform.MATRIX).endswith(".ogg")
+    assert build_auto_tts_output_path(Platform.FEISHU).endswith(".ogg")
+
+
+def test_output_paths_are_unique():
+    assert build_auto_tts_output_path("telegram") != build_auto_tts_output_path("telegram")
+
+
+# ---------------------------------------------------------------------------
+# Base-adapter auto-TTS block: explicit output_path, no contextvar reliance
+# ---------------------------------------------------------------------------
+
+async def _run_auto_tts(adapter: _DummyAdapter, platform: Platform):
+    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="tts-1"))
+    long_reply = "x" * 2000  # avoid the telegram caption-collapse path
+    adapter.set_message_handler(lambda _event: asyncio.sleep(0, result=long_reply))
+    event = _make_voice_event(platform)
+    requested = []
+
+    def fake_tts(*, text, output_path=None):
+        requested.append(output_path)
+        from pathlib import Path
+        Path(output_path).parent.mkdir(parents=True, exist_ok=True)
+        Path(output_path).write_bytes(b"fake audio")
+        return json.dumps({"success": True, "file_path": output_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)
+        )
+    return requested, adapter
+
+
+@pytest.mark.asyncio
+async def test_base_auto_tts_requests_ogg_on_opus_platform():
+    """Telegram (opus platform) gets an explicit .ogg path even though the
+    HERMES_SESSION_PLATFORM contextvar is cleared by the time the block runs."""
+    adapter = _DummyAdapter(Platform.TELEGRAM)
+    requested, adapter = await _run_auto_tts(adapter, Platform.TELEGRAM)
+
+    assert requested and requested[0] is not None
+    assert requested[0].endswith(".ogg")
+    adapter.play_tts.assert_awaited_once()
+    assert adapter.play_tts.await_args.kwargs["audio_path"].endswith(".ogg")
+
+
+@pytest.mark.asyncio
+async def test_base_auto_tts_keeps_mp3_on_non_opus_platform():
+    adapter = _DummyAdapter(Platform.DISCORD)
+    requested, adapter = await _run_auto_tts(adapter, Platform.DISCORD)
+
+    assert requested and requested[0] is not None
+    assert requested[0].endswith(".mp3")
+    adapter.play_tts.assert_awaited_once()
+    assert adapter.play_tts.await_args.kwargs["audio_path"].endswith(".mp3")
+
+
+@pytest.mark.asyncio
+async def test_base_auto_tts_skips_playback_when_tool_reports_failure():
+    """A success=False tool result must not deliver a stale/partial file."""
+    adapter = _DummyAdapter(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="tts-1"))
+    adapter.set_message_handler(lambda _event: asyncio.sleep(0, result="reply text"))
+    event = _make_voice_event(Platform.TELEGRAM)
+
+    def fake_tts(*, text, output_path=None):
+        from pathlib import Path
+        Path(output_path).parent.mkdir(parents=True, exist_ok=True)
+        Path(output_path).write_bytes(b"partial")
+        return json.dumps({"success": False, "error": "backend exploded"})
+
+    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)
+        )
+
+    adapter.play_tts.assert_not_awaited()
+    # Text reply still goes out.
+    assert adapter.sent and adapter.sent[0]["content"] == "reply text"

tests/gateway/test_telegram_voice_caption_markdown.py

@@ -0,0 +1,130 @@
+"""Telegram voice-message captions must render markdown (#32029).
+
+``TelegramAdapter.send_voice`` used to pass captions raw with no
+``parse_mode``, so auto-TTS captions carrying the agent's markdown reply
+showed literal *asterisks*, backticks and [links](...). It now formats the
+caption to MarkdownV2 (when the formatted text fits the 1024-char caption
+cap) and falls back to the plain truncated caption when the Bot API rejects
+the entities or formatting overflows.
+"""
+import sys
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+from gateway.config import PlatformConfig
+
+
+def _ensure_telegram_mock():
+    if "telegram" in sys.modules and hasattr(sys.modules["telegram"], "__file__"):
+        return
+    mod = MagicMock()
+    mod.error.NetworkError = type("NetworkError", (OSError,), {})
+    mod.error.TimedOut = type("TimedOut", (OSError,), {})
+    mod.error.BadRequest = type("BadRequest", (Exception,), {})
+    for name in ("telegram", "telegram.ext", "telegram.constants", "telegram.request"):
+        sys.modules.setdefault(name, mod)
+    sys.modules.setdefault("telegram.error", mod.error)
+
+
+_ensure_telegram_mock()
+
+from plugins.platforms.telegram import adapter as telegram_mod  # noqa: E402
+from plugins.platforms.telegram.adapter import TelegramAdapter  # noqa: E402
+
+
+def _make_adapter() -> TelegramAdapter:
+    adapter = TelegramAdapter(PlatformConfig(enabled=True, token="***"))
+    adapter._bot = MagicMock()
+    adapter._bot.send_voice = AsyncMock(return_value=MagicMock(message_id=1))
+    adapter._bot.send_audio = AsyncMock(return_value=MagicMock(message_id=2))
+    return adapter
+
+
+def _write_ogg(tmp_path):
+    audio = tmp_path / "reply.ogg"
+    audio.write_bytes(b"\x00" * 16)
+    return audio
+
+
+@pytest.mark.asyncio
+async def test_voice_caption_gets_markdown_parse_mode(monkeypatch, tmp_path):
+    """A markdown caption is MarkdownV2-formatted and sent with parse_mode."""
+    monkeypatch.setattr(
+        telegram_mod, "_probe_voice_duration_seconds", lambda _p: 3
+    )
+    adapter = _make_adapter()
+
+    result = await adapter.send_voice(
+        "123", str(_write_ogg(tmp_path)), caption="*bold* reply"
+    )
+
+    assert result.success is True
+    adapter._bot.send_voice.assert_awaited_once()
+    kwargs = adapter._bot.send_voice.await_args.kwargs
+    assert kwargs["parse_mode"] is not None
+    # format_message converts *bold* to MarkdownV2 bold; the raw caption must
+    # have gone through formatting rather than being passed verbatim.
+    assert kwargs["caption"] == adapter.format_message("*bold* reply")
+
+
+@pytest.mark.asyncio
+async def test_voice_caption_falls_back_to_plain_on_entity_rejection(
+    monkeypatch, tmp_path
+):
+    """A Bot API entity-parse rejection retries with the plain caption."""
+    monkeypatch.setattr(
+        telegram_mod, "_probe_voice_duration_seconds", lambda _p: 3
+    )
+    adapter = _make_adapter()
+    adapter._bot.send_voice = AsyncMock(
+        side_effect=[
+            Exception("Bad Request: can't parse entities"),
+            MagicMock(message_id=7),
+        ]
+    )
+
+    result = await adapter.send_voice(
+        "123", str(_write_ogg(tmp_path)), caption="*bold* reply"
+    )
+
+    assert result.success is True
+    assert adapter._bot.send_voice.await_count == 2
+    retry_kwargs = adapter._bot.send_voice.await_args_list[1].kwargs
+    assert retry_kwargs["parse_mode"] is None
+    assert retry_kwargs["caption"] == "*bold* reply"
+
+
+@pytest.mark.asyncio
+async def test_voice_caption_overflow_skips_formatting(monkeypatch, tmp_path):
+    """When the formatted caption exceeds 1024 UTF-16 units, send plain
+    (truncated) — MarkdownV2 escaping inflates length, and a truncated
+    formatted caption could cut an entity in half."""
+    monkeypatch.setattr(
+        telegram_mod, "_probe_voice_duration_seconds", lambda _p: 3
+    )
+    adapter = _make_adapter()
+    # Enough special chars that escaping pushes the formatted text over 1024.
+    caption = ("hello. " * 160)[:1020]
+
+    result = await adapter.send_voice("123", str(_write_ogg(tmp_path)), caption=caption)
+
+    assert result.success is True
+    kwargs = adapter._bot.send_voice.await_args.kwargs
+    assert kwargs["parse_mode"] is None
+    assert kwargs["caption"] == caption[:1024]
+
+
+@pytest.mark.asyncio
+async def test_voice_without_caption_unchanged(monkeypatch, tmp_path):
+    monkeypatch.setattr(
+        telegram_mod, "_probe_voice_duration_seconds", lambda _p: 3
+    )
+    adapter = _make_adapter()
+
+    result = await adapter.send_voice("123", str(_write_ogg(tmp_path)))
+
+    assert result.success is True
+    kwargs = adapter._bot.send_voice.await_args.kwargs
+    assert kwargs["caption"] is None
+    assert kwargs["parse_mode"] is None

tests/gateway/test_voice_command.py

@@ -1872,14 +1872,23 @@ class TestSendVoiceReplyFilename:
     """_send_voice_reply uses uuid for unique filenames."""
 
     def test_filename_uses_uuid(self):
-        """The method uses uuid in the filename, not time-based."""
+        """The path builder uses uuid in the filename, not time-based.
+
+        Filename construction moved into build_auto_tts_output_path
+        (gateway/platforms/base.py) when the path became platform-aware;
+        the uniqueness contract lives there now.
+        """
         import inspect
+        from gateway.platforms.base import build_auto_tts_output_path
         from gateway.run import GatewayRunner
-        source = inspect.getsource(GatewayRunner._send_voice_reply)
+        source = inspect.getsource(build_auto_tts_output_path)
         assert "uuid" in source, \
-            "_send_voice_reply should use uuid for unique filenames"
+            "build_auto_tts_output_path should use uuid for unique filenames"
         assert "int(time.time())" not in source, \
-            "_send_voice_reply should not use int(time.time()) — collision risk"
+            "build_auto_tts_output_path should not use int(time.time()) — collision risk"
+        runner_source = inspect.getsource(GatewayRunner._send_voice_reply)
+        assert "build_auto_tts_output_path" in runner_source, \
+            "_send_voice_reply should build its path via build_auto_tts_output_path"
 
     def test_filenames_are_unique(self):
         """Two calls produce different filenames."""