EventBus misbehaves when handlers change subscriptions

Usage by agent

Timeline

1. planner iteration 1 ✓Finished 3 calls · 9,683 tokens · $0.0013 · 12.7s

   Loading…

2. coder iteration 1 ✓Finished 5 calls · 34.3K tokens · $0.0029 · 20.2s

   Loading…

3. tester iteration 1 ✓Finished 3 calls · 24.1K tokens · $0.0018 · 12.8s

   Loading…

4. verify iteration 1 ✓Passed 0.5s

   .............                                                            [100%]
   13 passed in 0.02s
   

Final diff

diff --git a/events.py b/events.py
index 2e4dc6f..bc16318 100644
--- a/events.py
+++ b/events.py
@@ -1,18 +1,54 @@
+class PublishError(Exception):
+    """Raised after 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:
+    """One subscription. Identity-based, so duplicate subscriptions stay distinct."""
+
+    def __init__(self, event, handler, priority):
+        self.event = event
+        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])
+        sub = _Subscription(event, handler, priority)
+        handlers.append(sub)
+        # Stable sort: higher priority first, ties keep subscription order.
+        handlers.sort(key=lambda s: -s.priority)
 
         def unsubscribe():
-            handlers.remove(entry)
+            if not sub.active:
+                return
+            sub.active = False
+            handlers[:] = [s for s in handlers if s is not sub]
 
         return unsubscribe
 
     def publish(self, event, payload=None):
-        for _, handler in self._handlers.get(event, []):
-            handler(payload)
+        # Snapshot so handlers subscribed during this publish wait for the next one.
+        subs = list(self._handlers.get(event, []))
+        errors = []
+        for sub in subs:
+            # Skip handlers unsubscribed earlier in this publish.
+            if not sub.active:
+                continue
+            try:
+                sub.handler(payload)
+            except Exception as exc:
+                errors.append(exc)
+        if errors:
+            raise PublishError(errors)
diff --git a/test_events.py b/test_events.py
index c4297d4..0503fd4 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,157 @@ def test_unsubscribe():
     stop()
     bus.publish("a", 1)
     assert received == []
+
+
+def test_self_unsubscribe_does_not_skip_next():
+    calls = []
+    bus = EventBus()
+    stops = {}
+
+    def first(payload):
+        calls.append("first")
+        stops["first"]()
+
+    stops["first"] = bus.subscribe("a", first)
+    bus.subscribe("a", lambda p: calls.append("second"))
+    bus.publish("a")
+    assert calls == ["first", "second"]
+
+
+def test_subscribe_during_publish_runs_next_time():
+    calls = []
+    bus = EventBus()
+
+    def adder(payload):
+        calls.append("adder")
+        bus.subscribe("a", lambda p: calls.append("late"))
+
+    bus.subscribe("a", adder)
+    bus.publish("a")
+    assert calls == ["adder"]
+    bus.publish("a")
+    assert calls == ["adder", "adder", "late"]
+
+
+def test_unsubscribe_other_before_it_runs():
+    calls = []
+    bus = EventBus()
+    stops = {}
+
+    def first(payload):
+        calls.append("first")
+        stops["second"]()
+
+    bus.subscribe("a", first)
+    stops["second"] = bus.subscribe("a", lambda p: calls.append("second"))
+    bus.publish("a")
+    assert calls == ["first"]
+
+
+def test_priority_order_with_ties():
+    calls = []
+    bus = EventBus()
+    bus.subscribe("a", lambda p: calls.append("low1"), priority=0)
+    bus.subscribe("a", lambda p: calls.append("high"), priority=5)
+    bus.subscribe("a", lambda p: calls.append("low2"), priority=0)
+    bus.subscribe("a", lambda p: calls.append("mid"), priority=3)
+    bus.publish("a")
+    assert calls == ["high", "mid", "low1", "low2"]
+
+
+def test_failing_handlers_all_run_and_collect_errors():
+    calls = []
+    bus = EventBus()
+    e1, e2 = ValueError("one"), KeyError("two")
+
+    def bad1(p):
+        calls.append("bad1")
+        raise e1
+
+    def bad2(p):
+        calls.append("bad2")
+        raise e2
+
+    bus.subscribe("a", bad1)
+    bus.subscribe("a", lambda p: calls.append("ok"))
+    bus.subscribe("a", bad2)
+    with pytest.raises(PublishError) as info:
+        bus.publish("a")
+    assert calls == ["bad1", "ok", "bad2"]
+    assert info.value.errors == [e1, e2]
+
+
+def test_publish_with_no_subscribers_is_noop():
+    EventBus().publish("nothing", 1)
+
+
+def test_double_unsubscribe_is_noop():
+    received = []
+    bus = EventBus()
+    stop = bus.subscribe("a", received.append)
+    stop()
+    stop()
+    bus.publish("a", 1)
+    assert received == []
+
+
+def test_same_handler_subscribed_twice_independent_unsubscribe():
+    received = []
+    bus = EventBus()
+    stop1 = bus.subscribe("a", received.append)
+    bus.subscribe("a", received.append)
+    stop1()
+    bus.publish("a", 1)
+    assert received == [1]
+    stop1()
+    bus.publish("a", 2)
+    assert received == [1, 2]
+
+
+def test_single_failure_raises_publish_error_with_that_exception():
+    bus = EventBus()
+    err = RuntimeError("boom")
+
+    def bad(p):
+        raise err
+
+    bus.subscribe("a", bad)
+    with pytest.raises(PublishError) as info:
+        bus.publish("a")
+    assert info.value.errors == [err]
+
+
+def test_unsubscribing_already_run_handler_is_harmless():
+    calls = []
+    bus = EventBus()
+    stops = {}
+
+    def first(p):
+        calls.append("first")
+
+    def second(p):
+        calls.append("second")
+        stops["first"]()
+
+    stops["first"] = bus.subscribe("a", first)
+    bus.subscribe("a", second)
+    bus.publish("a")
+    bus.publish("a")
+    assert calls == ["first", "second", "second"]
+
+
+def test_late_high_priority_subscription_waits_for_next_publish():
+    calls = []
+    bus = EventBus()
+
+    def adder(p):
+        calls.append("adder")
+        bus.subscribe("a", lambda p: calls.append("high"), priority=10)
+
+    bus.subscribe("a", adder)
+    bus.subscribe("a", lambda p: calls.append("low"))
+    bus.publish("a")
+    assert calls == ["adder", "low"]
+    calls.clear()
+    bus.publish("a")
+    assert calls == ["high", "adder", "low"]