From db4aa564f48acf2e75905d62ba6fea7f338d3237 Mon Sep 17 00:00:00 2001
From: Nabi Bukhsh <nabibuksh.baloch01@gmail.com>
Date: Sat, 17 Oct 2026 00:17:25 +0500
Subject: [PATCH 1/5] fix(rtc): offload FFI handle disposal to thread in aclose
 and add sync close (#832)

- In AudioSource, AudioStream, VideoSource, and VideoStream, add a synchronous close() method.
- In AudioSource, VideoSource, AudioStream, and VideoStream, run self._ffi_handle.dispose() via asyncio.to_thread in aclose() so the synchronous native drop call does not block the asyncio event loop.
- In AudioSource, cancel any pending _join_handle timer and resolve _join_fut when close() or aclose() is called so waiters on wait_for_playout are immediately unblocked.
- In VideoSource.capture_frame, safely ignore frames if the handle is already disposed.

Fixes #832
---
 livekit-rtc/livekit/rtc/audio_source.py       | 17 +++-
 livekit-rtc/livekit/rtc/audio_stream.py       | 13 ++-
 livekit-rtc/livekit/rtc/video_source.py       | 18 +++-
 livekit-rtc/livekit/rtc/video_stream.py       |  8 +-
 livekit-rtc/tests/test_source_stream_close.py | 95 +++++++++++++++++++
 5 files changed, 145 insertions(+), 6 deletions(-)
 create mode 100644 livekit-rtc/tests/test_source_stream_close.py

diff --git a/livekit-rtc/livekit/rtc/audio_source.py b/livekit-rtc/livekit/rtc/audio_source.py
index 63cc1a5d..a63a2df0 100644
--- a/livekit-rtc/livekit/rtc/audio_source.py
+++ b/livekit-rtc/livekit/rtc/audio_source.py
@@ -163,6 +163,10 @@ async def wait_for_playout(self) -> None:
         await asyncio.shield(self._join_fut)
 
     def _release_waiter(self) -> None:
+        if self._join_handle is not None:
+            self._join_handle.cancel()
+            self._join_handle = None
+
         if self._join_fut is None:
             return  # could be None when clear_queue is called
 
@@ -173,9 +177,18 @@ def _release_waiter(self) -> None:
         self._q_size = 0.0
         self._join_fut = None
 
-    async def aclose(self) -> None:
-        """Close the audio source
+    def close(self) -> None:
+        """Close the audio source.
 
         This method cleans up resources associated with the audio source.
         """
+        self._release_waiter()
         self._ffi_handle.dispose()
+
+    async def aclose(self) -> None:
+        """Close the audio source without blocking the event loop.
+
+        This method cleans up resources associated with the audio source.
+        """
+        self._release_waiter()
+        await asyncio.to_thread(self._ffi_handle.dispose)
diff --git a/livekit-rtc/livekit/rtc/audio_stream.py b/livekit-rtc/livekit/rtc/audio_stream.py
index 97772f4f..b4411cad 100644
--- a/livekit-rtc/livekit/rtc/audio_stream.py
+++ b/livekit-rtc/livekit/rtc/audio_stream.py
@@ -315,15 +315,24 @@ async def _run(self) -> None:
 
         FfiClient.instance.queue.unsubscribe(self._ffi_queue)
 
+    def close(self) -> None:
+        """Synchronously close the audio stream and release its resources."""
+        if self._track is not None:
+            self._track._unregister_audio_stream(self)
+        self._ffi_handle.dispose()
+        FfiClient.instance.queue.unsubscribe(self._ffi_queue)
+        if self._processor is not None and self._processor_auto_close:
+            self._processor._close()
+
     async def aclose(self) -> None:
         """Asynchronously close the audio stream.
 
         This method cleans up resources associated with the audio stream and waits for
-        any pending operations to complete.
+        any pending operations to complete without blocking the event loop.
         """
         if self._track is not None:
             self._track._unregister_audio_stream(self)
-        self._ffi_handle.dispose()
+        await asyncio.to_thread(self._ffi_handle.dispose)
         await self._task
         if self._processor is not None and self._processor_auto_close:
             self._processor._close()
diff --git a/livekit-rtc/livekit/rtc/video_source.py b/livekit-rtc/livekit/rtc/video_source.py
index c8dd2a02..ad34fb2d 100644
--- a/livekit-rtc/livekit/rtc/video_source.py
+++ b/livekit-rtc/livekit/rtc/video_source.py
@@ -14,6 +14,8 @@
 
 from __future__ import annotations
 
+import asyncio
+
 from ._ffi_client import FfiHandle, FfiClient
 from ._proto import ffi_pb2 as proto_ffi
 from ._proto import video_frame_pb2 as proto_video
@@ -61,6 +63,9 @@ def capture_frame(
         metadata: proto_video.FrameMetadata | None = None,
     ) -> None:
         """Capture a frame, optionally attaching frame metadata."""
+        if self._ffi_handle.disposed:
+            return
+
         req = proto_ffi.FfiRequest()
         req.capture_video_frame.source_handle = self._ffi_handle.handle
         req.capture_video_frame.buffer.CopyFrom(frame._proto_info())
@@ -70,5 +75,16 @@ def capture_frame(
             req.capture_video_frame.metadata.CopyFrom(metadata)
         FfiClient.instance.request(req)
 
-    async def aclose(self) -> None:
+    def close(self) -> None:
+        """Close the video source.
+
+        This method cleans up resources associated with the video source.
+        """
         self._ffi_handle.dispose()
+
+    async def aclose(self) -> None:
+        """Close the video source without blocking the event loop.
+
+        This method cleans up resources associated with the video source.
+        """
+        await asyncio.to_thread(self._ffi_handle.dispose)
diff --git a/livekit-rtc/livekit/rtc/video_stream.py b/livekit-rtc/livekit/rtc/video_stream.py
index eeced550..8c686538 100644
--- a/livekit-rtc/livekit/rtc/video_stream.py
+++ b/livekit-rtc/livekit/rtc/video_stream.py
@@ -164,8 +164,14 @@ async def _run(self) -> None:
 
         FfiClient.instance.queue.unsubscribe(self._ffi_queue)
 
-    async def aclose(self) -> None:
+    def close(self) -> None:
+        """Synchronously close the video stream and release its resources."""
         self._ffi_handle.dispose()
+        FfiClient.instance.queue.unsubscribe(self._ffi_queue)
+
+    async def aclose(self) -> None:
+        """Asynchronously close the video stream without blocking the event loop."""
+        await asyncio.to_thread(self._ffi_handle.dispose)
         await self._task
 
     def _is_event(self, e: proto_ffi.FfiEvent) -> bool:
diff --git a/livekit-rtc/tests/test_source_stream_close.py b/livekit-rtc/tests/test_source_stream_close.py
new file mode 100644
index 00000000..cd84ac1d
--- /dev/null
+++ b/livekit-rtc/tests/test_source_stream_close.py
@@ -0,0 +1,95 @@
+# Copyright 2026 LiveKit, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from __future__ import annotations
+
+
+import pytest
+
+from livekit import rtc
+
+
+@pytest.mark.asyncio
+async def test_audio_source_aclose_releases_waiter_and_cancels_timer() -> None:
+    source = rtc.AudioSource(48000, 1)
+    frame = rtc.AudioFrame.create(48000, 1, 48000)
+    await source.capture_frame(frame)
+
+    assert source._join_fut is not None
+    assert source._join_handle is not None
+
+    await source.aclose()
+
+    assert source._ffi_handle.disposed
+    assert source._join_fut is None
+    assert source._join_handle is None
+
+    # After aclose, wait_for_playout should return immediately without hanging
+    await source.wait_for_playout()
+
+    # Repeated calls are idempotent
+    await source.aclose()
+
+
+def test_audio_source_sync_close() -> None:
+    source = rtc.AudioSource(48000, 1)
+    assert not source._ffi_handle.disposed
+
+    source.close()
+    assert source._ffi_handle.disposed
+
+    # Repeated close is idempotent
+    source.close()
+
+
+@pytest.mark.asyncio
+async def test_audio_source_capture_after_close_is_noop() -> None:
+    source = rtc.AudioSource(48000, 1)
+    source.close()
+
+    frame = rtc.AudioFrame.create(48000, 1, 480)
+    # Should safely return without error
+    await source.capture_frame(frame)
+
+
+def test_video_source_sync_close() -> None:
+    source = rtc.VideoSource(640, 480)
+    assert not source._ffi_handle.disposed
+
+    source.close()
+    assert source._ffi_handle.disposed
+
+    # Repeated close is idempotent
+    source.close()
+
+
+@pytest.mark.asyncio
+async def test_video_source_aclose_and_capture_after_close() -> None:
+    source = rtc.VideoSource(640, 480)
+    assert not source._ffi_handle.disposed
+
+    await source.aclose()
+    assert source._ffi_handle.disposed
+
+    # Capture after close is safely ignored
+    frame = rtc.VideoFrame(
+        640,
+        480,
+        rtc.VideoBufferType.RGBA,
+        b"\x00" * (640 * 480 * 4),
+    )
+    source.capture_frame(frame)
+
+    # Repeated aclose is idempotent
+    await source.aclose()

From 392126f61060674dd5d73eabf5732592e3421877 Mon Sep 17 00:00:00 2001
From: Nabi Bukhsh <nabibuksh.baloch01@gmail.com>
Date: Mon, 19 Oct 2026 01:06:17 +0500
Subject: [PATCH 2/5] fix(rtc): handle cross-thread close, safe mock
 attributes, and isolate FFI lifecycle test

---
 livekit-rtc/livekit/rtc/audio_source.py       | 30 +++++++++--
 livekit-rtc/livekit/rtc/audio_stream.py       | 13 +----
 livekit-rtc/livekit/rtc/video_source.py       |  5 +-
 livekit-rtc/livekit/rtc/video_stream.py       |  8 +--
 .../tests/test_ffi_handle_lifecycle.py        | 52 +++++++++++++------
 livekit-rtc/tests/test_source_stream_close.py | 27 +++++++---
 6 files changed, 88 insertions(+), 47 deletions(-)

diff --git a/livekit-rtc/livekit/rtc/audio_source.py b/livekit-rtc/livekit/rtc/audio_source.py
index a63a2df0..d3777b71 100644
--- a/livekit-rtc/livekit/rtc/audio_source.py
+++ b/livekit-rtc/livekit/rtc/audio_source.py
@@ -53,7 +53,17 @@ def __init__(
         """
         self._sample_rate = sample_rate
         self._num_channels = num_channels
-        self._loop = loop or asyncio.get_event_loop()
+
+        if loop is not None:
+            self._loop = loop
+        else:
+            try:
+                self._loop = asyncio.get_running_loop()
+            except RuntimeError:
+                try:
+                    self._loop = asyncio.get_event_loop()
+                except RuntimeError:
+                    self._loop = asyncio.new_event_loop()
 
         req = proto_ffi.FfiRequest()
         req.new_audio_source.type = proto_audio_frame.AudioSourceType.AUDIO_SOURCE_NATIVE
@@ -116,7 +126,7 @@ async def capture_frame(self, frame: AudioFrame) -> None:
             Exception: If there is an error during frame capture.
         """
 
-        if frame.samples_per_channel == 0 or self._ffi_handle.disposed:
+        if frame.samples_per_channel == 0 or getattr(self._ffi_handle, "disposed", False):
             return
 
         now = time.monotonic()
@@ -182,7 +192,18 @@ def close(self) -> None:
 
         This method cleans up resources associated with the audio source.
         """
-        self._release_waiter()
+        try:
+            running_loop = asyncio.get_running_loop()
+        except RuntimeError:
+            running_loop = None
+
+        if running_loop is self._loop:
+            self._release_waiter()
+        elif self._loop.is_running():
+            self._loop.call_soon_threadsafe(self._release_waiter)
+        else:
+            self._release_waiter()
+
         self._ffi_handle.dispose()
 
     async def aclose(self) -> None:
@@ -191,4 +212,5 @@ async def aclose(self) -> None:
         This method cleans up resources associated with the audio source.
         """
         self._release_waiter()
-        await asyncio.to_thread(self._ffi_handle.dispose)
+        if not getattr(self._ffi_handle, "disposed", False):
+            await asyncio.to_thread(self._ffi_handle.dispose)
diff --git a/livekit-rtc/livekit/rtc/audio_stream.py b/livekit-rtc/livekit/rtc/audio_stream.py
index b4411cad..97772f4f 100644
--- a/livekit-rtc/livekit/rtc/audio_stream.py
+++ b/livekit-rtc/livekit/rtc/audio_stream.py
@@ -315,24 +315,15 @@ async def _run(self) -> None:
 
         FfiClient.instance.queue.unsubscribe(self._ffi_queue)
 
-    def close(self) -> None:
-        """Synchronously close the audio stream and release its resources."""
-        if self._track is not None:
-            self._track._unregister_audio_stream(self)
-        self._ffi_handle.dispose()
-        FfiClient.instance.queue.unsubscribe(self._ffi_queue)
-        if self._processor is not None and self._processor_auto_close:
-            self._processor._close()
-
     async def aclose(self) -> None:
         """Asynchronously close the audio stream.
 
         This method cleans up resources associated with the audio stream and waits for
-        any pending operations to complete without blocking the event loop.
+        any pending operations to complete.
         """
         if self._track is not None:
             self._track._unregister_audio_stream(self)
-        await asyncio.to_thread(self._ffi_handle.dispose)
+        self._ffi_handle.dispose()
         await self._task
         if self._processor is not None and self._processor_auto_close:
             self._processor._close()
diff --git a/livekit-rtc/livekit/rtc/video_source.py b/livekit-rtc/livekit/rtc/video_source.py
index ad34fb2d..6b2c3947 100644
--- a/livekit-rtc/livekit/rtc/video_source.py
+++ b/livekit-rtc/livekit/rtc/video_source.py
@@ -63,7 +63,7 @@ def capture_frame(
         metadata: proto_video.FrameMetadata | None = None,
     ) -> None:
         """Capture a frame, optionally attaching frame metadata."""
-        if self._ffi_handle.disposed:
+        if getattr(self._ffi_handle, "disposed", False):
             return
 
         req = proto_ffi.FfiRequest()
@@ -87,4 +87,5 @@ async def aclose(self) -> None:
 
         This method cleans up resources associated with the video source.
         """
-        await asyncio.to_thread(self._ffi_handle.dispose)
+        if not getattr(self._ffi_handle, "disposed", False):
+            await asyncio.to_thread(self._ffi_handle.dispose)
diff --git a/livekit-rtc/livekit/rtc/video_stream.py b/livekit-rtc/livekit/rtc/video_stream.py
index 8c686538..eeced550 100644
--- a/livekit-rtc/livekit/rtc/video_stream.py
+++ b/livekit-rtc/livekit/rtc/video_stream.py
@@ -164,14 +164,8 @@ async def _run(self) -> None:
 
         FfiClient.instance.queue.unsubscribe(self._ffi_queue)
 
-    def close(self) -> None:
-        """Synchronously close the video stream and release its resources."""
-        self._ffi_handle.dispose()
-        FfiClient.instance.queue.unsubscribe(self._ffi_queue)
-
     async def aclose(self) -> None:
-        """Asynchronously close the video stream without blocking the event loop."""
-        await asyncio.to_thread(self._ffi_handle.dispose)
+        self._ffi_handle.dispose()
         await self._task
 
     def _is_event(self, e: proto_ffi.FfiEvent) -> bool:
diff --git a/livekit-rtc/tests/test_ffi_handle_lifecycle.py b/livekit-rtc/tests/test_ffi_handle_lifecycle.py
index 08750bfa..442a7bae 100644
--- a/livekit-rtc/tests/test_ffi_handle_lifecycle.py
+++ b/livekit-rtc/tests/test_ffi_handle_lifecycle.py
@@ -14,7 +14,6 @@
 
 from __future__ import annotations
 
-import gc
 import subprocess
 import sys
 
@@ -43,25 +42,46 @@ def test_active_runtime_invalid_handle_asserts() -> None:
 
 
 def test_live_handles_marked_consumed_and_no_assert_after_dispose() -> None:
-    handle_id = _allocate_real_handle()
-    handle = FfiHandle(handle_id)
-    assert not handle.disposed
+    code = """
+import gc
+from livekit.rtc._ffi_client import FfiClient, FfiHandle
+from livekit.rtc._proto import audio_frame_pb2, ffi_pb2
+
+req = ffi_pb2.FfiRequest()
+req.new_audio_source.type = audio_frame_pb2.AUDIO_SOURCE_NATIVE
+req.new_audio_source.sample_rate = 48000
+req.new_audio_source.num_channels = 1
+req.new_audio_source.queue_size_ms = 100
+resp = FfiClient.instance.request(req)
+handle_id = resp.new_audio_source.source.handle.id
+assert handle_id != 0
+
+handle = FfiHandle(int(handle_id))
+assert not handle.disposed
 
-    # Global dispose marks all live handles consumed and sets _ffi_disposed
-    FfiClient.instance._dispose()
-    assert handle.disposed
+# Global dispose marks all live handles consumed and sets _ffi_disposed
+FfiClient.instance._dispose()
+assert handle.disposed
 
-    # Explicit dispose or __del__ after global dispose should be a no-op and not assert
-    handle.dispose()
-    handle.__del__()
+# Explicit dispose or __del__ after global dispose should be a no-op and not assert
+handle.dispose()
+handle.__del__()
 
-    # Newly instantiated handle after global dispose also safely no-ops without assert
-    late_handle = FfiHandle(12345)
-    late_handle.dispose()
+# Newly instantiated handle after global dispose also safely no-ops without assert
+late_handle = FfiHandle(12345)
+late_handle.dispose()
 
-    del handle
-    del late_handle
-    gc.collect()
+del handle
+del late_handle
+gc.collect()
+"""
+    result = subprocess.run(
+        [sys.executable, "-c", code],
+        capture_output=True,
+        text=True,
+    )
+    assert result.returncode == 0
+    assert "AssertionError" not in result.stderr
 
 
 def test_subprocess_exit_with_undisposed_handle_has_no_assertion_error() -> None:
diff --git a/livekit-rtc/tests/test_source_stream_close.py b/livekit-rtc/tests/test_source_stream_close.py
index cd84ac1d..3b23dab7 100644
--- a/livekit-rtc/tests/test_source_stream_close.py
+++ b/livekit-rtc/tests/test_source_stream_close.py
@@ -14,6 +14,8 @@
 
 from __future__ import annotations
 
+import asyncio
+import threading
 
 import pytest
 
@@ -24,22 +26,22 @@
 async def test_audio_source_aclose_releases_waiter_and_cancels_timer() -> None:
     source = rtc.AudioSource(48000, 1)
     frame = rtc.AudioFrame.create(48000, 1, 48000)
-    await source.capture_frame(frame)
+    await asyncio.wait_for(source.capture_frame(frame), timeout=5.0)
 
     assert source._join_fut is not None
     assert source._join_handle is not None
 
-    await source.aclose()
+    await asyncio.wait_for(source.aclose(), timeout=5.0)
 
     assert source._ffi_handle.disposed
     assert source._join_fut is None
     assert source._join_handle is None
 
     # After aclose, wait_for_playout should return immediately without hanging
-    await source.wait_for_playout()
+    await asyncio.wait_for(source.wait_for_playout(), timeout=2.0)
 
     # Repeated calls are idempotent
-    await source.aclose()
+    await asyncio.wait_for(source.aclose(), timeout=2.0)
 
 
 def test_audio_source_sync_close() -> None:
@@ -53,6 +55,17 @@ def test_audio_source_sync_close() -> None:
     source.close()
 
 
+def test_audio_source_cross_thread_close() -> None:
+    source = rtc.AudioSource(48000, 1)
+    assert not source._ffi_handle.disposed
+
+    t = threading.Thread(target=source.close)
+    t.start()
+    t.join()
+
+    assert source._ffi_handle.disposed
+
+
 @pytest.mark.asyncio
 async def test_audio_source_capture_after_close_is_noop() -> None:
     source = rtc.AudioSource(48000, 1)
@@ -60,7 +73,7 @@ async def test_audio_source_capture_after_close_is_noop() -> None:
 
     frame = rtc.AudioFrame.create(48000, 1, 480)
     # Should safely return without error
-    await source.capture_frame(frame)
+    await asyncio.wait_for(source.capture_frame(frame), timeout=2.0)
 
 
 def test_video_source_sync_close() -> None:
@@ -79,7 +92,7 @@ async def test_video_source_aclose_and_capture_after_close() -> None:
     source = rtc.VideoSource(640, 480)
     assert not source._ffi_handle.disposed
 
-    await source.aclose()
+    await asyncio.wait_for(source.aclose(), timeout=5.0)
     assert source._ffi_handle.disposed
 
     # Capture after close is safely ignored
@@ -92,4 +105,4 @@ async def test_video_source_aclose_and_capture_after_close() -> None:
     source.capture_frame(frame)
 
     # Repeated aclose is idempotent
-    await source.aclose()
+    await asyncio.wait_for(source.aclose(), timeout=2.0)

From a861f73e3d8c8caa8d3cb49e4a84d990d17f76b5 Mon Sep 17 00:00:00 2001
From: Nabi Bukhsh <nabibuksh.baloch01@gmail.com>
Date: Mon, 19 Oct 2026 01:58:11 +0500
Subject: [PATCH 3/5] fix(rtc): dynamically bind active event loop in
 AudioSource to avoid capture hang

---
 livekit-rtc/livekit/rtc/audio_source.py       | 48 +++++++++++--------
 livekit-rtc/tests/test_source_stream_close.py | 33 +++++++++++++
 2 files changed, 60 insertions(+), 21 deletions(-)

diff --git a/livekit-rtc/livekit/rtc/audio_source.py b/livekit-rtc/livekit/rtc/audio_source.py
index d3777b71..903011a2 100644
--- a/livekit-rtc/livekit/rtc/audio_source.py
+++ b/livekit-rtc/livekit/rtc/audio_source.py
@@ -53,17 +53,12 @@ def __init__(
         """
         self._sample_rate = sample_rate
         self._num_channels = num_channels
-
-        if loop is not None:
-            self._loop = loop
-        else:
+        self._loop: asyncio.AbstractEventLoop | None = loop
+        if self._loop is None:
             try:
                 self._loop = asyncio.get_running_loop()
             except RuntimeError:
-                try:
-                    self._loop = asyncio.get_event_loop()
-                except RuntimeError:
-                    self._loop = asyncio.new_event_loop()
+                self._loop = None
 
         req = proto_ffi.FfiRequest()
         req.new_audio_source.type = proto_audio_frame.AudioSourceType.AUDIO_SOURCE_NATIVE
@@ -107,7 +102,12 @@ def clear_queue(self) -> None:
         req = proto_ffi.FfiRequest()
         req.clear_audio_buffer.source_handle = self._ffi_handle.handle
         _ = FfiClient.instance.request(req)
-        self._release_waiter()
+        self._release_waiter_threadsafe()
+
+    def _get_loop(self) -> asyncio.AbstractEventLoop:
+        if self._loop is None or self._loop.is_closed():
+            self._loop = asyncio.get_running_loop()
+        return self._loop
 
     async def capture_frame(self, frame: AudioFrame) -> None:
         """
@@ -129,6 +129,8 @@ async def capture_frame(self, frame: AudioFrame) -> None:
         if frame.samples_per_channel == 0 or getattr(self._ffi_handle, "disposed", False):
             return
 
+        loop = self._get_loop()
+
         now = time.monotonic()
         elapsed = 0.0 if self._last_capture == 0.0 else now - self._last_capture
         self._q_size += frame.samples_per_channel / self.sample_rate - elapsed
@@ -137,16 +139,20 @@ async def capture_frame(self, frame: AudioFrame) -> None:
         if self._join_handle:
             self._join_handle.cancel()
 
-        if self._join_fut is None:
-            self._join_fut = self._loop.create_future()
+        if (
+            self._join_fut is None
+            or self._join_fut.get_loop() is not loop
+            or self._join_fut.get_loop().is_closed()
+        ):
+            self._join_fut = loop.create_future()
 
-        self._join_handle = self._loop.call_later(self._q_size, self._release_waiter)
+        self._join_handle = loop.call_later(self._q_size, self._release_waiter)
 
         req = proto_ffi.FfiRequest()
         req.capture_audio_frame.source_handle = self._ffi_handle.handle
         req.capture_audio_frame.buffer.CopyFrom(frame._proto_info())
 
-        queue = FfiClient.instance.queue.subscribe(loop=self._loop)
+        queue = FfiClient.instance.queue.subscribe(loop=loop)
         try:
             resp = FfiClient.instance.request(req)
             cb: proto_ffi.FfiEvent = await queue.wait_for(
@@ -187,23 +193,23 @@ def _release_waiter(self) -> None:
         self._q_size = 0.0
         self._join_fut = None
 
-    def close(self) -> None:
-        """Close the audio source.
-
-        This method cleans up resources associated with the audio source.
-        """
+    def _release_waiter_threadsafe(self) -> None:
         try:
             running_loop = asyncio.get_running_loop()
         except RuntimeError:
             running_loop = None
 
-        if running_loop is self._loop:
-            self._release_waiter()
-        elif self._loop.is_running():
+        if self._loop is not None and self._loop.is_running() and running_loop is not self._loop:
             self._loop.call_soon_threadsafe(self._release_waiter)
         else:
             self._release_waiter()
 
+    def close(self) -> None:
+        """Close the audio source.
+
+        This method cleans up resources associated with the audio source.
+        """
+        self._release_waiter_threadsafe()
         self._ffi_handle.dispose()
 
     async def aclose(self) -> None:
diff --git a/livekit-rtc/tests/test_source_stream_close.py b/livekit-rtc/tests/test_source_stream_close.py
index 3b23dab7..ea51b46d 100644
--- a/livekit-rtc/tests/test_source_stream_close.py
+++ b/livekit-rtc/tests/test_source_stream_close.py
@@ -106,3 +106,36 @@ async def test_video_source_aclose_and_capture_after_close() -> None:
 
     # Repeated aclose is idempotent
     await asyncio.wait_for(source.aclose(), timeout=2.0)
+
+
+def test_audio_source_init_outside_loop_and_capture_in_asyncio_run() -> None:
+    source = rtc.AudioSource(48000, 1)
+    assert source._loop is None
+    frame = rtc.AudioFrame.create(48000, 1, 480)
+
+    async def _capture() -> None:
+        await asyncio.wait_for(source.capture_frame(frame), timeout=5.0)
+        assert source._loop is not None
+        assert source._loop.is_running()
+        await asyncio.wait_for(source.aclose(), timeout=5.0)
+
+    asyncio.run(_capture())
+
+
+def test_audio_source_cross_thread_clear_queue() -> None:
+    source = rtc.AudioSource(48000, 1)
+    frame = rtc.AudioFrame.create(48000, 1, 48000)
+
+    async def _run() -> None:
+        await source.capture_frame(frame)
+        assert source._join_fut is not None
+
+        t = threading.Thread(target=source.clear_queue)
+        t.start()
+        t.join()
+
+        # Playout waiter should be released by cross-thread clear_queue
+        await asyncio.wait_for(source.wait_for_playout(), timeout=2.0)
+        await source.aclose()
+
+    asyncio.run(_run())

From ecd42051f412bc00a9307a34c79e989c943567f2 Mon Sep 17 00:00:00 2001
From: Nabi Bukhsh <nabibuksh.baloch01@gmail.com>
Date: Mon, 19 Oct 2026 02:05:47 +0500
Subject: [PATCH 4/5] fix(rtc): track capture generation in AudioSource to
 isolate clear_queue

---
 livekit-rtc/livekit/rtc/audio_source.py       | 19 ++++++++----
 livekit-rtc/tests/test_source_stream_close.py | 30 +++++++++++++++++++
 2 files changed, 43 insertions(+), 6 deletions(-)

diff --git a/livekit-rtc/livekit/rtc/audio_source.py b/livekit-rtc/livekit/rtc/audio_source.py
index 903011a2..2625a3ab 100644
--- a/livekit-rtc/livekit/rtc/audio_source.py
+++ b/livekit-rtc/livekit/rtc/audio_source.py
@@ -72,6 +72,7 @@ def __init__(
 
         self._last_capture = 0.0
         self._q_size = 0.0
+        self._capture_gen = 0
         self._join_handle: asyncio.TimerHandle | None = None
         self._join_fut: asyncio.Future[None] | None = None
 
@@ -102,7 +103,7 @@ def clear_queue(self) -> None:
         req = proto_ffi.FfiRequest()
         req.clear_audio_buffer.source_handle = self._ffi_handle.handle
         _ = FfiClient.instance.request(req)
-        self._release_waiter_threadsafe()
+        self._release_waiter_threadsafe(gen=self._capture_gen)
 
     def _get_loop(self) -> asyncio.AbstractEventLoop:
         if self._loop is None or self._loop.is_closed():
@@ -131,6 +132,9 @@ async def capture_frame(self, frame: AudioFrame) -> None:
 
         loop = self._get_loop()
 
+        self._capture_gen += 1
+        current_gen = self._capture_gen
+
         now = time.monotonic()
         elapsed = 0.0 if self._last_capture == 0.0 else now - self._last_capture
         self._q_size += frame.samples_per_channel / self.sample_rate - elapsed
@@ -146,7 +150,7 @@ async def capture_frame(self, frame: AudioFrame) -> None:
         ):
             self._join_fut = loop.create_future()
 
-        self._join_handle = loop.call_later(self._q_size, self._release_waiter)
+        self._join_handle = loop.call_later(self._q_size, self._release_waiter, current_gen)
 
         req = proto_ffi.FfiRequest()
         req.capture_audio_frame.source_handle = self._ffi_handle.handle
@@ -178,7 +182,10 @@ async def wait_for_playout(self) -> None:
 
         await asyncio.shield(self._join_fut)
 
-    def _release_waiter(self) -> None:
+    def _release_waiter(self, gen: int | None = None) -> None:
+        if gen is not None and gen != self._capture_gen:
+            return
+
         if self._join_handle is not None:
             self._join_handle.cancel()
             self._join_handle = None
@@ -193,16 +200,16 @@ def _release_waiter(self) -> None:
         self._q_size = 0.0
         self._join_fut = None
 
-    def _release_waiter_threadsafe(self) -> None:
+    def _release_waiter_threadsafe(self, gen: int | None = None) -> None:
         try:
             running_loop = asyncio.get_running_loop()
         except RuntimeError:
             running_loop = None
 
         if self._loop is not None and self._loop.is_running() and running_loop is not self._loop:
-            self._loop.call_soon_threadsafe(self._release_waiter)
+            self._loop.call_soon_threadsafe(self._release_waiter, gen)
         else:
-            self._release_waiter()
+            self._release_waiter(gen)
 
     def close(self) -> None:
         """Close the audio source.
diff --git a/livekit-rtc/tests/test_source_stream_close.py b/livekit-rtc/tests/test_source_stream_close.py
index ea51b46d..cdd7277d 100644
--- a/livekit-rtc/tests/test_source_stream_close.py
+++ b/livekit-rtc/tests/test_source_stream_close.py
@@ -139,3 +139,33 @@ async def _run() -> None:
         await source.aclose()
 
     asyncio.run(_run())
+
+
+def test_audio_source_clear_queue_does_not_cancel_subsequent_capture() -> None:
+    source = rtc.AudioSource(48000, 1, queue_size_ms=10000)
+    frame1 = rtc.AudioFrame.create(48000, 1, 48000)
+    frame2 = rtc.AudioFrame.create(48000, 1, 48000)
+
+    async def _run() -> None:
+        await source.capture_frame(frame1)
+        gen1 = source._capture_gen
+
+        # Simulate clear_queue scheduled from another thread
+        loop = asyncio.get_running_loop()
+        loop.call_soon(source._release_waiter, gen1)
+
+        # Capture a subsequent frame before yielding
+        await source.capture_frame(frame2)
+        assert source._capture_gen > gen1
+        assert source._join_fut is not None
+
+        # Allow queued clear callback to execute
+        await asyncio.sleep(0.01)
+
+        # The subsequent frame's playout future and timer must not be cleared
+        assert source._join_fut is not None
+        assert not source._join_fut.done()
+
+        await source.aclose()
+
+    asyncio.run(_run())

From 7ca3bacb15d58576619b279a79a324d7da22d7cd Mon Sep 17 00:00:00 2001
From: Nabi Bukhsh <nabibuksh.baloch01@gmail.com>
Date: Mon, 19 Oct 2026 02:11:22 +0500
Subject: [PATCH 5/5] fix(rtc): synchronize audio source queue metrics with
 thread lock on clear

---
 livekit-rtc/livekit/rtc/audio_source.py       | 70 +++++++++++--------
 livekit-rtc/tests/test_source_stream_close.py | 27 +++++++
 2 files changed, 67 insertions(+), 30 deletions(-)

diff --git a/livekit-rtc/livekit/rtc/audio_source.py b/livekit-rtc/livekit/rtc/audio_source.py
index 2625a3ab..64180a43 100644
--- a/livekit-rtc/livekit/rtc/audio_source.py
+++ b/livekit-rtc/livekit/rtc/audio_source.py
@@ -14,8 +14,9 @@
 
 from __future__ import annotations
 
-import time
 import asyncio
+import threading
+import time
 
 from ._ffi_client import FfiHandle, FfiClient
 from ._proto import audio_frame_pb2 as proto_audio_frame
@@ -73,6 +74,7 @@ def __init__(
         self._last_capture = 0.0
         self._q_size = 0.0
         self._capture_gen = 0
+        self._lock = threading.Lock()
         self._join_handle: asyncio.TimerHandle | None = None
         self._join_fut: asyncio.Future[None] | None = None
 
@@ -89,7 +91,8 @@ def num_channels(self) -> int:
     @property
     def queued_duration(self) -> float:
         """The current duration (in seconds) of audio data queued for playback."""
-        return max(self._q_size - time.monotonic() + self._last_capture, 0.0)
+        with self._lock:
+            return max(self._q_size - time.monotonic() + self._last_capture, 0.0)
 
     def clear_queue(self) -> None:
         """
@@ -103,7 +106,11 @@ def clear_queue(self) -> None:
         req = proto_ffi.FfiRequest()
         req.clear_audio_buffer.source_handle = self._ffi_handle.handle
         _ = FfiClient.instance.request(req)
-        self._release_waiter_threadsafe(gen=self._capture_gen)
+        with self._lock:
+            self._last_capture = 0.0
+            self._q_size = 0.0
+            target_gen = self._capture_gen
+        self._release_waiter_threadsafe(gen=target_gen)
 
     def _get_loop(self) -> asyncio.AbstractEventLoop:
         if self._loop is None or self._loop.is_closed():
@@ -132,25 +139,27 @@ async def capture_frame(self, frame: AudioFrame) -> None:
 
         loop = self._get_loop()
 
-        self._capture_gen += 1
-        current_gen = self._capture_gen
+        with self._lock:
+            self._capture_gen += 1
+            current_gen = self._capture_gen
 
-        now = time.monotonic()
-        elapsed = 0.0 if self._last_capture == 0.0 else now - self._last_capture
-        self._q_size += frame.samples_per_channel / self.sample_rate - elapsed
-        self._last_capture = now
+            now = time.monotonic()
+            elapsed = 0.0 if self._last_capture == 0.0 else now - self._last_capture
+            self._q_size += frame.samples_per_channel / self.sample_rate - elapsed
+            self._last_capture = now
+            q_size = self._q_size
 
-        if self._join_handle:
-            self._join_handle.cancel()
+            if self._join_handle:
+                self._join_handle.cancel()
 
-        if (
-            self._join_fut is None
-            or self._join_fut.get_loop() is not loop
-            or self._join_fut.get_loop().is_closed()
-        ):
-            self._join_fut = loop.create_future()
+            if (
+                self._join_fut is None
+                or self._join_fut.get_loop() is not loop
+                or self._join_fut.get_loop().is_closed()
+            ):
+                self._join_fut = loop.create_future()
 
-        self._join_handle = loop.call_later(self._q_size, self._release_waiter, current_gen)
+            self._join_handle = loop.call_later(q_size, self._release_waiter, current_gen)
 
         req = proto_ffi.FfiRequest()
         req.capture_audio_frame.source_handle = self._ffi_handle.handle
@@ -183,22 +192,23 @@ async def wait_for_playout(self) -> None:
         await asyncio.shield(self._join_fut)
 
     def _release_waiter(self, gen: int | None = None) -> None:
-        if gen is not None and gen != self._capture_gen:
-            return
+        with self._lock:
+            if gen is not None and gen != self._capture_gen:
+                return
 
-        if self._join_handle is not None:
-            self._join_handle.cancel()
-            self._join_handle = None
+            if self._join_handle is not None:
+                self._join_handle.cancel()
+                self._join_handle = None
 
-        if self._join_fut is None:
-            return  # could be None when clear_queue is called
+            if self._join_fut is None:
+                return  # could be None when clear_queue is called
 
-        if not self._join_fut.done():
-            self._join_fut.set_result(None)
+            if not self._join_fut.done():
+                self._join_fut.set_result(None)
 
-        self._last_capture = 0.0
-        self._q_size = 0.0
-        self._join_fut = None
+            self._last_capture = 0.0
+            self._q_size = 0.0
+            self._join_fut = None
 
     def _release_waiter_threadsafe(self, gen: int | None = None) -> None:
         try:
diff --git a/livekit-rtc/tests/test_source_stream_close.py b/livekit-rtc/tests/test_source_stream_close.py
index cdd7277d..ab97b5db 100644
--- a/livekit-rtc/tests/test_source_stream_close.py
+++ b/livekit-rtc/tests/test_source_stream_close.py
@@ -169,3 +169,30 @@ async def _run() -> None:
         await source.aclose()
 
     asyncio.run(_run())
+
+
+def test_audio_source_cross_thread_clear_queue_resets_queued_duration() -> None:
+    source = rtc.AudioSource(48000, 1, queue_size_ms=10000)
+    frame1 = rtc.AudioFrame.create(48000, 1, 48000)
+    frame2 = rtc.AudioFrame.create(48000, 1, 48000)
+
+    async def _run() -> None:
+        await source.capture_frame(frame1)
+        assert source.queued_duration > 0.8
+
+        t = threading.Thread(target=source.clear_queue)
+        t.start()
+        t.join()
+
+        # Queued duration should be reset immediately
+        assert source.queued_duration == 0.0
+
+        # Capture a new frame immediately after cross-thread clear
+        await source.capture_frame(frame2)
+
+        # Queued duration should reflect only the new frame (~1.0s), not ~2.0s
+        assert 0.8 < source.queued_duration <= 1.05
+
+        await source.aclose()
+
+    asyncio.run(_run())