fix(eventbus): only drain events in the waiting chain, not all buses (#5509)

[Ethanz11-creat]

Summary

Fixes browser-use/browser-use#5509 — cross-loop contamination in the EventBus drain loop.

When a handler on bus A awaits a child event (e.g. via await event inside a handler context while holding the global lock), the __await__ drain loop iterated over all EventBus.all_instances and processed queued events from every bus — not just the waiting chain. With multiple parallel agent sessions (each with its own bus), an unrelated event queued on bus B during that window was processed inside bus A's handler context, stealing bus B's scheduling. With many parallel sessions this drives the capacity errors / agent run failures reported in the issue.

Root cause

BaseEvent.__await__ → wait_for_handlers_to_complete_then_return_event() in bubus/models.py:

for bus in list(EventBus.all_instances):
    if bus.event_queue.qsize() > 0:
        event = bus.event_queue.get_nowait()
        await bus.process_event(event)
        ...

Every queued event on every bus was consumed, regardless of whether it belonged to the waiting chain.

Fix

• Compute the ancestry chain of the awaited event once ({self} ∪ {parents}).
• When draining a bus, locate the first event in its queue that belongs to that chain and process only that one; unrelated events keep their FIFO position untouched and are left for their own bus's run loop (which gets control via the existing sleep(0) yield). The underlying deque is indexed directly (same approach as the existing memory-usage check); task_done() pairs with the put() that enqueued the event.
• Cross-bus forwarded child events still get processed (they are part of the ancestry chain), so the deadlock-prevention semantics are preserved.

Tests

• New tests/test_issue5509_cross_loop.py: a bus A handler chain forwards its grandchild to bus B, while an unrelated event is queued on bus B first. Asserts the bus B event is not processed inside bus A's drain loop (fails before the fix with order == ['a_drain_enter', 'b_handled', 'a_drain_exit'], passes after with the bus B event handled by bus B's own run loop afterwards).
• Full suite: 136 passed (135 existing + 1 new). ruff check and codespell clean.

Disclosure

This PR was developed with AI assistance (analysis and test-driven implementation), reviewed and verified by the author.

[cubic-dev-ai]

All reported issues were addressed across 2 files

<sub>Reply with feedback, questions, or to request a fix.

Fix all with cubic | Re-trigger cubic</sub>

[Ethanz11-creat]

Thanks for the review — all three points addressed in the latest push:

P1 (sibling drain): Correct — matching on event_parent_id in ancestor_ids would also drain siblings that merely share a parent with the awaited event. The check now walks the candidate's parent chain and only matches if the chain contains the awaited event itself (self.event_id) — i.e. the candidate is a descendant of the awaited event. This still covers cross-bus forwarded descendants (the original reproduction), while siblings stay queued for normal scheduling.

P2 (task_done ordering): Correct — task_done() is now called in a finally after process_event(candidate) completes, so wait_until_idle()/queue joins cannot observe the task done before processing finished, and handler failures still release the queue task.

P3 (test mechanism): Correct — the test previously enqueued the independent event after the grandchild. It now parks bus B's consumer on the global lock (bus B's run loop get()s its first event and blocks acquiring the lock held by bus A), then queues a second independent event that only the drain loop can reach. The assertion checks both bus B events are handled strictly after the drain window. Full suite: 136 passed, ruff + codespell clean.

[cubic-dev-ai]

All reported issues were addressed across 2 files (changes from recent commits).

<sub>Reply with feedback, questions, or to request a fix.

Fix all with cubic | Re-trigger cubic</sub>

[Ethanz11-creat]

Addressed in 6ef3c3e: extracted a shared _find_event_by_id() helper (scanning the bus snapshot once per lookup) and reused it in both the ancestry walk and the per-candidate descendant walk. EventBus.all_instances is now snapshotted once before the drain loop instead of rebuilt per hop — the drain is no longer O(candidates × bus_count × chain_depth) per iteration. Full suite still passes (136 tests), ruff + codespell clean.

[cubic-dev-ai]

1 issue found across 1 file (changes from recent commits).

Prompt for AI agents (unresolved issues)

Check if these issues are valid — if so, understand the root cause of each and fix them. If appropriate, use sub-agents to investigate and fix each issue separately.


<file name="bubus/models.py">

<violation number="1" location="bubus/models.py:303">
P2: The lookup snapshot is frozen once, but the drain loop below still re-enumerates `list(EventBus.all_instances)` on every iteration (`for bus in list(EventBus.all_instances)`). A bus created mid-drain is iterated by the drain but is invisible to `_find_event_by_id`, which only scans `buses_snapshot`. Because the snapshot was taken before those buses existed, any of their queued events that belong to this waiting chain get `parent_candidate = None` during the descendant walk and are never drained, while unrelated events on them are scanned in the loop — so the drain's behavior silently depends on when the snapshot happened and contradicts the stated "snapshot once per drain" intent.</violation>
</file>

<sub>Tip: Review your code locally with the cubic CLI to iterate faster.

Fix all with cubic | Re-trigger cubic</sub>

[cubic-dev-ai]

P2: The lookup snapshot is frozen once, but the drain loop below still re-enumerates list(EventBus.all_instances) on every iteration (for bus in list(EventBus.all_instances)). A bus created mid-drain is iterated by the drain but is invisible to _find_event_by_id, which only scans buses_snapshot. Because the snapshot was taken before those buses existed, any of their queued events that belong to this waiting chain get parent_candidate = None during the descendant walk and are never drained, while unrelated events on them are scanned in the loop — so the drain's behavior silently depends on when the snapshot happened and contradicts the stated "snapshot once per drain" intent.

Prompt for AI agents

Check if this issue is valid — if so, understand the root cause and fix it. At bubus/models.py, line 303:

<comment>The lookup snapshot is frozen once, but the drain loop below still re-enumerates `list(EventBus.all_instances)` on every iteration (`for bus in list(EventBus.all_instances)`). A bus created mid-drain is iterated by the drain but is invisible to `_find_event_by_id`, which only scans `buses_snapshot`. Because the snapshot was taken before those buses existed, any of their queued events that belong to this waiting chain get `parent_candidate = None` during the descendant walk and are never drained, while unrelated events on them are scanned in the loop — so the drain's behavior silently depends on when the snapshot happened and contradicts the stated "snapshot once per drain" intent.</comment>

<file context>
@@ -295,6 +295,19 @@ async def wait_for_handlers_to_complete_then_return_event():
+                # events across this set, and rebuilding it per hop would make
+                # the drain O(candidates × bus_count × chain_depth) per
+                # iteration (review feedback on #5509).
+                buses_snapshot = list(EventBus.all_instances)
+
+                def _find_event_by_id(event_id: str) -> BaseEvent[Any] | None:
</file context>