feat: add debug audio dump functionality for ElevenLabs ASR provider

deployment/aimaq.env.production

@@ -68,6 +68,7 @@ AI_VOICE_ASR_ELEVENLABS_LANGUAGE=ru
 AI_VOICE_ASR_ELEVENLABS_REALTIME_MODEL_ID=scribe_v2_realtime
 AI_VOICE_ASR_ELEVENLABS_REALTIME_AUDIO_FORMAT=pcm_16000
 AI_VOICE_ASR_ELEVENLABS_REALTIME_COMMIT_STRATEGY=vad
+AI_VOICE_ASR_ELEVENLABS_REALTIME_DEBUG_DUMP_AUDIO=1
 AI_VOICE_ASR_ELEVENLABS_REALTIME_FINALIZE_TIMEOUT_SECONDS=2.5
 AI_VOICE_ASR_YANDEX_API_KEY=AQWJe16Rdlx_peHUppZ79csG49f1gP4wwQyDzeQ_
 AI_VOICE_ASR_YANDEX_FOLDER_ID=ao7hkif5pvc7vfnmbl0d

services/ai_voice_runtime_service/providers/asr.py

@@ -13,6 +13,7 @@ import uuid
 import wave
 from collections.abc import Callable
 from dataclasses import dataclass, field
+from pathlib import Path
 
 import httpx
 
@@ -137,6 +138,18 @@ def _elevenlabs_realtime_commit_strategy() -> str:
     return raw if raw in {"manual", "vad"} else "manual"
 
 
+def _elevenlabs_realtime_debug_dump_audio_enabled() -> bool:
+    raw = os.getenv("AI_VOICE_ASR_ELEVENLABS_REALTIME_DEBUG_DUMP_AUDIO", "").strip().lower()
+    return raw in {"1", "true", "yes", "on"}
+
+
+def _elevenlabs_realtime_debug_dump_audio_dir() -> Path:
+    explicit = os.getenv("AI_VOICE_ASR_ELEVENLABS_REALTIME_DEBUG_DUMP_DIR", "").strip()
+    if explicit:
+        return Path(explicit)
+    return Path(os.getenv("CC_DATA_DIR", ".data_local").strip() or ".data_local") / "asr_debug_dumps"
+
+
 def _elevenlabs_realtime_finalize_timeout_seconds() -> float:
     raw_ms = os.getenv("AI_VOICE_V2_STREAMING_FINAL_HARD_TIMEOUT_MS", "").strip()
     if raw_ms:
@@ -539,6 +552,8 @@ class _ElevenLabsRealtimeStreamState:
     resample_lock: threading.Lock = field(default_factory=threading.Lock)
     committed_segments: list[str] = field(default_factory=list)
     finalize_requested: bool = False
+    debug_pcm: bytearray = field(default_factory=bytearray)
+    debug_lock: threading.Lock = field(default_factory=threading.Lock)
 
 
 class ElevenLabsRealtimeStreamingASRProvider(StreamingASRProvider):
@@ -556,6 +571,8 @@ class ElevenLabsRealtimeStreamingASRProvider(StreamingASRProvider):
         audio_format: str | None = None,
         commit_strategy: str | None = None,
         websocket_factory: Callable[..., object] | None = None,
+        debug_dump_audio: bool | None = None,
+        debug_dump_dir: str | Path | None = None,
     ) -> None:
         self._api_base = str(api_base or _elevenlabs_asr_api_base()).strip().rstrip("/")
         self._api_key = str(api_key if api_key is not None else _elevenlabs_asr_api_key()).strip()
@@ -580,6 +597,10 @@ class ElevenLabsRealtimeStreamingASRProvider(StreamingASRProvider):
         self._websocket_factory = websocket_factory
         self._streams: dict[str, _ElevenLabsRealtimeStreamState] = {}
         self._lock = threading.Lock()
+        self._debug_dump_audio = (
+            debug_dump_audio if debug_dump_audio is not None else _elevenlabs_realtime_debug_dump_audio_enabled()
+        )
+        self._debug_dump_dir = Path(debug_dump_dir) if debug_dump_dir is not None else _elevenlabs_realtime_debug_dump_audio_dir()
 
     def _websocket_url(self, *, language: str) -> str:
         base = self._api_base
