diff --git a/EvoScientist/cli/channel.py b/EvoScientist/cli/channel.py index 78383a9..0b713c4 100644 --- a/EvoScientist/cli/channel.py +++ b/EvoScientist/cli/channel.py @@ -1,8 +1,20 @@ -"""Background channel management — bus mode with ChannelManager.""" +"""Background channel management — bus mode with ChannelManager. + +Architecture: + Bus thread: runs ChannelManager + all channels + inbound consumer. + Main CLI thread: runs agent invocations (to avoid event-loop conflicts). + +The inbound consumer does NOT call the agent directly. Instead it +enqueues a ``ChannelMessage`` on a thread-safe ``queue.Queue`` and waits +for the main thread to set a response via ``_set_channel_response()``. +""" import asyncio import logging +import queue import threading +import uuid +from dataclasses import dataclass from typing import Any, Optional from rich.panel import Panel @@ -14,7 +26,62 @@ from ..stream.display import console _channel_logger = logging.getLogger(__name__) +# --------------------------------------------------------------------------- +# Queue bridge: bus thread ⇄ main CLI thread +# --------------------------------------------------------------------------- + +@dataclass +class ChannelMessage: + """A message from a channel, enqueued for the main CLI thread.""" + msg_id: str + content: str + sender: str + channel_type: str + metadata: Any = None + # Filled by the bus consumer so the main thread can send callbacks + channel_ref: Any = None # Channel instance (for thinking / todo / file) + bus_ref: Any = None # MessageBus (for publishing outbound) + chat_id: str = "" + message_id: str | None = None + + +# Thread-safe queue: bus → main +_message_queue: queue.Queue[ChannelMessage] = queue.Queue() + +# Pending responses: main → bus (msg_id → {"event": Event, "response": str|None}) +_pending_responses: dict[str, dict] = {} +_response_lock = threading.Lock() + + +def _enqueue_channel_message(msg: ChannelMessage) -> threading.Event: + """Enqueue a channel message for the main thread and return a wait event.""" + event = threading.Event() + with _response_lock: + _pending_responses[msg.msg_id] = {"event": event, "response": None} + _message_queue.put(msg) + return event + + +def _set_channel_response(msg_id: str, response: str) -> None: + """Set the response for a channel message and unblock the bus consumer.""" + with _response_lock: + slot = _pending_responses.get(msg_id) + if slot: + slot["response"] = response + slot["event"].set() + + +def _pop_channel_response(msg_id: str) -> str | None: + """Retrieve and remove the response for a channel message.""" + with _response_lock: + slot = _pending_responses.pop(msg_id, None) + return slot["response"] if slot else None + + +# --------------------------------------------------------------------------- # Module-level channel state (bus mode) +# --------------------------------------------------------------------------- + _manager: Optional[Any] = None # ChannelManager _bus_loop: Optional[asyncio.AbstractEventLoop] = None _bus_thread: Optional[threading.Thread] = None @@ -103,7 +170,7 @@ def _start_channels_bus_mode(config, agent, thread_id: str, show_thinking: bool async def _run(): consumer = asyncio.create_task( - _bus_inbound_consumer(mgr.bus, mgr, agent, thread_id, show_thinking) + _bus_inbound_consumer(mgr.bus, mgr, show_thinking) ) try: await mgr.start_all() @@ -152,27 +219,15 @@ def _add_channel_to_running_bus(channel_type: str, config) -> None: async def _bus_inbound_consumer( - bus, manager, agent, thread_id: str, show_thinking: bool = True, + bus, manager, show_thinking: bool = True, ) -> None: - """Core bridge: consume inbound messages from bus and run agent. + """Consume inbound messages from bus and bridge to the main CLI thread. - Streams agent events on the bus loop with Rich Live real-time display - (identical to interactive CLI) and sends thinking / todo / answer to - the originating channel via direct ``await`` calls. + This does NOT invoke the agent. It enqueues a ``ChannelMessage`` on + the thread-safe queue and waits for the main thread to set a response. + Once the response arrives it publishes the outbound message on the bus. """ - from ..stream import events as _stream_events_mod - from ..stream.display import ( - console, create_streaming_display, - ) - from ..stream.state import StreamState - from ..channels.consumer import _format_todo_list from ..channels.bus.events import OutboundMessage - from rich.live import Live - from rich.text import Text as _Text - - def _print_separator(): - width = console.size.width - console.print(_Text("\u2500" * width, style="dim")) while True: try: @@ -183,95 +238,45 @@ async def _bus_inbound_consumer( break _channel_logger.info( - f"[bus] Processing from {msg.channel}:{msg.sender_id}: " + f"[bus] Received from {msg.channel}:{msg.sender_id}: " f"{msg.content[:60]}..." ) manager.record_message(msg.channel, "received") - # CLI: show query from channel (mirrors interactive prompt) - source_label = _Text() - source_label.append(f"[{msg.channel}] ", style="cyan bold") - source_label.append(msg.content) - console.print(source_label) - channel = manager.get_channel(msg.channel) - state = StreamState() - thinking_sent = False - todo_sent = False - if channel: await channel.start_typing(msg.chat_id) + # Enqueue for main CLI thread to process with its own event loop + cm = ChannelMessage( + msg_id=str(uuid.uuid4()), + content=msg.content, + sender=msg.sender_id, + channel_type=msg.channel, + metadata=msg.metadata, + channel_ref=channel, + bus_ref=bus, + chat_id=msg.chat_id, + message_id=msg.message_id, + ) + event = _enqueue_channel_message(cm) + + # Wait (non-blocking for asyncio) until main thread sets response + await asyncio.to_thread(event.wait) + response = _pop_channel_response(cm.msg_id) or "No response" + + # Publish the response back through the bus → channel try: - with Live(console=console, refresh_per_second=10, transient=False) as live: - live.update(create_streaming_display(is_waiting=True)) - - async for event in _stream_events_mod.stream_agent_events( - agent, msg.content, thread_id, - ): - etype = state.handle_event(event) - - # Channel: send thinking on transition - if (etype != "thinking" - and not thinking_sent - and state.thinking_text): - if channel and show_thinking: - await channel.send_thinking_message( - msg.sender_id, state.thinking_text, msg.metadata, - ) - thinking_sent = True - - # Channel: send todo list - if (etype == "tool_call" - and event.get("name") == "write_todos" - and not todo_sent - and state.todo_items): - if channel: - await channel.send_todo_message( - msg.sender_id, - _format_todo_list(state.todo_items), - msg.metadata, - ) - todo_sent = True - - # CLI: Live update - live.update(create_streaming_display( - **state.get_display_args(), - show_thinking=show_thinking, - )) - if etype in ( - "tool_call", "tool_result", - "subagent_start", "subagent_tool_call", - "subagent_tool_result", "subagent_end", - ): - live.refresh() - - # Flush remaining thinking - if (not thinking_sent - and state.thinking_text): - if channel and show_thinking: - await channel.send_thinking_message( - msg.sender_id, state.thinking_text, msg.metadata, - ) - - # Channel: publish answer await bus.publish_outbound(OutboundMessage( channel=msg.channel, chat_id=msg.chat_id, - content=state.response_text or "No response", + content=response, reply_to=msg.message_id or None, metadata=msg.metadata, )) manager.record_message(msg.channel, "sent") - console.print(_Text("> ", style="blue bold"), end="") except Exception as e: - _channel_logger.error(f"[bus] Agent error: {e}") - await bus.publish_outbound(OutboundMessage( - channel=msg.channel, - chat_id=msg.chat_id, - content=f"Error processing message: {e}", - metadata=msg.metadata, - )) + _channel_logger.error(f"[bus] Outbound error: {e}") finally: if channel: await channel.stop_typing(msg.chat_id) diff --git a/EvoScientist/cli/interactive.py b/EvoScientist/cli/interactive.py index ecaf323..b5648f0 100644 --- a/EvoScientist/cli/interactive.py +++ b/EvoScientist/cli/interactive.py @@ -2,6 +2,7 @@ import asyncio import os +import queue import sys from datetime import datetime, timezone from typing import Any @@ -32,10 +33,13 @@ from ..sessions import ( from ..stream.display import console, _run_streaming from .agent import _shorten_path, _create_session_workspace, _load_agent from .channel import ( + ChannelMessage, _channels_is_running, _cmd_channel, _cmd_channel_stop, _auto_start_channel, + _message_queue, + _set_channel_response, ) import EvoScientist.cli.channel as _ch_mod from .mcp_ui import _cmd_mcp @@ -429,8 +433,132 @@ def cmd_interactive( else: print_banner(state["thread_id"], state["workspace_dir"], memory_dir, mode, model, provider) - # Start background queue checker - # (no longer needed — bus mode handles messages internally) + # ---- Channel queue processing (bus → main thread) ---- + + async def _process_channel_message(msg: ChannelMessage) -> None: + """Process a single channel message with real-time streaming. + + Clears the waiting prompt line and reprints the message as if + the user typed it after ❯, then streams the agent response + with Rich Live display. + + Display: + ❯ message content + [channel: Received from sender] + ───────────────── + (real-time streaming output) + [channel: Replied to sender] + ───────────────── + """ + # Clear the waiting ❯ prompt line + sys.stdout.write("\r\033[2K") + sys.stdout.flush() + + # Reprint as if user typed it after ❯ + prompt_line = Text() + prompt_line.append("\u276f ", style="bold blue") + prompt_line.append(msg.content) + console.print(prompt_line) + rx = Text() + rx.append(f"[{msg.channel_type}: Received from ", style="dim") + rx.append(msg.sender, style="cyan") + rx.append("]", style="dim") + console.print(rx) + _print_separator() + console.print() + + def _send_thinking_to_channel(thinking: str) -> None: + """Send thinking text to the channel (sync callback).""" + ch = msg.channel_ref + loop = _ch_mod._bus_loop + if ch and loop and ch.send_thinking: + try: + future = asyncio.run_coroutine_threadsafe( + ch.send_thinking_message( + sender=msg.chat_id, + thinking=thinking, + metadata=msg.metadata, + ), + loop, + ) + future.result(timeout=15) + except Exception: + pass + + def _send_todo_to_channel(items: list[dict]) -> None: + """Send todo list to the channel (sync callback).""" + from ..channels.consumer import _format_todo_list + ch = msg.channel_ref + loop = _ch_mod._bus_loop + if ch and loop: + try: + future = asyncio.run_coroutine_threadsafe( + ch.send_todo_message( + sender=msg.chat_id, + content=_format_todo_list(items), + metadata=msg.metadata, + ), + loop, + ) + future.result(timeout=15) + except Exception: + pass + + def _send_media_to_channel(file_path: str) -> None: + """Send media file back through the channel (sync callback).""" + ch = msg.channel_ref + loop = _ch_mod._bus_loop + if ch and loop: + try: + future = asyncio.run_coroutine_threadsafe( + ch.send_media( + recipient=msg.chat_id, + file_path=file_path, + metadata=msg.metadata, + ), + loop, + ) + future.result(timeout=30) + except Exception as e: + console.print(f"[dim]Media send failed: {e}[/dim]") + + meta = _build_metadata(state["workspace_dir"], model) + try: + response = _run_streaming( + state["agent"], msg.content, state["thread_id"], + show_thinking, interactive=True, metadata=meta, + on_thinking=_send_thinking_to_channel, + on_todo=_send_todo_to_channel, + on_file_write=_send_media_to_channel, + ) + except Exception as e: + response = f"Error: {e}" + console.print(f"[red]Channel error: {e}[/red]") + + _set_channel_response(msg.msg_id, response) + + tx = Text() + tx.append(f"[{msg.channel_type}: Replied to ", style="dim") + tx.append(msg.sender, style="cyan") + tx.append("]", style="dim") + console.print(tx) + _print_separator() + + # Redraw the ❯ prompt on a new line after separator + sys.stdout.write("\n\033[34;1m\u276f\033[0m ") + sys.stdout.flush() + + async def _check_channel_queue() -> None: + """Poll the channel message queue and dispatch to the agent.""" + while True: + try: + msg = _message_queue.get_nowait() + except queue.Empty: + await asyncio.sleep(0.1) + continue + await _process_channel_message(msg) + + queue_task = asyncio.create_task(_check_channel_queue()) # Auto-start channel if enabled in config from ..config import load_config @@ -483,6 +611,10 @@ def cmd_interactive( state["agent"] = _load_agent(workspace_dir=state["workspace_dir"], checkpointer=checkpointer) state["thread_id"] = generate_thread_id() state["resumed"] = False + # Sync channel refs so the queue checker uses the new agent + if _channels_is_running(): + _ch_mod._cli_agent = state["agent"] + _ch_mod._cli_thread_id = state["thread_id"] console.print(f"[green]New session:[/green] [yellow]{state['thread_id']}[/yellow]") if state["workspace_dir"]: console.print(f"[dim]Workspace:[/dim] [cyan]{_shorten_path(state['workspace_dir'])}[/cyan]\n") @@ -552,7 +684,11 @@ def cmd_interactive( else: console.print(f"[red]Error: {e}[/red]") finally: - pass + queue_task.cancel() + try: + await queue_task + except asyncio.CancelledError: + pass # Run the async main loop try: diff --git a/EvoScientist/middleware/memory.py b/EvoScientist/middleware/memory.py index 01e1bbf..88fa37a 100644 --- a/EvoScientist/middleware/memory.py +++ b/EvoScientist/middleware/memory.py @@ -564,11 +564,29 @@ class EvoMemoryMiddleware(AgentMiddleware): conversation="\n".join(conv_parts), ) + @staticmethod + def _disable_thinking(model: BaseChatModel) -> BaseChatModel: + """Return a copy of the model with thinking/reasoning disabled. + + Anthropic's API does not allow extended thinking when tool_choice + forces tool use (as with_structured_output does). Similarly, + OpenAI reasoning can conflict. Strip these settings so extraction + works reliably. + """ + rebind: dict[str, Any] = {} + model_kwargs = getattr(model, "model_kwargs", {}) or {} + if getattr(model, "thinking", None) or "thinking" in model_kwargs: + rebind["thinking"] = {"type": "disabled"} + if getattr(model, "reasoning", None) or "reasoning" in model_kwargs: + rebind["reasoning"] = None + return model.bind(**rebind) if rebind else model + def _extract(self, model: BaseChatModel, memory: str, messages: list[AnyMessage]) -> dict[str, Any]: """Run LLM extraction on recent messages using structured output.""" prompt = self._build_extraction_prompt(memory, messages) try: - structured_model = model.with_structured_output(ExtractedMemory) + plain_model = self._disable_thinking(model) + structured_model = plain_model.with_structured_output(ExtractedMemory) result = structured_model.invoke(prompt) return result.model_dump(exclude_none=True) except Exception as e: # noqa: BLE001 @@ -579,7 +597,8 @@ class EvoMemoryMiddleware(AgentMiddleware): """Async: Run LLM extraction on recent messages using structured output.""" prompt = self._build_extraction_prompt(memory, messages) try: - structured_model = model.with_structured_output(ExtractedMemory) + plain_model = self._disable_thinking(model) + structured_model = plain_model.with_structured_output(ExtractedMemory) result = await structured_model.ainvoke(prompt) return result.model_dump(exclude_none=True) except Exception as e: # noqa: BLE001 diff --git a/tests/test_bus_integration.py b/tests/test_bus_integration.py index 8d73ff4..a40fb7f 100644 --- a/tests/test_bus_integration.py +++ b/tests/test_bus_integration.py @@ -1,8 +1,12 @@ -"""Tests for bus-mode agent integration (_bus_inbound_consumer).""" +"""Tests for bus-mode queue bridge (_bus_inbound_consumer). + +The consumer no longer calls the agent directly. Instead it enqueues a +``ChannelMessage`` on a thread-safe queue and waits for the main CLI +thread to set a response via ``_set_channel_response()``. +""" import asyncio - from EvoScientist.channels.bus.events import InboundMessage from EvoScientist.channels.bus.message_bus import MessageBus from EvoScientist.channels.channel_manager import ChannelManager @@ -18,6 +22,15 @@ def _run(coro): loop.close() +def _drain_queue(q): + """Drain a queue.Queue before a test to avoid cross-test leaks.""" + while not q.empty(): + try: + q.get_nowait() + except Exception: + break + + class _FakeConfig: text_chunk_limit = 4096 allowed_senders = None @@ -56,37 +69,17 @@ class FakeChannel(Channel): pass -def _mock_stream_events(content, reply): - """Create a mock stream_agent_events that yields text then done.""" - async def _stream(agent, message, thread_id): - yield {"type": "text", "content": reply} - yield {"type": "done", "response": reply} - return _stream - - -def _mock_stream_events_error(error_msg): - """Create a mock stream_agent_events that raises.""" - async def _stream(agent, message, thread_id): - raise RuntimeError(error_msg) - yield # make it an async generator # pragma: no cover - return _stream - - -def _mock_stream_events_with_thinking(thinking_text, reply): - """Create a mock stream_agent_events that yields thinking then done.""" - async def _stream(agent, message, thread_id): - yield {"type": "thinking", "content": thinking_text} - yield {"type": "text", "content": reply} - yield {"type": "done", "content": reply} - return _stream - - class TestBusInboundConsumer: - """Test the _bus_inbound_consumer bridge function.""" + """Test the _bus_inbound_consumer queue bridge.""" def test_processes_inbound_and_publishes_outbound(self): - """InboundMessage -> agent -> OutboundMessage flow.""" - from EvoScientist.cli.channel import _bus_inbound_consumer + """InboundMessage -> queue -> response -> OutboundMessage flow.""" + from EvoScientist.cli.channel import ( + _bus_inbound_consumer, + _message_queue, + _set_channel_response, + ) + _drain_queue(_message_queue) async def _test(): bus = MessageBus() @@ -94,48 +87,54 @@ class TestBusInboundConsumer: ch = FakeChannel() manager.register(ch) - mock_stream = _mock_stream_events( - "hello agent", "Reply to: hello agent", + consumer = asyncio.create_task( + _bus_inbound_consumer(bus, manager, False) ) - import EvoScientist.stream.events as events_mod - original = events_mod.stream_agent_events - events_mod.stream_agent_events = mock_stream + await bus.publish_inbound(InboundMessage( + channel="fake", + sender_id="user1", + chat_id="chat1", + content="hello agent", + )) + # Wait for consumer to enqueue the message + for _ in range(20): + if not _message_queue.empty(): + break + await asyncio.sleep(0.05) + + msg = _message_queue.get_nowait() + assert msg.content == "hello agent" + assert msg.sender == "user1" + assert msg.channel_type == "fake" + + # Simulate main-thread response + _set_channel_response(msg.msg_id, "Reply to: hello agent") + + outbound = await asyncio.wait_for( + bus.consume_outbound(), timeout=2.0, + ) + assert outbound.channel == "fake" + assert outbound.chat_id == "chat1" + assert "Reply to: hello agent" in outbound.content + + consumer.cancel() try: - consumer = asyncio.create_task( - _bus_inbound_consumer(bus, manager, None, "test-thread", False) - ) - - await bus.publish_inbound(InboundMessage( - channel="fake", - sender_id="user1", - chat_id="chat1", - content="hello agent", - )) - - await asyncio.sleep(0.5) - - outbound = await asyncio.wait_for( - bus.consume_outbound(), timeout=2.0, - ) - assert outbound.channel == "fake" - assert outbound.chat_id == "chat1" - assert "Reply to: hello agent" in outbound.content - - consumer.cancel() - try: - await consumer - except asyncio.CancelledError: - pass - finally: - events_mod.stream_agent_events = original + await consumer + except asyncio.CancelledError: + pass _run(_test()) - def test_agent_error_publishes_error_outbound(self): - """When agent raises, an error message is published outbound.""" - from EvoScientist.cli.channel import _bus_inbound_consumer + def test_no_response_fallback(self): + """Empty response is replaced with 'No response' fallback.""" + from EvoScientist.cli.channel import ( + _bus_inbound_consumer, + _message_queue, + _set_channel_response, + ) + _drain_queue(_message_queue) async def _test(): bus = MessageBus() @@ -143,46 +142,47 @@ class TestBusInboundConsumer: ch = FakeChannel() manager.register(ch) - mock_stream = _mock_stream_events_error("agent crashed") + consumer = asyncio.create_task( + _bus_inbound_consumer(bus, manager, False) + ) - import EvoScientist.stream.events as events_mod - original = events_mod.stream_agent_events - events_mod.stream_agent_events = mock_stream + await bus.publish_inbound(InboundMessage( + channel="fake", + sender_id="user1", + chat_id="chat1", + content="test", + )) + for _ in range(20): + if not _message_queue.empty(): + break + await asyncio.sleep(0.05) + + msg = _message_queue.get_nowait() + # Set empty response — falsy, so consumer falls back to "No response" + _set_channel_response(msg.msg_id, "") + + outbound = await asyncio.wait_for( + bus.consume_outbound(), timeout=2.0, + ) + assert outbound.content == "No response" + + consumer.cancel() try: - consumer = asyncio.create_task( - _bus_inbound_consumer(bus, manager, None, "test-thread", False) - ) - - await bus.publish_inbound(InboundMessage( - channel="fake", - sender_id="user1", - chat_id="chat1", - content="crash me", - )) - - await asyncio.sleep(0.5) - - outbound = await asyncio.wait_for( - bus.consume_outbound(), timeout=2.0, - ) - assert outbound.channel == "fake" - assert "Error" in outbound.content or "error" in outbound.content.lower() - assert "agent crashed" in outbound.content - - consumer.cancel() - try: - await consumer - except asyncio.CancelledError: - pass - finally: - events_mod.stream_agent_events = original + await consumer + except asyncio.CancelledError: + pass _run(_test()) def test_message_counting(self): """Messages are counted via record_message.""" - from EvoScientist.cli.channel import _bus_inbound_consumer + from EvoScientist.cli.channel import ( + _bus_inbound_consumer, + _message_queue, + _set_channel_response, + ) + _drain_queue(_message_queue) async def _test(): bus = MessageBus() @@ -190,97 +190,89 @@ class TestBusInboundConsumer: ch = FakeChannel() manager.register(ch) - mock_stream = _mock_stream_events("test", "ok") + consumer = asyncio.create_task( + _bus_inbound_consumer(bus, manager, False) + ) - import EvoScientist.stream.events as events_mod - original = events_mod.stream_agent_events - events_mod.stream_agent_events = mock_stream + await bus.publish_inbound(InboundMessage( + channel="fake", + sender_id="u1", + chat_id="c1", + content="test", + )) + for _ in range(20): + if not _message_queue.empty(): + break + await asyncio.sleep(0.05) + + msg = _message_queue.get_nowait() + _set_channel_response(msg.msg_id, "ok") + + await asyncio.wait_for(bus.consume_outbound(), timeout=2.0) + + assert manager._message_counts["fake"]["received"] == 1 + assert manager._message_counts["fake"]["sent"] == 1 + + consumer.cancel() try: - consumer = asyncio.create_task( - _bus_inbound_consumer(bus, manager, None, "test-thread", False) - ) - - await bus.publish_inbound(InboundMessage( - channel="fake", - sender_id="u1", - chat_id="c1", - content="test", - )) - - await asyncio.sleep(0.5) - await asyncio.wait_for(bus.consume_outbound(), timeout=2.0) - - assert manager._message_counts["fake"]["received"] == 1 - assert manager._message_counts["fake"]["sent"] == 1 - - consumer.cancel() - try: - await consumer - except asyncio.CancelledError: - pass - finally: - events_mod.stream_agent_events = original + await consumer + except asyncio.CancelledError: + pass _run(_test()) - def test_thinking_sent_to_channel(self): - """Thinking messages are sent to the channel when show_thinking=True.""" - from EvoScientist.cli.channel import _bus_inbound_consumer + def test_channel_message_carries_metadata(self): + """ChannelMessage carries metadata, chat_id, and message_id.""" + from EvoScientist.cli.channel import ( + _bus_inbound_consumer, + _message_queue, + _set_channel_response, + ) + _drain_queue(_message_queue) async def _test(): bus = MessageBus() manager = ChannelManager(bus) ch = FakeChannel() - server = manager.register(ch) - server.send_thinking = True + manager.register(ch) - long_thinking = "A" * 250 # >= _MIN_THINKING_LEN (200) - mock_stream = _mock_stream_events_with_thinking( - long_thinking, "final answer", + consumer = asyncio.create_task( + _bus_inbound_consumer(bus, manager, False) ) - import EvoScientist.stream.events as events_mod - original = events_mod.stream_agent_events - events_mod.stream_agent_events = mock_stream + await bus.publish_inbound(InboundMessage( + channel="fake", + sender_id="user1", + chat_id="chat1", + content="with metadata", + metadata={"key": "value"}, + message_id="msg-123", + )) + for _ in range(20): + if not _message_queue.empty(): + break + await asyncio.sleep(0.05) + + msg = _message_queue.get_nowait() + assert msg.content == "with metadata" + assert msg.metadata == {"key": "value"} + assert msg.chat_id == "chat1" + assert msg.message_id == "msg-123" + assert msg.channel_ref is ch + + _set_channel_response(msg.msg_id, "done") + + outbound = await asyncio.wait_for( + bus.consume_outbound(), timeout=2.0, + ) + assert outbound.reply_to == "msg-123" + + consumer.cancel() try: - consumer = asyncio.create_task( - _bus_inbound_consumer( - bus, manager, None, "test-thread", True, - ) - ) - - await bus.publish_inbound(InboundMessage( - channel="fake", - sender_id="user1", - chat_id="chat1", - content="think about this", - metadata={"chat_id": "chat1"}, - )) - - await asyncio.sleep(0.5) - - # Drain outbound (final answer) - outbound = await asyncio.wait_for( - bus.consume_outbound(), timeout=2.0, - ) - assert "final answer" in outbound.content - - # Check that thinking was sent via channel.send - thinking_msgs = [ - m for m in ch._sent - if "\U0001f9e0" in m.content - ] - assert len(thinking_msgs) == 1 - assert long_thinking in thinking_msgs[0].content - - consumer.cancel() - try: - await consumer - except asyncio.CancelledError: - pass - finally: - events_mod.stream_agent_events = original + await consumer + except asyncio.CancelledError: + pass _run(_test())