Don't drain other event loops' buses in BaseEvent.__await__

bubus/models.py

@@ -295,6 +295,8 @@ async def wait_for_handlers_to_complete_then_return_event():
                 max_iterations = 1000  # Prevent infinite loops
                 iterations = 0
 
+                current_loop = asyncio.get_running_loop()
+
                 try:
                     while not self.event_completed_signal.is_set() and iterations < max_iterations:
                         iterations += 1
@@ -306,6 +308,17 @@ async def wait_for_handlers_to_complete_then_return_event():
                             if not bus or not bus.event_queue:
                                 continue
 
+                            # Only drain running buses that belong to the current event loop.
+                            # Draining a bus owned by another loop runs its handlers on the wrong
+                            # loop, where they hang forever and pile up until the bus hits its
+                            # capacity limit (cross-loop contamination, browser-use/browser-use#5509).
+                            # Each bus's own _run_loop drains it on its own loop. Skip buses that
+                            # haven't started (_loop is None) or have been stopped (_is_running is
+                            # False) — a stopped bus can keep _loop set with events still queued, and
+                            # nothing should run its handlers after stop().
+                            if not bus._is_running or bus._loop is not current_loop:
+                                continue
+
                             # Process one event from this bus if available
                             try:
                                 if bus.event_queue.qsize() > 0:

bubus/service.py

@@ -274,6 +274,7 @@ class EventBus:
     _is_running: bool = False
     _runloop_task: asyncio.Task[None] | None = None
     _on_idle: asyncio.Event | None = None
+    _loop: asyncio.AbstractEventLoop | None = None  # the event loop this bus's run loop was started on
 
     def __init__(
         self,
@@ -733,6 +734,7 @@ def close_with_cleanup() -> None:
                     self._on_idle.clear()  # Start in a busy state unless we confirm queue is empty by running step() at least once
 
                 # Create and start the run loop task
+                self._loop = loop
                 self._runloop_task = loop.create_task(self._run_loop(), name=f'{self}._run_loop')
                 self._is_running = True
             except RuntimeError:

tests/test_cross_loop_isolation.py

@@ -0,0 +1,123 @@
+# pyright: basic
+"""Regression test for cross-loop contamination in BaseEvent.__await__.
+
+When a handler awaits an event, __await__ enters a drain loop that used to iterate
+*every* EventBus in the process and run its queued handlers on the awaiting task's
+event loop. In a multi-loop application (e.g. several parallel agent sessions, each
+on its own loop) that ran one bus's handlers on another bus's loop, where they hung
+and piled up until the bus hit its capacity limit:
+
+    RuntimeError: EventBus at capacity: 100 pending events (100 max)
+
+The drain loop must only process buses that belong to the current running loop.
+See browser-use/browser-use#5509.
+"""
+
+import asyncio
+import threading
+
+from bubus import BaseEvent, EventBus
+
+
+class BlockerEvent(BaseEvent):
+    pass
+
+
+class ProbeEvent(BaseEvent):
+    pass
+
+
+class ParentEvent(BaseEvent):
+    pass
+
+
+def _spin_loop(loop: asyncio.AbstractEventLoop) -> None:
+    asyncio.set_event_loop(loop)
+    loop.run_forever()
+
+
+async def test_await_drain_does_not_process_other_loops_buses():
+    loop_a = asyncio.get_running_loop()
+
+    # --- Bus B lives on its own event loop, running in a background thread ---
+    loop_b = asyncio.new_event_loop()
+    thread = threading.Thread(target=_spin_loop, args=(loop_b,), daemon=True)
+    thread.start()
+
+    ran_on: dict[str, int] = {}
+    release_blocker = threading.Event()
+
+    async def blocker_handler(event: BlockerEvent) -> None:
+        # Occupy bus B's serial processing so its own run loop cannot advance the
+        # ProbeEvent. That way the *only* thing that could move the ProbeEvent while
+        # the blocker is held is a (buggy) cross-loop drain from loop A.
+        while not release_blocker.is_set():
+            await asyncio.sleep(0.01)
+
+    async def probe_handler(event: ProbeEvent) -> None:
+        ran_on['probe'] = id(asyncio.get_running_loop())
+
+    async def build_bus_b() -> EventBus:
+        b = EventBus(name='BusB')
+        b.on(BlockerEvent, blocker_handler)
+        b.on(ProbeEvent, probe_handler)
+        b.dispatch(BlockerEvent())  # starts B's run loop and keeps it busy
+        return b
+
+    bus_b = asyncio.run_coroutine_threadsafe(build_bus_b(), loop_b).result(timeout=5)
+
+    # Create the probe on loop B and keep a reference so loop A can await the same
+    # object. It sits queued behind the blocker.
+    probe = await asyncio.wrap_future(
+        asyncio.run_coroutine_threadsafe(_dispatch_probe(bus_b), loop_b)
+    )
+    await asyncio.sleep(0.1)
+    assert bus_b.event_queue is not None and bus_b.event_queue.qsize() >= 1
+
+    # --- Bus A: a handler on loop A awaits bus B's probe event -> enters the drain ---
+    bus_a = EventBus(name='BusA')
+
+    async def parent_handler(event: ParentEvent) -> None:
+        # Awaiting from inside a handler (holding the global lock) triggers the
+        # cross-bus drain loop. Give it a bounded window to (wrongly) pick up the
+        # probe from bus B, then stop waiting so the test can assert.
+        try:
+            await asyncio.wait_for(_await_probe(probe), timeout=0.4)
+        except asyncio.TimeoutError:
+            pass
+
+    bus_a.on(ParentEvent, parent_handler)
+    try:
+        await bus_a.dispatch(ParentEvent())
+
+        # The probe belongs to bus B's loop. Loop A must NOT have run it.
+        assert ran_on.get('probe') != id(loop_a), (
+            "ProbeEvent from bus B ran on bus A's loop — cross-loop contamination"
+        )
+
+        # Once bus B is free, its own loop processes the probe — on loop B.
+        release_blocker.set()
+        for _ in range(100):
+            if 'probe' in ran_on:
+                break
+            await asyncio.sleep(0.02)
+        assert ran_on.get('probe') == id(loop_b)
+    finally:
+        # Always tear down, even if an assertion above fails, so a failing run
+        # never leaks bus B's still-spinning run loop or its background thread
+        # into later tests.
+        release_blocker.set()
+        try:
+            asyncio.run_coroutine_threadsafe(bus_b.stop(), loop_b).result(timeout=5)
+        finally:
+            await bus_a.stop()
+            loop_b.call_soon_threadsafe(loop_b.stop)
+            thread.join(timeout=5)
+
+
+async def _dispatch_probe(bus: EventBus) -> ProbeEvent:
+    return bus.dispatch(ProbeEvent())
+
+
+async def _await_probe(probe: ProbeEvent) -> None:
+    await probe