fix(stt): preprocess .silk voice notes before transcription

pyproject.toml

@@ -176,6 +176,7 @@ voice = [
   "faster-whisper==1.2.1",
   "sounddevice==0.5.5",
   "numpy==2.4.3",
+  "pilk>=0.2.4,<1",
 ]
 pty = [
   # Kept as a no-op back-compat alias — `ptyprocess` and `pywinpty` are now

tests/tools/test_transcription_tools.py

@@ -53,6 +53,14 @@ def sample_ogg(tmp_path):
     ogg_path.write_bytes(b"fake audio data")
     return str(ogg_path)
 
+@pytest.fixture
+def sample_silk(tmp_path):
+    """Create a fake WeChat .silk file for preprocessing tests."""
+    silk_path = tmp_path / "voice.silk"
+    silk_path.write_bytes(b"\x02#!SILK_V3fake")
+    return str(silk_path)
+
+
 
 pytestmark = pytest.mark.usefixtures("disable_lazy_stt_install")
 
@@ -909,6 +917,64 @@ class TestTranscribeAudioDispatch:
 
         assert mock_local.call_args[0][1] == "large-v3"
 
+    def test_converts_silk_before_dispatch(self, sample_silk):
+        with patch("tools.transcription_tools._prepare_audio_for_transcription",
+                   return_value=("/tmp/converted.wav", "/tmp/hermes-silk-123", None),
+                   create=True) as mock_prepare, \
+             patch("tools.transcription_tools._validate_audio_file", return_value=None), \
+             patch("tools.transcription_tools._load_stt_config", return_value={}), \
+             patch("tools.transcription_tools._get_provider", return_value="local"), \
+             patch("tools.transcription_tools._transcribe_local",
+                   return_value={"success": True, "transcript": "hi"}) as mock_local, \
+             patch("tools.transcription_tools.shutil.rmtree") as mock_rmtree:
+            from tools.transcription_tools import transcribe_audio
+            result = transcribe_audio(sample_silk)
+
+        assert result["success"] is True
+        mock_prepare.assert_called_once_with(sample_silk)
+        mock_local.assert_called_once_with("/tmp/converted.wav", "base")
+        mock_rmtree.assert_called_once_with("/tmp/hermes-silk-123", ignore_errors=True)
+
+    def test_silk_symlink_is_rejected_before_preprocessing(self, tmp_path):
+        """A Silk symlink must not reach the decoder before path safety validation."""
+        if not hasattr(os, "symlink"):
+            pytest.skip("symlinks are not supported on this platform")
+
+        target = tmp_path / "voice.silk"
+        target.write_bytes(b"\x02#!SILK_V3fake")
+        link = tmp_path / "linked.silk"
+        try:
+            os.symlink(target, link)
+        except (OSError, NotImplementedError) as exc:
+            pytest.skip(f"symlink creation unavailable: {exc}")
+
+        with patch(
+            "tools.transcription_tools._prepare_audio_for_transcription", create=True
+        ) as mock_prepare:
+            from tools.transcription_tools import transcribe_audio
+            result = transcribe_audio(str(link))
+
+        assert result["success"] is False
+        assert "symbolic link" in result["error"]
+        mock_prepare.assert_not_called()
+
+    def test_oversized_silk_is_rejected_before_preprocessing(self, tmp_path):
+        """A Silk source over the upload limit must not reach the decoder."""
+        silk_path = tmp_path / "oversized.silk"
+        from tools.transcription_tools import MAX_FILE_SIZE
+        with silk_path.open("wb") as audio_file:
+            audio_file.truncate(MAX_FILE_SIZE + 1)
+
+        with patch(
+            "tools.transcription_tools._prepare_audio_for_transcription", create=True
+        ) as mock_prepare:
+            from tools.transcription_tools import transcribe_audio
+            result = transcribe_audio(str(silk_path))
+
+        assert result["success"] is False
+        assert "File too large" in result["error"]
+        mock_prepare.assert_not_called()
+
     def test_default_model_used_when_none(self, sample_ogg):
         with patch("tools.transcription_tools._load_stt_config", return_value={}), \
              patch("tools.transcription_tools._get_provider", return_value="groq"), \

tools/transcription_tools.py

@@ -79,6 +79,7 @@ def _safe_find_spec(module_name: str) -> bool:
 _HAS_FASTER_WHISPER = _safe_find_spec("faster_whisper")
 _HAS_OPENAI = _safe_find_spec("openai")
 _HAS_MISTRAL = _safe_find_spec("mistralai")
+_HAS_PILK = _safe_find_spec("pilk")
 
 # ---------------------------------------------------------------------------
 # Constants
@@ -1019,8 +1020,8 @@ def _dispatch_to_plugin_provider(
 # ---------------------------------------------------------------------------
 
 
-def _validate_audio_file(file_path: str) -> Optional[Dict[str, Any]]:
-    """Validate the audio file.  Returns an error dict or None if OK."""
+def _validate_audio_source_file(file_path: str) -> Optional[Dict[str, Any]]:
+    """Validate source path safety and size before any decoder is invoked."""
     audio_path = Path(file_path)
 
     if os.path.islink(audio_path):
@@ -1029,25 +1030,67 @@ def _validate_audio_file(file_path: str) -> Optional[Dict[str, Any]]:
         return {"success": False, "transcript": "", "error": f"Audio file not found: {file_path}"}
     if not audio_path.is_file():
         return {"success": False, "transcript": "", "error": f"Path is not a file: {file_path}"}
+    try:
+        file_size = audio_path.stat().st_size
+    except OSError as exc:
+        return {"success": False, "transcript": "", "error": f"Failed to access file: {exc}"}
+    if file_size > MAX_FILE_SIZE:
+        return {
+            "success": False,
+            "transcript": "",
+            "error": f"File too large: {file_size / (1024*1024):.1f}MB (max {MAX_FILE_SIZE / (1024*1024):.0f}MB)",
+        }
+    return None
+
+
+def _validate_audio_file(file_path: str) -> Optional[Dict[str, Any]]:
+    """Validate a supported, decoder-safe audio file."""
+    source_error = _validate_audio_source_file(file_path)
+    if source_error:
+        return source_error
+
+    audio_path = Path(file_path)
     if audio_path.suffix.lower() not in SUPPORTED_FORMATS:
         return {
             "success": False,
             "transcript": "",
             "error": f"Unsupported format: {audio_path.suffix}. Supported: {', '.join(sorted(SUPPORTED_FORMATS))}",
         }
-    try:
-        file_size = audio_path.stat().st_size
-        if file_size > MAX_FILE_SIZE:
-            return {
-                "success": False,
-                "transcript": "",
-                "error": f"File too large: {file_size / (1024*1024):.1f}MB (max {MAX_FILE_SIZE / (1024*1024):.0f}MB)",
-            }
-    except OSError as e:
-        return {"success": False, "transcript": "", "error": f"Failed to access file: {e}"}
-
     return None
 
+
+def _prepare_audio_for_transcription(
+    file_path: str,
+) -> tuple[Optional[str], Optional[str], Optional[Dict[str, Any]]]:
+    """Convert a decoder-safe .silk source to a temporary supported WAV file."""
+    audio_path = Path(file_path)
+    if audio_path.suffix.lower() != ".silk":
+        return file_path, None, None
+    if not _HAS_PILK:
+        return None, None, {
+            "success": False,
+            "transcript": "",
+            "error": "Unsupported format: .silk. Install the optional 'pilk' dependency to enable WeChat voice transcription.",
+        }
+
+    temp_dir = tempfile.mkdtemp(prefix="hermes-silk-")
+    converted_path = os.path.join(temp_dir, f"{audio_path.stem}.wav")
+    try:
+        import pilk
+
+        pilk.silk_to_wav(file_path, converted_path)
+        if not Path(converted_path).is_file() or Path(converted_path).stat().st_size == 0:
+            raise RuntimeError("pilk did not produce a readable WAV file")
+        return converted_path, temp_dir, None
+    except Exception as exc:
+        shutil.rmtree(temp_dir, ignore_errors=True)
+        logger.error("Failed to convert .silk audio %s: %s", file_path, exc, exc_info=True)
+        return None, None, {
+            "success": False,
+            "transcript": "",
+            "error": f"Failed to convert .silk audio for transcription: {exc}",
+        }
+
 # ---------------------------------------------------------------------------
 # Provider: local (faster-whisper)
 # ---------------------------------------------------------------------------
@@ -1621,7 +1664,7 @@ def _transcribe_elevenlabs(file_path: str, model_name: str) -> Dict[str, Any]:
 # ---------------------------------------------------------------------------
 
 
-def transcribe_audio(file_path: str, model: Optional[str] = None) -> Dict[str, Any]:
+def _transcribe_prepared_audio(file_path: str, model: Optional[str] = None) -> Dict[str, Any]:
     """
     Transcribe an audio file using the configured STT provider.
 
@@ -1640,10 +1683,7 @@ def transcribe_audio(file_path: str, model: Optional[str] = None) -> Dict[str, A
           - "error" (str, optional): Error message if success is False
           - "provider" (str, optional): Which provider was used
     """
-    # Validate input
-    error = _validate_audio_file(file_path)
-    if error:
-        return error
+    # `file_path` has already passed source and extension validation.
 
     # Load config and determine provider
     stt_config = _load_stt_config()
@@ -1751,6 +1791,32 @@ def transcribe_audio(file_path: str, model: Optional[str] = None) -> Dict[str, A
     }
 
 
+def transcribe_audio(file_path: str, model: Optional[str] = None) -> Dict[str, Any]:
+    """Safely validate, preprocess supported inputs, and dispatch transcription."""
+    source_error = _validate_audio_source_file(file_path)
+    if source_error:
+        return source_error
+
+    prepared_path, cleanup_dir, prep_error = _prepare_audio_for_transcription(file_path)
+    if prep_error:
+        return prep_error
+    if prepared_path is None:
+        return {
+            "success": False,
+            "transcript": "",
+            "error": "Audio preprocessing did not produce a file for transcription.",
+        }
+
+    try:
+        prepared_error = _validate_audio_file(prepared_path)
+        if prepared_error:
+            return prepared_error
+        return _transcribe_prepared_audio(prepared_path, model)
+    finally:
+        if cleanup_dir:
+            shutil.rmtree(cleanup_dir, ignore_errors=True)
+
+
 def _resolve_openai_audio_client_config() -> tuple[str, str]:
     """Return direct OpenAI audio config or a managed gateway fallback."""
     stt_config = _load_stt_config()

uv.lock

@@ -1715,6 +1715,7 @@ vertex = [
 voice = [
     { name = "faster-whisper" },
     { name = "numpy" },
+    { name = "pilk" },
     { name = "sounddevice" },
 ]
 web = [
@@ -1810,6 +1811,7 @@ requires-dist = [
     { name = "packaging", specifier = "==26.0" },
     { name = "parallel-web", marker = "extra == 'parallel-web'", specifier = "==0.4.2" },
     { name = "pathspec", specifier = "==1.1.1" },
+    { name = "pilk", marker = "extra == 'voice'", specifier = ">=0.2.4,<1" },
     { name = "pillow", specifier = "==12.2.0" },
     { name = "prompt-toolkit", specifier = "==3.0.52" },
     { name = "psutil", specifier = "==7.2.2" },
@@ -2951,6 +2953,18 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/f1/d9/7fb5aa316bc299258e68c73ba3bddbc499654a07f151cba08f6153988714/pathspec-1.1.1-py3-none-any.whl", hash = "sha256:a00ce642f577bf7f473932318056212bc4f8bfdf53128c78bbd5af0b9b20b189", size = 57328, upload-time = "2026-04-27T01:46:07.06Z" },
 ]
 
+[[package]]
+name = "pilk"
+version = "0.2.4"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "typing-extensions" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/66/bb/938dd697b6bc2d851ffec4ffe82ed20078e58bd0ef049e84f4d038d6f991/pilk-0.2.4.tar.gz", hash = "sha256:d4a1bcf93dc6ef5e95e0cfd728ed4ef4d49f9c0476d70816fecbe456cc762e7f", size = 226451, upload-time = "2023-05-26T03:32:34.133Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/ff/06/80cac61bc7f791bcaae552ba90fb868f7af7503ef1c74e423cc20fca1a53/pilk-0.2.4-cp311-cp311-win_amd64.whl", hash = "sha256:a6692096607e8d77d348aeec00df633788c85b527aa83cbd803ddf508936db7f", size = 127024, upload-time = "2023-05-26T03:32:24.524Z" },
+]
+
 [[package]]
 name = "pillow"
 version = "12.2.0"