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

@@ -2067,8 +2067,7 @@ def _interrupt_by_audio_activity(
                 self._session._update_agent_state("listening")
                 if self._audio_recognition:
                     self._audio_recognition._on_end_of_agent_speech(
-                        ignore_user_transcript_until=ignore_user_transcript_until or time.time(),
-                        paused=True,
+                        ignore_user_transcript_until=ignore_user_transcript_until or time.time()
                     )
                 if self.interruption_enabled:
                     self._restore_interruption_by_audio_activity()
@@ -2192,8 +2191,7 @@ def on_interruption(self, ev: inference.OverlappingSpeechEvent) -> None:
         # flush held transcripts again if possible
         if self._audio_recognition:
             self._audio_recognition._on_end_of_agent_speech(
-                ignore_user_transcript_until=ev.overlap_started_at or ev.detected_at,
-                paused=self._paused_speech is not None,
+                ignore_user_transcript_until=ev.overlap_started_at or ev.detected_at
             )
 
     def on_interim_transcript(self, ev: stt.SpeechEvent, *, speaking: bool | None) -> None:
@@ -3466,6 +3464,12 @@ async def _next_segment() -> _SpeechSegment | None:
 
         if not speech_handle.interrupted and len(tool_output.output) > 0:
             self._session._update_agent_state("thinking")
+            if self._audio_recognition:
+                self._audio_recognition._on_end_of_agent_speech(
+                    ignore_user_transcript_until=time.time()
+                )
+            if self.interruption_enabled:
+                self._restore_interruption_by_audio_activity()
         elif self._session.agent_state == "speaking":
             self._session._update_agent_state("listening")
             if self._audio_recognition:
@@ -4380,9 +4384,7 @@ def _on_false_interruption() -> None:
                     otel_context=self._paused_speech.handle._agent_turn_context,
                 )
                 if self._audio_recognition and self._paused_speech.agent_state == "speaking":
-                    self._audio_recognition._on_start_of_agent_speech(
-                        started_at=time.time(), resumed=True
-                    )
+                    self._audio_recognition._on_start_of_agent_speech(started_at=time.time())
                 if self.interruption_enabled:
                     self._disable_vad_interruption_soon()
                 audio_output.resume()
