fix(eventbus): prevent cross-loop contamination in bubus EventBus - #5512

Check if this issue is valid — if so, understand the root cause and fix it. At browser_use/bubus_compat.py, line 63:

<comment>When an already-running bus receives `dispatch()` from another event loop, `_original_start()` is a no-op but this wrapper overwrites `_loop` with the caller's loop. Record ownership only when `_start()` transitions the bus from stopped to running.</comment>

<file context>
@@ -0,0 +1,160 @@
+
+	def _patched_start(self: EventBus) -> None:
+		"""Record the owning event loop so BaseEvent.__await__ can stay on-loop."""
+		_original_start(self)
+		try:
+			loop = asyncio.get_running_loop()
</file context>

Check if this issue is valid — if so, understand the root cause and fix it. At browser_use/bubus_compat.py, line 113:

<comment>When an `EventBus` started before this patch is still running, it has no `_loop`, so this condition skips its queued child events even when it is the current loop's bus. A nested `await` can therefore hit the 1,000-iteration limit and return before the child completes; preserve the pre-patch fallback for untagged buses or initialize their owning loop.</comment>

<file context>
@@ -0,0 +1,160 @@
+							# 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 getattr(bus, '_loop', None) is not current_loop:
+								continue
+
</file context>

Check if this issue is valid — if so, understand the root cause and fix it. At browser_use/bubus_compat.py, line 144:

<comment>The `if iterations >= max_iterations:` block does nothing — its body is a commented-out `logger.error` and a `pass`. Since the drain loop can legitimately exhaust 1000 iterations, this placeholder silently swallows the case instead of reporting it. Remove the block, or replace it with a real `_logger.warning(...)` so hitting the iteration cap is observably logged.</comment>

<file context>
@@ -0,0 +1,160 @@
+					_logger.debug(f'Polling loop cancelled for {self}')
+					raise
+
+				if iterations >= max_iterations:
+					# logger.error(f'Max iterations reached while waiting for {self}')
+					pass
</file context>

Check if this issue is valid — if so, understand the root cause and fix it. At tests/ci/test_bubus_cross_loop_isolation.py, line 133:

<comment>loop_b is never close()d and thread.join(timeout=5) is silently ignored, so a failing run can leave the background thread running loop_b and leak into later tests, contradicting the finally block's comment. Also, `bus_b = ...result(timeout=5)` sits outside the try, so any failure before it (build_bus_b raising, or the 5s timeout) skips the finally entirely and leaks the still-spinning loop and thread unconditionally. Move loop_b/thread creation inside the guarded region and call loop_b.close() once the thread has exited.</comment>

<file context>
@@ -0,0 +1,133 @@
+			if bus_a is not None:
+				await bus_a.stop()
+			loop_b.call_soon_threadsafe(loop_b.stop)
+			thread.join(timeout=5)
</file context>