fix(gateway): support Feishu voice TTS opus delivery

plugins/platforms/feishu/adapter.py

@@ -58,6 +58,8 @@
 import mimetypes
 import os
 import re
+import shutil
+import subprocess
 import threading
 import time
 import uuid
@@ -4500,12 +4502,18 @@ async def _send_uploaded_file_message(
             file_path=display_name,
             requested_message_type=outbound_message_type,
         )
+        upload_duration_ms = (
+            self._probe_audio_duration_ms(file_path)
+            if upload_file_type == "opus"
+            else None
+        )
         try:
             with open(file_path, "rb") as file_obj:
                 body = self._build_file_upload_body(
                     file_type=upload_file_type,
                     file_name=display_name,
                     file=file_obj,
+                    duration=upload_duration_ms,
                 )
                 request = self._build_file_upload_request(body)
                 upload_response = await self._run_blocking(self._client.im.v1.file.create, request)
@@ -4531,10 +4539,13 @@ async def _send_uploaded_file_message(
                     metadata=metadata,
                 )
             else:
+                message_payload = {"file_key": file_key}
+                if resolved_message_type == "audio" and upload_duration_ms and upload_duration_ms > 0:
+                    message_payload["duration"] = upload_duration_ms
                 message_response = await self._feishu_send_with_retry(
                     chat_id=chat_id,
                     msg_type=resolved_message_type,
-                    payload=json.dumps({"file_key": file_key}, ensure_ascii=False),
+                    payload=json.dumps(message_payload, ensure_ascii=False),
                     reply_to=reply_to,
                     metadata=metadata,
                 )
@@ -4925,16 +4936,24 @@ def _build_image_upload_request(request_body: Any) -> Any:
         return SimpleNamespace(request_body=request_body)
 
     @staticmethod
-    def _build_file_upload_body(*, file_type: str, file_name: str, file: Any) -> Any:
+    def _build_file_upload_body(
+        *,
+        file_type: str,
+        file_name: str,
+        file: Any,
+        duration: Optional[int] = None,
+    ) -> Any:
         if "CreateFileRequestBody" in globals():
