EventBus misbehaves when handlers change subscriptions

Usage by agent

Timeline

1. planner iteration 1 ✓Finished 3 calls · 9,662 tokens · $0.0013 · 12.5s

   Loading…

2. coder iteration 1 ✓Finished 7 calls · 48.9K tokens · $0.0032 · 29.1s

   Loading…

3. tester iteration 1 ✓Finished 3 calls · 26.7K tokens · $0.0020 · 10.9s

   Loading…

4. verify iteration 1 ✓Passed 0.5s

   ..............                                                           [100%]
   14 passed in 0.03s
   

5. reviewer iteration 1 ✓Finished 1 calls · 7,914 tokens · $0.0012 · 5.8s

   Loading…

Final diff

diff --git a/events.py b/events.py
index 2e4dc6f..a17e49d 100644
--- a/events.py
+++ b/events.py
@@ -1,18 +1,55 @@
+class PublishError(Exception):
+    """Raised by EventBus.publish when one or more handlers raised.
+
+    ``errors`` holds the exceptions in the order they were raised.
+    """
+
+    def __init__(self, errors):
+        self.errors = list(errors)
+        super().__init__(f"{len(self.errors)} handler(s) raised an exception")
+
+
+class _Subscription:
+    __slots__ = ("handler", "priority", "active")
+
+    def __init__(self, handler, priority):
+        self.handler = handler
+        self.priority = priority
+        self.active = True
+
+
 class EventBus:
     def __init__(self):
         self._handlers = {}
 
     def subscribe(self, event, handler, priority=0):
         handlers = self._handlers.setdefault(event, [])
-        entry = (priority, handler)
-        handlers.append(entry)
-        handlers.sort(key=lambda item: item[0])
+        record = _Subscription(handler, priority)
+        handlers.append(record)
 
         def unsubscribe():
-            handlers.remove(entry)
+            if not record.active:
+                return
+            record.active = False
+            for i, existing in enumerate(handlers):
+                if existing is record:
+                    del handlers[i]
+                    break
 
         return unsubscribe
 
     def publish(self, event, payload=None):
-        for _, handler in self._handlers.get(event, []):
-            handler(payload)
+        # Snapshot at start: handlers subscribed during this publish wait for
+        # the next one. Stable sort keeps subscription order for equal priority.
+        snapshot = sorted(self._handlers.get(event, []), key=lambda r: -r.priority)
+        errors = []
+        for record in snapshot:
+            # Skip handlers unsubscribed earlier in this publish.
+            if not record.active:
+                continue
+            try:
+                record.handler(payload)
+            except Exception as exc:  # BaseException still propagates immediately
+                errors.append(exc)
+        if errors:
+            raise PublishError(errors)
diff --git a/test_events.py b/test_events.py
index c4297d4..9a250a9 100644
--- a/test_events.py
+++ b/test_events.py
@@ -1,4 +1,6 @@
-from events import EventBus
+import pytest
+
+from events import EventBus, PublishError
 
 
 def test_publish():
@@ -16,3 +18,179 @@ def test_unsubscribe():
     stop()
     bus.publish("a", 1)
     assert received == []
