fix(events): restore event bus close and bridge semantics

src/cleveragents/a2a/events.py

@@ -290,25 +290,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:

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

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

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