@@ -860,6 +881,9 @@ class ElevenLabsRealtimeStreamingASRProvider(StreamingASRProvider):
                     output_rate_hz=self._sample_rate_hz,
                     state=state.resample_state,
                 )
+        if self._debug_dump_audio:
+            with state.debug_lock:
+                state.debug_pcm.extend(pcm_bytes)
         self._send_json(
             state,
             {
@@ -870,6 +894,29 @@ class ElevenLabsRealtimeStreamingASRProvider(StreamingASRProvider):
             },
         )
 
+    def _write_debug_dump(self, state: _ElevenLabsRealtimeStreamState) -> None:
+        with state.debug_lock:
+            pcm_bytes = bytes(state.debug_pcm)
+        if not pcm_bytes:
+            return
+        try:
+            self._debug_dump_dir.mkdir(parents=True, exist_ok=True)
+            path = self._debug_dump_dir / f"{state.stream_id}.wav"
+            path.write_bytes(pcm16le_to_wav_bytes(pcm_bytes, sample_rate_hz=self._sample_rate_hz))
+        except OSError as exc:
+            logger.warning(
+                "asr.elevenlabs_realtime.debug_dump_failed stream_id=%s error=%s",
+                state.stream_id,
+                str(exc)[:300],
+            )
+            return
+        logger.info(
+            "asr.elevenlabs_realtime.debug_dump_written stream_id=%s path=%s bytes=%s",
+            state.stream_id,
+            path,
+            len(pcm_bytes),
+        )
+
     def poll_partial(self, stream_id: str) -> StreamingASRPartial | None:
         state = self._state(stream_id)
         self._drain_updates(state)
@@ -955,6 +1002,8 @@ class ElevenLabsRealtimeStreamingASRProvider(StreamingASRProvider):
         thread = state.thread
         if thread is not None and thread.is_alive():
             thread.join(timeout=0.5)
+        if self._debug_dump_audio:
+            self._write_debug_dump(state)
         with self._lock:
             self._streams.pop(stream_id, None)
         logger.info("asr.elevenlabs_realtime.stream_close stream_id=%s", stream_id)

tests/test_ai_voice_asr_provider.py

@@ -3,6 +3,7 @@ from __future__ import annotations
 import json
 import queue
 import time
+import wave
 
 import httpx
 import pytest
@@ -529,6 +530,51 @@ def test_elevenlabs_realtime_streaming_provider_accumulates_autonomous_vad_commi
     assert final.language == "ru"
 
 
+def test_elevenlabs_realtime_streaming_provider_writes_debug_audio_dump(tmp_path):
+    class _FakeRealtimeWebSocket:
+        def __init__(self) -> None:
+            self.incoming: queue.Queue[str | None] = queue.Queue()
+
+        def send(self, raw_payload: str) -> None:
+            payload = json.loads(raw_payload)
+            if payload.get("commit"):
+                self.incoming.put(
+                    json.dumps({"message_type": "committed_transcript", "text": "тест", "language_code": "ru"})
+                )
+
+        def recv(self) -> str:
+            item = self.incoming.get(timeout=1)
+            if item is None:
+                raise RuntimeError("closed")
+            return item
+
+        def close(self) -> None:
+            self.incoming.put(None)
+
+    websocket = _FakeRealtimeWebSocket()
+    dump_dir = tmp_path / "asr_debug_dumps"
+
+    provider = asr_module.ElevenLabsRealtimeStreamingASRProvider(
+        api_base="https://api.elevenlabs.example",
+        api_key="asr-key",
+        timeout_seconds=1,
+        finalize_timeout_seconds=1,
+        websocket_factory=lambda url, *, header, timeout: websocket,
+        debug_dump_audio=True,
+        debug_dump_dir=dump_dir,
+    )
+    stream_id = provider.open_stream("session-1", language_hint="ru")
+    provider.push_pcm(stream_id, b"\x10\x00" * 160)
+    provider.finalize(stream_id)
+    provider.close_stream(stream_id)
+
+    dumped = dump_dir / f"{stream_id}.wav"
+    assert dumped.exists()
+    with wave.open(str(dumped), "rb") as wav_file:
+        assert wav_file.getframerate() == 16000
+        assert wav_file.getnframes() > 0
+
+
 def test_yandex_grpc_streaming_provider_returns_partial_and_final():
     calls: list[dict] = []