+
+
+def test_self_unsubscribe_does_not_skip_next():
+    received = []
+    bus = EventBus()
+    stops = []
+
+    def first(payload):
+        received.append("first")
+        stops[0]()
+
+    stops.append(bus.subscribe("a", first))
+    bus.subscribe("a", lambda p: received.append("second"))
+    bus.publish("a")
+    assert received == ["first", "second"]
+
+
+def test_subscribe_during_publish_deferred():
+    received = []
+    bus = EventBus()
+
+    def adder(payload):
+        received.append("adder")
+        bus.subscribe("a", lambda p: received.append("late"))
+
+    bus.subscribe("a", adder)
+    bus.publish("a")
+    assert received == ["adder"]
+    bus.publish("a")
+    assert received == ["adder", "adder", "late"]
+
+
+def test_unsubscribe_other_before_run():
+    received = []
+    bus = EventBus()
+    stops = {}
+
+    def a(payload):
+        received.append("a")
+        stops["b"]()
+
+    stops["a"] = bus.subscribe("x", a)
+    stops["b"] = bus.subscribe("x", lambda p: received.append("b"))
+    bus.publish("x")
+    assert received == ["a"]
+
+
+def test_priority_order_ties_in_subscribe_order():
+    received = []
+    bus = EventBus()
+    bus.subscribe("x", lambda p: received.append("low1"), priority=0)
+    bus.subscribe("x", lambda p: received.append("high1"), priority=5)
+    bus.subscribe("x", lambda p: received.append("low2"), priority=0)
+    bus.subscribe("x", lambda p: received.append("high2"), priority=5)
+    bus.publish("x")
+    assert received == ["high1", "high2", "low1", "low2"]
+
+
+def test_exceptions_collected_in_order():
+    received = []
+    bus = EventBus()
+
+    def boom1(p):
+        raise ValueError("one")
+
+    def ok(p):
+        received.append("ok")
+
+    def boom2(p):
+        raise KeyError("two")
+
+    bus.subscribe("x", boom1)
+    bus.subscribe("x", ok)
+    bus.subscribe("x", boom2)
+    bus.subscribe("x", lambda p: received.append("last"))
+    with pytest.raises(PublishError) as info:
+        bus.publish("x")
+    assert received == ["ok", "last"]
+    assert [type(e) for e in info.value.errors] == [ValueError, KeyError]
+    assert [e.args for e in info.value.errors] == [("one",), ("two",)]
+
+
+def test_double_unsubscribe_is_noop():
+    received = []
+    bus = EventBus()
+    stop = bus.subscribe("x", lambda p: received.append("a"))
+    other = bus.subscribe("x", lambda p: received.append("b"))
+    stop()
+    stop()
+    bus.publish("x")
+    assert received == ["b"]
+    other()
+
+
+def test_duplicate_subscriptions_unsubscribe_independently():
+    received = []
+    bus = EventBus()
+
+    def h(p):
+        received.append(p)
+
+    stop1 = bus.subscribe("x", h)
+    stop2 = bus.subscribe("x", h)
+    bus.publish("x", 1)
+    assert received == [1, 1]
+    stop1()
+    received.clear()
+    bus.publish("x", 2)
+    assert received == [2]
+    stop2()
+    received.clear()
+    bus.publish("x", 3)
+    assert received == []
+
+
+def test_double_unsubscribe_does_not_remove_duplicate_twin():
+    received = []
+    bus = EventBus()
+
+    def h(p):
+        received.append(p)
+
+    stop1 = bus.subscribe("x", h)
+    bus.subscribe("x", h)
+    stop1()
+    stop1()
+    bus.publish("x", 1)
+    assert received == [1]
+
+
+def test_unsubscribe_second_duplicate_mid_publish():
+    received = []
+    bus = EventBus()
+    stops = []
+
+    def h(p):
+        received.append("h")
+        if len(stops) > 1:
+            stops[1]()
+
+    stops.append(bus.subscribe("x", h))
+    stops.append(bus.subscribe("x", h))
+    bus.publish("x")
+    assert received == ["h"]
+
+
+def test_base_exception_propagates_immediately():
+    received = []
+    bus = EventBus()
+
+    def interrupt(p):
+        raise KeyboardInterrupt()
+
+    bus.subscribe("x", interrupt)
+    bus.subscribe("x", lambda p: received.append("after"))
+    with pytest.raises(KeyboardInterrupt):
+        bus.publish("x")
+    assert received == []
+
+
+def test_publish_error_message_and_errors_type():
+    bus = EventBus()
+    bus.subscribe("x", lambda p: 1 / 0)
+    bus.subscribe("x", lambda p: 1 / 0)
+    with pytest.raises(PublishError) as info:
+        bus.publish("x")
+    assert isinstance(info.value.errors, list)
+    assert len(info.value.errors) == 2
+    assert all(isinstance(e, ZeroDivisionError) for e in info.value.errors)
+    assert "2" in str(info.value)
+
+
+def test_no_error_when_no_handler_fails():
+    bus = EventBus()
+    bus.subscribe("x", lambda p: None)
+    assert bus.publish("x") is None