Replace spin-polling with anyio.Event in ToolEventBroadcaster

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()

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]