Count unique unfinished events for EventBus capacity

.github/workflows/claude-code-review.yml

@@ -36,6 +36,7 @@ jobs:
         uses: anthropics/claude-code-action@beta
         with:
           anthropic_api_key: ${{ secrets.ANTHROPIC_API_KEY }}
+          model: claude-sonnet-4-6
 
           # Optional: Specify model (defaults to Claude Sonnet 4, uncomment for Claude Opus 4.1)
           # model: "claude-opus-4-1-20250805"

bubus/service.py

@@ -47,6 +47,10 @@ class QueueShutDown(Exception):
     pass
 
 
+class EventBusCapacityError(asyncio.QueueFull, RuntimeError):
+    """A bounded bus cannot admit another event."""
+
+
 QueueEntryType = TypeVar('QueueEntryType', bound='BaseEvent[Any]')
 T_ExpectedEvent = TypeVar('T_ExpectedEvent', bound='BaseEvent[Any]')
 
@@ -330,6 +334,7 @@ def __init__(
 
         self.event_queue = None
         self.event_history = {}
+        self._outstanding_events: dict[int, BaseEvent[Any]] = {}
         self.handlers = defaultdict(list)
         self.parallel_handlers = parallel_handlers
         self.wal_path = Path(wal_path) if wal_path else None
@@ -542,18 +547,18 @@ def dispatch(self, event: T_ExpectedEvent) -> T_ExpectedEvent:
             f'Event.event_path must be a list of valid EventBus names, got: {event.event_path}'
         )
 
-        # Check hard limit on total pending events (queue + in-progress)
-        # Only enforce if we have memory limits set
+        # Outstanding admission is independent of truncated diagnostic history.
+        self._outstanding_events = {
+            key: value for key, value in self._outstanding_events.items() if value.event_status in ('pending', 'started')
+        }
         if self.max_history_size is not None:
             queue_size = self.event_queue.qsize() if self.event_queue else 0