@@ -4453,13 +4455,6 @@ async def _cancel_speech_pause(
         if not self._paused_speech:
             return
 
-        # the pause withheld end-of-agent-speech for a resume; interrupting ends the turn
-        # instead. the audio stopped when it was paused, so no playout is left to wait for
-        if interrupt and self._audio_recognition:
-            self._audio_recognition._on_end_of_agent_speech(
-                ignore_user_transcript_until=time.time()
-            )
-
         if (
             interrupt
             and not self._paused_speech.handle.interrupted

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

@@ -296,6 +296,7 @@ def __init__(
         self._ignore_user_transcript_until: NotGivenOr[float] = NOT_GIVEN
         self._transcript_buffer: deque[SpeechEvent] = deque()
         self._interruption_enabled: bool = interruption_detection is not None and vad is not None
+        # Tracks active audio playout, independently of the generation lifecycle.
         self._agent_speaking: bool = False
         self._agent_speech_started_at: float | None = None
         # turn-scoped backchannel-over-agent verdict from adaptive interruption, consumed and reset at end of turn
@@ -462,7 +463,12 @@ def _cancel_backchannel_boundary(self) -> None:
 
     # endregion
 
-    def _on_start_of_agent_speech(self, started_at: float, *, resumed: bool = False) -> None:
+    def _on_start_of_agent_speech(self, started_at: float) -> None:
+        """Mark the start of active agent speech.
+
+        This lifecycle follows audible playout, not the generation. Resuming paused playout
+        starts a new active-speech interval.
+        """
         self._agent_speaking = True
         self._agent_speech_started_at = started_at
         self._endpointing.on_start_of_agent_speech(started_at=started_at)
@@ -476,31 +482,35 @@ def _on_start_of_agent_speech(self, started_at: float, *, resumed: bool = False)
                 start_cooldown, self._on_backchannel_boundary_done
             )
 
-        # a resume re-enters the same agent turn; restarting would discard the open overlap
-        if self._adaptive_interruption_active and not resumed:
+        if self._adaptive_interruption_active:
             self._interruption_ch.send_nowait(_AgentSpeechStartedSentinel())  # type: ignore[union-attr]
 
-    def _on_end_of_agent_speech(
-        self, *, ignore_user_transcript_until: float, paused: bool = False
-    ) -> None:
+        if self._speaking:
+            self._on_start_of_overlap_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:
+        """Mark the end of active agent speech.
+
+        This can occur while the generation remains active, such as when playout is paused.
+        """
         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
 
-        # a pause is provisional: keep the inference running until the overlap has a verdict
-        if not paused:
-            self._interruption_ch.send_nowait(_AgentSpeechEndedSentinel())  # type: ignore[union-attr]
-
         if self._agent_speaking:
-            # no interruption is detected, end the inference (idempotent)
-            if not paused and not is_given(self._ignore_user_transcript_until):
-                self._on_end_of_overlap_speech(ended_at=time.time(), agent_ended=True)
+            # close any unresolved overlap before resetting the detector
+            self._on_end_of_overlap_speech(ended_at=time.time(), agent_ended=True)
+
+        self._interruption_ch.send_nowait(_AgentSpeechEndedSentinel())  # type: ignore[union-attr]
 
+        if self._agent_speaking:
             end_cooldown: float = (
                 self._backchannel_boundary[1] if self._backchannel_boundary else 0.0
             )
@@ -524,9 +534,6 @@ def _on_end_of_agent_speech(
             task.add_done_callback(lambda _: self._tasks.discard(task))
             self._tasks.add(task)
 
-        if not paused:
-            # the sentinel sent above resets the detector stream, dropping any open overlap
-            self._overlap_open = False
         self._agent_speaking = False
 
     def _on_start_of_speech(
@@ -543,6 +550,18 @@ def _on_start_of_speech(
         if not self._agent_speaking:
             self._overlap_in_current_turn = False
 
+        self._on_start_of_overlap_speech(
+            started_at=started_at,
+            speech_duration=speech_duration,
+            user_speaking_span=user_speaking_span,
+        )
+
+    def _on_start_of_overlap_speech(
+        self,
+        started_at: float,
+        speech_duration: float = 0.0,
+        user_speaking_span: trace.Span | None = None,
+    ) -> None:
         if not self._adaptive_interruption_active or not self._agent_speaking:
             return
         # overlap over agent speech started this turn; gates verdict acceptance below

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

@@ -696,6 +696,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

@@ -331,11 +331,32 @@ 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[0] == "thinking"
+    assert all(state == "listening" for state in agent_speech_end_states[1:])
 
     assert len(agent_state_events) == 6
     assert agent_state_events[0].old_state == "initializing"

tests/test_false_interruption_resume.py

@@ -298,8 +298,9 @@ async def test_server_side_turn_detection_keeps_the_resume_armed(
     assert activity._paused_speech is None
 
 
-async def test_interrupting_the_pause_ends_the_agent_turn(monkeypatch: pytest.MonkeyPatch) -> None:
-    # the pipeline paths that report it are gated on a speaking state the pause already left
+async def test_interrupting_paused_speech_does_not_end_agent_twice(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
     monkeypatch.setenv("LIVEKIT_API_KEY", "k")
     monkeypatch.setenv("LIVEKIT_API_SECRET", "s")
 
@@ -312,7 +313,7 @@ async def test_interrupting_the_pause_ends_the_agent_turn(monkeypatch: pytest.Mo
     await session.aclose()
 
     handle.interrupt.assert_called_once()
-    activity._audio_recognition._on_end_of_agent_speech.assert_called_once()
+    activity._audio_recognition._on_end_of_agent_speech.assert_not_called()
 
 
 async def test_handing_over_a_paused_speech_does_not_end_the_agent_turn(

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

tests/test_realtime_adaptive_interruption.py

@@ -317,51 +317,63 @@ def _recognition_with_interruption_ch() -> tuple[AudioRecognition, _RecordingCha
     ar._user_silence_ev = asyncio.Event()
     ar._user_silence_ev.set()
     ar._hooks = MagicMock()
+    ar._session = MagicMock()
     return ar, ch
 
 
 def _sentinel_names(ch: _RecordingChan) -> list[str]:
     return [type(item).__name__ for item in ch.sent]
 
 
-async def test_pause_keeps_overlap_inference_alive() -> None:
-    # pausing is provisional: the verdict for this overlap is what decides whether the
-    # pause becomes an interruption, so the inference must survive it
+async def test_agent_speech_end_closes_overlap_before_reset() -> None:
     ar, ch = _recognition_with_interruption_ch()
     ar._on_start_of_agent_speech(started_at=time.time())
     ar._on_start_of_speech(started_at=time.time())
     ch.sent.clear()
 
-    ar._on_end_of_agent_speech(ignore_user_transcript_until=time.time(), paused=True)
+    ar._on_end_of_agent_speech(ignore_user_transcript_until=time.time())
 
-    assert _sentinel_names(ch) == []
+    assert _sentinel_names(ch) == [
+        "_OverlapSpeechEndedSentinel",
+        "_AgentSpeechEndedSentinel",
+    ]
+    assert ch.sent[0]._agent_ended is True  # type: ignore[attr-defined]
+    assert ar._overlap_open is False
 
 
-async def test_pause_still_lets_the_user_close_the_overlap() -> None:
-    # the user finishing their utterance during the pause is what produces the verdict
+async def test_user_speech_ending_after_agent_end_does_not_close_overlap_again() -> None:
     ar, ch = _recognition_with_interruption_ch()
     ar._on_start_of_agent_speech(started_at=time.time())
     ar._on_start_of_speech(started_at=time.time())
-    ar._on_end_of_agent_speech(ignore_user_transcript_until=time.time(), paused=True)
+    ar._on_end_of_agent_speech(ignore_user_transcript_until=time.time())
     ch.sent.clear()
 
     ar._on_end_of_speech(ended_at=time.time())
 
-    assert _sentinel_names(ch) == ["_OverlapSpeechEndedSentinel"]
-    assert ch.sent[0]._agent_ended is False  # type: ignore[attr-defined]
+    assert _sentinel_names(ch) == []
 
 
-async def test_resume_does_not_restart_the_detector() -> None:
-    # a resume re-enters the same agent turn; restarting would reset the open overlap
+async def test_resume_restarts_the_detector() -> None:
     ar, ch = _recognition_with_interruption_ch()
     ar._on_start_of_agent_speech(started_at=time.time())
-    ar._on_start_of_speech(started_at=time.time())
-    ar._on_end_of_agent_speech(ignore_user_transcript_until=time.time(), paused=True)
+    user_started_at = time.time()
+    ar._speaking = True
+    ar._on_start_of_speech(started_at=user_started_at)
+    ar._on_end_of_agent_speech(ignore_user_transcript_until=time.time())
     ch.sent.clear()
 
-    ar._on_start_of_agent_speech(started_at=time.time(), resumed=True)
-
-    assert _sentinel_names(ch) == []
+    resumed_at = time.time()
+    ar._on_start_of_agent_speech(started_at=resumed_at)
+
+    assert _sentinel_names(ch) == [
+        "_AgentSpeechStartedSentinel",
+        "_OverlapSpeechStartedSentinel",
+    ]
+    assert ch.sent[1]._speech_duration == 0.0  # type: ignore[attr-defined]
+    assert ch.sent[1]._started_at == resumed_at  # type: ignore[attr-defined]
+    ar._endpointing.on_start_of_speech.assert_called_once_with(
+        started_at=user_started_at, overlapping=True
+    )
 
 
 async def test_a_resolved_overlap_is_not_closed_again() -> None:
@@ -377,19 +389,6 @@ async def test_a_resolved_overlap_is_not_closed_again() -> None:
     assert _sentinel_names(ch) == []
 
 
-async def test_interrupting_a_paused_speech_tears_down() -> None:
-    # the pause withheld the teardown for a possible resume; an interrupt ends the turn instead
-    ar, ch = _recognition_with_interruption_ch()
-    ar._on_start_of_agent_speech(started_at=time.time())
-    ar._on_start_of_speech(started_at=time.time())
-    ar._on_end_of_agent_speech(ignore_user_transcript_until=time.time(), paused=True)
-    ch.sent.clear()
-
-    ar._on_end_of_agent_speech(ignore_user_transcript_until=time.time())
-
-    assert _sentinel_names(ch) == ["_AgentSpeechEndedSentinel"]
-
-
 async def test_real_end_of_agent_speech_still_tears_down() -> None:
     # the agent turn genuinely ending must stop the inference
     ar, ch = _recognition_with_interruption_ch()
@@ -399,4 +398,8 @@ async def test_real_end_of_agent_speech_still_tears_down() -> None:
 
     ar._on_end_of_agent_speech(ignore_user_transcript_until=time.time())
 
-    assert "_AgentSpeechEndedSentinel" in _sentinel_names(ch)
+    assert _sentinel_names(ch) == [
+        "_OverlapSpeechEndedSentinel",
+        "_AgentSpeechEndedSentinel",
+    ]
+    assert ch.sent[0]._agent_ended is True  # type: ignore[attr-defined]