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

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

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]
@@ -189,6 +190,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 +233,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
@@ -230,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"
@@ -240,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"
@@ -250,18 +277,67 @@ 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"
 
 
+@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(cast(EventType, "plan.created"), lambda e: None)
+    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, cast(Callable[[DomainEvent], None], "not-callable")
+        )
+    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."""
 
-    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")
@@ -269,6 +345,24 @@ 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):
+        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)
+
+
 # ---------------------------------------------------------------------------
 # DecisionService event emission steps
 # ---------------------------------------------------------------------------
@@ -477,56 +571,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"

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
@@ -24,6 +25,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)
@@ -261,20 +263,35 @@ class EventBusBridge:
         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"):
+        if not hasattr(self._event_bus, "subscribe"):
+            return
+        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)
-            logger.info("a2a.event_bridge.started")
+        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 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")
+        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."""

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/logging_bus.py

@@ -105,5 +105,38 @@ 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: Callable to remove (same object passed to
+                :meth:`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"]

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

src/cleveragents/infrastructure/events/reactive.py

@@ -119,7 +119,7 @@ class ReactiveEventBus:
 
         if self._closed:
             raise RuntimeError(
-                "ReactiveEventBus.emit() called after close() — "
+                "ReactiveEventBus.emit() called after close() - "
                 "the bus has been shut down and the RxPY Subject is completed"
             )
 
@@ -174,6 +174,39 @@ 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: Callable to remove (same object passed to
+                :meth:`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.
@@ -188,11 +221,10 @@ class ReactiveEventBus:
         """Complete the RxPY stream and prevent further event emission.
 
         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`.
+        finalize their state. After ``close()``, any call to :meth:`emit`
+        raises :exc:`RuntimeError`.
 
-        This method is idempotent — calling it more than once is safe.
+        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
@@ -207,11 +239,7 @@ class ReactiveEventBus:
         self._audit_log.clear()
 
     def __enter__(self) -> ReactiveEventBus:
-        """Support use as a context manager.
-
-        Returns:
-            The bus instance itself.
-        """
+        """Support use as a context manager."""
         return self
 
     def __exit__(
@@ -220,13 +248,7 @@ class ReactiveEventBus:
         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.
-        """
+        """Call :meth:`close` on context manager exit."""
         self.close()
 
 

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