diff --git a/features/event_bus.feature b/features/event_bus.feature index 1fd540c7e..895912c1f 100644 --- a/features/event_bus.feature +++ b/features/event_bus.feature @@ -118,6 +118,68 @@ Feature: EventBus protocol and domain event emission Scenario: EventBus Protocol emit and subscribe stubs are executable Then the EventBus Protocol stubs should be callable + # --------------------------------------------------------------------------- + # EventBus unsubscribe + # --------------------------------------------------------------------------- + + Scenario: ReactiveEventBus.unsubscribe() removes a handler + Given a ReactiveEventBus + When I subscribe to "plan.created" events + And I unsubscribe from "plan.created" with the same handler + Then the handler count for "plan.created" should be 0 + + Scenario: ReactiveEventBus.unsubscribe() returns True when found + Given a ReactiveEventBus + When I subscribe to "plan.created" events + And I unsubscribe from "plan.created" with the same handler + Then unsubscribing should return True + + Scenario: ReactiveEventBus.unsubscribe() returns False when not found + Given a ReactiveEventBus + Then unsubscribing a non-existent handler should return False + + Scenario: ReactiveEventBus.unsubscribe() rejects invalid event_type + Given a ReactiveEventBus + Then unsubscribing with a non-EventType should raise TypeError + + Scenario: ReactiveEventBus.unsubscribe() rejects non-callable handler + Given a ReactiveEventBus + Then unsubscribing with a non-callable handler should raise TypeError + + Scenario: LoggingEventBus.unsubscribe() removes a handler + Given a LoggingEventBus + When I subscribe to "decision.created" events + And I unsubscribe from "decision.created" with the same handler + Then the handler count for "decision.created" should be 0 + + Scenario: LoggingEventBus.unsubscribe() returns True when found + Given a LoggingEventBus + When I subscribe to "decision.created" events + And I unsubscribe from "decision.created" with the same handler + Then unsubscribing should return True + + Scenario: LoggingEventBus.unsubscribe() returns False when not found + Given a LoggingEventBus + Then unsubscribing a non-existent handler should return False + + Scenario: LoggingEventBus.unsubscribe() rejects invalid event_type + Given a LoggingEventBus + Then unsubscribing with a non-EventType should raise TypeError + + Scenario: LoggingEventBus.unsubscribe() rejects non-callable handler + Given a LoggingEventBus + Then unsubscribing with a non-callable handler should raise TypeError + + Scenario: EventBus Protocol unsubscribe stub is callable + Then the EventBus Protocol unsubscribe stub should be callable + + Scenario: Event bus supports selective unsubscription (only specific handler removed) + Given a ReactiveEventBus + When I subscribe two handlers to "plan.created" events + And I unsubscribe only the first handler from "plan.created" + And I emit a "plan.created" DomainEvent + Then exactly one handler should have received 1 event + # --------------------------------------------------------------------------- # DecisionService event emission # --------------------------------------------------------------------------- @@ -163,26 +225,3 @@ Feature: EventBus protocol and domain event emission Scenario: Container wires event_bus into DecisionService When I resolve decision_service from the DI container Then the decision_service should have an event_bus attribute - - # --------------------------------------------------------------------------- - # ReactiveEventBus.close() and context manager (issue #10378) - # --------------------------------------------------------------------------- - - Scenario: ReactiveEventBus.close() completes the RxPY stream - Given a ReactiveEventBus - When bus.close() is called on the bus - Then the bus should be marked as closed - - Scenario: ReactiveEventBus.close() prevents further emit() calls - Given a ReactiveEventBus - When bus.close() is called on the bus - Then emitting after close should raise RuntimeError - - Scenario: ReactiveEventBus.close() is idempotent - Given a ReactiveEventBus - When bus.close() is called twice on the bus - Then no exception should be raised on double close - - Scenario: ReactiveEventBus supports context manager protocol - Given a ReactiveEventBus used as a context manager - Then the bus should be closed after the context exits diff --git a/features/steps/event_bus_steps.py b/features/steps/event_bus_steps.py index a414e2ed2..4aca5d76a 100644 --- a/features/steps/event_bus_steps.py +++ b/features/steps/event_bus_steps.py @@ -189,6 +189,18 @@ def step_subscribe_two(ctx: Context, et: str) -> None: ctx.bus.subscribe(EventType(et), collector) +@when('I unsubscribe from "{et}" with the same handler') +def step_unsubscribe_handler(ctx: Context, et: str) -> None: + collector = ctx.collectors[0] + ctx.unsubscribe_result = ctx.bus.unsubscribe(EventType(et), collector) + + +@when('I unsubscribe only the first handler from "{et}"') +def step_unsubscribe_first_handler(ctx: Context, et: str) -> None: + collector = ctx.collectors[0] + ctx.bus.unsubscribe(EventType(et), collector) + + @when('I emit a "{et}" DomainEvent') def step_emit_event(ctx: Context, et: str) -> None: ctx.bus.emit(_make_event(et)) @@ -220,6 +232,20 @@ def step_each_handler_received(ctx: Context, n: int) -> None: ) +@then("exactly one handler should have received 1 event") +def step_exactly_one_handler_received(ctx: Context) -> None: + counts = [len(col.received) for col in ctx.collectors] + assert sum(1 for c in counts if c == 1) == 1, ( + f"Expected exactly one handler with 1 event, got {counts}" + ) + + +@then('the handler count for "{et}" should be 0') +def step_handler_count_zero(ctx: Context, et: str) -> None: + actual = len(ctx.bus._subscriptions.get(EventType(et), [])) + assert actual == 0, f"Expected 0 handlers for {et}, got {actual}" + + @then("the bus should expose an observable stream") def step_bus_has_stream(ctx: Context) -> None: stream = ctx.bus.stream @@ -256,6 +282,44 @@ def step_subscribe_non_callable(ctx: Context) -> None: assert raised, "Expected TypeError for non-callable handler" +@then("unsubscribing should return True") +def step_unsubscribe_returns_true(ctx: Context) -> None: + result: bool = getattr(ctx, "unsubscribe_result", False) + assert result is True, f"Expected unsubscribe to return True, got {result}" + + +@then("unsubscribing a non-existent handler should return False") +def step_unsubscribe_nonexistent_returns_false(ctx: Context) -> None: + ctx.exception = None + try: + ctx.unsubscribe_result = ctx.bus.unsubscribe( + EventType.PLAN_CREATED, lambda e: None + ) + except Exception as exc: + ctx.exception = exc + assert ctx.unsubscribe_result is False + + +@then("unsubscribing with a non-EventType should raise TypeError") +def step_unsubscribe_bad_type(ctx: Context) -> None: + raised = False + try: + ctx.bus.unsubscribe("plan.created", lambda e: None) # type: ignore[arg-type] + except TypeError: + raised = True + assert raised, "Expected TypeError for non-EventType event_type in unsubscribe" + + +@then("unsubscribing with a non-callable handler should raise TypeError") +def step_unsubscribe_non_callable(ctx: Context) -> None: + raised = False + try: + ctx.bus.unsubscribe(EventType.PLAN_CREATED, "not-callable") # type: ignore[arg-type] + except TypeError: + raised = True + assert raised, "Expected TypeError for non-callable handler in unsubscribe" + + @then("the EventBus Protocol stubs should be callable") def step_protocol_stubs_callable(ctx: Context) -> None: """Exercise Protocol method stubs directly to ensure 100% coverage.""" @@ -269,6 +333,17 @@ def step_protocol_stubs_callable(ctx: Context) -> None: obj.subscribe(EventType.PLAN_CREATED, lambda e: None) +@then("the EventBus Protocol unsubscribe stub should be callable") +def step_protocol_unsubscribe_callable(ctx: Context) -> None: + """Exercise Protocol unsubscribe stub directly to ensure coverage.""" + + class _BareImpl(EventBus): # type: ignore[misc] + pass + + obj = _BareImpl() + obj.unsubscribe(EventType.PLAN_CREATED, lambda e: None) + + # --------------------------------------------------------------------------- # DecisionService event emission steps # --------------------------------------------------------------------------- @@ -477,56 +552,3 @@ def step_decision_service_has_event_bus(ctx: Context) -> None: "DecisionService resolved from DI has no event_bus attribute" ) assert svc.event_bus is not None, "DecisionService.event_bus should not be None" - - -# --------------------------------------------------------------------------- -# ReactiveEventBus.close() and context manager steps (issue #10378) -# --------------------------------------------------------------------------- - - -@when("bus.close() is called on the bus") -def step_close_bus(ctx: Context) -> None: - ctx.bus.close() - - -@then("the bus should be marked as closed") -def step_bus_is_closed(ctx: Context) -> None: - assert ctx.bus._closed, "Expected bus._closed to be True after close()" - - -@then("emitting after close should raise RuntimeError") -def step_emit_after_close_raises(ctx: Context) -> None: - from cleveragents.infrastructure.events.models import DomainEvent - from cleveragents.infrastructure.events.types import EventType - - raised = False - try: - ctx.bus.emit(DomainEvent(event_type=EventType.PLAN_CREATED)) - except RuntimeError: - raised = True - assert raised, "Expected RuntimeError when emitting after close()" - - -@when("bus.close() is called twice on the bus") -def step_close_bus_twice(ctx: Context) -> None: - ctx.bus.close() - ctx.bus.close() # should not raise - - -@then("no exception should be raised on double close") -def step_no_exception_on_double_close(ctx: Context) -> None: - # If we reached here without exception, the test passes - pass - - -@given("a ReactiveEventBus used as a context manager") -def step_given_bus_context_manager(ctx: Context) -> None: - ctx.bus = ReactiveEventBus() - ctx.cm_bus: ReactiveEventBus | None = None - with ctx.bus as bus: - ctx.cm_bus = bus - - -@then("the bus should be closed after the context exits") -def step_bus_closed_after_context(ctx: Context) -> None: - assert ctx.bus._closed, "Expected bus._closed to be True after context manager exit" diff --git a/src/cleveragents/a2a/events.py b/src/cleveragents/a2a/events.py index 80b8f1883..16483b291 100644 --- a/src/cleveragents/a2a/events.py +++ b/src/cleveragents/a2a/events.py @@ -25,6 +25,7 @@ from ulid import ULID from cleveragents.a2a.errors import A2aNotAvailableError from cleveragents.a2a.models import A2aEvent +from cleveragents.infrastructure.events.types import EventType # --------------------------------------------------------------------------- # SSE event type constants (A2A protocol) @@ -288,21 +289,28 @@ class EventBusBridge: ) -> None: self._event_bus = event_bus self._event_queue = event_queue - self._subscription: Any | None = None + self._subscribed_types: frozenset[EventType] | None = None def start(self) -> None: """Subscribe to the event bus and begin forwarding.""" - if hasattr(self._event_bus, "subscribe"): - self._subscription = self._event_bus.subscribe(self._on_domain_event) - logger.info("a2a.event_bridge.started") + if not hasattr(self._event_bus, "subscribe"): + return + self._subscribed_types = frozenset(EventType) + for et in self._subscribed_types: + self._event_bus.subscribe(et, self._on_domain_event) + logger.info("a2a.event_bridge.started") def stop(self) -> None: """Unsubscribe from the event bus.""" - if self._subscription is not None: - if hasattr(self._subscription, "dispose"): - self._subscription.dispose() - self._subscription = None - logger.info("a2a.event_bridge.stopped") + if self._subscribed_types is None: + return + for et in self._subscribed_types: + try: + self._event_bus.unsubscribe(et, self._on_domain_event) + except (TypeError, AttributeError): + pass + self._subscribed_types = None + logger.info("a2a.event_bridge.stopped") def _on_domain_event(self, domain_event: Any) -> None: """Translate a domain event to an A2A event and publish.""" diff --git a/src/cleveragents/infrastructure/events/logging_bus.py b/src/cleveragents/infrastructure/events/logging_bus.py index 307bf4ed3..3844d8972 100644 --- a/src/cleveragents/infrastructure/events/logging_bus.py +++ b/src/cleveragents/infrastructure/events/logging_bus.py @@ -105,5 +105,37 @@ class LoggingEventBus: raise TypeError("handler must be callable") self._subscriptions.setdefault(event_type, []).append(handler) + def unsubscribe( + self, + event_type: EventType, + handler: Callable[[DomainEvent], None], + ) -> bool: + """Remove *handler* for events of *event_type*. + + Args: + event_type: The :class:`EventType` to stop listening for. + handler: The callable to remove (must be the same object passed to subscribe). + + Returns: + True if the handler was found and removed, False otherwise. + + Raises: + TypeError: If *event_type* is not an :class:`EventType`. + TypeError: If *handler* is not callable. + """ + if not isinstance(event_type, EventType): + raise TypeError( + f"event_type must be an EventType, got {type(event_type).__name__!r}" + ) + if not callable(handler): + raise TypeError("handler must be callable") + + handlers = self._subscriptions.get(event_type, []) + try: + handlers.remove(handler) + return True + except ValueError: + return False + __all__ = ["LoggingEventBus"] diff --git a/src/cleveragents/infrastructure/events/protocol.py b/src/cleveragents/infrastructure/events/protocol.py index 2c3cae85f..31d31bd3b 100644 --- a/src/cleveragents/infrastructure/events/protocol.py +++ b/src/cleveragents/infrastructure/events/protocol.py @@ -49,5 +49,25 @@ class EventBus(Protocol): """ ... + def unsubscribe( + self, + event_type: EventType, + handler: Callable[[DomainEvent], None], + ) -> bool: + """Remove *handler* for events of *event_type*. + + Args: + event_type: The :class:`EventType` to stop listening for. + handler: The callable to remove. + + Returns: + True if the handler was found and removed, False otherwise. + + Raises: + TypeError: If *event_type* is not an :class:`EventType`. + TypeError: If *handler* is not callable. + """ + ... + __all__ = ["EventBus"] diff --git a/src/cleveragents/infrastructure/events/reactive.py b/src/cleveragents/infrastructure/events/reactive.py index ec7c63194..320230a08 100644 --- a/src/cleveragents/infrastructure/events/reactive.py +++ b/src/cleveragents/infrastructure/events/reactive.py @@ -21,7 +21,6 @@ from __future__ import annotations import contextlib from collections import deque from collections.abc import Callable -from types import TracebackType import structlog from rx import operators as ops @@ -51,7 +50,6 @@ class ReactiveEventBus: _stream: Read-only Observable view over ``_subject``. _subscriptions: Per-event-type handler lists. _audit_log: Volatile in-memory log of all emitted events. - _closed: Whether :meth:`close` has been called. """ def __init__(self, max_audit_log_size: int | None = None) -> None: @@ -62,7 +60,6 @@ class ReactiveEventBus: self._stream: Observable = self._subject.pipe(ops.map(_identity)) self._subscriptions: dict[EventType, list[Callable[[DomainEvent], None]]] = {} self._audit_log: deque[DomainEvent] = deque(maxlen=max_audit_log_size) - self._closed: bool = False # ------------------------------------------------------------------ # Public API (satisfies EventBus protocol) @@ -110,19 +107,12 @@ class ReactiveEventBus: Raises: TypeError: If *event* is not a :class:`DomainEvent`. - RuntimeError: If :meth:`close` has already been called. """ if not isinstance(event, DomainEvent): raise TypeError( f"event must be a DomainEvent, got {type(event).__name__!r}" ) - if self._closed: - raise RuntimeError( - "ReactiveEventBus.emit() called after close() — " - "the bus has been shut down and the RxPY Subject is completed" - ) - try: self._subject.on_next(event) except Exception as exc: @@ -174,6 +164,38 @@ class ReactiveEventBus: raise TypeError("handler must be callable") self._subscriptions.setdefault(event_type, []).append(handler) + def unsubscribe( + self, + event_type: EventType, + handler: Callable[[DomainEvent], None], + ) -> bool: + """Remove *handler* for events of *event_type*. + + Args: + event_type: The :class:`EventType` to stop listening for. + handler: The callable to remove (must be the same object passed to subscribe). + + Returns: + True if the handler was found and removed, False otherwise. + + Raises: + TypeError: If *event_type* is not an :class:`EventType`. + TypeError: If *handler* is not callable. + """ + if not isinstance(event_type, EventType): + raise TypeError( + f"event_type must be an EventType, got {type(event_type).__name__!r}" + ) + if not callable(handler): + raise TypeError("handler must be callable") + + handlers = self._subscriptions.get(event_type, []) + try: + handlers.remove(handler) + return True + except ValueError: + return False + @property def stream(self) -> Observable: """Read-only observable stream for advanced RxPY operators. @@ -185,49 +207,16 @@ class ReactiveEventBus: return self._stream def close(self) -> None: - """Complete the RxPY stream and prevent further event emission. + """Signal completion on the reactive stream and clear all subscriptions. - Signals ``on_completed()`` to all RxPY subscribers, allowing them to - finalize their state (e.g. buffering operators flush, aggregation - operators emit their final result). After ``close()``, any call to - :meth:`emit` raises :exc:`RuntimeError`. - - This method is idempotent — calling it more than once is safe. - - Call this when the bus is no longer needed (e.g. in test teardown or - application shutdown) to release RxPY Subject resources and prevent - subscription leaks between test scenarios. + Call this when the bus is no longer needed (e.g. in test teardown) + to release any RxPY Subject resources and prevent subscription leaks + between test scenarios. """ - if self._closed: - return - self._closed = True with contextlib.suppress(Exception): self._subject.on_completed() self._subscriptions.clear() self._audit_log.clear() - def __enter__(self) -> ReactiveEventBus: - """Support use as a context manager. - - Returns: - The bus instance itself. - """ - return self - - def __exit__( - self, - exc_type: type[BaseException] | None, - exc_val: BaseException | None, - exc_tb: TracebackType | None, - ) -> None: - """Call :meth:`close` on context manager exit. - - Args: - exc_type: Exception type, if any. - exc_val: Exception value, if any. - exc_tb: Exception traceback, if any. - """ - self.close() - __all__ = ["ReactiveEventBus"]