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
@@ -450,20 +450,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,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 ──────────────────────────────────────────
 

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

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]:

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 == []