-            pending_in_history = sum(1 for e in self.event_history.values() if e.event_status in ('pending', 'started'))
-            total_pending = queue_size + pending_in_history
-
-            if total_pending >= 100:
-                raise RuntimeError(
+            total_pending = len(self._outstanding_events)
+            if total_pending >= 100 or (self.event_queue and self.event_queue.full()):
+                raise EventBusCapacityError(
                     f'EventBus at capacity: {total_pending} pending events (100 max). '
-                    f'Queue: {queue_size}, Processing: {pending_in_history}. '
-                    f'Cannot accept new events until some complete.'
+                    f'Queue: {queue_size}, Processing: {max(0, total_pending - queue_size)}. '
+                    f'Queue limit: 50. Cannot accept new events until some complete.'
                 )
 
         # Auto-start if needed
@@ -565,6 +570,7 @@ def dispatch(self, event: T_ExpectedEvent) -> T_ExpectedEvent:
                 self.event_queue.put_nowait(event)
                 # Only add to history after successfully queuing
                 self.event_history[event.event_id] = event
+                self._outstanding_events[id(event)] = event
                 logger.info(
                     f'🗣️ {self}.dispatch({event.event_type}) ➡️ {event.event_type}#{event.event_id[-4:]} (#{self.event_queue.qsize()} {event.event_status})'
                 )
@@ -789,6 +795,7 @@ async def stop(self, timeout: float | None = None, clear: bool = False) -> None:
         # Clear event history and handlers if requested (for memory cleanup)
         if clear:
             self.event_history.clear()
+            self._outstanding_events.clear()
             self.handlers.clear()
             # Remove from global instance tracking
             if self in EventBus.all_instances:
@@ -980,6 +987,8 @@ async def process_event(self, event: 'BaseEvent[Any]', timeout: float | None = N
 
         # Mark event as complete if all handlers are done
         event.event_mark_complete_if_all_handlers_completed()
+        if event.event_status not in ('pending', 'started'):
+            self._outstanding_events.pop(id(event), None)
 
         # After processing this event, check if any parent events can now be marked complete
         # We do this by walking up the parent chain

tests/test_unique_capacity.py

@@ -0,0 +1,119 @@
+import asyncio
+
+import pytest
+
+from bubus import BaseEvent, EventBus
+from bubus.service import EventBusCapacityError
+
+
+async def test_history_eviction_cannot_hide_outstanding_events():
+    bus = EventBus(max_history_size=1)
+    admitted = []
+    try:
+        for _ in range(100):
+            event = bus.dispatch(BaseEvent())
+            admitted.append(event)
+            assert bus.event_queue.get_nowait() is event
+            bus.event_queue.task_done()
+        assert len(bus.event_history) == 1
+        assert len({e.event_id for e in admitted}) == 100
+        with pytest.raises(EventBusCapacityError, match='capacity: 100 pending'):
+            bus.dispatch(BaseEvent())
+        assert len(bus._outstanding_events) == 100
+    finally:
+        await bus.stop(timeout=0, clear=True)
+    assert not bus._outstanding_events
+
+
+async def test_queue_bound_completion_and_exception_compatibility():
+    bus = EventBus(max_history_size=1)
+    seen = []
+
+    async def consume(event: BaseEvent):
+        seen.append(event.event_id)
+
+    bus.on(BaseEvent, consume)
+    try:
+        events = [bus.dispatch(BaseEvent()) for _ in range(50)]
+        with pytest.raises(EventBusCapacityError, match='capacity: 50 pending') as error:
+            bus.dispatch(BaseEvent())
+        assert isinstance(error.value, (RuntimeError, asyncio.QueueFull))
+        assert bus.event_queue.maxsize == 50
+        await bus.wait_until_idle(timeout=5)
+        assert sorted(seen) == sorted(e.event_id for e in events)
+        assert not bus._outstanding_events
+        await bus.dispatch(BaseEvent())
+        assert len(seen) == 51
+    finally:
+        await bus.stop(timeout=2, clear=True)
+
+
+async def test_started_event_survives_history_eviction_in_capacity_count():
+    bus = EventBus(max_history_size=1)
+    entered, release = asyncio.Event(), asyncio.Event()
+    completed = []
+
+    async def hold(event: BaseEvent):
+        entered.set()
+        await release.wait()
+        completed.append(event.event_id)
+
+    bus.on(BaseEvent, hold)
+    try:
+        first = bus.dispatch(BaseEvent())
+        await entered.wait()
+        queued = [bus.dispatch(BaseEvent()) for _ in range(50)]
+        assert first.event_id not in bus.event_history
+        with pytest.raises(EventBusCapacityError, match='51 pending.*Queue: 50, Processing: 1'):
+            bus.dispatch(BaseEvent())
+        release.set()
+        await bus.wait_until_idle(timeout=5)
+        assert sorted(completed) == sorted(e.event_id for e in [first, *queued])
+    finally:
+        release.set()
+        await bus.stop(timeout=2, clear=True)
+
+
+async def test_distinct_instances_with_same_event_id_remain_counted():
+    bus = EventBus(max_history_size=1)
+    entered, release = asyncio.Event(), asyncio.Event()
+    completed = []
+
+    async def hold(event: BaseEvent):
+        entered.set()
+        await release.wait()
+        completed.append(id(event))
+
+    bus.on(BaseEvent, hold)
+    try:
+        first = bus.dispatch(BaseEvent())
+        await entered.wait()
+        second = bus.dispatch(BaseEvent(event_id=first.event_id))
+        assert len(bus._outstanding_events) == 2
+        release.set()
+        await bus.wait_until_idle(timeout=5)
+        assert sorted(completed) == sorted([id(first), id(second)])
+        assert not bus._outstanding_events
+    finally:
+        release.set()
+        await bus.stop(timeout=2, clear=True)
+
+
+async def test_repeated_same_instance_preserves_one_handler_execution():
+    bus = EventBus()
+    calls = []
+
+    async def consume(event: BaseEvent):
+        calls.append(id(event))
+
+    bus.on(BaseEvent, consume)
+    try:
+        event = BaseEvent()
+        bus.dispatch(event)
+        bus.dispatch(event)
+        assert bus.event_queue.qsize() == 2
+        await bus.wait_until_idle(timeout=5)
+        assert calls == [id(event)]
+        assert not bus._outstanding_events
+    finally:
+        await bus.stop(timeout=2, clear=True)