-            return (
+            builder = (
                 CreateFileRequestBody.builder()
                 .file_type(file_type)
                 .file_name(file_name)
                 .file(file)
-                .build()
             )
-        return SimpleNamespace(file_type=file_type, file_name=file_name, file=file)
+            if duration is not None:
+                builder = builder.duration(duration)
+            return builder.build()
+        return SimpleNamespace(file_type=file_type, file_name=file_name, file=file, duration=duration)
 
     @staticmethod
     def _build_file_upload_request(request_body: Any) -> Any:
@@ -4951,6 +4970,34 @@ def _build_media_post_payload(self, *, caption: str, media_tag: Dict[str, str])
         content.append([media_tag])
         return json.dumps(payload, ensure_ascii=False)
 
+    @staticmethod
+    def _probe_audio_duration_ms(file_path: str) -> Optional[int]:
+        ffmpeg = shutil.which("ffmpeg")
+        if not ffmpeg:
+            return None
+        try:
+            result = subprocess.run(
+                [ffmpeg, "-i", file_path, "-f", "null", "-"],
+                stdout=subprocess.DEVNULL,
+                stderr=subprocess.PIPE,
+                text=True,
+                timeout=10,
+            )
+        except (FileNotFoundError, subprocess.TimeoutExpired, OSError):
+            return None
+
+        match = re.search(
+            r"Duration:\s*(\d+):(\d+):(\d+(?:\.\d+)?)",
+            result.stderr or "",
+        )
+        if not match:
+            return None
+
+        hours = int(match.group(1))
+        minutes = int(match.group(2))
+        seconds = float(match.group(3))
+        return int(round(((hours * 60 + minutes) * 60 + seconds) * 1000))
+
     @staticmethod
     def _resolve_outbound_file_routing(
         *,

tests/gateway/test_feishu.py

@@ -2481,15 +2481,22 @@ async def _direct(func, *args, **kwargs):
             audio_path = tmp.name
 
         try:
-            with patch("plugins.platforms.feishu.adapter.asyncio.to_thread", side_effect=_direct):
+            with (
+                patch("plugins.platforms.feishu.adapter.asyncio.to_thread", side_effect=_direct),
+                patch.object(FeishuAdapter, "_probe_audio_duration_ms", return_value=1234),
+            ):
                 result = asyncio.run(adapter.send_voice(chat_id="oc_chat", audio_path=audio_path))
         finally:
             os.unlink(audio_path)
 
         self.assertTrue(result.success)
         self.assertEqual(captured["upload_request"].request_body.file_type, "opus")
+        self.assertEqual(captured["upload_request"].request_body.duration, 1234)
         self.assertEqual(captured["message_request"].request_body.msg_type, "audio")
-        self.assertEqual(captured["message_request"].request_body.content, '{"file_key": "file_audio_123"}')
+        self.assertEqual(
+            json.loads(captured["message_request"].request_body.content),
+            {"file_key": "file_audio_123", "duration": 1234},
+        )
 
     @patch.dict(os.environ, {}, clear=True)
     def test_build_post_payload_extracts_title_and_links(self):

tests/tools/test_tts_command_providers.py

@@ -204,7 +204,7 @@ def test_output_format_rejects_unknown(self):
         assert _get_command_tts_output_format({"format": "m4a"}) == DEFAULT_COMMAND_TTS_OUTPUT_FORMAT
 
     def test_output_format_supported_set(self):
-        assert COMMAND_TTS_OUTPUT_FORMATS == frozenset({"mp3", "wav", "ogg", "flac"})
+        assert COMMAND_TTS_OUTPUT_FORMATS == frozenset({"mp3", "wav", "ogg", "opus", "flac"})
 
     def test_voice_compatible_boolean(self):
         assert _is_command_tts_voice_compatible({"voice_compatible": True}) is True

tools/tts_tool.py

@@ -401,7 +401,7 @@ def _get_provider(tts_config: Dict[str, Any]) -> str:
 
 DEFAULT_COMMAND_TTS_TIMEOUT_SECONDS = 120
 DEFAULT_COMMAND_TTS_OUTPUT_FORMAT = "mp3"
-COMMAND_TTS_OUTPUT_FORMATS = frozenset({"mp3", "wav", "ogg", "flac"})
+COMMAND_TTS_OUTPUT_FORMATS = frozenset({"mp3", "wav", "ogg", "opus", "flac"})
 DEFAULT_COMMAND_TTS_MAX_TEXT_LENGTH = 5000
 
 
@@ -613,7 +613,7 @@ def _get_command_tts_output_format(
     config: Dict[str, Any],
     output_path: Optional[str] = None,
 ) -> str:
-    """Return the validated output format (mp3/wav/ogg/flac)."""
+    """Return the validated output format (mp3/wav/ogg/opus/flac)."""
     if output_path:
         suffix = Path(output_path).suffix.lower().strip().lstrip(".")
         if suffix in COMMAND_TTS_OUTPUT_FORMATS:
@@ -933,6 +933,11 @@ def _convert_to_opus(mp3_path: str) -> Optional[str]:
     return None
 
 
+def _is_opus_voice_file(file_path: str) -> bool:
+    """Return True when *file_path* already points to an Opus voice payload."""
+    return file_path.lower().endswith((".ogg", ".opus"))
+
+
 # ===========================================================================
 # Provider: Edge TTS (free)
 # ===========================================================================
@@ -987,7 +992,7 @@ def _generate_elevenlabs(text: str, output_path: str, tts_config: Dict[str, Any]
     model_id = el_config.get("model_id", DEFAULT_ELEVENLABS_MODEL_ID)
 
     # Determine output format based on file extension
-    if output_path.endswith(".ogg"):
+    if output_path.lower().endswith((".ogg", ".opus")):
         output_format = "opus_48000_64"
     else:
         output_format = "mp3_44100_128"
@@ -1048,7 +1053,7 @@ def _generate_openai_tts(text: str, output_path: str, tts_config: Dict[str, Any]
         model = DEFAULT_OPENAI_MODEL
 
     # Determine response format from extension
-    if output_path.endswith(".ogg"):
+    if output_path.lower().endswith((".ogg", ".opus")):
         response_format = "opus"
     else:
         response_format = "mp3"
@@ -1447,7 +1452,7 @@ def _generate_mistral_tts(text: str, output_path: str, tts_config: Dict[str, Any
     model = mi_config.get("model", DEFAULT_MISTRAL_TTS_MODEL)
     voice_id = mi_config.get("voice_id") or DEFAULT_MISTRAL_TTS_VOICE_ID
 
-    if output_path.endswith(".ogg"):
+    if output_path.lower().endswith((".ogg", ".opus")):
         response_format = "opus"
     elif output_path.endswith(".wav"):
         response_format = "wav"
@@ -1788,7 +1793,7 @@ def _generate_gemini_tts(text: str, output_path: str, tts_config: Dict[str, Any]
         if ffmpeg:
             # For .ogg output, force libopus encoding (Telegram voice bubbles
             # require Opus specifically; ffmpeg's default for .ogg is Vorbis).
-            if output_path.lower().endswith(".ogg"):
+            if output_path.lower().endswith((".ogg", ".opus")):
                 cmd = [
                     ffmpeg, "-i", wav_path,
                     "-acodec", "libopus", "-ac", "1",
@@ -2194,12 +2199,12 @@ 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.
+    # Telegram and Feishu voice bubbles prefer Opus audio. OpenAI and
+    # ElevenLabs can produce Opus natively (no ffmpeg needed). Edge TTS
+    # always outputs MP3 and needs ffmpeg for conversion.
     from gateway.session_context import get_session_env
     platform = get_session_env("HERMES_SESSION_PLATFORM", "").lower()
-    want_opus = (platform == "telegram")
+    want_opus = platform in {"telegram", "feishu"}
 
     # Determine output path
     if output_path:
@@ -2236,8 +2241,8 @@ def text_to_speech_tool(
         if command_provider_config is not None:
             fmt = _get_command_tts_output_format(command_provider_config)
             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).
+        # Use .ogg for platforms with native Opus voice delivery, otherwise
+        # fall back to .mp3 (Edge TTS will attempt ffmpeg conversion later).
         elif want_opus and provider in {"openai", "elevenlabs", "mistral", "gemini"}:
             file_path = out_dir / f"tts_{timestamp}.ogg"
         else:
@@ -2390,7 +2395,7 @@ def text_to_speech_tool(
                 "error": f"TTS generation produced no output (provider: {provider})"
             }, ensure_ascii=False)
 
-        # Try Opus conversion for Telegram compatibility.
+        # Try Opus conversion for voice-message compatibility.
         # Edge TTS outputs MP3, NeuTTS/KittenTTS output WAV. Keep those native
         # formats for local/CLI playback and only convert when the current
         # platform actually needs Opus voice delivery.
@@ -2400,34 +2405,38 @@ def text_to_speech_tool(
             # delivery only kicks in when the user explicitly opts in
             # via ``voice_compatible: true`` in their provider config.
             if _is_command_tts_voice_compatible(command_provider_config):
-                if not file_str.endswith(".ogg"):
+                if not _is_opus_voice_file(file_str):
                     opus_path = _convert_to_opus(file_str)
                     if opus_path:
                         file_str = opus_path
-                voice_compatible = file_str.endswith(".ogg")
+                voice_compatible = _is_opus_voice_file(file_str)
         elif provider not in BUILTIN_TTS_PROVIDERS:
             # Plugin-registered provider (issue #30398). Voice-bubble
             # delivery opts in via ``TTSProvider.voice_compatible``
             # (mirrors the command-provider opt-in). Plugins that
             # already write Opus skip the ffmpeg conversion.
             plugin_voice_compatible = _plugin_provider_is_voice_compatible(provider)
             if plugin_voice_compatible:
-                if not file_str.endswith(".ogg"):
+                if not _is_opus_voice_file(file_str):
                     opus_path = _convert_to_opus(file_str)
                     if opus_path:
                         file_str = opus_path
-                voice_compatible = file_str.endswith(".ogg")
+                voice_compatible = _is_opus_voice_file(file_str)
         elif (
             want_opus
             and provider in {"edge", "neutts", "minimax", "xai", "kittentts", "piper"}
-            and not file_str.endswith(".ogg")
+            and not _is_opus_voice_file(file_str)
         ):
             opus_path = _convert_to_opus(file_str)
             if opus_path:
                 file_str = opus_path
                 voice_compatible = True
         elif provider in {"elevenlabs", "openai", "mistral", "gemini"}:
-            voice_compatible = want_opus and file_str.endswith(".ogg")
+            if want_opus and not _is_opus_voice_file(file_str):
+                opus_path = _convert_to_opus(file_str)
+                if opus_path:
+                    file_str = opus_path
+            voice_compatible = want_opus and _is_opus_voice_file(file_str)
 
         file_size = os.path.getsize(file_str)
         logger.info("TTS audio saved: %s (%s bytes, provider: %s)", file_str, f"{file_size:,}", provider)