From 4e20d7ff09492ede9f7fa65539d5913674082093 Mon Sep 17 00:00:00 2001
From: "raghav.mehndiratta" <raghav.mehndiratta@servicenow.com>
Date: Mon, 27 Jul 2026 13:17:52 -0700
Subject: [PATCH 1/4] consolidate tool calling code

---
 src/eva/assistant/agentic/audit_log.py   | 27 ------------------------
 src/eva/assistant/agentic/system.py      | 10 ++-------
 src/eva/assistant/base_server.py         | 21 ++++++++----------
 src/eva/assistant/tools/tool_executor.py | 27 ++++++++++++++++++++++++
 tests/unit/assistant/test_audit_log.py   |  6 ------
 5 files changed, 38 insertions(+), 53 deletions(-)

diff --git a/src/eva/assistant/agentic/audit_log.py b/src/eva/assistant/agentic/audit_log.py
index 14a366ea..fb74982e 100644
--- a/src/eva/assistant/agentic/audit_log.py
+++ b/src/eva/assistant/agentic/audit_log.py
@@ -343,33 +343,6 @@ def append_tool_response(self, tool_name: str, response: dict[str, Any]) -> None
         self.transcript.append(tool_response_entry)
         logger.debug(f"Audit: tool response for {tool_name}")
 
-    def append_realtime_tool_call(
-        self,
-        tool_name: str,
-        parameters: dict[str, Any],
-    ) -> None:
-        """Record a tool call from the realtime pipeline (no AgentConfig/AgentTool required).
-
-        Note: the S2S model processes raw audio and can call tools *before* transcription.completed fires,
-        so this may be appended before the corresponding user entry.  Correct chronological order is guaranteed
-        by ``save()`` sorting the transcript by timestamp — the user entry carries the ``speech_started`` wall-clock
-        which is always earlier than the tool call's ``current_timestamp_ms()``.
-        """
-        tool_call_entry = {
-            "value": {"tool": tool_name, "parameters": parameters},
-            "displayName": "Tool",
-            "type": "tool_call",
-            "isBotMessage": True,
-            "timestamp": current_timestamp_ms(),
-            "message_type": "tool_call",
-        }
-        self.transcript.append(tool_call_entry)
-        self._tool_calls_count += 1
-        if tool_name not in self._tools_called:
-            self._tools_called.append(tool_name)
-        self._last_tool_call = tool_name
-        logger.debug(f"Audit: realtime tool call - {tool_name}")
-
     def get_conversation_messages(
         self,
         max_messages: int | None = None,
diff --git a/src/eva/assistant/agentic/system.py b/src/eva/assistant/agentic/system.py
index e61530f3..2d0cec53 100644
--- a/src/eva/assistant/agentic/system.py
+++ b/src/eva/assistant/agentic/system.py
@@ -17,7 +17,7 @@
     LLMCall,
     MessageRole,
 )
-from eva.assistant.tools.tool_executor import ToolExecutor
+from eva.assistant.tools.tool_executor import ToolExecutor, execute_and_log_tool
 from eva.models.agents import AgentConfig
 from eva.utils.conversation_checks import LLM_GENERIC_ERROR_MESSAGE as GENERIC_ERROR
 from eva.utils.error_handler import categorize_error
