feat: implement channel message queue and response handling in bus mode

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)

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:

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

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())