.forgejo/workflows/master.yml

@@ -3,8 +3,6 @@ name: CI
 on:
     push:
         branches: [master, develop]
-    pull_request:
-        branches: [master, develop]
 
 vars:
     docker_prefix: "http://harbor.cleverthis.com/docker/"

CHANGELOG.md

@@ -5,6 +5,17 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
 
 ## [Unreleased]
 
+### Added
+
+- **ReactiveEventBus close() lifecycle and context manager protocol** (#10916):
+  Added `_closed` flag and `is_closed` public property to `ReactiveEventBus`.
+  `emit()` now raises `RuntimeError` when called after `close()`. Added
+  `__enter__`/`__exit__` context manager protocol for automatic cleanup.
+  Added BDD scenarios covering: close marks bus as closed and clears subscriptions,
+  emit raises RuntimeError after close, and context manager protocol. Fixed
+  single-quote style inconsistency in step decorators and removed accidentally
+  committed `add_close_steps.py` patch script.
+
 ### Changed
 
 - **CI coverage job now waits for unit_tests** (#10714): Added `unit_tests` to the

features/event_bus.feature

@@ -119,6 +119,37 @@ Feature: EventBus protocol and domain event emission
     Then the EventBus Protocol stubs should be callable
 
   # ---------------------------------------------------------------------------
+
+  # ---------------------------------------------------------------------------
+  # ReactiveEventBus close() - RxPY subject lifecycle
+  # ---------------------------------------------------------------------------
+
+  Scenario: ReactiveEventBus close marks bus as closed and clears subscriptions
+    Given a ReactiveEventBus
+    When I subscribe to "plan.created" events
+    And I emit a "plan.created" DomainEvent
+    And I close the ReactiveEventBus
+    Then the subscribed handler should have received 1 event
+    And the subscriptions should be cleared
+    And the bus should be closed
+
+  Scenario: ReactiveEventBus close clears the audit log
+    Given a ReactiveEventBus
+    When I emit a "plan.created" DomainEvent
+    And I emit a "decision.created" DomainEvent
+    And I close the ReactiveEventBus
+    Then the audit log should be empty
+
+  Scenario: ReactiveEventBus emit raises RuntimeError after close
+    Given a ReactiveEventBus
+    When I close the ReactiveEventBus
+    And I emit a "plan.created" DomainEvent after close
+    Then a RuntimeError should be raised
+
+  Scenario: ReactiveEventBus supports context manager protocol
+    When I use the ReactiveEventBus as a context manager
+    Then the bus should be closed after the context manager exits
+
   # DecisionService event emission
   # ---------------------------------------------------------------------------
 

features/steps/event_bus_steps.py

@@ -256,6 +256,79 @@ def step_subscribe_non_callable(ctx: Context) -> None:
     assert raised, "Expected TypeError for non-callable handler"
 
 
+
+# ---------------------------------------------------------------------------
+# ReactiveEventBus close() steps
+# ---------------------------------------------------------------------------
+
+
+@when("I close the ReactiveEventBus")
+def step_close_reactive_bus(ctx: Context) -> None:
+    ctx.bus.close()
+
+
+@then("the subscribed handler should have received {n:d} event")
+def step_subscribed_handler_received(ctx: Context, n: int) -> None:
+    assert ctx.collectors, "No collectors registered"
+    count = len(ctx.collectors[0].received)
+    assert count == n, f"Expected {n} events, got {count}"
+
+
+@then("the bus should be closed")
+def step_bus_is_closed(ctx: Context) -> None:
+    assert ctx.bus.is_closed, "Expected bus.is_closed to be True after close()"
+
+
+@then("the subscriptions should be cleared")
+def step_subscriptions_cleared(ctx: Context) -> None:
+    # Verify the bus is closed (which guarantees subscriptions are cleared)
+    assert ctx.bus.is_closed, "Expected bus.is_closed to be True after close()"
+    # Also verify via audit_log that no further events can be dispatched
+    assert len(ctx.bus.audit_log) == 0, (
+        "Expected empty audit log after close (subscriptions and audit log cleared)"
+    )
+
+
+@then("the audit log should be empty")
+def step_audit_log_empty(ctx: Context) -> None:
+    assert len(ctx.bus.audit_log) == 0, (
+        f"Expected empty audit log after close, found {len(ctx.bus.audit_log)} events"
+    )
+
+
+@when('I emit a "{et}" DomainEvent after close')
+def step_emit_after_close(ctx: Context, et: str) -> None:
+    ctx.exception = None
+    try:
+        ctx.bus.emit(_make_event(et))
+    except RuntimeError as exc:
+        ctx.exception = exc
+
+
+@then("a RuntimeError should be raised")
+def step_runtime_error_raised(ctx: Context) -> None:
+    assert isinstance(ctx.exception, RuntimeError), (
+        f"Expected RuntimeError, got {type(ctx.exception).__name__!r}: {ctx.exception}"
+    )
+
+
+@when("I use the ReactiveEventBus as a context manager")
+def step_use_as_context_manager(ctx: Context) -> None:
+    ctx.context_manager_bus: ReactiveEventBus | None = None
+    with ReactiveEventBus() as bus:
+        ctx.context_manager_bus = bus
+        bus.subscribe(EventType.PLAN_CREATED, lambda e: None)
+        bus.emit(_make_event("plan.created"))
+    ctx.closed_bus = bus
+
+
+@then("the bus should be closed after the context manager exits")
+def step_bus_closed_after_context_manager(ctx: Context) -> None:
+    assert ctx.closed_bus.is_closed, (
+        "Expected bus.is_closed to be True after context manager exit"
+    )
+
+
 @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."""

src/cleveragents/infrastructure/events/reactive.py

@@ -60,6 +60,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)
@@ -93,6 +94,11 @@ class ReactiveEventBus:
         """Clear all retained in-memory audit events."""
         self._audit_log.clear()
 
+    @property
+    def is_closed(self) -> bool:
+        """Return ``True`` if :meth:`close` has been called on this bus."""
+        return self._closed
+
     def emit(self, event: DomainEvent) -> None:
         """Publish *event* to the reactive stream and all type handlers.
 
@@ -107,7 +113,12 @@ class ReactiveEventBus:
 
         Raises:
             TypeError: If *event* is not a :class:`DomainEvent`.
+            RuntimeError: If :meth:`close` has already been called.
         """
+        if self._closed:
+            raise RuntimeError(
+                "Cannot emit events on a closed ReactiveEventBus"
+            )
         if not isinstance(event, DomainEvent):
             raise TypeError(
                 f"event must be a DomainEvent, got {type(event).__name__!r}"
@@ -179,11 +190,35 @@ class ReactiveEventBus:
         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.
+
+        After calling ``close()``, any subsequent call to :meth:`emit` will
+        raise :exc:`RuntimeError`.
         """
+        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:
+            This :class:`ReactiveEventBus` instance.
+        """
+        return self
+
+    def __exit__(
+        self,
+        exc_type: type[BaseException] | None,
+        exc_val: BaseException | None,
+        exc_tb: object,
+    ) -> None:
+        """Close the bus on context manager exit.
+
+        Calls :meth:`close` regardless of whether an exception occurred.
+        """
+        self.close()
+
 
 __all__ = ["ReactiveEventBus"]

vulture_whitelist.py

@@ -1164,6 +1164,9 @@ _build_analyzer_registry  # noqa: B018, F821
 # ReactiveEventBus.audit_log — public property for event audit trail (#587)
 audit_log  # noqa: B018, F821
 
+# ReactiveEventBus.is_closed — public property for closed state inspection (#10916)
+is_closed  # noqa: B018, F821
+
 # Test doubles — public API for BDD/Robot test infrastructure
 TrackingEventBus  # noqa: B018, F821