From 390f8674ffcba995d901cebfb62c1d0843d0f508 Mon Sep 17 00:00:00 2001
From: Matt Apperson <me@mattapperson.com>
Date: Fri, 3 Apr 2026 13:31:41 -0400
Subject: [PATCH] Replace spin-polling with anyio.Event in ToolEventBroadcaster

The _BroadcastConsumer.__anext__ method was using await anyio.sleep(0.001)
in a tight loop, burning CPU while waiting for new events. Replace this with
an anyio.Event that is set on push() and complete(), so consumers block
efficiently until signaled. The event is replaced after each set() to support
repeated wake cycles.

Adds concurrent producer/consumer tests to verify the event-driven behavior.
---
 .../_tool_event_broadcaster.py                |  6 +-
 tests/test_tool_event_broadcaster.py          | 56 +++++++++++++++++++
 2 files changed, 61 insertions(+), 1 deletion(-)

diff --git a/src/openrouter_agent/_tool_event_broadcaster.py b/src/openrouter_agent/_tool_event_broadcaster.py
index 6ae2f20..ec72188 100644
--- a/src/openrouter_agent/_tool_event_broadcaster.py
+++ b/src/openrouter_agent/_tool_event_broadcaster.py
@@ -21,17 +21,21 @@ def __init__(self) -> None:
         self._buffer: list[T] = []
         self._done = False
         self._error: BaseException | None = None
+        self._event = anyio.Event()
 
     def push(self, event: T) -> None:
         """Push an event to all consumers."""
         if self._done:
             raise RuntimeError("Cannot push to completed broadcaster")
         self._buffer.append(event)
+        self._event.set()
+        self._event = anyio.Event()
 
     def complete(self, error: BaseException | None = None) -> None:
         """Mark the broadcaster as complete."""
         self._done = True
         self._error = error
+        self._event.set()
 
     def create_consumer(self) -> AsyncIterator[T]:
         """Create a new independent consumer."""
@@ -64,4 +68,4 @@ async def __anext__(self) -> T:
                     raise self._broadcaster._error
                 raise StopAsyncIteration
 
-            await anyio.sleep(0.001)
+            await self._broadcaster._event.wait()
diff --git a/tests/test_tool_event_broadcaster.py b/tests/test_tool_event_broadcaster.py
index 9b8f68d..4228edb 100644
--- a/tests/test_tool_event_broadcaster.py
+++ b/tests/test_tool_event_broadcaster.py
@@ -1,5 +1,8 @@
 """Tests for ToolEventBroadcaster."""
 
+from collections.abc import AsyncIterator
+
+import anyio
 import pytest
 
 from openrouter_agent import ToolEventBroadcaster
@@ -58,3 +61,56 @@ def test_broadcaster_push_after_complete_raises():
     b.complete()
     with pytest.raises(RuntimeError):
         b.push("too late")
+
+
+@pytest.mark.anyio
+async def test_broadcaster_concurrent_push_and_consume():
+    """Verify consumers wake via event signaling, not spin-polling."""
+    b: ToolEventBroadcaster[int] = ToolEventBroadcaster()
+    consumer = b.create_consumer()
+    collected: list[int] = []
+
+    async def producer() -> None:
+        for i in range(5):
+            await anyio.sleep(0.01)
+            b.push(i)
+        b.complete()
+
+    async def consume() -> None:
+        async for event in consumer:
+            collected.append(event)
+
+    async with anyio.create_task_group() as tg:
+        tg.start_soon(producer)
+        tg.start_soon(consume)
+
+    assert collected == [0, 1, 2, 3, 4]
+
+
+@pytest.mark.anyio
+async def test_broadcaster_multiple_concurrent_consumers():
+    """Multiple consumers receiving events concurrently."""
+    b: ToolEventBroadcaster[int] = ToolEventBroadcaster()
+    c1 = b.create_consumer()
+    c2 = b.create_consumer()
+    r1: list[int] = []
+    r2: list[int] = []
+
+    async def producer() -> None:
+        for i in range(3):
+            await anyio.sleep(0.01)
+            b.push(i)
+        b.complete()
+
+    async with anyio.create_task_group() as tg:
+        tg.start_soon(producer)
+
+        async def consume(consumer: AsyncIterator[int], results: list[int]) -> None:
+            async for event in consumer:
+                results.append(event)
+
+        tg.start_soon(consume, c1, r1)
+        tg.start_soon(consume, c2, r2)
+
+    assert r1 == [0, 1, 2]
+    assert r2 == [0, 1, 2]