diff --git a/features/mocks/uko_indexer_mocks.py b/features/mocks/uko_indexer_mocks.py index 1717f87be..96d3bfc00 100644 --- a/features/mocks/uko_indexer_mocks.py +++ b/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) diff --git a/robot/helper_event_bus.py b/robot/helper_event_bus.py index 2c6b9179c..e7cf6dc58 100644 --- a/robot/helper_event_bus.py +++ b/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, diff --git a/src/cleveragents/infrastructure/events/__init__.py b/src/cleveragents/infrastructure/events/__init__.py index f05d55c81..362f2ef43 100644 --- a/src/cleveragents/infrastructure/events/__init__.py +++ b/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", diff --git a/src/cleveragents/infrastructure/events/logging_bus.py b/src/cleveragents/infrastructure/events/logging_bus.py index 307bf4ed3..e5c8c23ab 100644 --- a/src/cleveragents/infrastructure/events/logging_bus.py +++ b/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"] diff --git a/src/cleveragents/infrastructure/events/protocol.py b/src/cleveragents/infrastructure/events/protocol.py index 2c3cae85f..821bd6c12 100644 --- a/src/cleveragents/infrastructure/events/protocol.py +++ b/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"] diff --git a/src/cleveragents/infrastructure/events/reactive.py b/src/cleveragents/infrastructure/events/reactive.py index ec7c63194..1d7acdad0 100644 --- a/src/cleveragents/infrastructure/events/reactive.py +++ b/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: