remove dead audio accumulation

src/eva/assistant/base_server.py

@@ -20,7 +20,7 @@
 from eva.assistant.tools.tool_executor import ToolExecutor, execute_and_log_tool
 from eva.models.agents import AgentConfig
 from eva.models.config import ModelConfig
-from eva.utils.audio_utils import pcm16_mix, save_pcm_as_wav
+from eva.utils.audio_utils import pcm16_mix, save_audio_track
 from eva.utils.culture import get_initial_message
 from eva.utils.logging import get_logger
 from eva.utils.prompt_manager import PromptManager
@@ -156,23 +156,7 @@ async def stop(self) -> asyncio.Task | None:
 
         # Auto-compute mixed audio from tracks if not already populated (S2S servers
         # populate user/assistant tracks but not the mixed buffer directly).
-        if not self._audio_buffer:
-            if self.user_audio_buffer and self.assistant_audio_buffer:
-                diff_bytes = abs(len(self.user_audio_buffer) - len(self.assistant_audio_buffer))
-                diff_ms = diff_bytes / (2 * self._audio_sample_rate) * 1000
-                if diff_ms > 500:
-                    logger.warning(
-                        f"Audio buffer length mismatch: user={len(self.user_audio_buffer)} "
-                        f"assistant={len(self.assistant_audio_buffer)} "
-                        f"diff={diff_ms:.0f}ms — mixed recording may be temporally skewed"
-                    )
-                self._audio_buffer = bytearray(
-                    pcm16_mix(bytes(self.user_audio_buffer), bytes(self.assistant_audio_buffer))
-                )
-            elif self.user_audio_buffer:
-                self._audio_buffer = bytearray(self.user_audio_buffer)
-            elif self.assistant_audio_buffer:
-                self._audio_buffer = bytearray(self.assistant_audio_buffer)
+        self._ensure_mixed_audio()
 
         # Extract bytes and clear in-memory buffers so the caller can release its
         # concurrency slot while audio writes happen in a background thread.
@@ -264,20 +248,22 @@ def _save_transcript(self) -> None:
         """
         self.audit_log.save_transcript_jsonl(self.output_dir / "transcript.jsonl")
 
-    def _save_audio(self) -> None:
-        """Save accumulated audio buffers to WAV files.
+    def _ensure_mixed_audio(self) -> None:
+        """Populate ``_audio_buffer`` (mixed track) from the per-channel tracks.
 
-        If _audio_buffer (mixed) is empty but user and assistant buffers are
-        available, compute mixed audio automatically via sample-wise addition.
+        No-op if the mixed buffer is already populated. When only user + assistant
+        tracks exist (S2S/realtime servers populate those, not the mixed buffer),
+        mix them sample-wise; when only one track exists, use it as-is.
 
         NOTE: user_audio_buffer and assistant_audio_buffer must be time-aligned
-        (same total length in samples) before this method is called.  S2s/realtime
-        servers are responsible for calling ``sync_buffer_to_position`` during
-        streaming so the two tracks stay aligned.  A length mismatch produces a
-        usable but temporally skewed mixed recording.
+        (same total length in samples) before mixing. S2S/realtime servers are
+        responsible for calling ``sync_buffer_to_position`` during streaming so the
+        two tracks stay aligned. A length mismatch produces a usable but temporally
+        skewed mixed recording.
         """
-        # Auto-compute mixed audio from user + assistant tracks when not populated
-        if not self._audio_buffer and self.user_audio_buffer and self.assistant_audio_buffer:
+        if self._audio_buffer:
+            return
+        if self.user_audio_buffer and self.assistant_audio_buffer:
             diff_bytes = abs(len(self.user_audio_buffer) - len(self.assistant_audio_buffer))
             diff_ms = diff_bytes / (2 * self._audio_sample_rate) * 1000  # 16-bit PCM → 2 bytes/sample
             if diff_ms > 500:
