fix(audio): don't report a stopped playback as a decode error

livekit-agents/livekit/agents/utils/audio.py

@@ -233,6 +233,7 @@ async def audio_frames_from_file(
     decoder = AudioStreamDecoder(sample_rate=sample_rate, num_channels=num_channels)
 
     async def file_reader() -> None:
+        aborted = False
         try:
             async with aiofiles.open(file_path, mode="rb") as f:
                 while True:
@@ -241,8 +242,14 @@ async def file_reader() -> None:
                         break
 
                     decoder.push(chunk)
+        except asyncio.CancelledError:
+            aborted = True
+            raise
         finally:
-            decoder.end_input()
+            # a cancelled read leaves a truncated file, not an end of input: signalling EOF
+            # would make the decoder report the abort as invalid audio. aclose() closes it.
+            if not aborted:
+                decoder.end_input()
 
     reader_task = asyncio.create_task(file_reader())
 

livekit-agents/livekit/agents/utils/codecs/decoder.py

@@ -462,7 +462,9 @@ def _decode_loop(self) -> None:
                     self._emit_av_frame(f)
 
         except Exception:
-            logger.exception("error decoding audio")
+            # a close tears the input down mid-decode, which PyAV reports as invalid data
+            if not self._closed:
+                logger.exception("error decoding audio")
         finally:
             self._loop.call_soon_threadsafe(self._output_ch.close)
             if container:
@@ -489,15 +491,22 @@ async def aclose(self) -> None:
         if self._closed:
             return
 
-        self.end_input()
-        self._closed = True
+        if self._is_wav:
+            # decoded inline, there is no worker thread to wind down
+            self.end_input()
+            self._closed = True
+            return
 
-        if self._input_buf is not None:
-            self._input_buf.close()
+        # set before tearing the input down, so the decode thread can tell this close
+        # from a real decode failure
+        self._closed = True
 
-        if not self._started:
+        if self._input_buf is None:
+            self._output_ch.close()  # nothing was ever pushed, no frame will come
             return
 
+        self._input_buf.close()
+
         try:
             async for _ in self._output_ch:
                 pass

tests/test_audio_decoder.py

@@ -1,4 +1,7 @@
+import asyncio
+import contextlib
 import io
+import logging
 import os
 import struct
 import threading
@@ -10,6 +13,7 @@
 
 from livekit.agents import inference
 from livekit.agents.stt import SpeechEventType
+from livekit.agents.utils.audio import audio_frames_from_file
 from livekit.agents.utils.codecs import AudioStreamDecoder, StreamBuffer
 from livekit.agents.utils.misc import is_cloud
 
@@ -483,3 +487,60 @@ async def test_wav_multi_chunk_with_resampling():
     expected = samples_per_chunk * num_chunks * out_rate // src_rate
     assert abs(total_samples - expected) <= out_rate // 50  # within 20ms tolerance
     await decoder.aclose()
+
+
+@pytest.mark.asyncio
+async def test_aclose_while_probing_is_not_an_error(caplog: pytest.LogCaptureFixture) -> None:
+    """Closing a decoder mid-probe is intentional and must not be reported as a decode failure."""
+    with open(TEST_AUDIO_FILEPATH, "rb") as f:
+        head = f.read(16)  # far less than av.open needs to probe the container
+
+    decoder = AudioStreamDecoder()
+    decoder.push(head)
+    await asyncio.sleep(0.05)  # the decode thread is now blocked reading inside av.open
+
+    with caplog.at_level(logging.DEBUG, logger="livekit.agents"):
+        await decoder.aclose()
+        await asyncio.sleep(0.05)
+
+    assert [r.getMessage() for r in caplog.records if "error decoding" in r.getMessage()] == []
+
+
+@pytest.mark.asyncio
+async def test_aborted_read_does_not_signal_end_of_input(monkeypatch: pytest.MonkeyPatch) -> None:
+    """The BackgroundAudioPlayer stop() path: an aborted read is a truncated file, not an EOF.
+
+    Signalling the end of input hands the partial file to PyAV as a complete container, which
+    it then reports as invalid data.
+    """
+    real_push, real_end_input = AudioStreamDecoder.push, AudioStreamDecoder.end_input
+    pushes: list[None] = []
+    ends: list[None] = []
+
+    def slow_push(self: AudioStreamDecoder, chunk: bytes) -> None:
+        pushes.append(None)
+        time.sleep(0.005)  # hold the reader mid-file for the whole test
+        real_push(self, chunk)
+
+    def spy_end_input(self: AudioStreamDecoder) -> None:
+        ends.append(None)
+        real_end_input(self)
+
+    monkeypatch.setattr(AudioStreamDecoder, "push", slow_push)
+    monkeypatch.setattr(AudioStreamDecoder, "end_input", spy_end_input)
+
+    gen = audio_frames_from_file(TEST_AUDIO_FILEPATH)
+
+    async def drain() -> None:
+        async for _ in gen:
+            pass
+
+    task = asyncio.create_task(drain())
+    await asyncio.sleep(0.005)  # one chunk in, many chunks from the end
+    task.cancel()
+    with contextlib.suppress(asyncio.CancelledError):
+        await task
+    await gen.aclose()  # what PlayHandle.stop() ultimately triggers
+
+    assert pushes, "the reader never ran, so the abort was never exercised"
+    assert ends == []