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`

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.

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

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
 # ---------------------------------------------------------------------------

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

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

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"]

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"]

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.