@@ -287,33 +273,11 @@ def _save_audio(self) -> None:
                     f"diff={diff_ms:.0f}ms — mixed recording may be temporally skewed"
                 )
             self._audio_buffer = bytearray(pcm16_mix(bytes(self.user_audio_buffer), bytes(self.assistant_audio_buffer)))
-        elif not self._audio_buffer and self.user_audio_buffer:
+        elif self.user_audio_buffer:
             self._audio_buffer = bytearray(self.user_audio_buffer)
-        elif not self._audio_buffer and self.assistant_audio_buffer:
+        elif self.assistant_audio_buffer:
             self._audio_buffer = bytearray(self.assistant_audio_buffer)
 
-        if self._audio_buffer:
-            save_pcm_as_wav(
-                bytes(self._audio_buffer),
-                self.output_dir / "audio_mixed.wav",
-                self._audio_sample_rate,
-                1,
-            )
-        if self.user_audio_buffer:
-            save_pcm_as_wav(
-                bytes(self.user_audio_buffer),
-                self.output_dir / "audio_user.wav",
-                self._audio_sample_rate,
-                1,
-            )
-        if self.assistant_audio_buffer:
-            save_pcm_as_wav(
-                bytes(self.assistant_audio_buffer),
-                self.output_dir / "audio_assistant.wav",
-                self._audio_sample_rate,
-                1,
-            )
-
     def _save_audio_deferred(
         self,
         mixed_audio: bytes,
@@ -322,12 +286,9 @@ def _save_audio_deferred(
         sample_rate: int,
     ) -> None:
         """Write pre-extracted audio bytes to WAV files off the event loop."""
-        if mixed_audio:
-            save_pcm_as_wav(mixed_audio, self.output_dir / "audio_mixed.wav", sample_rate, 1)
-        if user_audio:
-            save_pcm_as_wav(user_audio, self.output_dir / "audio_user.wav", sample_rate, 1)
-        if assistant_audio:
-            save_pcm_as_wav(assistant_audio, self.output_dir / "audio_assistant.wav", sample_rate, 1)
+        save_audio_track(mixed_audio, self.output_dir / "audio_mixed.wav", sample_rate)
+        save_audio_track(user_audio, self.output_dir / "audio_user.wav", sample_rate)
+        save_audio_track(assistant_audio, self.output_dir / "audio_assistant.wav", sample_rate)
         if mixed_audio or user_audio or assistant_audio:
             logger.info(f"Saved audio files to {self.output_dir} ({len(mixed_audio)} bytes mixed)")
 

src/eva/user_simulator/audio_bridge.py

@@ -256,7 +256,6 @@ def output(self, audio: bytes) -> None:
                     audio = self._perturbator.apply(audio)
                 self.send_queue.put_nowait(audio)
                 if self.record_callback:
-                    self.record_callback("user", audio)
                     self.record_callback("user_clean", clean_audio)
             except asyncio.QueueFull:
                 logger.warning("Send queue full, dropping audio")
@@ -796,7 +795,6 @@ async def _send_to_assistant(self) -> None:
                             next_send_time = silence_start_time + (silence_chunks_sent * send_interval)
                             # Record only after successful send to prevent double-recording on retry
                             if self.record_callback:
-                                self.record_callback("assistant", silence_pcm)
                                 self.record_callback("user_clean", silence_pcm)
                             if silence_chunks_sent % LOG_INTERVAL_SILENCE == 0:
                                 actual_elapsed = current_time - silence_start_time

src/eva/user_simulator/base.py

@@ -14,6 +14,7 @@
 from eva.models.config import LANGUAGE_DISPLAY_NAMES, PerturbationConfig
 from eva.user_simulator.event_logger import UserSimulatorEventLogger
 from eva.user_simulator.perturbation import AudioPerturbator
+from eva.utils.audio_utils import save_audio_track
 from eva.utils.culture import add_user_language_directive
 from eva.utils.logging import current_record_id, get_logger
 from eva.utils.prompt_manager import PromptManager
@@ -82,9 +83,7 @@ def __init__(
             provider=provider,
         )
 
-        self._user_audio_chunks: list[bytes] = []
-        self._assistant_audio_chunks: list[bytes] = []
-        self._user_clean_audio_chunks: list[bytes] = []
+        self._user_clean_audio = bytearray()
         self._record_id = current_record_id.get()
 
     @abstractmethod
@@ -163,21 +162,21 @@ def _on_assistant_speaks(self, transcript: str) -> None:
     def _record_audio(self, source: str, audio_data: bytes) -> None:
         """Record audio for later analysis.
 
+        Only the clean (unperturbed) user track is persisted — it is the one
+        artifact the assistant server never sees and therefore cannot record.
+        Other sources are captured by the assistant server's own recording path.
+
         Args:
-            source: "user", "assistant", or "user_clean"
+            source: recording channel; only "user_clean" is retained
             audio_data: Raw audio bytes
         """
-        if source == "user":
-            self._user_audio_chunks.append(audio_data)
-        elif source == "assistant":
-            self._assistant_audio_chunks.append(audio_data)
-        elif source == "user_clean":
-            self._user_clean_audio_chunks.append(audio_data)
-
-    def get_recorded_audio(self) -> tuple[bytes, bytes]:
-        """Get the recorded audio.
-
-        Returns:
-            Tuple of (user_audio, assistant_audio) as raw bytes
+        if source == "user_clean":
+            self._user_clean_audio.extend(audio_data)
+
+    def _save_clean_user_audio(self, sample_rate: int) -> None:
+        """Persist the recorded clean user track to ``audio_user_clean.wav``.
+
+        Shared by all providers; skips writing when no clean audio was recorded.
         """
-        return b"".join(self._user_audio_chunks), b"".join(self._assistant_audio_chunks)
+        if save_audio_track(bytes(self._user_clean_audio), self.output_dir / "audio_user_clean.wav", sample_rate):
+            logger.info(f"Saved clean user audio to {self.output_dir / 'audio_user_clean.wav'}")

src/eva/user_simulator/elevenlabs.py

@@ -19,7 +19,6 @@
 from eva.models.config import PerturbationConfig
 from eva.user_simulator.audio_bridge import ELEVENLABS_OUTPUT_RATE, ElevenLabsAudioInterface
 from eva.user_simulator.base import AbstractUserSimulator
-from eva.utils.audio_utils import save_pcm_as_wav
 from eva.utils.logging import current_record_id, get_logger
 
 logger = get_logger(__name__)
@@ -249,15 +248,7 @@ async def _run_elevenlabs_conversation(self, api_key: str) -> str:
                         )
                     logger.info(f"Saved {len(latencies)} response latencies to {latency_file}")
 
-            if self._user_clean_audio_chunks:
-                clean_audio_path = self.output_dir / "audio_user_clean.wav"
-                save_pcm_as_wav(
-                    b"".join(self._user_clean_audio_chunks),
-                    clean_audio_path,
-                    sample_rate=ELEVENLABS_OUTPUT_RATE,
-                    num_channels=1,
-                )
-                logger.info(f"Saved clean user audio to {clean_audio_path}")
+            self._save_clean_user_audio(ELEVENLABS_OUTPUT_RATE)
 
             # Grace period: keep the WebSocket open so the assistant pipeline
             # (Pipecat STT) can finish processing the last user utterance.

src/eva/user_simulator/openai_realtime.py

@@ -20,7 +20,6 @@
 from eva.models.config import OpenAIRealtimeSimulatorConfig, PerturbationConfig
 from eva.user_simulator.audio_bridge import BotToBotAudioBridge
 from eva.user_simulator.base import AbstractUserSimulator
-from eva.utils.audio_utils import save_pcm_as_wav
 from eva.utils.logging import get_logger
 
 logger = get_logger(__name__)
@@ -214,7 +213,7 @@ async def _run_openai_conversation(self, api_key: str) -> None:
                     await self._cancel_background_task(task)
             await client.close()
             await self._audio_interface.stop_async()
-            self._save_user_audio()
+            self._save_clean_user_audio(BRIDGE_SAMPLE_RATE)
             self.event_logger.log_connection_state("session_ended", {"reason": self._end_reason})
 
     @staticmethod
@@ -425,13 +424,3 @@ def _flush_caller_output(self) -> None:
         if self._caller_audio_seen and self._audio_interface is not None:
             self._audio_interface.output(b"\x00\x00")
             self._caller_audio_seen = False
-
-    def _save_user_audio(self) -> None:
-        if not self._user_clean_audio_chunks:
-            return
-        save_pcm_as_wav(
-            b"".join(self._user_clean_audio_chunks),
-            self.output_dir / "audio_user_clean.wav",
-            sample_rate=BRIDGE_SAMPLE_RATE,
-            num_channels=1,
-        )

src/eva/utils/audio_utils.py

@@ -152,3 +152,21 @@ def save_pcm_as_wav(
         logger.debug(f"Audio saved to {file_path} ({len(audio_data)} bytes)")
     except Exception as e:
         logger.error(f"Error saving audio to {file_path}: {e}")
+
+
+def save_audio_track(
+    audio_bytes: bytes,
+    file_path: Path,
+    sample_rate: int,
+    num_channels: int = 1,
+) -> bool:
+    """Save a single-track PCM recording to a WAV file, skipping empty audio.
+
+    Returns True if a file was written, False if there was no audio to save.
+    This is the shared entry point for both the assistant server's deferred
+    audio saving and the user simulator's clean-track saving.
+    """
+    if not audio_bytes:
+        return False
+    save_pcm_as_wav(audio_bytes, file_path, sample_rate, num_channels)
+    return True

tests/unit/user_simulator/test_elevenlabs.py

@@ -5,6 +5,7 @@
 """
 
 import asyncio
+import wave
 from pathlib import Path
 from types import SimpleNamespace
 from unittest.mock import AsyncMock, MagicMock, patch
@@ -357,16 +358,35 @@ async def test_cancelled_error_propagates(self, tmp_path):
                 await sim._keep_alive_task()
 
 
-class TestRecordAndRetrieveAudio:
-    """Test the full record → retrieve flow with interleaved audio."""
+class TestRecordAndSaveCleanAudio:
+    """Test the clean-user-track record → save flow."""
 
-    def test_interleaved_audio_preserved_in_order(self, tmp_path):
+    def test_only_clean_source_is_retained(self, tmp_path):
         sim = _make_simulator(tmp_path)
+        # Non-clean sources are no longer persisted by the simulator.
         sim._record_audio("user", b"\x01\x02")
         sim._record_audio("assistant", b"\xaa")
-        sim._record_audio("user", b"\x03")
-        sim._record_audio("assistant", b"\xbb\xcc")
+        sim._record_audio("user_clean", b"\x03\x04")
+        sim._record_audio("user_clean", b"\x05\x06")
 
-        user_audio, assistant_audio = sim.get_recorded_audio()
-        assert user_audio == b"\x01\x02\x03"
-        assert assistant_audio == b"\xaa\xbb\xcc"
+        assert sim._user_clean_audio == b"\x03\x04\x05\x06"
+
+    def test_save_clean_user_audio_writes_wav(self, tmp_path):
+        sim = _make_simulator(tmp_path)
+        sim._record_audio("user_clean", b"\x01\x02\x03\x04")
+
+        sim._save_clean_user_audio(sample_rate=16000)
+
+        wav_path = tmp_path / "audio_user_clean.wav"
+        assert wav_path.exists()
+        with wave.open(str(wav_path), "rb") as wav_file:
+            assert wav_file.getframerate() == 16000
+            assert wav_file.getnchannels() == 1
+            assert wav_file.readframes(wav_file.getnframes()) == b"\x01\x02\x03\x04"
+
+    def test_save_clean_user_audio_skips_when_empty(self, tmp_path):
+        sim = _make_simulator(tmp_path)
+
+        sim._save_clean_user_audio(sample_rate=16000)
+
+        assert not (tmp_path / "audio_user_clean.wav").exists()