fix(events): dispatch foreign loops through serving loop

backend/core/event_bus.py

@@ -64,31 +64,27 @@ def emit(kind: str, payload: dict[str, Any] | None = None) -> None:
         **(payload or {}),
     }
     event_str = json.dumps(event)
-    loop = asyncio.get_running_loop() if _on_event_loop() else _serving_loop
-    if loop is None:
+    try:
+        caller_loop = asyncio.get_running_loop()
+    except RuntimeError:
+        caller_loop = None
+    target_loop = _serving_loop or caller_loop
+    if target_loop is None:
         # No serving loop yet — nobody to notify; dropping is correct.
         logger.debug("No event loop — event dropped: %s", kind)
         return
     try:
-        if _on_event_loop():
-            loop.create_task(_broadcast(event_str))
+        if caller_loop is target_loop:
+            target_loop.create_task(_broadcast(event_str))
         else:
-            # Threadpool worker (sync endpoint body): the only thread-safe way in.
-            loop.call_soon_threadsafe(_schedule_broadcast, event_str)
+            # Sync endpoints and async producers on a foreign loop must both
+            # hand off: the lock and listener queues belong to serving_loop.
+            target_loop.call_soon_threadsafe(_schedule_broadcast, event_str)
     except RuntimeError:
         # The serving loop closed between capture and use (app shutdown).
         logger.debug("Event loop closed — event dropped: %s", kind)
 
 
-def _on_event_loop() -> bool:
-    """True when called from the running event loop (async-context emit)."""
-    try:
-        asyncio.get_running_loop()
-    except RuntimeError:
-        return False
-    return True
-
-
 def _schedule_broadcast(event_str: str) -> None:
     """Run `_broadcast` on the serving loop; called via call_soon_threadsafe."""
     asyncio.get_running_loop().create_task(_broadcast(event_str))

tests/test_event_bus_thread_emit.py

@@ -66,6 +66,58 @@ def test_emit_from_thread_reaches_serving_loop(bus, tmp_path):
         loop.close()
 
 
+def test_emit_from_foreign_running_loop_reaches_serving_loop(bus):
+    """An async producer may run on a worker loop, but listener state belongs
+    to the WebSocket serving loop and must only be touched there."""
+    serving_loop = asyncio.new_event_loop()
+    serving_loop.set_debug(True)
+    received: list[str] = []
+    started = threading.Event()
+    done = threading.Event()
+
+    async def serve_with_waiter_ready():
+        q = await bus.subscribe()
+        waiter = asyncio.create_task(q.get())
+        await asyncio.sleep(0)  # q.get() has installed its serving-loop Future
+        started.set()
+        try:
+            received.append(await asyncio.wait_for(waiter, 0.5))
+        except asyncio.TimeoutError:
+            pass
+        finally:
+            done.set()
+            await bus.unsubscribe(q)
+
+    def run_serving_loop():
+        asyncio.set_event_loop(serving_loop)
+        serving_loop.run_until_complete(serve_with_waiter_ready())
+
+    serving_thread = threading.Thread(
+        target=run_serving_loop, name="test-serving-loop"
+    )
+    serving_thread.start()
+    try:
+        assert started.wait(2.0), "serving loop never subscribed"
+
+        async def foreign_async_caller():
+            assert asyncio.get_running_loop() is not serving_loop
+            bus.emit("profiles", {"action": "updated", "id": "foreign-loop"})
+            await asyncio.sleep(0.05)
+
+        asyncio.run(foreign_async_caller())
+
+        assert done.wait(2.0), (
+            "emit() ran listener delivery on the caller's foreign loop"
+        )
+        assert received, "foreign-loop event never reached the serving loop"
+        payload = json.loads(received[0])
+        assert payload["id"] == "foreign-loop"
+    finally:
+        serving_loop.call_soon_threadsafe(done.set)
+        serving_thread.join(2.0)
+        serving_loop.close()
+
+
 async def _serve(bus, received: list[str], started: threading.Event, done: threading.Event):
     q = await bus.subscribe()
     started.set()