Python SDK BLE callback errors never reach the awaiting listener #21264

diff --git a/sdks/python/omi/ble.py b/sdks/python/omi/ble.py
index 36aa00d8..d6ccecd7 100644
--- a/sdks/python/omi/ble.py
+++ b/sdks/python/omi/ble.py
@@ -41,33 +41,60 @@ async def listen(
     char_uuid: str = AUDIO_DATA_UUID,
     service_uuid: Optional[str] = None,
 ) -> None:
-    """Connect and notify on audio characteristic until cancelled or disconnected."""
+    """Notify until cancelled, disconnected, or a packet callback raises.
 
-    async def _handler(_sender, data: bytearray) -> None:
-        raw = bytes(data)
-        result = on_packet(raw)
-        if inspect.isawaitable(result):
-            await result
+    Callback errors propagate to the caller after in-flight callbacks are cancelled.
+    """
 
-    disconnected = asyncio.Event()
+    stopped = asyncio.Event()
+    callback_error: Optional[Exception] = None
+    callback_tasks = set()
+
+    async def _handler(_sender, data: bytearray) -> None:
+        nonlocal callback_error
+        if stopped.is_set():
+            return
+        task = asyncio.current_task()
+        callback_tasks.add(task)
+        try:
+            result = on_packet(bytes(data))
+            if inspect.isawaitable(result):
+                await result
+        except Exception as error:
+            # Bleak dispatches async callbacks in detached tasks. Forward the
+            # first failure to the listener instead of leaking a task exception.
+            if callback_error is None:
+                callback_error = error
+            stopped.set()
+        finally:
+            callback_tasks.discard(task)
 
     def _on_disconnect(_client: BleakClient) -> None:
-        disconnected.set()
+        stopped.set()
 
     async with BleakClient(device_id, disconnected_callback=_on_disconnect) as client:
-        services = getattr(client, "services", None)
-        if services is not None and service_uuid:
-            service = services.get_service(service_uuid)
-            if service is None:
-                raise BleakError(f"Service {service_uuid} was not found")
-            characteristic = service.get_characteristic(char_uuid)
-            if characteristic is None:
-                raise BleakError(f"Characteristic {char_uuid} was not found in service {service_uuid}")
-            await client.start_notify(characteristic, _handler)
-        else:
-            await client.start_notify(char_uuid, _handler)
-        while not disconnected.is_set():
-            await asyncio.sleep(1)
+        try:
+            services = getattr(client, "services", None)
+            if services is not None and service_uuid:
+                service = services.get_service(service_uuid)
+                if service is None:
+                    raise BleakError(f"Service {service_uuid} was not found")
+                characteristic = service.get_characteristic(char_uuid)
+                if characteristic is None:
+                    raise BleakError(f"Characteristic {char_uuid} was not found in service {service_uuid}")
+                await client.start_notify(characteristic, _handler)
+            else:
+                await client.start_notify(char_uuid, _handler)
+            while not stopped.is_set():
+                await asyncio.sleep(1)
+        finally:
+            stopped.set()
+            pending = list(callback_tasks)
+            for task in pending:
+                task.cancel()
+            await asyncio.gather(*pending, return_exceptions=True)
+    if callback_error is not None:
+        raise callback_error
     raise ConnectionError(f"Device {device_id} disconnected")
 
 
diff --git a/sdks/python/tests/test_ble_callbacks.py b/sdks/python/tests/test_ble_callbacks.py
index e59ecfe1..de43c803 100644
--- a/sdks/python/tests/test_ble_callbacks.py
+++ b/sdks/python/tests/test_ble_callbacks.py
@@ -3,8 +3,6 @@ from __future__ import annotations
 import asyncio
 from unittest.mock import patch
 
-import pytest
-
 from omi.ble import listen, listen_payload
 from omi.constants import PACKET_HEADER_BYTES
 