@@ -450,7 +450,7 @@ async def _run_tool_loop(
                     self.audit_log.append_assistant_output(transfer_message, reasoning=reasoning_content)
                     return
 
-                result = await self.tool_handler.execute(tool_name, params)
+                result = await execute_and_log_tool(self.tool_handler, self.audit_log, tool_name, params)
 
                 if result.get("status") == "error":
                     logger.warning(f"❌ Tool error: {tool_name} - {result.get('message', 'Unknown error')}")
@@ -458,12 +458,6 @@ async def _run_tool_loop(
                     logger.info(f"✅ Tool response: {tool_name}")
                     logger.info(f"   Result: {json.dumps(result, indent=2, ensure_ascii=False)}")
 
-                self.audit_log.append_tool_call(
-                    tool_name=tool_name,
-                    parameters=params,
-                    response=result,
-                )
-
                 # Add tool response to messages
                 tool_content = json.dumps(result, ensure_ascii=False)
                 messages.append(
diff --git a/src/eva/assistant/base_server.py b/src/eva/assistant/base_server.py
index 43371fce..179a8460 100644
--- a/src/eva/assistant/base_server.py
+++ b/src/eva/assistant/base_server.py
@@ -17,7 +17,7 @@
 
 from eva.assistant.agentic.audit_log import AuditLog
 from eva.assistant.audio_bridge import FrameworkLogWriter, MetricsLogWriter
-from eva.assistant.tools.tool_executor import ToolExecutor
+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 save_pcm_as_wav
@@ -225,19 +225,16 @@ def get_final_scenario_db(self) -> dict[str, Any]:
     async def execute_tool(self, tool_name: str, arguments: dict) -> Any:
         """Execute a tool call and record it in the audit log.
 
-        Logs the call and response as separate timestamped entries so latency
-        between them is preserved.  Use this whenever the server handles tool
-        calls directly (s2s/realtime events, or any custom cascade that
-        doesn't delegate to AgenticSystem).
+        Thin wrapper over the shared ``execute_and_log_tool`` helper (the single
+        assistant-side tool-execution path). Use this whenever the server
+        handles tool calls directly (s2s/realtime events, or any custom cascade
+        that doesn't delegate to AgenticSystem).
 
-        Note: AgenticSystem has its own tool execution + logging loop
-        (``append_tool_call``), so Pipecat cascade pipelines that use
-        AgenticSystem should *not* also call this method.
+        Note: AgenticSystem routes through the same ``execute_and_log_tool``
+        helper, so Pipecat cascade pipelines that use AgenticSystem should
+        *not* also call this method (it would double-log).
         """
-        self.audit_log.append_realtime_tool_call(tool_name, arguments)
-        result = await self.tool_handler.execute(tool_name, arguments)
-        self.audit_log.append_tool_response(tool_name, result)
-        return result
+        return await execute_and_log_tool(self.tool_handler, self.audit_log, tool_name, arguments)
 
     # ── Shared output helpers ──────────────────────────────────────────
 
diff --git a/src/eva/assistant/tools/tool_executor.py b/src/eva/assistant/tools/tool_executor.py
index 32bd6c57..04c65997 100644
--- a/src/eva/assistant/tools/tool_executor.py
+++ b/src/eva/assistant/tools/tool_executor.py
@@ -5,15 +5,42 @@
 import json
 from collections.abc import Callable
 from pathlib import Path
+from typing import TYPE_CHECKING
 
 import yaml
 from pipecat.services.llm_service import FunctionCallParams
 
 from eva.utils.logging import get_logger
 
+if TYPE_CHECKING:
+    from eva.assistant.agentic.audit_log import AuditLog
+
 logger = get_logger(__name__)
 
 
+async def execute_and_log_tool(
+    tool_handler: "ToolExecutor",
+    audit_log: "AuditLog",
+    tool_name: str,
+    params: dict,
+) -> dict:
+    """Single assistant-side tool-execution path: log call, execute, log response.
+
+    Logs the call entry *before* execution and the response entry *after* it, so
+    the audit log preserves the call→response latency. This is the one place a
+    tool is executed and recorded on the assistant side; both the cascade
+    (``AgenticSystem``) and the realtime/S2S servers route through it so they
+    produce identical audit entries.
+
+    In a later refactor phase (see docs/refactor-step1.md) this function becomes
+    the body of ``AssistantRole.handle_tool_call_request``.
+    """
+    audit_log.append_tool_call(tool_name, params)
+    result = await tool_handler.execute(tool_name, params)
+    audit_log.append_tool_response(tool_name, result)
+    return result
+
+
 class ToolExecutor:
     """Python function-based tool executor.
 
diff --git a/tests/unit/assistant/test_audit_log.py b/tests/unit/assistant/test_audit_log.py
index 2b6689c0..f669942f 100644
--- a/tests/unit/assistant/test_audit_log.py
+++ b/tests/unit/assistant/test_audit_log.py
@@ -207,12 +207,6 @@ def test_last_tool_call_tracked(self):
         self.log.append_tool_call("book", {})
         assert self.log._last_tool_call == "book"
 
-    def test_append_realtime_tool_call(self):
-        self.log.append_realtime_tool_call("get_flight", {"id": "123"})
-        assert len(self.log.transcript) == 1
-        assert self.log._tool_calls_count == 1
-        assert self.log._tools_called == ["get_flight"]
-
     def test_get_conversation_messages_empty(self):
         result = self.log.get_conversation_messages()
         assert result == []

From 78feef1feb4be0c5b9b1725f448103edde049e44 Mon Sep 17 00:00:00 2001
From: "raghav.mehndiratta" <raghav.mehndiratta@servicenow.com>
Date: Mon, 27 Jul 2026 14:14:30 -0700
Subject: [PATCH 2/4] update with fallback

---
 src/eva/backend/base.py   | 30 +++++++++-------
 src/eva/role/assistant.py | 76 ++++++++++++++++++++++++---------------
 2 files changed, 65 insertions(+), 41 deletions(-)

diff --git a/src/eva/backend/base.py b/src/eva/backend/base.py
index 786931ba..63bbd278 100644
--- a/src/eva/backend/base.py
+++ b/src/eva/backend/base.py
@@ -126,15 +126,19 @@ class BackendEvent:
     being present across providers.
 
     Convention (not enforced by this contract): a backend that proactively
-    re-engages after an idle period (see ``AssistantRole``'s
-    ``self_nudge_timeout_seconds``) may set ``metadata["is_nudge"] = True`` on
-    the ``AUDIO_OUTPUT``/``TRANSCRIPT`` event it emits for that turn, purely
-    so callers that want to distinguish a self-initiated nudge from an
-    ordinary model turn (e.g. for audit logging) can do so. This is *not* a
-    new event type -- a nudge is just an ordinary turn from the backend's
-    model, triggered by the backend noticing its own idle timeout rather than
-    by new input; it flows through the same ``receive()`` surface as
-    anything else."""
+    re-engages after a dropped user turn (the turn-end fallback; see
+    ``AssistantRole``'s ``turn_end_fallback_seconds`` and the shipped
+    ``eva.assistant.pipeline.fallback``) tags the ``AUDIO_OUTPUT``/
+    ``TRANSCRIPT`` event it emits for that turn so callers can distinguish a
+    fallback nudge from an ordinary model turn (e.g. for audit logging and so
+    downstream metrics can zero it). The shipped feature records the transcript
+    marker with ``message_type="turn_fallback"``; a backend surfacing the same
+    turn here should carry an equivalent flag in ``metadata`` (e.g.
+    ``metadata["turn_fallback"] = True``). This is *not* a new event type -- a
+    nudge is just an ordinary turn from the backend's model, triggered by the
+    backend noticing that a user turn was never detected within the fallback
+    window rather than by new input; it flows through the same ``receive()``
+    surface as anything else."""
 
 
 class Backend(ABC):
@@ -199,11 +203,11 @@ async def open(self, *, system_prompt: str, tools: list[dict[str, Any]] | None,
                 validates its own config shape; the abstract contract does
                 not prescribe one, since a native S2S config and a cascade
                 config share little structure. An ``AssistantRole`` backend
-                configured to self-nudge (see
-                ``AssistantRole.self_nudge_timeout_seconds``) reads its
+                configured for the turn-end fallback (see
+                ``AssistantRole.turn_end_fallback_seconds``) reads its
                 threshold from this blob (e.g. a
-                ``config["self_nudge_timeout_seconds"]`` key) the same way --
-                self-nudging needs no dedicated typed parameter or new
+                ``config["turn_end_fallback_seconds"]`` key) the same way --
+                the fallback needs no dedicated typed parameter or new
                 ``Backend`` method, since the resulting nudge is just an
                 ordinary outbound turn (see ``BackendEvent.metadata``).
 
diff --git a/src/eva/role/assistant.py b/src/eva/role/assistant.py
index f8eb00b8..897dba14 100644
--- a/src/eva/role/assistant.py
+++ b/src/eva/role/assistant.py
@@ -40,20 +40,37 @@ class AssistantRole(Role):
     (constructed by subclasses, not by this contract) to fulfill
     ``handle_tool_call_request``.
 
-    Self-nudge: if the caller goes quiet for too long mid-call, the assistant
-    itself proactively re-engages ("are you still there?") rather than
-    waiting forever -- this is assistant-initiated, unlike a caller nudging an
-    unresponsive agent. Unlike the tool-call/idle-detection seams elsewhere in
-    this contract, self-nudging needs no new ``Role`` method and no new
-    ``Backend`` event type: the nudge is just an ordinary outbound turn that
-    this role's backend produces on its own after
-    ``self_nudge_timeout_seconds`` of inactivity, using the same
-    ``system_prompt``/instructions already established at ``open()`` time
-    (see ``Backend.open``'s ``config`` docstring). Whether the *other* side
-    (a ``UserRole``) needs to do anything special upon receiving it, versus
-    just treating it as an ordinary assistant turn through its existing
-    ``run()`` loop, is left open -- see docs/refactor-step1.md discussion;
-    nothing here requires ``UserRole`` changes to handle it correctly today.
+    Turn-end fallback (self-nudge): the assistant's backstop for a *dropped
+    user turn*. When VAD / turn detection silently fails to fire for a real
+    user utterance, the call would otherwise hang until the provider's
+    inactivity timeout ends it. After the assistant stops speaking, if no user
+    turn is detected within ``turn_end_fallback_seconds``, the assistant
+    proactively re-engages with a nudge (acknowledge-and-answer if partial
+    user speech/audio was captured, otherwise ask the caller to repeat). This
+    is the seam already shipped as the pipeline-side ``TurnEndFallbackTimer``
+    (see ``eva.assistant.pipeline.fallback`` and ``EVA_TURN_END_FALLBACK_TIME``);
+    it works for both cascade and audio-LLM pipelines.
+
+    Two policies the backend owns, carried over from the shipped feature:
+    - Give up after a small number of *consecutive* nudges without a real user
+      turn resetting the count (``MAX_CONSECUTIVE_FALLBACK_NUDGES``), then let
+      the provider's inactivity backstop end the call.
+    - Never nudge once the call is ending (a nudge during teardown produces a
+      phantom assistant turn after the conversation is logically closed).
+
+    Unlike the tool-call/idle-detection seams elsewhere in this contract, the
+    fallback needs no new ``Role`` method and no new ``Backend`` event type:
+    the nudge is just an ordinary outbound turn that this role's backend
+    produces on its own after the timeout, using the same
+    ``system_prompt``/instructions already established at ``open()`` time (see
+    ``Backend.open``'s ``config`` docstring). It is surfaced through the normal
+    ``receive()`` stream and tagged so downstream metrics can identify and zero
+    it (the shipped feature records the transcript marker with
+    ``message_type="turn_fallback"``; see ``BackendEvent.metadata``). Whether
+    the *other* side (a ``UserRole``) needs to do anything special upon
+    receiving it, versus just treating it as an ordinary assistant turn through
+    its existing ``run()`` loop, is left open -- see docs/refactor-step1.md
+    discussion; nothing here requires ``UserRole`` changes to handle it today.
     """
 
     def __init__(
@@ -65,7 +82,7 @@ def __init__(
         agent_config_path: str,
         scenario_db_path: str,
         current_date_time: str,
-        self_nudge_timeout_seconds: float | None = None,
+        turn_end_fallback_seconds: float | None = None,
     ) -> None:
         """Initialize the assistant role.
 
@@ -83,25 +100,28 @@ def __init__(
                 prompt construction and tool execution (mirrors existing
                 ``current_date_time`` plumbing throughout the assistant
                 stack).
-            self_nudge_timeout_seconds: How long the assistant backend should
-                wait without hearing from the caller before proactively
-                speaking again, or ``None`` to disable self-nudging entirely.
-                This is an ``AssistantRole``-level knob, not a
+            turn_end_fallback_seconds: How long after the assistant stops
+                speaking to wait for a user turn before firing a turn-end
+                fallback nudge, or ``None`` to disable the fallback entirely
+                (preserving the old behavior of waiting for the provider's
+                inactivity timeout). Mirrors the shipped
+                ``EVA_TURN_END_FALLBACK_TIME`` knob. This is an
+                ``AssistantRole``-level tuning value, not a
                 ``BackendCapabilities`` flag (capabilities describe what a
-                backend *can* do, statically; this is a per-run tuning
-                value). Wiring it into the constructed ``self.backend``'s own
-                config (via ``backend_config`` / ``Backend.open(config=...)``)
-                is left to the concrete subclass's constructor, same as
-                elsewhere in this contract -- a ``Role`` does not otherwise
-                reach into backend config after construction. A backend with
-                no notion of provider-driven idle timing (e.g. a thin
-                end-to-end backend) may simply ignore this value.
+                backend *can* do, statically). Wiring it into the constructed
+                ``self.backend``'s own config (via ``backend_config`` /
+                ``Backend.open(config=...)``) is left to the concrete
+                subclass's constructor, same as elsewhere in this contract --
+                a ``Role`` does not otherwise reach into backend config after
+                construction. A backend with no notion of idle timing (e.g. a
+                thin end-to-end backend that relies on its own provider
+                backstop) may simply ignore this value.
         """
         super().__init__(backend_factory=backend_factory, backend_name=backend_name, backend_config=backend_config)
         self.agent_config_path = agent_config_path
         self.scenario_db_path = scenario_db_path
         self.current_date_time = current_date_time
-        self.self_nudge_timeout_seconds = self_nudge_timeout_seconds
+        self.turn_end_fallback_seconds = turn_end_fallback_seconds
 
     @abstractmethod
     def get_final_scenario_db(self) -> dict[str, Any]:

From 7bb05212d8eb545ef981802f78c83cc51f8649b3 Mon Sep 17 00:00:00 2001
From: "raghav.mehndiratta" <raghav.mehndiratta@servicenow.com>
Date: Wed, 29 Jul 2026 10:34:28 -0700
Subject: [PATCH 3/4] remove dead audio accumulation

---
 src/eva/assistant/base_server.py             | 81 +++++---------------
 src/eva/user_simulator/audio_bridge.py       |  2 -
 src/eva/user_simulator/base.py               | 29 +++----
 src/eva/user_simulator/elevenlabs.py         | 11 +--
 src/eva/user_simulator/openai_realtime.py    | 12 +--
 src/eva/utils/audio_utils.py                 | 20 +++++
 tests/unit/user_simulator/test_elevenlabs.py | 36 +++++++--
 7 files changed, 86 insertions(+), 105 deletions(-)

diff --git a/src/eva/assistant/base_server.py b/src/eva/assistant/base_server.py
index 179a8460..76267f40 100644
--- a/src/eva/assistant/base_server.py
+++ b/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 save_pcm_as_wav
+from eva.utils.audio_utils import 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,25 +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"
-                    )
-                from eva.assistant.audio_bridge import pcm16_mix  # lazy: avoids circular import at module load
-
-                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.
@@ -266,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:
@@ -288,36 +272,14 @@ def _save_audio(self) -> None:
                     f"assistant={len(self.assistant_audio_buffer)} "
                     f"diff={diff_ms:.0f}ms — mixed recording may be temporally skewed"
                 )
-            from eva.assistant.audio_bridge import pcm16_mix
+            from eva.assistant.audio_bridge import pcm16_mix  # lazy: avoids circular import at module load
 
             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,
@@ -326,12 +288,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)")
 
diff --git a/src/eva/user_simulator/audio_bridge.py b/src/eva/user_simulator/audio_bridge.py
index d1e75415..15a949df 100644
--- a/src/eva/user_simulator/audio_bridge.py
+++ b/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
diff --git a/src/eva/user_simulator/base.py b/src/eva/user_simulator/base.py
index fb65227d..37372039 100644
--- a/src/eva/user_simulator/base.py
+++ b/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,8 +83,6 @@ def __init__(
             provider=provider,
         )
 
-        self._user_audio_chunks: list[bytes] = []
-        self._assistant_audio_chunks: list[bytes] = []
         self._user_clean_audio_chunks: list[bytes] = []
         self._record_id = current_record_id.get()
 
@@ -163,21 +162,25 @@ 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":
+        if source == "user_clean":
             self._user_clean_audio_chunks.append(audio_data)
 
-    def get_recorded_audio(self) -> tuple[bytes, bytes]:
-        """Get the recorded audio.
+    def _save_clean_user_audio(self, sample_rate: int) -> None:
+        """Persist the recorded clean user track to ``audio_user_clean.wav``.
 
-        Returns:
-            Tuple of (user_audio, assistant_audio) as raw bytes
+        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(
+            self._user_clean_audio_chunks,
+            self.output_dir / "audio_user_clean.wav",
+            sample_rate,
+        ):
+            logger.info(f"Saved clean user audio to {self.output_dir / 'audio_user_clean.wav'}")
diff --git a/src/eva/user_simulator/elevenlabs.py b/src/eva/user_simulator/elevenlabs.py
index 8e2ac90a..9fd2a85b 100644
--- a/src/eva/user_simulator/elevenlabs.py
+++ b/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.
diff --git a/src/eva/user_simulator/openai_realtime.py b/src/eva/user_simulator/openai_realtime.py
index e056fd89..32128f58 100644
--- a/src/eva/user_simulator/openai_realtime.py
+++ b/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
@@ -426,12 +425,3 @@ def _flush_caller_output(self) -> 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,
-        )
diff --git a/src/eva/utils/audio_utils.py b/src/eva/utils/audio_utils.py
index 77c82eef..68e78549 100644
--- a/src/eva/utils/audio_utils.py
+++ b/src/eva/utils/audio_utils.py
@@ -25,3 +25,23 @@ 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(
+    data: bytes | list[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.
+
+    Accepts either raw PCM bytes or a list of PCM chunks (which are joined).
+    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.
+    """
+    audio_bytes = b"".join(data) if isinstance(data, list) else data
+    if not audio_bytes:
+        return False
+    save_pcm_as_wav(audio_bytes, file_path, sample_rate, num_channels)
+    return True
diff --git a/tests/unit/user_simulator/test_elevenlabs.py b/tests/unit/user_simulator/test_elevenlabs.py
index 24b7735d..97bb3ac2 100644
--- a/tests/unit/user_simulator/test_elevenlabs.py
+++ b/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_chunks == [b"\x03\x04", b"\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()

From 38d51858a9975843ea7c90205c0678fa489be150 Mon Sep 17 00:00:00 2001
From: raghavm243512 <44511569+raghavm243512@users.noreply.github.com>
Date: Thu, 30 Jul 2026 21:03:57 +0000
Subject: [PATCH 4/4] Apply pre-commit

---
 src/eva/user_simulator/openai_realtime.py | 1 -
 1 file changed, 1 deletion(-)

diff --git a/src/eva/user_simulator/openai_realtime.py b/src/eva/user_simulator/openai_realtime.py
index 32128f58..71c1c60b 100644
--- a/src/eva/user_simulator/openai_realtime.py
+++ b/src/eva/user_simulator/openai_realtime.py
@@ -424,4 +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
-