fix: preserve adaptive interruption across tool calls

livekit-agents/livekit/agents/inference/interruption.py

@@ -667,7 +667,7 @@ async def _metrics_monitor_task(
                 prediction_duration=ev.prediction_duration,
                 detection_delay=ev.detection_delay,
                 num_interruptions=1 if ev.is_interruption else 0,
-                num_backchannels=1 if not ev.is_interruption else 0,
+                num_backchannels=1 if not ev.is_interruption and not ev.agent_ended else 0,
                 num_requests=ev.num_requests,
                 metadata=Metadata(
                     model_name=self._model.model, model_provider=self._model.provider

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

@@ -2608,8 +2608,9 @@ def _to_instructions(v: Instructions | str) -> Instructions:
 
     def _on_pipeline_reply_done(self, _: asyncio.Task[None]) -> None:
         if not self._speech_q and (not self._current_speech or self._current_speech.done()):
+            was_speaking = self._session.agent_state == "speaking"
             self._session._update_agent_state("listening")
-            if self._audio_recognition:
+            if self._audio_recognition and was_speaking:
                 self._audio_recognition._on_end_of_agent_speech(
                     ignore_user_transcript_until=time.time()
                 )
@@ -3331,7 +3332,14 @@ async def _next_segment() -> _SpeechSegment | None:
             current_span.set_attribute(trace_types.ATTR_RESPONSE_TEXT, forwarded_text)
 
         if not speech_handle.interrupted and len(tool_output.output) > 0:
+            was_speaking = self._session.agent_state == "speaking"
             self._session._update_agent_state("thinking")
+            if self._audio_recognition and was_speaking:
+                self._audio_recognition._on_end_of_agent_speech(
+                    ignore_user_transcript_until=time.time()
+                )
+            if self.interruption_enabled and was_speaking:
+                self._restore_interruption_by_audio_activity()
         elif self._session.agent_state == "speaking":
             self._session._update_agent_state("listening")
             if self._audio_recognition:

livekit-agents/livekit/agents/voice/audio_recognition.py

@@ -448,6 +448,9 @@ def _cancel_backchannel_boundary(self) -> None:
     # endregion
 
     def _on_start_of_agent_speech(self, started_at: float) -> None:
+        if self._agent_speaking:
+            return
+
         self._agent_speaking = True
         self._agent_speech_started_at = started_at
         self._endpointing.on_start_of_agent_speech(started_at=started_at)
@@ -464,22 +467,30 @@ def _on_start_of_agent_speech(self, started_at: float) -> None:
         if self._adaptive_interruption_active:
             self._interruption_ch.send_nowait(_AgentSpeechStartedSentinel())  # type: ignore[union-attr]
 
+        if self._speaking:
+            # Agent speech can begin mid-utterance without a new VAD onset, such as
+            # when playout resumes after a tool call.
+            self._on_start_of_speech(
+                started_at=started_at,
+                user_speaking_span=self._session._user_speaking_span,
+            )
+
     def _on_end_of_agent_speech(self, *, ignore_user_transcript_until: float) -> None:
         self._cancel_backchannel_boundary()
 
         if self._agent_speaking:
             self._endpointing.on_end_of_agent_speech(ended_at=time.time())
-
         if not self._adaptive_interruption_active:
             self._agent_speaking = False
             return
 
-        self._interruption_ch.send_nowait(_AgentSpeechEndedSentinel())  # type: ignore[union-attr]
-
         if self._agent_speaking:
+            ended_at = time.time()
             # no interruption is detected, end the inference (idempotent)
-            if not is_given(self._ignore_user_transcript_until):
-                self._on_end_of_overlap_speech(ended_at=time.time(), agent_ended=True)
+            if self._speaking and not is_given(self._ignore_user_transcript_until):
+                self._on_end_of_overlap_speech(ended_at=ended_at, agent_ended=True)
+
+            self._interruption_ch.send_nowait(_AgentSpeechEndedSentinel())  # type: ignore[union-attr]
 
             end_cooldown: float = (
                 self._backchannel_boundary[1] if self._backchannel_boundary else 0.0

livekit-agents/livekit/agents/voice/remote_session.py

@@ -595,6 +595,7 @@ def _on_overlapping_speech(self, event: OverlappingSpeechEvent) -> None:
             overlap_started_at = Timestamp()
             overlap_started_at.FromNanoseconds(int(event.overlap_started_at * 1e9))
 
+        # TODO(AGT-3180): Forward agent_ended when the remote-session protocol supports it.
         pb = agent_pb.AgentSessionEvent.OverlappingSpeech(
             is_interruption=event.is_interruption,
             detection_delay=event.detection_delay,

tests/test_agent_session.py

@@ -259,11 +259,29 @@ async def test_tool_call() -> None:
     session.on("function_tools_executed", tool_executed_events.append)
     session.output.audio.on("playback_finished", playback_finished_events.append)
 
-    t_origin = await asyncio.wait_for(run_session(session, agent), timeout=SESSION_TIMEOUT)
+    agent_speech_end_states: list[str] = []
+    on_end_of_agent_speech = AudioRecognition._on_end_of_agent_speech
+
+    def _record_agent_speech_end(
+        recognition: AudioRecognition, *, ignore_user_transcript_until: float
+    ) -> None:
+        agent_speech_end_states.append(session.agent_state)
+        on_end_of_agent_speech(
+            recognition,
+            ignore_user_transcript_until=ignore_user_transcript_until,
+        )
+
+    with patch.object(
+        AudioRecognition,
+        "_on_end_of_agent_speech",
+        _record_agent_speech_end,
+    ):
+        t_origin = await asyncio.wait_for(run_session(session, agent), timeout=SESSION_TIMEOUT)
 
     assert len(playback_finished_events) == 2
     check_timestamp(playback_finished_events[0].playback_position, 2.0, speed_factor=speed)
     check_timestamp(playback_finished_events[1].playback_position, 3.0, speed_factor=speed)
+    assert agent_speech_end_states == ["thinking", "listening"]
 
     assert len(agent_state_events) == 6
     assert agent_state_events[0].old_state == "initializing"

tests/test_audio_recognition_interruption_signals.py

@@ -0,0 +1,59 @@
+from __future__ import annotations
+
+import asyncio
+from collections import deque
+from unittest.mock import MagicMock
+
+import pytest
+
+from livekit.agents import NOT_GIVEN
+from livekit.agents.inference.interruption import (
+    _AgentSpeechEndedSentinel,
+    _AgentSpeechStartedSentinel,
+    _OverlapSpeechEndedSentinel,
+    _OverlapSpeechStartedSentinel,
+)
+from livekit.agents.voice.audio_recognition import AudioRecognition
+
+pytestmark = pytest.mark.unit
+
+
+async def test_agent_segments_emit_complete_interruption_boundaries() -> None:
+    recognition = AudioRecognition.__new__(AudioRecognition)
+    recognition._agent_speaking = False
+    recognition._agent_speech_started_at = None
+    recognition._endpointing = MagicMock()
+    recognition._backchannel_boundary = None
+    recognition._backchannel_boundary_timer = None
+    recognition._backchannel_boundary_callback = None
+    recognition._interruption_enabled = True
+    recognition._interruption_ch = MagicMock()
+    recognition._interruption_ch.closed = False
+    recognition._ignore_user_transcript_until = NOT_GIVEN
+    recognition._transcript_buffer = deque()
+    recognition._tasks = set()
+    recognition._overlap_in_current_turn = False
+    recognition._turn_backchannel_over_agent = False
+    recognition._user_silence_ev = asyncio.Event()
+    recognition._user_silence_ev.set()
+    recognition._session = MagicMock()
+
+    recognition._on_start_of_agent_speech(started_at=1.0)
+    recognition._speaking = True
+    recognition._on_start_of_speech(started_at=2.0)
+    recognition._on_end_of_agent_speech(ignore_user_transcript_until=3.0)
+    recognition._on_start_of_agent_speech(started_at=4.0)
+
+    frames = [call.args[0] for call in recognition._interruption_ch.send_nowait.call_args_list]
+    assert [type(frame) for frame in frames] == [
+        _AgentSpeechStartedSentinel,
+        _OverlapSpeechStartedSentinel,
+        _OverlapSpeechEndedSentinel,
+        _AgentSpeechEndedSentinel,
+        _AgentSpeechStartedSentinel,
+        _OverlapSpeechStartedSentinel,
+    ]
+    assert frames[-1]._speech_duration == 0.0
+    assert frames[-1]._started_at == 4.0
+
+    await asyncio.gather(*recognition._tasks)

tests/test_interruption/test_overlapping_speech_event.py

@@ -1,7 +1,10 @@
+from unittest.mock import MagicMock
+
 import numpy as np
 import pytest
 
 from livekit.agents.inference import OverlappingSpeechEvent
+from livekit.agents.inference.interruption import InterruptionWebSocketStream
 
 pytestmark = pytest.mark.unit
 
@@ -12,3 +15,17 @@ def test_interruption_event_serialization() -> None:
     assert ev.model_dump()["speech_input"] is None
     assert ev.model_dump(mode="json")["speech_input"] is None
     assert ev.speech_input is not None
+
+
+async def test_agent_ended_overlap_is_not_counted_as_backchannel() -> None:
+    stream = InterruptionWebSocketStream.__new__(InterruptionWebSocketStream)
+    stream._model = MagicMock(model="test-model", provider="test-provider")
+
+    async def _events():
+        yield OverlappingSpeechEvent(is_interruption=False, agent_ended=True)
+
+    await stream._metrics_monitor_task(_events())
+
+    metrics = stream._model.emit.call_args.args[1]
+    assert metrics.num_interruptions == 0
+    assert metrics.num_backchannels == 0