@@ -186,58 +184,3 @@ def test_listen_payload_awaits_futures_and_custom_awaitables():
                         pass
 
     asyncio.run(_test())
-
-
-@pytest.mark.parametrize("listen_fn", [listen, listen_payload])
-def test_callback_exceptions_propagate(listen_fn):
-    async def _test():
-        client_instance = MockBleakClient("fake-device")
-        loop = asyncio.get_running_loop()
-
-        async def coro_handler(_data: bytes) -> None:
-            raise ValueError("coro error")
-
-        def future_handler(_data: bytes):
-            fut = loop.create_future()
-            loop.call_soon(fut.set_exception, ValueError("future error"))
-            return fut
-
-        class FailingAwaitable:
-            def __await__(self):
-                async def _run():
-                    raise ValueError("custom error")
-
-                return _run().__await__()
-
-        def custom_handler(_data: bytes):
-            return FailingAwaitable()
-
-        def sync_handler(_data: bytes) -> None:
-            raise ValueError("sync error")
-
-        handlers = [
-            ("sync", sync_handler, "sync error"),
-            ("coro", coro_handler, "coro error"),
-            ("future", future_handler, "future error"),
-            ("custom", custom_handler, "custom error"),
-        ]
-
-        with patch("omi.ble.BleakClient", return_value=client_instance):
-            for _name, handler, message in handlers:
-                if listen_fn is listen_payload:
-                    packet = b"\x00" * PACKET_HEADER_BYTES + b"\x03\x04"
-                else:
-                    packet = b"\x00\x01\x02\x03\x04"
-
-                async def fake_sleep(_sec, expected_message=message, current_handler=handler):
-                    with pytest.raises(ValueError, match=expected_message):
-                        await client_instance.notify_handler("sender", bytearray(packet))
-                    raise asyncio.CancelledError()
-
-                with patch("asyncio.sleep", side_effect=fake_sleep):
-                    try:
-                        await listen_fn("fake-device", handler)
-                    except asyncio.CancelledError:
-                        pass
-
-    asyncio.run(_test())
diff --git a/sdks/python/tests/test_ble_notification_errors.py b/sdks/python/tests/test_ble_notification_errors.py
new file mode 100644
index 00000000..46e5db91
--- /dev/null
+++ b/sdks/python/tests/test_ble_notification_errors.py
@@ -0,0 +1,164 @@
+"""Exercise Bleak's real notification dispatch with an in-memory BLE backend."""
+
+from __future__ import annotations
+
+import asyncio
+from functools import partial
+from types import SimpleNamespace
+from unittest.mock import patch
+
+import pytest
+from bleak import BleakClient
+
+from omi.ble import listen, listen_payload
+from omi.constants import PACKET_HEADER_BYTES
+
+
+class NotificationBackend:
+    """Replace only device I/O; Bleak still schedules notification callbacks."""
+
+    def __init__(self, address, *, disconnected_callback, ready, **_kwargs):
+        self.address = address
+        self.disconnected_callback = disconnected_callback
+        self.ready = ready
+        self.is_connected = False
+        self.disconnect_count = 0
+        self.notification = None
+        self.services = SimpleNamespace(
+            get_characteristic=lambda _uuid: SimpleNamespace(handle=1)
+        )
+
+    async def connect(self, **_kwargs):
+        self.is_connected = True
+        return True
+
+    async def disconnect(self):
+        self.disconnect_count += 1
+        self.is_connected = False
+        return True
+
+    async def start_notify(self, _characteristic, callback, **_kwargs):
+        self.notification = callback
+        self.ready.set_result(self)
+
+    def emit(self, payload=b"audio"):
+        self.notification(bytearray(b"\x00" * PACKET_HEADER_BYTES + payload))
+
+
+class FailingAwaitable:
+    def __init__(self, error):
+        self.error = error
+
+    def __await__(self):
+        async def fail():
+            raise self.error
+
+        return fail().__await__()
+
+
+@pytest.mark.parametrize("listen_fn", [listen, listen_payload])
+@pytest.mark.parametrize(
+    "callback_kind", ["sync", "coroutine", "future", "task", "custom"]
+)
+def test_notification_error_reaches_listener(listen_fn, callback_kind):
+    async def run():
+        loop = asyncio.get_running_loop()
+        ready = loop.create_future()
+        unhandled = []
+        loop.set_exception_handler(lambda _loop, context: unhandled.append(context))
+        failure = ValueError("audio consumer failed")
+
+        async def fail():
+            await asyncio.sleep(0)
+            raise failure
+
+        def on_packet(packet):
+            expected = (
+                b"audio"
+                if listen_fn is listen_payload
+                else b"\x00" * PACKET_HEADER_BYTES + b"audio"
+            )
+            assert packet == expected
+            if callback_kind == "sync":
+                raise failure
+            if callback_kind == "coroutine":
+                return fail()
+            if callback_kind == "future":
+                future = loop.create_future()
+                loop.call_soon(future.set_exception, failure)
+                return future
+            if callback_kind == "task":
+                return asyncio.create_task(fail())
+            return FailingAwaitable(failure)
+
+        client_factory = partial(BleakClient, backend=NotificationBackend, ready=ready)
+        with patch("omi.ble.BleakClient", client_factory):
+            listener = asyncio.create_task(listen_fn("test-device", on_packet))
+            try:
+                backend = await asyncio.wait_for(ready, timeout=2)
+                # Backend notifications are synchronous. Real Bleak creates the
+                # async callback task, so this never awaits the SDK's handler.
+                backend.emit()
+                with pytest.raises(ValueError) as raised:
+                    await asyncio.wait_for(listener, timeout=2)
+                assert raised.value is failure
+                assert backend.disconnect_count == 1
+                assert not backend.is_connected
+                await asyncio.sleep(0)
+                assert unhandled == []
+            finally:
+                listener.cancel()
+                await asyncio.gather(listener, return_exceptions=True)
+
+    asyncio.run(run())
+
+
+@pytest.mark.parametrize("ending", ["callback_error", "cancel", "disconnect"])
+def test_listener_drains_inflight_notifications(ending):
+    async def run():
+        loop = asyncio.get_running_loop()
+        ready = loop.create_future()
+        entered = asyncio.Event()
+        cleaned_up = asyncio.Event()
+        calls = []
+        failure = ValueError("second packet failed")
+
+        async def on_packet(packet):
+            calls.append(packet)
+            if packet.endswith(b"fail"):
+                raise failure
+            entered.set()
+            try:
+                await asyncio.Event().wait()
+            finally:
+                cleaned_up.set()
+
+        client_factory = partial(BleakClient, backend=NotificationBackend, ready=ready)
+        with patch("omi.ble.BleakClient", client_factory):
+            listener = asyncio.create_task(listen("test-device", on_packet))
+            try:
+                backend = await asyncio.wait_for(ready, timeout=2)
+                backend.emit(b"pending")
+                await asyncio.wait_for(entered.wait(), timeout=2)
+                if ending == "callback_error":
+                    backend.emit(b"fail")
+                    expected_error = ValueError
+                elif ending == "cancel":
+                    listener.cancel()
+                    expected_error = asyncio.CancelledError
+                else:
+                    backend.disconnected_callback()
+                    expected_error = ConnectionError
+                with pytest.raises(expected_error):
+                    await asyncio.wait_for(listener, timeout=2)
+                assert cleaned_up.is_set()
+                assert backend.disconnect_count == 1
+                previous_calls = len(calls)
+                backend.emit(b"late")
+                await asyncio.sleep(0)
+                assert len(calls) == previous_calls
+            finally:
+                listener.cancel()
+                await asyncio.gather(listener, return_exceptions=True)
+
+    asyncio.run(run())