fix(voice): preserve e2e latency across agent handoffs

livekit-agents/livekit/agents/voice/agent_activity.py

@@ -115,6 +115,9 @@
 _AgentActivityContextVar = contextvars.ContextVar["AgentActivity"]("agents_activity")
 _SpeechHandleContextVar = contextvars.ContextVar["SpeechHandle"]("agents_speech_handle")
 _IdleHoldContextVar = contextvars.ContextVar[bool]("agents_idle_hold", default=False)
+_UserMetricsContextVar = contextvars.ContextVar[llm.MetricsReport | None](
+    "agents_user_metrics", default=None
+)
 
 
 async def _aligned_transcript_or_text(
@@ -1624,6 +1627,9 @@ def _generate_reply(
         if self.llm is None:
             raise RuntimeError("trying to generate reply without an LLM model")
 
+        handoff_user_metrics = self._session._handoff_user_metrics
+        self._session._handoff_user_metrics = None
+
         task = asyncio.current_task()
         if not is_given(tool_choice) and task is not None:
             if task_info := _get_activity_task_info(task):
@@ -1684,6 +1690,7 @@ def _generate_reply(
                     tools=resolved_tools if is_given(resolved_tools) else all_tools,
                     new_message=user_message if is_given(user_message) else None,
                     instructions=instructions or None,
+                    _previous_user_metrics=handoff_user_metrics,
                     model_settings=ModelSettings(
                         tool_choice=tool_choice
                         if utils.is_given(tool_choice) or self._tool_choice is None
@@ -3494,15 +3501,19 @@ def _tool_execution_completed_cb(out: ToolExecutionOutput) -> None:
             speech_handle._item_added([out.fnc_call_out])
 
         # start to execute tools (only after play())
-        exe_task, tool_output = perform_tool_executions(
-            session=self._session,
-            speech_handle=speech_handle,
-            tool_ctx=tool_ctx,
-            tool_choice=model_settings.tool_choice,
-            function_stream=llm_gen_data.function_ch,
-            tool_execution_started_cb=_tool_execution_started_cb,
-            tool_execution_completed_cb=_tool_execution_completed_cb,
-        )
+        user_metrics_token = _UserMetricsContextVar.set(user_metrics)
+        try:
+            exe_task, tool_output = perform_tool_executions(
+                session=self._session,
+                speech_handle=speech_handle,
+                tool_ctx=tool_ctx,
+                tool_choice=model_settings.tool_choice,
+                function_stream=llm_gen_data.function_ch,
+                tool_execution_started_cb=_tool_execution_started_cb,
+                tool_execution_completed_cb=_tool_execution_completed_cb,
+            )
+        finally:
+            _UserMetricsContextVar.reset(user_metrics_token)
 
         # use the TTS-aligned timing for the transcript instead of the raw text, when
         # the TTS supports it (resolved per segment below)
@@ -3722,6 +3733,7 @@ async def _next_segment() -> _SpeechSegment | None:
 
             draining = self.scheduling_paused
             if fnc_executed_ev._handoff_required and new_agent_task and not ignore_task_switch:
+                self._session._handoff_user_metrics = user_metrics
                 self._session.update_agent(new_agent_task)
                 draining = True
 

livekit-agents/livekit/agents/voice/agent_session.py

@@ -698,6 +698,9 @@ def __init__(
         self._foreground_guards: set[asyncio.Future[None]] = set()
         # TODO(theomonnom): need a better way to expose early assistant metrics
         self._early_assistant_metrics: MetricsReport | None = None
+        # Metrics for the user turn that triggered an in-flight agent handoff.
+        # The next activity consumes this when its on_enter callback generates a reply.
+        self._handoff_user_metrics: MetricsReport | None = None
 
         # trace
         self._user_speaking_span: trace.Span | None = None
@@ -1622,6 +1625,11 @@ def commit_user_turn(
     def update_agent(self, agent: Agent) -> None:
         self._agent = agent
 
+        from .agent_activity import _UserMetricsContextVar
+
+        if (user_metrics := _UserMetricsContextVar.get()) is not None:
+            self._handoff_user_metrics = user_metrics
+
         if self._started:
             # immediately block the old activity from accepting new user turns
             # during the transition window (before drain() formally pauses scheduling)

tests/test_update_agent_long_on_enter.py

@@ -8,6 +8,7 @@
 from livekit.agents.llm import FunctionToolCall
 
 from .fake_llm import FakeLLM, FakeLLMResponse
+from .fake_session import FakeActions, create_session, run_session
 
 pytestmark = [pytest.mark.unit, pytest.mark.virtual_time, pytest.mark.no_concurrent]
 
@@ -146,3 +147,54 @@ async def test_update_agent_long_on_enter_no_deadlock():
         # the next turn completes the task
         second_result = await asyncio.wait_for(sess.run(user_input="Bob"), timeout=5.0)
         second_result.expect.contains_function_call(name="record_name")
+
+
+@pytest.mark.asyncio
+async def test_handoff_reply_preserves_user_metrics() -> None:
+    class HandoffTarget(Agent):
+        def __init__(self) -> None:
+            super().__init__(instructions="handoff target")
+
+        async def on_enter(self) -> None:
+            await self.session.generate_reply(instructions="handoff_reply")
+
+    class HandoffSource(Agent):
+        def __init__(self) -> None:
+            super().__init__(instructions="handoff source")
+
+        @function_tool
+        async def handoff(self, ctx: RunContext) -> None:
+            """Hand the current turn to the next agent."""
+            target = HandoffTarget()
+            self.session.update_agent(target)
+
+    actions = FakeActions()
+    actions.add_user_speech(0.5, 2.5, "go")
+    actions.add_llm(
+        content="",
+        tool_calls=[FunctionToolCall(name="handoff", arguments="{}", call_id="call_1")],
+    )
+    actions.add_llm("hello from the new agent", input="handoff_reply")
+    actions.add_tts(1.0)
+
+    session = create_session(actions)
+    conversation_events = []
+    session.on("conversation_item_added", conversation_events.append)
+
+    await run_session(session, HandoffSource(), drain_delay=1.0)
+
+    assistant_messages = [
+        event.item
+        for event in conversation_events
+        if event.item.type == "message" and event.item.role == "assistant"
+    ]
+    user_messages = [
+        event.item
+        for event in conversation_events
+        if event.item.type == "message" and event.item.role == "user"
+    ]
+    assert len(user_messages) == 1
+    assert "stopped_speaking_at" in user_messages[0].metrics
+    assert len(assistant_messages) == 1
+    assert assistant_messages[0].text_content == "hello from the new agent"
+    assert "e2e_latency" in assistant_messages[0].metrics