Refactor/phase1 code path consolidation

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,

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
@@ -459,20 +459,14 @@ 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')}")
                 else:
                     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(

src/eva/assistant/base_server.py

@@ -17,10 +17,10 @@
 
 from eva.assistant.agentic.audit_log import AuditLog
 from eva.assistant.pipeline.observers 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 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.
@@ -223,19 +207,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 ──────────────────────────────────────────
 
@@ -267,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:
@@ -290,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,
@@ -325,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/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.
 

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``).