From d506de50a92b93de533e08550ed47fb9c2a4002a Mon Sep 17 00:00:00 2001 From: CleverThis Date: Thu, 14 May 2026 05:24:49 +0000 Subject: [PATCH 1/8] fix(events): add unsubscribe() to EventBus protocol and implementations --- features/event_bus.feature | 85 ++++++++---- features/steps/event_bus_steps.py | 128 ++++++++++-------- src/cleveragents/a2a/events.py | 26 ++-- .../infrastructure/events/logging_bus.py | 32 +++++ .../infrastructure/events/protocol.py | 20 +++ .../infrastructure/events/reactive.py | 83 +++++------- 6 files changed, 242 insertions(+), 132 deletions(-) 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 f17ae21e7..7b410c011 100644 --- a/src/cleveragents/a2a/events.py +++ b/src/cleveragents/a2a/events.py @@ -24,6 +24,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) @@ -260,21 +261,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"] -- 2.52.0 From 6d429b127683c87aa16594c0d46ee9aa3b3c3a6c Mon Sep 17 00:00:00 2001 From: CleverThis Bot Date: Fri, 15 May 2026 01:44:13 +0000 Subject: [PATCH 2/8] fix(ci): resolve ruff SIM105, E501, W291 lint errors in PR #11197 changes - events.py: add missing contextlib import, use contextlib.suppress for TypeError/AttributeError try/except (SIM105) - reactive.py: fix docstring line length (E501), remove trailing whitespace (W291) - logging_bus.py: fix docstring line length (E501), remove trailing whitespace (W291) --- src/cleveragents/a2a/events.py | 5 ++--- src/cleveragents/infrastructure/events/logging_bus.py | 3 ++- src/cleveragents/infrastructure/events/reactive.py | 3 ++- 3 files changed, 6 insertions(+), 5 deletions(-) diff --git a/src/cleveragents/a2a/events.py b/src/cleveragents/a2a/events.py index 7b410c011..431f945f6 100644 --- a/src/cleveragents/a2a/events.py +++ b/src/cleveragents/a2a/events.py @@ -15,6 +15,7 @@ publishes translated :class:`A2aEvent` instances to an event queue. from __future__ import annotations +import contextlib import json from collections.abc import Callable from typing import Any, ClassVar @@ -277,10 +278,8 @@ class EventBusBridge: if self._subscribed_types is None: return for et in self._subscribed_types: - try: + with contextlib.suppress(TypeError, AttributeError): self._event_bus.unsubscribe(et, self._on_domain_event) - except (TypeError, AttributeError): - pass self._subscribed_types = None logger.info("a2a.event_bridge.stopped") diff --git a/src/cleveragents/infrastructure/events/logging_bus.py b/src/cleveragents/infrastructure/events/logging_bus.py index 3844d8972..b1b604dea 100644 --- a/src/cleveragents/infrastructure/events/logging_bus.py +++ b/src/cleveragents/infrastructure/events/logging_bus.py @@ -114,7 +114,8 @@ class LoggingEventBus: Args: event_type: The :class:`EventType` to stop listening for. - handler: The callable to remove (must be the same object passed to subscribe). + handler: Callable to remove (same object passed to + :meth:`subscribe`). Returns: True if the handler was found and removed, False otherwise. diff --git a/src/cleveragents/infrastructure/events/reactive.py b/src/cleveragents/infrastructure/events/reactive.py index 320230a08..e0dfd86a7 100644 --- a/src/cleveragents/infrastructure/events/reactive.py +++ b/src/cleveragents/infrastructure/events/reactive.py @@ -173,7 +173,8 @@ class ReactiveEventBus: Args: event_type: The :class:`EventType` to stop listening for. - handler: The callable to remove (must be the same object passed to subscribe). + handler: Callable to remove (same object passed to + :meth:`subscribe`). Returns: True if the handler was found and removed, False otherwise. -- 2.52.0 From fa4337dd1ccc6876004cd111d581735bb96a7416 Mon Sep 17 00:00:00 2001 From: CleverThis Date: Fri, 15 May 2026 11:09:43 +0000 Subject: [PATCH 3/8] fix(events): eliminate inline type ignores in event_bus_steps.py Replace all # type: ignore[arg-type] and # type: ignore[misc] comments with pyright-compatible alternatives: typing.cast() for intentional type mismatches in error-handling tests, and explicit stub method bodies for Protocol subclasses. --- features/steps/event_bus_steps.py | 39 +++++++++++++++++++++++-------- 1 file changed, 29 insertions(+), 10 deletions(-) diff --git a/features/steps/event_bus_steps.py b/features/steps/event_bus_steps.py index 4aca5d76a..371fc9ad9 100644 --- a/features/steps/event_bus_steps.py +++ b/features/steps/event_bus_steps.py @@ -6,7 +6,8 @@ LoggingEventBus, service event emission, and DI container registration. from __future__ import annotations -from typing import Any +from collections.abc import Callable +from typing import Any, cast from unittest.mock import MagicMock from behave import given, then, when # type: ignore[import-untyped] @@ -256,7 +257,7 @@ def step_bus_has_stream(ctx: Context) -> None: def step_emit_non_domain_event(ctx: Context) -> None: raised = False try: - ctx.bus.emit("not-an-event") # type: ignore[arg-type] + ctx.bus.emit(cast(DomainEvent, "not-an-event")) except TypeError: raised = True assert raised, "Expected TypeError for non-DomainEvent" @@ -266,7 +267,7 @@ def step_emit_non_domain_event(ctx: Context) -> None: def step_subscribe_bad_type(ctx: Context) -> None: raised = False try: - ctx.bus.subscribe("plan.created", lambda e: None) # type: ignore[arg-type] + ctx.bus.subscribe(cast(EventType, "plan.created"), lambda e: None) except TypeError: raised = True assert raised, "Expected TypeError for non-EventType event_type" @@ -276,7 +277,9 @@ def step_subscribe_bad_type(ctx: Context) -> None: def step_subscribe_non_callable(ctx: Context) -> None: raised = False try: - ctx.bus.subscribe(EventType.PLAN_CREATED, "not-callable") # type: ignore[arg-type] + ctx.bus.subscribe( + EventType.PLAN_CREATED, cast(Callable[[DomainEvent], None], "not-callable") + ) except TypeError: raised = True assert raised, "Expected TypeError for non-callable handler" @@ -304,7 +307,7 @@ def step_unsubscribe_nonexistent_returns_false(ctx: Context) -> None: def step_unsubscribe_bad_type(ctx: Context) -> None: raised = False try: - ctx.bus.unsubscribe("plan.created", lambda e: None) # type: ignore[arg-type] + ctx.bus.unsubscribe(cast(EventType, "plan.created"), lambda e: None) except TypeError: raised = True assert raised, "Expected TypeError for non-EventType event_type in unsubscribe" @@ -314,7 +317,9 @@ def step_unsubscribe_bad_type(ctx: Context) -> None: def step_unsubscribe_non_callable(ctx: Context) -> None: raised = False try: - ctx.bus.unsubscribe(EventType.PLAN_CREATED, "not-callable") # type: ignore[arg-type] + ctx.bus.unsubscribe( + EventType.PLAN_CREATED, cast(Callable[[DomainEvent], None], "not-callable") + ) except TypeError: raised = True assert raised, "Expected TypeError for non-callable handler in unsubscribe" @@ -324,8 +329,15 @@ def step_unsubscribe_non_callable(ctx: Context) -> None: def step_protocol_stubs_callable(ctx: Context) -> None: """Exercise Protocol method stubs directly to ensure 100% coverage.""" - class _BareImpl(EventBus): # type: ignore[misc] - pass + class _BareImpl(EventBus): + def emit(self, event: DomainEvent) -> None: ... + def subscribe( + self, event_type: EventType, handler: Callable[[DomainEvent], None] + ) -> None: ... + def unsubscribe( + self, event_type: EventType, handler: Callable[[DomainEvent], None] + ) -> bool: + return True obj = _BareImpl() event = _make_event("plan.created") @@ -337,8 +349,15 @@ def step_protocol_stubs_callable(ctx: Context) -> None: def step_protocol_unsubscribe_callable(ctx: Context) -> None: """Exercise Protocol unsubscribe stub directly to ensure coverage.""" - class _BareImpl(EventBus): # type: ignore[misc] - pass + class _BareImpl(EventBus): + def emit(self, event: DomainEvent) -> None: ... + def subscribe( + self, event_type: EventType, handler: Callable[[DomainEvent], None] + ) -> None: ... + def unsubscribe( + self, event_type: EventType, handler: Callable[[DomainEvent], None] + ) -> bool: + return True obj = _BareImpl() obj.unsubscribe(EventType.PLAN_CREATED, lambda e: None) -- 2.52.0 From 0054c761f608ba506273d7e6f56965a48b38fb79 Mon Sep 17 00:00:00 2001 From: controller-ci-rerun Date: Thu, 11 Jun 2026 00:15:26 -0400 Subject: [PATCH 4/8] chore: re-trigger CI [controller] -- 2.52.0 From 8db127d3d80b7b2ccfe23c43d617091e7512d4df Mon Sep 17 00:00:00 2001 From: controller-ci-rerun Date: Sat, 13 Jun 2026 09:36:35 -0400 Subject: [PATCH 5/8] chore: re-trigger CI [controller] -- 2.52.0 From 11d47c240ebabf1b8b3059941c28c62cf98ae892 Mon Sep 17 00:00:00 2001 From: controller-ci-rerun Date: Tue, 16 Jun 2026 18:18:02 -0400 Subject: [PATCH 6/8] chore: re-trigger CI [controller] -- 2.52.0 From 755c78ce366a3df90ed32e936b8a8593723c7ed5 Mon Sep 17 00:00:00 2001 From: controller-ci-rerun Date: Wed, 17 Jun 2026 14:32:46 -0400 Subject: [PATCH 7/8] chore: re-trigger CI [controller] -- 2.52.0 From 2254991776ec01c08a2ecbb947fff7d4df1c5629 Mon Sep 17 00:00:00 2001 From: CleverAgents Bot Date: Thu, 18 Jun 2026 06:17:09 -0400 Subject: [PATCH 8/8] fix(events): restore event bus close and bridge semantics --- src/cleveragents/a2a/events.py | 28 ++-- .../services/fix_then_revalidate.py | 5 +- .../infrastructure/events/reactive.py | 40 +++++- tests/events/test_event_bus_semantics.py | 133 ++++++++++++++++++ 4 files changed, 191 insertions(+), 15 deletions(-) create mode 100644 tests/events/test_event_bus_semantics.py diff --git a/src/cleveragents/a2a/events.py b/src/cleveragents/a2a/events.py index 431f945f6..9e9d36dcd 100644 --- a/src/cleveragents/a2a/events.py +++ b/src/cleveragents/a2a/events.py @@ -262,25 +262,35 @@ 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 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) + subscribed_types = frozenset(EventType) + try: + for et in subscribed_types: + self._event_bus.subscribe(et, self._on_domain_event) + except TypeError: + self._subscribed_types = None + self._subscription = self._event_bus.subscribe(self._on_domain_event) + else: + self._subscribed_types = subscribed_types logger.info("a2a.event_bridge.started") def stop(self) -> None: """Unsubscribe from the event bus.""" - if self._subscribed_types is None: - return - for et in self._subscribed_types: - with contextlib.suppress(TypeError, AttributeError): - self._event_bus.unsubscribe(et, self._on_domain_event) - self._subscribed_types = None + if self._subscribed_types is not None: + for et in self._subscribed_types: + with contextlib.suppress(TypeError, AttributeError): + self._event_bus.unsubscribe(et, self._on_domain_event) + self._subscribed_types = None + if self._subscription is not None: + if hasattr(self._subscription, "dispose"): + self._subscription.dispose() + self._subscription = None logger.info("a2a.event_bridge.stopped") def _on_domain_event(self, domain_event: Any) -> None: diff --git a/src/cleveragents/application/services/fix_then_revalidate.py b/src/cleveragents/application/services/fix_then_revalidate.py index 5af537a0f..fa91fabed 100644 --- a/src/cleveragents/application/services/fix_then_revalidate.py +++ b/src/cleveragents/application/services/fix_then_revalidate.py @@ -42,6 +42,7 @@ import threading from collections import defaultdict from collections.abc import Callable from datetime import UTC, datetime +from typing import cast import structlog from pydantic import BaseModel, ConfigDict, Field, model_validator @@ -287,14 +288,14 @@ class FixThenRevalidateOrchestrator: "auto_validation_fix must be between 0.0 and 1.0, " f"got {auto_validation_fix}" ) - if event_bus is not None and not isinstance(event_bus, EventBus): + if event_bus is not None and not callable(getattr(event_bus, "emit", None)): raise ValidationError("event_bus must be an EventBus instance or None") self._pipeline = validation_pipeline self._max_retries = max_retries self._auto_strategy_revision = float(auto_strategy_revision) self._auto_validation_fix = float(auto_validation_fix) - self._event_bus = event_bus + self._event_bus = cast(EventBus | None, event_bus) # Per-(plan_id, validation_key) retry counters. # Inner key is (validation_name, resource_id) to prevent # same-named validations on different resources from sharing diff --git a/src/cleveragents/infrastructure/events/reactive.py b/src/cleveragents/infrastructure/events/reactive.py index e0dfd86a7..03068594a 100644 --- a/src/cleveragents/infrastructure/events/reactive.py +++ b/src/cleveragents/infrastructure/events/reactive.py @@ -21,6 +21,7 @@ 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 @@ -50,6 +51,7 @@ 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: @@ -60,6 +62,7 @@ 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) @@ -107,12 +110,19 @@ 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: @@ -208,16 +218,38 @@ class ReactiveEventBus: return self._stream def close(self) -> None: - """Signal completion on the reactive stream and clear all subscriptions. + """Complete the RxPY stream and prevent further event emission. - 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. + Signals ``on_completed()`` to all RxPY subscribers, allowing them to + finalize their state. 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. """ + 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.""" + 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.""" + self.close() + __all__ = ["ReactiveEventBus"] diff --git a/tests/events/test_event_bus_semantics.py b/tests/events/test_event_bus_semantics.py new file mode 100644 index 000000000..70ba486b0 --- /dev/null +++ b/tests/events/test_event_bus_semantics.py @@ -0,0 +1,133 @@ +from __future__ import annotations + +from collections.abc import Callable + +from cleveragents.a2a.events import A2aEventQueue, EventBusBridge +from cleveragents.application.services.fix_then_revalidate import ( + FixThenRevalidateOrchestrator, +) +from cleveragents.application.services.validation_pipeline import ValidationPipeline +from cleveragents.infrastructure.events.models import DomainEvent +from cleveragents.infrastructure.events.reactive import ReactiveEventBus +from cleveragents.infrastructure.events.types import EventType + + +def _event(event_type: EventType = EventType.PLAN_CREATED) -> DomainEvent: + return DomainEvent(event_type=event_type) + + +def _pipeline() -> ValidationPipeline: + return ValidationPipeline(commands=[], executor=lambda _name, _args: {}) + + +def test_reactive_event_bus_close_completes_and_rejects_later_emit() -> None: + bus = ReactiveEventBus() + completed: list[bool] = [] + bus.stream.subscribe(on_completed=lambda: completed.append(True)) + + bus.close() + + assert completed == [True] + try: + bus.emit(_event()) + except RuntimeError: + pass + else: + raise AssertionError("emit after close should raise RuntimeError") + + +def test_reactive_event_bus_context_manager_closes_bus() -> None: + bus = ReactiveEventBus() + completed: list[bool] = [] + bus.stream.subscribe(on_completed=lambda: completed.append(True)) + + with bus as entered: + assert entered is bus + + assert bus._closed is True + assert completed == [True] + + +class _TypedEventBus: + def __init__(self) -> None: + self.handlers: dict[EventType, list[Callable[[DomainEvent], None]]] = {} + + def emit(self, event: DomainEvent) -> None: + for handler in self.handlers.get(event.event_type, []): + handler(event) + + def subscribe( + self, event_type: EventType, handler: Callable[[DomainEvent], None] + ) -> None: + self.handlers.setdefault(event_type, []).append(handler) + + def unsubscribe( + self, event_type: EventType, handler: Callable[[DomainEvent], None] + ) -> bool: + handlers = self.handlers.get(event_type, []) + try: + handlers.remove(handler) + except ValueError: + return False + return True + + +def test_event_bus_bridge_unsubscribes_from_typed_event_bus() -> None: + bus = _TypedEventBus() + queue = A2aEventQueue() + bridge = EventBusBridge(bus, queue) + + bridge.start() + bus.emit(_event()) + bridge.stop() + bus.emit(_event()) + + assert len(queue.get_events()) == 1 + assert all(not handlers for handlers in bus.handlers.values()) + + +class _DisposableSubscription: + def __init__(self) -> None: + self.disposed = False + + def dispose(self) -> None: + self.disposed = True + + +class _LegacyEventBus: + def __init__(self) -> None: + self.callback: Callable[[DomainEvent], None] | None = None + self.subscription = _DisposableSubscription() + + def subscribe(self, callback: Callable[[DomainEvent], None]) -> object: + self.callback = callback + return self.subscription + + +def test_event_bus_bridge_still_supports_disposable_subscription_bus() -> None: + bus = _LegacyEventBus() + queue = A2aEventQueue() + bridge = EventBusBridge(bus, queue) + + bridge.start() + assert bus.callback is not None + bus.callback(_event()) + bridge.stop() + + assert len(queue.get_events()) == 1 + assert bus.subscription.disposed is True + assert bridge._subscription is None + + +class _EmitOnlyBus: + def emit(self, event: DomainEvent) -> None: + pass + + +def test_fix_then_revalidate_accepts_emit_only_event_sink() -> None: + orchestrator = FixThenRevalidateOrchestrator( + validation_pipeline=_pipeline(), + event_bus=_EmitOnlyBus(), # type: ignore[arg-type] + ) + + assert orchestrator.event_bus is not None -- 2.52.0