From 4e20d7ff09492ede9f7fa65539d5913674082093 Mon Sep 17 00:00:00 2001 From: "raghav.mehndiratta" Date: Mon, 27 Jul 2026 13:17:52 -0700 Subject: [PATCH 1/2] 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" Date: Mon, 27 Jul 2026 14:14:30 -0700 Subject: [PATCH 2/2] 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]: