fix(events): add unsubscribe() to EventBus protocol and implementations (#10356)

Add individual handler unsubscription to the EventBus protocol and all
implementations (ReactiveEventBus, LoggingEventBus, TrackingEventBus).

Changes:
- EventBus protocol adds `unsubscribe(subscription_id) -> bool` and changes
  `subscribe()` return type from None to EventSubID (ULID string)
- EventSubID = str type alias for subscription IDs
- ReactiveEventBus stores subscriptions as (sub_id, handler) tuples per
  EventType, enabling individual unsubscription by ULID
- LoggingEventBus uses the same pattern for consistent protocol conformance
- TrackingEventBus mock updated with subscribe/unsubscribe support
- EventBusBridge.start() now receives a proper subscription ID return from
  subscribe(), resolving existing subscription handle tracking
- robot/helper_event_bus.py gain unsubscribe test commands

This resolves issue #10356 and makes all bus implementations satisfy the
full EventBus protocol with per-handler lifecycle management.

features/mocks/uko_indexer_mocks.py

@@ -117,10 +117,7 @@ class FailingTextBackend:
         raise RuntimeError("Simulated text failure")
 
     def search(
-        self,
-        project: str,
-        query: str,
-        limit: int = 20,
+        self, project: str, query: str, limit: int = 20,
     ) -> list[SearchResult]:
         return []
 
@@ -170,11 +167,16 @@ class FailingVectorBackend:
 
 
 class TrackingEventBus:
-    """Minimal EventBus stub that records emitted events."""
+    """Minimal EventBus stub that records emitted events and subscribable handlers.
+
+    Implements the :class:`EventBus` protocol including :meth:`subscribe`
+    (returns a subscription ID) and :meth:`unsubscribe`.
+    """
 
     def __init__(self) -> None:
         self.events: list[DomainEvent] = []
         self._event = threading.Event()
+        self._subscriptions: dict[str, dict[str, Any]] = {}
 
     def emit(self, event: DomainEvent) -> None:
         self.events.append(event)
@@ -184,8 +186,17 @@ class TrackingEventBus:
         self,
         event_type: EventType,
         handler: Callable[..., Any],
-    ) -> None:
-        _ = event_type, handler  # Not needed for tests
+    ) -> str:
+        """Register a handler and return a subscription ID."""
+        _ = event_type
+        sub_id = f"track-{handler.__qualname__ if hasattr(handler, '__qualname__') else id(handler)}"
+        self._subscriptions[sub_id] = {"handler": handler}
+        return sub_id
+
+    def unsubscribe(self, subscription_id: str) -> bool:
+        """Remove a subscription by ID."""
+        removed = self._subscriptions.pop(subscription_id, None) is not None
+        return removed
 
     def wait(self, timeout: float = 2.0) -> bool:
         return self._event.wait(timeout=timeout)

robot/helper_event_bus.py

@@ -104,6 +104,78 @@ def reactive_type_filtering() -> None:
     print("reactive-type-filtering-ok")
 
 
+def reactive_unsubscribe() -> None:
+    """Verify ReactiveEventBus.unsubscribe removes specific handler."""
+    received: list[DomainEvent] = []
+    bus = ReactiveEventBus()
+    sub_id = bus.subscribe(EventType.PLAN_CREATED, received.append)
+
+    # Verify subscribe returns a non-empty string (ULID subscription ID)
+    if not isinstance(sub_id, str) or len(sub_id) == 0:
+        print(f"FAIL: expected non-empty subscription ID, got {sub_id!r}", file=sys.stderr)
+        sys.exit(1)
+
+    # Emit and verify handler was called
+    bus.emit(DomainEvent(event_type=EventType.PLAN_CREATED))
+    if len(received) != 1:
+        print(f"FAIL: expected 1 event before unsubscribe, got {len(received)}", file=sys.stderr)
+        sys.exit(1)
+
+    # Unsubscribe and verify success
+    result = bus.unsubscribe(sub_id)
+    if result is not True:
+        print(f"FAIL: expected True from unsubscribe, got {result}", file=sys.stderr)
+        sys.exit(1)
+
+    # Emit again and verify handler was NOT called
+    bus.emit(DomainEvent(event_type=EventType.PLAN_CREATED))
+    if len(received) != 1:
+        print(f"FAIL: handler should not be called after unsubscribe, got {len(received)}", file=sys.stderr)
+        sys.exit(1)
+
+    # Unsubscribe again with same ID should return False (already removed)
+    result = bus.unsubscribe(sub_id)
+    if result is not False:
+        print(f"FAIL: expected False from double-unsubscribe, got {result}", file=sys.stderr)
+        sys.exit(1)
+
+    print("reactive-unsubscribe-ok")
+
+
+def logging_unsubscribe() -> None:
+    """Verify LoggingEventBus.unsubscribe removes specific handler."""
+    received: list[DomainEvent] = []
+    bus = LoggingEventBus()
+    sub_id = bus.subscribe(EventType.DECISION_CREATED, received.append)
+
+    if not isinstance(sub_id, str) or len(sub_id) == 0:
+        print(f"FAIL: expected non-empty subscription ID, got {sub_id!r}", file=sys.stderr)
+        sys.exit(1)
+
+    bus.emit(DomainEvent(event_type=EventType.DECISION_CREATED))
+    if len(received) != 1:
+        print(f"FAIL: expected 1 event before unsubscribe, got {len(received)}", file=sys.stderr)
+        sys.exit(1)
+
+    result = bus.unsubscribe(sub_id)
+    if result is not True:
+        print(f"FAIL: expected True from unsubscribe, got {result}", file=sys.stderr)
+        sys.exit(1)
+
+    bus.emit(DomainEvent(event_type=EventType.DECISION_CREATED))
+    if len(received) != 1:
+        print(f"FAIL: handler should not be called after unsubscribe, got {len(received)}", file=sys.stderr)
+        sys.exit(1)
+
+    # Unsubscribe with invalid ID should return False
+    result = bus.unsubscribe("nonexistent-id")
+    if result is not False:
+        print(f"FAIL: expected False for unknown ID, got {result}", file=sys.stderr)
+        sys.exit(1)
+
+    print("logging-unsubscribe-ok")
+
+
 def decision_service_emits_event() -> None:
     """Verify DecisionService emits DECISION_CREATED via injected bus."""
     import sys
@@ -200,6 +272,8 @@ _COMMANDS = {
     "reactive_emit_subscribe": reactive_emit_subscribe,
     "logging_emit_subscribe": logging_emit_subscribe,
     "reactive_type_filtering": reactive_type_filtering,
+    "reactive_unsubscribe": reactive_unsubscribe,
+    "logging_unsubscribe": logging_unsubscribe,
     "decision_service_emits_event": decision_service_emits_event,
     "plan_lifecycle_emits_event": plan_lifecycle_emits_event,
     "container_event_bus_singleton": container_event_bus_singleton,

src/cleveragents/infrastructure/events/__init__.py

@@ -7,6 +7,7 @@ This package provides:
 - :class:`EventBus` — structural Protocol that implementations satisfy.
 - :class:`ReactiveEventBus` — RxPY-backed in-process bus.
 - :class:`LoggingEventBus` — structured-log-only bus.
+- :type: EventSubID — subscription ID type alias (ULID string).
 
 Typical usage::
 
@@ -18,19 +19,21 @@ Typical usage::
     )
 
     bus = ReactiveEventBus()
-    bus.subscribe(EventType.DECISION_CREATED, lambda e: print(e))
+    sub_id = bus.subscribe(EventType.DECISION_CREATED, lambda e: print(e))
     bus.emit(DomainEvent(event_type=EventType.DECISION_CREATED))
+    bus.unsubscribe(sub_id)
 """
 
 from cleveragents.infrastructure.events.logging_bus import LoggingEventBus
 from cleveragents.infrastructure.events.models import DomainEvent
-from cleveragents.infrastructure.events.protocol import EventBus
+from cleveragents.infrastructure.events.protocol import EventBus, EventSubID
 from cleveragents.infrastructure.events.reactive import ReactiveEventBus
 from cleveragents.infrastructure.events.types import EventType
 
 __all__ = [
     "DomainEvent",
     "EventBus",
+    "EventSubID",
     "EventType",
     "LoggingEventBus",
     "ReactiveEventBus",

src/cleveragents/infrastructure/events/logging_bus.py

@@ -16,9 +16,11 @@ from __future__ import annotations
 from collections.abc import Callable
 
 import structlog
+from ulid import ULID
 
 from cleveragents.infrastructure.events.models import DomainEvent
 from cleveragents.infrastructure.events.types import EventType
+from cleveragents.infrastructure.events.protocol import EventSubID
 
 _logger = structlog.get_logger(__name__)
 
@@ -43,7 +45,8 @@ class LoggingEventBus:
     """
 
     def __init__(self) -> None:
-        self._subscriptions: dict[EventType, list[Callable[[DomainEvent], None]]] = {}
+        # Per-event-type handler lists, each entry a (sub_id, handler) tuple
+        self._subscriptions: dict[EventType, list[tuple[str, Callable[[DomainEvent], None]]]] = {}
 
     # ------------------------------------------------------------------
     # Public API (satisfies EventBus protocol)
@@ -70,7 +73,7 @@ class LoggingEventBus:
             actor_name=event.actor_name,
             details=event.details,
         )
-        for handler in list(self._subscriptions.get(event.event_type, ())):
+        for _, handler in list(self._subscriptions.get(event.event_type, ())):
             try:
                 handler(event)
             except Exception as exc:
@@ -86,13 +89,17 @@ class LoggingEventBus:
         self,
         event_type: EventType,
         handler: Callable[[DomainEvent], None],
-    ) -> None:
+    ) -> str:
         """Register *handler* for events of *event_type*.
 
         Args:
             event_type: The :class:`EventType` to listen for.
             handler: Callable invoked for each matching :class:`DomainEvent`.
 
+        Returns:
+            A subscription ID (ULID string) that can be used with
+            :meth:`unsubscribe` to remove this specific handler.
+
         Raises:
             TypeError: If *event_type* is not an :class:`EventType`.
             TypeError: If *handler* is not callable.
@@ -103,7 +110,33 @@ class LoggingEventBus:
             )
         if not callable(handler):
             raise TypeError("handler must be callable")
-        self._subscriptions.setdefault(event_type, []).append(handler)
+        sub_id = str(ULID())
+        self._subscriptions.setdefault(event_type, []).append((sub_id, handler))
+        return sub_id
+
+    def unsubscribe(self, subscription_id: EventSubID) -> bool:
+        """Remove a previously registered subscription.
+
+        Only removes the matching *handler* for a specific event type —
+        other handlers registered for events of the same :class:`EventType`
+        remain untouched.
+
+        Args:
+            subscription_id: The ULID string returned by :meth:`subscribe`.
+
+        Returns:
+            ``True`` if the subscription was found and removed, ``False``
+            otherwise (e.g. invalid or already-removed ID).
+        """
+        for event_type_list in self._subscriptions.values():
+            index = next(
+                (i for i, (sid, _) in enumerate(event_type_list) if sid == subscription_id),
+                None,
+            )
+            if index is not None:
+                event_type_list.pop(index)
+                return True
+        return False
 
 
 __all__ = ["LoggingEventBus"]

src/cleveragents/infrastructure/events/protocol.py

@@ -18,6 +18,9 @@ from typing import Protocol, runtime_checkable
 from cleveragents.infrastructure.events.models import DomainEvent
 from cleveragents.infrastructure.events.types import EventType
 
+#: Unique identifier for an event bus subscription, as returned by :meth:`EventBus.subscribe`.
+EventSubID = str
+
 
 @runtime_checkable
 class EventBus(Protocol):
@@ -40,14 +43,30 @@ class EventBus(Protocol):
         self,
         event_type: EventType,
         handler: Callable[[DomainEvent], None],
-    ) -> None:
+    ) -> EventSubID:
         """Register *handler* to receive events of *event_type*.
 
         Args:
             event_type: The :class:`EventType` to listen for.
             handler: Callable invoked synchronously for each matching event.
