fix: attribute playback evidence to its own segment on the same audio output

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

@@ -2769,10 +2769,18 @@ def _on_first_frame(fut: asyncio.Future[float] | asyncio.Future[None]) -> None:
         if speech_handle.interrupted and audio_output is not None:
             playback_ev = await audio_output.wait_for_playout()
 
-            if (
+            # A reported playback position is proof of partial playback even when
+            # the playback-started notification hasn't arrived yet (remote avatar
+            # outputs deliver it via RPC, which can race with the interruption).
+            # It only counts as evidence when THIS segment bumped the output's
+            # segment count (see _AudioOutput.captured_segments_before).
+            played_own_frame = (
                 audio_out is not None
-                and audio_out.first_frame_fut.done()
-                and not audio_out.first_frame_fut.cancelled()
+                and audio_output.captured_playout_segments > audio_out.captured_segments_before
+            )
+            if audio_out is not None and (
+                (audio_out.first_frame_fut.done() and not audio_out.first_frame_fut.cancelled())
+                or (played_own_frame and playback_ev.playback_position > 0)
             ):
                 if playback_ev.synchronized_transcript is not None:
                     forwarded_text = playback_ev.synchronized_transcript

livekit-agents/livekit/agents/voice/generation.py

@@ -450,23 +450,49 @@ class _AudioOutput:
     first_frame_fut: asyncio.Future[float]
     """Future that will be set with the timestamp of the first frame's capture"""
 
+    captured_segments_before: int
+    """`AudioOutput.captured_playout_segments` snapshot taken when forwarding was set up.
+
+    The interrupted-commit gate compares the current count against this baseline: a
+    bump proves a frame of THIS segment made it through `AudioOutput.capture_frame`
+    — the same condition under which `wait_for_playout()` waits for this segment's
+    playback event instead of returning a stale one — so the reported playback
+    position can be trusted as evidence of partial playback.
+    """
+
     started_forwarding_at: float | None = None
 
-    def _resolve_first_frame_fut(self, ev: io.PlaybackStartedEvent) -> None:
-        if not self.first_frame_fut.done():
-            self.first_frame_fut.set_result(ev.created_at)
+    has_captured_own_frame: bool = False
+    """Whether this segment has pushed at least one of its own frames to `capture_frame`.
+
+    Set *before* the first `capture_frame` await: sinks emit `playback_started`
+    synchronously inside the first counted capture, so a flag set after the await
+    would miss the segment's own event.
+    """
 
 
 def perform_audio_forwarding(
     *,
     audio_output: io.AudioOutput,
     tts_output: AsyncIterable[rtc.AudioFrame],
 ) -> tuple[asyncio.Task[None], _AudioOutput]:
-    out = _AudioOutput(audio=[], first_frame_fut=asyncio.Future())
+    out = _AudioOutput(
+        audio=[],
+        first_frame_fut=asyncio.Future(),
+        captured_segments_before=audio_output.captured_playout_segments,
+    )
+
+    def _on_playback_started(ev: io.PlaybackStartedEvent) -> None:
+        # The audio output is shared across overlapping segments. Only honor the
+        # event once this segment has captured its own first frame, so a stray event
+        # from another segment can't resolve our first_frame_fut prematurely.
+        if out.has_captured_own_frame and not out.first_frame_fut.done():
+            out.first_frame_fut.set_result(ev.created_at)
+
     # out.first_frame_fut should be cancelled in the caller after the playout is finished or interrupted
-    audio_output.on("playback_started", out._resolve_first_frame_fut)
+    audio_output.on("playback_started", _on_playback_started)
     out.first_frame_fut.add_done_callback(
-        lambda _: audio_output.off("playback_started", out._resolve_first_frame_fut)
+        lambda _: audio_output.off("playback_started", _on_playback_started)
     )
     task = asyncio.create_task(_audio_forwarding_task(audio_output, tts_output, out))
     return task, out
@@ -501,6 +527,8 @@ async def _audio_forwarding_task(
                     num_channels=frame.num_channels,
                 )
 
+            out.has_captured_own_frame = True
+
             if resampler:
                 for f in resampler.push(frame):
                     await audio_output.capture_frame(f)
@@ -596,10 +624,20 @@ async def forward_generation(
             if audio_output is not None:
                 audio_output.clear_buffer()
                 playback_ev = await audio_output.wait_for_playout()
-                if (
+                # A reported playback position is proof of partial playback even when
+                # the playback-started notification hasn't arrived yet (remote avatar
+                # outputs deliver it via RPC, which can race with the interruption).
+                # It only counts as evidence when THIS segment bumped the output's
+                # segment count — the same condition under which wait_for_playout waits
+                # for this segment's playback event instead of returning a stale one
+                # from a previous segment (see _AudioOutput.captured_segments_before).
+                played_own_frame = (
                     audio_out is not None
-                    and audio_out.first_frame_fut.done()
-                    and not audio_out.first_frame_fut.cancelled()
+                    and audio_output.captured_playout_segments > audio_out.captured_segments_before
+                )
+                if audio_out is not None and (
+                    (audio_out.first_frame_fut.done() and not audio_out.first_frame_fut.cancelled())
+                    or (played_own_frame and playback_ev.playback_position > 0)
                 ):
                     out.played = "partial"
                     out.playback_position = playback_ev.playback_position

livekit-agents/livekit/agents/voice/io.py

@@ -243,6 +243,17 @@ def _pending_playback_count(self) -> int:
         """Number of captured segments that haven't reported playback_finished yet."""
         return self.__playback_segments_count - self.__playback_finished_count
 
+    @property
+    def captured_playout_segments(self) -> int:
+        """Count of playback segments captured so far.
+
+        Increments when the first frame of a new segment is accepted by
+        ``capture_frame`` (segments are delimited by ``flush``/``clear_buffer``).
+        Lets callers detect — free of races with concurrent finishes — whether a
+        frame they forwarded was actually accepted into a counted playout segment.
+        """
+        return self.__playback_segments_count
+
     @property
     def sample_rate(self) -> int | None:
         """The sample rate required by the audio sink, if None, any sample rate is accepted"""

tests/test_playback_started_attribution.py

@@ -0,0 +1,210 @@
+"""Tests for playback_started attribution and interrupted position evidence.
+
+Mirrors agents-js#1966: the shared AudioOutput emits playback_started with no
+segment identity, so the per-segment listener only resolves after this segment
+has captured its own frame; interrupted commits also accept playback_position > 0
+when this segment bumped the output's segment counter.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import time
+from collections.abc import AsyncIterable
+
+import pytest
+
+from livekit import rtc
+from livekit.agents.voice.generation import forward_generation, perform_audio_forwarding
+from livekit.agents.voice.io import AudioOutput
+from livekit.agents.voice.speech_handle import SpeechHandle
+
+from .fake_io import FakeAudioOutput
+
+pytestmark = pytest.mark.unit
+
+SAMPLE_RATE = 16000
+
+
+def _make_frame(duration: float = 0.1) -> rtc.AudioFrame:
+    num_samples = int(SAMPLE_RATE * duration + 0.5)
+    return rtc.AudioFrame(
+        data=b"\x00\x00" * num_samples,
+        sample_rate=SAMPLE_RATE,
+        num_channels=1,
+        samples_per_channel=num_samples,
+    )
+
+
+class _ForwardFirstWrapper(AudioOutput):
+    """Mimics _SyncedAudioOutput / recorder ordering.
+
+    Forwards to the leaf (which emits playback_started synchronously inside the
+    first capture) BEFORE counting its own segment, so the forwarded event fires
+    while this output's counter still reads the pre-capture snapshot.
+    """
+
+    def __init__(self, next_in_chain: AudioOutput) -> None:
+        super().__init__(
+            label="ForwardFirstWrapper",
+            next_in_chain=next_in_chain,
+            capabilities=next_in_chain._capabilities,
+        )
+
+    async def capture_frame(self, frame: rtc.AudioFrame) -> None:
+        assert self.next_in_chain is not None
+        await self.next_in_chain.capture_frame(frame)
+        await super().capture_frame(frame)
+
+    def flush(self) -> None:
+        super().flush()
+        assert self.next_in_chain is not None
+        self.next_in_chain.flush()
+
+    def clear_buffer(self) -> None:
+        assert self.next_in_chain is not None
+        self.next_in_chain.clear_buffer()
+
+
+class _NoStartNotifyOutput(FakeAudioOutput):
+    """A sink whose playback-started notification is delayed/lost.
+
+    Simulates a remote (avatar) output that delivers playback_started via RPC:
+    frames are accepted and counted, playback genuinely progresses (so the
+    playback-finished event reports a real position), but no playback_started
+    is emitted locally.
+    """
+
+    async def capture_frame(self, frame: rtc.AudioFrame) -> None:
+        await AudioOutput.capture_frame(self, frame)  # count the playout segment
+        self._pushed_duration += frame.duration
+        if self._started_at is None:
+            self._started_at = time.time()  # playing, but never notifies
+
+
+async def _drive_forwarding(
+    audio_output: AudioOutput, frames_ch: asyncio.Queue[rtc.AudioFrame | None]
+):
+    async def _source() -> AsyncIterable[rtc.AudioFrame]:
+        while True:
+            frame = await frames_ch.get()
+            if frame is None:
+                return
+            yield frame
+
+    return perform_audio_forwarding(audio_output=audio_output, tts_output=_source())
+
+
+async def test_own_playback_started_resolves_first_frame_fut() -> None:
+    audio_output = FakeAudioOutput()
+    frames_ch: asyncio.Queue[rtc.AudioFrame | None] = asyncio.Queue()
+    task, out = await _drive_forwarding(audio_output, frames_ch)
+
+    frames_ch.put_nowait(_make_frame())
+    frames_ch.put_nowait(None)
+    await task
+
+    assert out.first_frame_fut.done()
+    assert audio_output.captured_playout_segments > out.captured_segments_before
+    audio_output.clear_buffer()
+
+
+async def test_own_event_forwarded_before_wrapper_counts_still_resolves() -> None:
+    # Chained outputs (transcript sync, recorder) forward the leaf's synchronous
+    # playback_started before counting their own segment: the event must still be
+    # attributed once has_captured_own_frame is set.
+    audio_output = _ForwardFirstWrapper(FakeAudioOutput())
+    frames_ch: asyncio.Queue[rtc.AudioFrame | None] = asyncio.Queue()
+    task, out = await _drive_forwarding(audio_output, frames_ch)
+
+    frames_ch.put_nowait(_make_frame())
+    frames_ch.put_nowait(None)
+    await task
+
+    assert out.first_frame_fut.done()
+    audio_output.clear_buffer()
+
+
+async def test_stale_event_before_own_capture_is_ignored() -> None:
+    audio_output = FakeAudioOutput()
+    frames_ch: asyncio.Queue[rtc.AudioFrame | None] = asyncio.Queue()
+    task, out = await _drive_forwarding(audio_output, frames_ch)
+    await asyncio.sleep(0)  # let the forwarding task attach the listener
+
+    # a stale event (e.g. an avatar RPC from a previous, interrupted segment)
+    # arrives before this segment captured anything
+    audio_output.on_playback_started(created_at=time.time())
+
+    assert not out.first_frame_fut.done()
+
+    frames_ch.put_nowait(None)
+    await task
+    out.first_frame_fut.cancel()
+
+
+async def test_interrupted_commit_uses_position_evidence_without_started_event() -> None:
+    # Avatar-style race: frames genuinely played, but the started notification
+    # never arrived before the interruption. The playback position reported by
+    # the finished event is proof of partial playback when this segment bumped
+    # the output's segment count.
+    audio_output = _NoStartNotifyOutput()
+    speech_handle = SpeechHandle.create()
+
+    frame_captured = asyncio.Event()
+
+    async def _audio_source() -> AsyncIterable[rtc.AudioFrame]:
+        yield _make_frame()
+        frame_captured.set()
+        await asyncio.Event().wait()  # keep the TTS stream open until interrupted
+
+    forward_task = asyncio.create_task(
+        forward_generation(
+            speech_handle=speech_handle,
+            audio_output=audio_output,
+            text_output=None,
+            audio_source=_audio_source(),
+            text_source=None,
+            on_first_frame=lambda _fut, _out: None,
+        )
+    )
+
+    await asyncio.wait_for(frame_captured.wait(), timeout=5)
+    await asyncio.sleep(0.05)  # accrue some playback position
+    speech_handle.interrupt()
+    out = await asyncio.wait_for(forward_task, timeout=5)
+
+    assert out.audio_out is not None
+    assert not out.audio_out.first_frame_fut.done() or out.audio_out.first_frame_fut.cancelled()
+    assert audio_output.captured_playout_segments > out.audio_out.captured_segments_before
+    assert out.played == "partial"
+    assert out.playback_position > 0
+
+
+async def test_interrupted_commit_stays_skipped_without_any_capture() -> None:
+    # Interrupted before any frame was counted: no started event, no segment bump
+    # — the segment must stay "skipped" (nothing reached the user).
+    audio_output = FakeAudioOutput()
+    speech_handle = SpeechHandle.create()
+
+    async def _audio_source() -> AsyncIterable[rtc.AudioFrame]:
+        await asyncio.Event().wait()  # never yields
+        yield _make_frame()  # pragma: no cover
+
+    forward_task = asyncio.create_task(
+        forward_generation(
+            speech_handle=speech_handle,
+            audio_output=audio_output,
+            text_output=None,
+            audio_source=_audio_source(),
+            text_source=None,
+            on_first_frame=lambda _fut, _out: None,
+        )
+    )
+
+    await asyncio.sleep(0.05)
+    speech_handle.interrupt()
+    out = await asyncio.wait_for(forward_task, timeout=5)
+
+    assert out.audio_out is not None
+    assert audio_output.captured_playout_segments == out.audio_out.captured_segments_before
+    assert out.played == "skipped"