diff --git a/EvoScientist/channels/bus/message_bus.py b/EvoScientist/channels/bus/message_bus.py index 89afcc3..4af4d60 100644 --- a/EvoScientist/channels/bus/message_bus.py +++ b/EvoScientist/channels/bus/message_bus.py @@ -2,23 +2,20 @@ Channels push messages to the inbound queue; the agent (or any consumer) reads from inbound, processes, and pushes responses to the outbound queue. -A background dispatcher routes outbound messages to the correct channel -via subscriber callbacks. +``ChannelManager._dispatch_outbound`` routes outbound messages to the +correct channel by looking up its registered :class:`Channel` instance. Deduplication is handled at the Channel level (single dedup point). """ import asyncio import logging -from collections.abc import Awaitable, Callable from ..debug import TraceMixin, debug_trace_enabled from .events import InboundMessage, OutboundMessage logger = logging.getLogger(__name__) -OutboundCallback = Callable[[OutboundMessage], Awaitable[None]] - class MessageBus(TraceMixin): """Async message bus that decouples chat channels from the agent core.""" @@ -28,8 +25,6 @@ class MessageBus(TraceMixin): def __init__(self): self.inbound: asyncio.Queue[InboundMessage] = asyncio.Queue(maxsize=5000) self.outbound: asyncio.Queue[OutboundMessage] = asyncio.Queue(maxsize=5000) - self._outbound_subscribers: dict[str, list[OutboundCallback]] = {} - self._running = False self._debug_trace = debug_trace_enabled() self._trace_logger = logger @@ -53,58 +48,6 @@ class MessageBus(TraceMixin): """Consume the next outbound message (blocks until available).""" return await self.outbound.get() - # ── subscriber routing ── - - def subscribe_outbound( - self, - channel: str, - callback: OutboundCallback, - ) -> None: - """Register a callback for outbound messages targeting *channel*.""" - if channel not in self._outbound_subscribers: - self._outbound_subscribers[channel] = [] - self._outbound_subscribers[channel].append(callback) - - async def dispatch_outbound(self) -> None: - """Route outbound messages to subscribed channels. - - Run as a background task — loops until :meth:`stop` is called. - """ - self._running = True - while self._running: - try: - msg = await asyncio.wait_for( - self.outbound.get(), - timeout=1.0, - ) - except TimeoutError: - continue - subscribers = self._outbound_subscribers.get(msg.channel, []) - if not subscribers: - self._trace_event( - "bus_dispatch_drop", - target_channel=msg.channel, - reason="no_subscriber", - chat_id=msg.chat_id, - ) - logger.warning(f"No subscriber for channel: {msg.channel}") - continue - for callback in subscribers: - try: - await callback(msg) - except Exception as e: - self._trace_event( - "bus_dispatch_error", - target_channel=msg.channel, - chat_id=msg.chat_id, - error_type=type(e).__name__, - ) - logger.error(f"Error dispatching to {msg.channel}: {e}") - - def stop(self) -> None: - """Stop the dispatcher loop.""" - self._running = False - @property def inbound_size(self) -> int: return self.inbound.qsize() diff --git a/EvoScientist/cli/channel.py b/EvoScientist/cli/channel.py index fec8e2b..49e544f 100644 --- a/EvoScientist/cli/channel.py +++ b/EvoScientist/cli/channel.py @@ -741,8 +741,6 @@ def _channels_stop( future.result(timeout=10) except Exception as e: _channel_logger.debug(f"Error stopping channels: {e}") - if _manager: - _manager.bus.stop() if _bus_thread: _bus_thread.join(timeout=5) _manager = None diff --git a/tests/test_channel_comprehensive.py b/tests/test_channel_comprehensive.py index 3866d74..4af92c6 100644 --- a/tests/test_channel_comprehensive.py +++ b/tests/test_channel_comprehensive.py @@ -1510,6 +1510,13 @@ class TestInboundConsumerErrorHandling: class TestMessageBus: + """Covers the bus as a pure pub/sub queue. + + Outbound routing (subscriber dispatch, error handling, stop semantics) + is owned by ``ChannelManager._dispatch_outbound`` — see + ``TestChannelManagerDispatch`` for that coverage. + """ + def test_publish_consume_inbound(self): async def _test(): bus = MessageBus() @@ -1535,57 +1542,6 @@ class TestMessageBus: _run(_test()) - def test_subscriber_dispatch(self): - async def _test(): - bus = MessageBus() - received = [] - bus.subscribe_outbound("tg", lambda m: received.append(m)) - - task = asyncio.create_task(bus.dispatch_outbound()) - await bus.publish_outbound( - BusOutbound( - channel="tg", - chat_id="c1", - content="hello", - ) - ) - await asyncio.sleep(0.1) - bus.stop() - await asyncio.sleep(0.1) - task.cancel() - try: - await task - except asyncio.CancelledError: - pass - - assert len(received) == 1 - - _run(_test()) - - def test_no_subscriber_logs_warning(self): - """Messages to unsubscribed channels should warn, not crash.""" - - async def _test(): - bus = MessageBus() - task = asyncio.create_task(bus.dispatch_outbound()) - await bus.publish_outbound( - BusOutbound( - channel="unknown", - chat_id="c1", - content="lost", - ) - ) - await asyncio.sleep(0.1) - bus.stop() - await asyncio.sleep(0.1) - task.cancel() - try: - await task - except asyncio.CancelledError: - pass - - _run(_test()) - def test_queue_sizes(self): async def _test(): bus = MessageBus() @@ -1603,41 +1559,6 @@ class TestMessageBus: _run(_test()) - def test_subscriber_error_does_not_crash_dispatch(self): - async def _test(): - bus = MessageBus() - - async def bad_callback(msg): - raise RuntimeError("subscriber crash") - - bus.subscribe_outbound("tg", bad_callback) - - task = asyncio.create_task(bus.dispatch_outbound()) - await bus.publish_outbound( - BusOutbound( - channel="tg", - chat_id="c1", - content="trigger", - ) - ) - await asyncio.sleep(0.1) - bus.stop() - await asyncio.sleep(0.1) - task.cancel() - try: - await task - except asyncio.CancelledError: - pass - # dispatch should survive the error - - _run(_test()) - - def test_stop_flag(self): - bus = MessageBus() - assert bus._running is False - bus.stop() - assert bus._running is False - # ═══════════════════════════════════════════════════════════════════ # 9. Event dataclasses