+
+        Returns:
+            A subscription ID (:class:`~EventSubID`) that can be used with
+            :meth:`unsubscribe` to remove this specific handler.
+        """
+        ...
+
+    def unsubscribe(self, subscription_id: EventSubID) -> bool:
+        """Remove a previously registered subscription.
+
+        Args:
+            subscription_id: The ULID string returned by :meth:`subscribe`.
+
+        Returns:
+            ``True`` if the subscription was found and removed, ``False``
+            otherwise (e.g. invalid or already-removed ID).
         """
         ...
 
 
-__all__ = ["EventBus"]
+__all__ = ["EventBus", "EventSubID"]

src/cleveragents/infrastructure/events/reactive.py

@@ -27,9 +27,11 @@ import structlog
 from rx import operators as ops
 from rx.core.observable.observable import Observable
 from rx.subject.subject import Subject
+from ulid import ULID
 
 from cleveragents.infrastructure.events.models import DomainEvent
 from cleveragents.infrastructure.events.types import EventType
+from cleveragents.infrastructure.events.protocol import EventSubID
 
 _logger = structlog.get_logger(__name__)
 
@@ -49,7 +51,8 @@ class ReactiveEventBus:
     Attributes:
         _subject: Hot RxPY Subject — the raw observable stream.
         _stream: Read-only Observable view over ``_subject``.
-        _subscriptions: Per-event-type handler lists.
+        _subscriptions: Per-event-type handler lists, each entry a list of
+            ``(sub_id, handler)`` tuples enabling individual unsubscription.
         _audit_log: Volatile in-memory log of all emitted events.
         _closed: Whether :meth:`close` has been called.
     """
@@ -60,7 +63,8 @@ class ReactiveEventBus:
 
         self._subject: Subject = Subject()
         self._stream: Observable = self._subject.pipe(ops.map(_identity))
-        self._subscriptions: dict[EventType, list[Callable[[DomainEvent], None]]] = {}
+        # Changed: subscriptions now map event_type -> list of (sub_id, handler)
+        self._subscriptions: dict[EventType, list[tuple[str, Callable[[DomainEvent], None]]]] = {}
         self._audit_log: deque[DomainEvent] = deque(maxlen=max_audit_log_size)
         self._closed: bool = False
 
@@ -135,7 +139,7 @@ class ReactiveEventBus:
 
         self._audit_log.append(event.model_copy(deep=True))
 
-        for handler in list(self._subscriptions.get(event.event_type, ())):
+        for _, handler in list(self._subscriptions.get(event.event_type, ())):
             try:
                 handler(event)
             except Exception as exc:
@@ -152,7 +156,7 @@ class ReactiveEventBus:
         self,
         event_type: EventType,
         handler: Callable[[DomainEvent], None],
-    ) -> None:
+    ) -> str:
         """Register *handler* for events of *event_type*.
 
         Handlers are called synchronously inside :meth:`emit` in the order
@@ -162,6 +166,10 @@ class ReactiveEventBus:
             event_type: The :class:`EventType` to listen for.
             handler: Callable invoked for each matching :class:`DomainEvent`.
 
+        Returns:
+            A subscription ID (ULID string) that can be used with
+            :meth:`unsubscribe` to remove this specific handler.
+
         Raises:
             TypeError: If *event_type* is not an :class:`EventType`.
             TypeError: If *handler* is not callable.
@@ -172,7 +180,33 @@ class ReactiveEventBus:
             )
         if not callable(handler):
             raise TypeError("handler must be callable")
-        self._subscriptions.setdefault(event_type, []).append(handler)
+        sub_id = str(ULID())
+        self._subscriptions.setdefault(event_type, []).append((sub_id, handler))
+        return sub_id
+
+    def unsubscribe(self, subscription_id: EventSubID) -> bool:
+        """Remove a previously registered subscription.
+
+        Only removes the matching ``handler`` for a specific event type —
+        other handlers registered for events of the same :class:`EventType`
+        remain untouched.
+
+        Args:
+            subscription_id: The ULID string returned by :meth:`subscribe`.
+
+        Returns:
+            ``True`` if the subscription was found and removed, ``False``
+            otherwise (e.g. invalid or already-removed ID).
+        """
+        for event_type_list in self._subscriptions.values():
+            index = next(
+                (i for i, (sid, _handler) in enumerate(event_type_list) if sid == subscription_id),
+                None,
+            )
+            if index is not None:
+                event_type_list.pop(index)
+                return True
+        return False
 
     @property
     def stream(self) -> Observable: