diff --git a/CHANGELOG.md b/CHANGELOG.md index 2468a0029..a244431b5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -252,6 +252,19 @@ ensuring data is stored with proper parameter values. traceback in the structlog warning entry. Removed `@tdd_expected_fail` tag from the TDD test so both scenarios run as normal regression guards. (#988) +### Fixed + +- **EventBus protocol and implementations now support `unsubscribe()`** (#10356): + Added the missing ``unsubscribe()`` method to the ``EventBus`` Protocol in + ``protocol.py`` and both concrete implementations. ``ReactiveEventBus.unsubscribe()`` + uses ``contextlib.suppress(ValueError)`` to silently ignore handlers that were + never registered; ``LoggingEventBus.unsubscribe()`` uses a try/except pattern. + Unsubscribed handlers are no longer called on subsequent :meth:`emit` calls, + breaking the reference cycle that prevented garbage collection of owning objects. + Includes comprehensive BDD regression tests covering unsubscription for both + implementations, no-op behavior for non-registered handlers, multi-handler + targeting, and Protocol stub coverage. + ### Added - **`pr-review-worker` review-started notification** (#11028): The `first_review` diff --git a/CONTRIBUTORS.md b/CONTRIBUTORS.md index e6049bf9e..a85ac6b7e 100644 --- a/CONTRIBUTORS.md +++ b/CONTRIBUTORS.md @@ -36,6 +36,7 @@ Below are some of the specific details of various contributions. * HAL 9000 has contributed the plan artifacts JSON completeness fix (#9084): ensured `validation_summary` and `apply_summary` are correctly included in `_build_artifacts_dict`, removing stale `@tdd_expected_fail` tags from Behave scenarios to enable full regression test coverage. * HAL 9000 has contributed the bug-hunt-pool-supervisor non-blocking tracking fix (#7875 / PR #7957): updated step 5 to be best-effort and added rule 9 to prevent the automation-tracking-manager call from blocking the main supervisor loop. * Jeffrey Phillips Freeman has contributed the complete AUTO-BUG-POOL to AUTO-BUG-SUP tracking prefix fix across agent-system-specification.md, automation-tracking.md documentation and agent-system-specification.md spec document, replaced with correct `AUTO-BUG-SUP` prefix used by the bug-hunt-pool-supervisor agent (#7875). +* Jeffrey Phillips Freeman has implemented the missing ``unsubscribe()`` method in the EventBus Protocol and both implementations (``ReactiveEventBus``, ``LoggingEventBus``), preventing memory leaks from handler references in long-running processes with comprehensive BDD test coverage (#10356). * HAL 9000 has contributed the plugin entry point security hardening fix (#7476): enforced entry point allowlist validation before importing plugin modules to prevent malicious plugin loading. * HAL 9000 has contributed the benchmark workflow separation (#9040): moved the benchmark-regression job out of the default PR workflow into a dedicated scheduled workflow, reducing median PR CI turnaround time from 99-132 minutes to under 30 minutes. * HAL 9000 has contributed the plan tree JSON/YAML command envelope fix (#9163): wrapped `agents plan tree --format json/yaml` output in the spec-required command envelope structure, added summary statistics, decision_ids mapping, child_plans list, and accurate timing measurement. diff --git a/features/event_bus.feature b/features/event_bus.feature index 1fd540c7e..bcd56739d 100644 --- a/features/event_bus.feature +++ b/features/event_bus.feature @@ -186,3 +186,37 @@ Feature: EventBus protocol and domain event emission Scenario: ReactiveEventBus supports context manager protocol Given a ReactiveEventBus used as a context manager Then the bus should be closed after the context exits + + # --------------------------------------------------------------------------- + # EventBus unsubscribe (issue #10356, TDD issue #10354) + # --------------------------------------------------------------------------- + + Scenario: ReactiveEventBus supports unsubscribing a handler + Given a ReactiveEventBus + When I subscribe to "plan.created" events + And the handler is unsubscribed + And I emit a "plan.created" DomainEvent + Then the handler should have received 0 event + + Scenario: LoggingEventBus supports unsubscribing a handler + Given a LoggingEventBus + When I subscribe to "decision.created" events + And the handler is unsubscribed + And I emit a "decision.created" DomainEvent + Then the handler should have received 0 event + + Scenario: Unsubscribing a non-registered handler is a no-op + Given a ReactiveEventBus + When an unknown handler is unsubscribed for "plan.created" events + Then the unsubscribe call should not raise an error + + Scenario: Multiple handlers — unsubscribe removes only the target + Given a ReactiveEventBus + When I subscribe ten unique handlers to "tool.invoked" events + And I unsubscribe the first handler + And I emit a "tool.invoked" DomainEvent + Then the unsubscribed handler should have received 0 event + And each remaining handler should have received 1 event + + Scenario: EventBus Protocol unsubscribe stub is callable + Then the EventBus Protocol unsubscribe stub should be executable diff --git a/features/steps/event_bus_steps.py b/features/steps/event_bus_steps.py index a414e2ed2..b18c442ad 100644 --- a/features/steps/event_bus_steps.py +++ b/features/steps/event_bus_steps.py @@ -179,6 +179,7 @@ def step_subscribe_single(ctx: Context, et: str) -> None: collector = _EventCollector() ctx.collectors.append(collector) ctx.bus.subscribe(EventType(et), collector) + ctx.last_subscribed_type: str = et # type: ignore[attr-defined] @when('I subscribe two handlers to "{et}" events') @@ -187,6 +188,16 @@ def step_subscribe_two(ctx: Context, et: str) -> None: collector = _EventCollector() ctx.collectors.append(collector) ctx.bus.subscribe(EventType(et), collector) + ctx.last_subscribed_type = et # type: ignore[attr-defined] + + +@when('I subscribe ten unique handlers to "{et}" events') +def step_subscribe_ten(ctx: Context, et: str) -> None: + for _ in range(10): + collector = _EventCollector() + ctx.collectors.append(collector) + ctx.bus.subscribe(EventType(et), collector) + ctx.last_subscribed_type = et # type: ignore[attr-defined] @when('I emit a "{et}" DomainEvent') @@ -269,6 +280,110 @@ def step_protocol_stubs_callable(ctx: Context) -> None: obj.subscribe(EventType.PLAN_CREATED, lambda e: None) +@then("the EventBus Protocol unsubscribe stub should be executable") +def step_protocol_unsubscribe_callable(ctx: Context) -> None: + """Exercise the unsubscribe Protocol stub to ensure 100% coverage.""" + + class _BareImpl(EventBus): # type: ignore[misc] + pass + + obj = _BareImpl() + obj.unsubscribe(EventType.PLAN_CREATED, lambda e: None) + + +# --------------------------------------------------------------------------- +# EventBus unsubscribe steps (issue #10356) +# --------------------------------------------------------------------------- + + +@when("the handler is unsubscribed") +def step_unsubscribe_handler(ctx: Context) -> None: + """Unsubscribe the first collector registered on ctx.bus.""" + if not ctx.collectors: + raise AssertionError("No collectors registered to unsubscribe") + collector = ctx.collectors[0] + ctx.bus.unsubscribe( # type: ignore[attr-defined] + EventType(ctx.last_subscribed_type), # type: ignore[attr-defined] + collector, + ) + ctx._unsubscribed_index: int = 0 # type: ignore[attr-defined] + + +@when("I unsubscribe the first handler") +def step_unsubscribe_first_handler(ctx: Context) -> None: + """Unsubscribe ``ctx.collectors[0]`` from the most-recently-used event type. + + Mirrors :func:`step_unsubscribe_handler` but reads naturally in scenarios + that subscribe multiple handlers before removing one. + """ + if not ctx.collectors: + raise AssertionError("No collectors registered to unsubscribe") + collector = ctx.collectors[0] + ctx.bus.unsubscribe( # type: ignore[attr-defined] + EventType(ctx.last_subscribed_type), # type: ignore[attr-defined] + collector, + ) + ctx._unsubscribed_index = 0 + + +@when('an unknown handler is unsubscribed for "{et}" events') +def step_unsubscribe_unknown_handler(ctx: Context, et: str) -> None: + """Unsubscribe a handler that was never registered (no-op test).""" + + def _unknown_handler(event: DomainEvent) -> None: + pass # pragma: no cover + + try: + ctx.bus.unsubscribe(EventType(et), _unknown_handler) # type: ignore[attr-defined] + ctx._unsubscribe_raised = False + except Exception as exc: + ctx._unsubscribe_raised = True + ctx._unsubscribe_exception = exc + + +@then("the unsubscribe call should not raise an error") +def step_no_unsubscribe_exception(ctx: Context) -> None: + """Verify that a no-op unsubscribe did not raise.""" + raised = getattr(ctx, "_unsubscribe_raised", True) + assert not raised, ( + f"Expected no-op unsubscribe to not raise, but got {ctx._unsubscribe_exception}" # type: ignore[attr-defined] + ) + + +@then("the unsubscribed handler should have received 0 event") +def step_unsub_handler_received_zero(ctx: Context) -> None: + """The first collector was unsubscribed — it should have received nothing.""" + assert ctx.collectors, "No collectors registered" + count = len(ctx.collectors[0].received) + assert count == 0, f"Expected 0 events for unsubscribed handler, got {count}" + + +@then("each remaining handler should have received {n:d} event") +def step_remaining_handlers_received(ctx: Context, n: int) -> None: + """Verify that non-unsubscribed handlers still receive events. + + Skips the collector at ``ctx._unsubscribed_index`` (set by an + ``unsubscribe`` step) so we don't check the handler we removed. + """ + skip_idx = getattr(ctx, "_unsubscribed_index", None) + all_passed = True + for i, col in enumerate(ctx.collectors): + if i == skip_idx: + continue + count = len(col.received) + if count != n: + ctx._handler_failure_count = i + ctx._handler_failure_expected = n + ctx._handler_failure_actual = count + all_passed = False + + if not all_passed: + idx = ctx._handler_failure_count # type: ignore[attr-defined] + exp = ctx._handler_failure_expected # type: ignore[attr-defined] + got = ctx._handler_failure_actual # type: ignore[attr-defined] + raise AssertionError(f"Handler {idx} received {got}, expected {exp}") + + # --------------------------------------------------------------------------- # DecisionService event emission steps # --------------------------------------------------------------------------- diff --git a/features/steps/fix_then_revalidate_coverage_boost_steps.py b/features/steps/fix_then_revalidate_coverage_boost_steps.py index c900cafa1..e3665be0c 100644 --- a/features/steps/fix_then_revalidate_coverage_boost_steps.py +++ b/features/steps/fix_then_revalidate_coverage_boost_steps.py @@ -547,6 +547,13 @@ class _MockEventBus: ) -> None: pass # Not needed for this test + def unsubscribe( + self, + event_type: EventType, + handler: Any, + ) -> None: + pass # Not needed for this test + class _FailingEventBus: """Mock EventBus whose emit always raises.""" @@ -561,6 +568,13 @@ class _FailingEventBus: ) -> None: pass + def unsubscribe( + self, + event_type: EventType, + handler: Any, + ) -> None: + pass + # --------------------------------------------------------------------------- # Given steps for new review-fix scenarios diff --git a/features/steps/fix_then_revalidate_coverage_r3_steps.py b/features/steps/fix_then_revalidate_coverage_r3_steps.py index d6ec44448..b0a9fb61f 100644 --- a/features/steps/fix_then_revalidate_coverage_r3_steps.py +++ b/features/steps/fix_then_revalidate_coverage_r3_steps.py @@ -118,6 +118,13 @@ class _FailingEventBus: ) -> None: pass + def unsubscribe( + self, + event_type: EventType, + handler: Any, + ) -> None: + pass + # --------------------------------------------------------------------------- # Given steps diff --git a/src/cleveragents/infrastructure/events/logging_bus.py b/src/cleveragents/infrastructure/events/logging_bus.py index 307bf4ed3..7911b4e12 100644 --- a/src/cleveragents/infrastructure/events/logging_bus.py +++ b/src/cleveragents/infrastructure/events/logging_bus.py @@ -13,6 +13,7 @@ Based on: from __future__ import annotations +import contextlib from collections.abc import Callable import structlog @@ -105,5 +106,22 @@ 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], + ) -> None: + """Remove *handler* for events of *event_type*. + + If the handler was not registered, this is a no-op. + + Args: + event_type: The :class:`EventType` the handler was registered for. + handler: The previously registered callable to remove. + """ + handlers = self._subscriptions.get(event_type, []) + with contextlib.suppress(ValueError): + handlers.remove(handler) + __all__ = ["LoggingEventBus"] diff --git a/src/cleveragents/infrastructure/events/protocol.py b/src/cleveragents/infrastructure/events/protocol.py index 2c3cae85f..79e2a144e 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], + ) -> None: + """Remove a previously registered handler for *event_type*. + + If the handler was not registered, this is a no-op (silent pass). + + Implementing classes must remove the handler from their internal + subscription storage so that it will not be invoked by subsequent + :meth:`emit` calls and no longer holds a reference to the handler's + owning object. + + Args: + event_type: The :class:`EventType` the handler was registered for. + handler: The previously registered callable to remove. + """ + ... + __all__ = ["EventBus"] diff --git a/src/cleveragents/infrastructure/events/reactive.py b/src/cleveragents/infrastructure/events/reactive.py index ec7c63194..f5b4b724d 100644 --- a/src/cleveragents/infrastructure/events/reactive.py +++ b/src/cleveragents/infrastructure/events/reactive.py @@ -174,6 +174,23 @@ 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], + ) -> None: + """Remove *handler* for events of *event_type*. + + If the handler was not registered, this is a no-op. + + Args: + event_type: The :class:`EventType` the handler was registered for. + handler: The previously registered callable to remove. + """ + handlers = self._subscriptions.get(event_type, []) + with contextlib.suppress(ValueError): + handlers.remove(handler) + @property def stream(self) -> Observable: """Read-only observable stream for advanced RxPY operators.