diff --git a/EvoScientist/cli/channel.py b/EvoScientist/cli/channel.py index 52c1741..9886773 100644 --- a/EvoScientist/cli/channel.py +++ b/EvoScientist/cli/channel.py @@ -447,6 +447,137 @@ _ASK_USER_TIMEOUT = ( _STOP_COMMANDS = frozenset(("/stop", "/cancel")) +# --------------------------------------------------------------------------- +# Per-thread channel-origin registry +# --------------------------------------------------------------------------- +# When a channel-originated message starts an agent turn, the turn's +# thread_id is remembered against its (channel_type, chat_id, metadata). +# Later, when an async sub-agent notification fires a synthetic agent turn +# for that same thread_id, the notifier path pushes the synthesized final +# response back to the same chat — otherwise the follow-up would only render +# locally and the channel user would never see it. v1 forwards only the +# final response (no mid-turn thinking/todo/media). + + +@dataclass(frozen=True) +class _ChannelOrigin: + """Channel destination remembered for a thread, for notifier push-back.""" + + channel_type: str + chat_id: str + sender: str + metadata: dict | None = None + + +_thread_channel_origins: dict[str, _ChannelOrigin] = {} +_thread_channel_origins_lock = threading.Lock() + + +def remember_channel_origin(thread_id: str | None, msg: ChannelMessage) -> None: + """Record that ``thread_id`` is currently bound to ``msg``'s channel chat. + + Called on entry to each channel-triggered agent turn (Rich CLI / TUI / + serve). The latest channel turn for a given thread wins — re-registering + is intentional, since the user can keep talking on the same thread from + the same channel and we always want the most recent metadata. + """ + if not thread_id: + return + with _thread_channel_origins_lock: + _thread_channel_origins[thread_id] = _ChannelOrigin( + channel_type=msg.channel_type, + chat_id=msg.chat_id, + sender=msg.sender, + metadata=dict(msg.metadata) if msg.metadata else None, + ) + + +def get_channel_origin(thread_id: str | None) -> _ChannelOrigin | None: + """Return the channel origin remembered for ``thread_id``, or ``None``.""" + if not thread_id: + return None + with _thread_channel_origins_lock: + return _thread_channel_origins.get(thread_id) + + +def forget_channel_origin(thread_id: str | None) -> None: + """Drop the registry entry for ``thread_id`` (e.g. on ``/new`` rotation).""" + if not thread_id: + return + with _thread_channel_origins_lock: + _thread_channel_origins.pop(thread_id, None) + + +def publish_to_channel_origin(thread_id: str | None, content: str) -> bool: + """Schedule pushing ``content`` to the channel remembered for ``thread_id``. + + Fire-and-forget: returns ``True`` iff a publish coroutine was scheduled + on the bus loop; returns ``False`` if no origin is registered, the bus + isn't running, ``content`` is empty/whitespace, or scheduling itself + fails. The publish runs asynchronously — failures inside the coroutine + are logged via a done-callback so callers (which are often on event + loops that must not block) don't pay any latency. + """ + from ..channels.bus.events import OutboundMessage + + if not content or not content.strip(): + return False + origin = get_channel_origin(thread_id) + if origin is None: + return False + loop = _bus_loop + manager = _manager + if loop is None or manager is None: + return False + bus = getattr(manager, "bus", None) + if bus is None: + return False + + async def _publish_and_record() -> None: + await bus.publish_outbound( + OutboundMessage( + channel=origin.channel_type, + chat_id=origin.chat_id, + content=content, + metadata=origin.metadata or {}, + ) + ) + # Mirror the normal channel-reply path, which records a "sent" + # message after a successful publish so per-channel stats stay + # accurate for forwarded notifications too. + manager.record_message(origin.channel_type, "sent") + + try: + future = asyncio.run_coroutine_threadsafe(_publish_and_record(), loop) + except Exception as exc: + _channel_logger.warning( + "Async notification publish to %s:%s failed to schedule: %s", + origin.channel_type, + origin.chat_id, + exc, + ) + return False + + def _on_publish_done(fut) -> None: + """Log any exception raised by the fire-and-forget publish coroutine.""" + # A cancelled future raises CancelledError from .exception() rather + # than returning it (e.g. bus loop torn down mid-publish); treat that + # as a benign shutdown, not a failure to log. + if fut.cancelled(): + return + exc = fut.exception() + if exc is not None: + _channel_logger.warning( + "Async notification publish to %s:%s failed: %s", + origin.channel_type, + origin.chat_id, + exc, + ) + + future.add_done_callback(_on_publish_done) + return True + + def _is_stop_command(content: str | None) -> bool: """Whether incoming content is a stop/cancel slash command.""" return (content or "").strip().lower() in _STOP_COMMANDS diff --git a/EvoScientist/cli/commands.py b/EvoScientist/cli/commands.py index 9195aad..cb5477c 100644 --- a/EvoScientist/cli/commands.py +++ b/EvoScientist/cli/commands.py @@ -37,6 +37,10 @@ from .channel import ( channel_ask_user_prompt, channel_hitl_prompt, dispatch_channel_slash_command, + forget_channel_origin, + get_channel_origin, + publish_to_channel_origin, + remember_channel_origin, ) from .mcp_ui import ( _mcp_add_server_from_kwargs, @@ -790,6 +794,7 @@ def _make_serve_start_new_session_cb( from ..sessions import generate_thread_id new_tid = generate_thread_id() + forget_channel_origin(agent_holder.get("thread_id")) agent_holder["thread_id"] = new_tid if channel_runtime is not None: channel_runtime.thread_id = new_tid @@ -837,6 +842,7 @@ def _make_serve_cmd_completed_hook( new_tid = getattr(ctx, "thread_id", None) thread_changed = bool(new_tid) and new_tid != agent_holder.get("thread_id") if thread_changed: + forget_channel_origin(agent_holder.get("thread_id")) agent_holder["thread_id"] = new_tid if channel_runtime is not None: channel_runtime.thread_id = new_tid @@ -896,6 +902,8 @@ def _serve_process_message( if not _claim_or_complete_channel_request(msg): return + remember_channel_origin(agent_holder.get("thread_id"), msg) + runtime_workspace = agent_holder.get("workspace_dir") or workspace_dir console.print( @@ -1015,6 +1023,10 @@ def _serve_process_message( return if _slash_handled: + # A channel-issued /new or /resume rotates the thread inside the + # dispatch above; re-bind the now-current thread to this channel + # so async-notifier turns on it still forward back here. + remember_channel_origin(agent_holder["thread_id"], msg) console.print(f"[dim][{msg.channel_type}] Replied to {msg.sender}[/dim]") return @@ -1079,18 +1091,29 @@ def _serve_drain_notifications( # session-rebind callback), falling back to the startup value. runtime_workspace = agent_holder.get("workspace_dir") or workspace_dir meta = build_metadata(runtime_workspace, model) + tid = agent_holder["thread_id"] try: - run_streaming( + response = run_streaming( ui_backend="cli", agent=agent_holder["agent"], message=text, - thread_id=agent_holder["thread_id"], + thread_id=tid, show_thinking=show_thinking, interactive=True, metadata=meta, ) except Exception as exc: _serve_logger.warning("Notification agent turn failed: %s", exc) + return + if publish_to_channel_origin(tid, response or ""): + # Mirror a normal channel turn's closing "Replied to" line so the + # forwarded notification reads as terminated in the serve log. + origin = get_channel_origin(tid) + if origin is not None: + console.print( + f"[dim][{origin.channel_type}] Replied to " + f"{origin.sender or origin.chat_id}[/dim]" + ) async def _run_notification_message_async(text: str, notifs: list) -> None: await _aio.to_thread(_run_notification_message, text, notifs) diff --git a/EvoScientist/cli/interactive.py b/EvoScientist/cli/interactive.py index 0d7bd3e..9c4e6d8 100644 --- a/EvoScientist/cli/interactive.py +++ b/EvoScientist/cli/interactive.py @@ -604,6 +604,7 @@ def cmd_interactive( and kick off background agent reload. The dispatch block refreshes the status bar post-execute (symmetric with /compact).""" + _ch_mod.forget_channel_origin(state.get("thread_id")) if not workspace_fixed: state["workspace_dir"] = _create_session_workspace(run_name) state["thread_id"] = generate_thread_id() @@ -665,6 +666,11 @@ def cmd_interactive( console.print(f"[red]{exc}[/red]") return state["workspace_dir"] = workspace_dir + if thread_id != state.get("thread_id"): + # Only drop the origin on a real thread change — resuming + # the already-active thread must keep its live origin so a + # later async-notifier turn still forwards to the channel. + _ch_mod.forget_channel_origin(state.get("thread_id")) state["thread_id"] = thread_id state["resumed"] = True state["status_started_at"] = datetime.now() @@ -801,6 +807,8 @@ def cmd_interactive( if not _ch_mod._claim_or_complete_channel_request(msg): return + _ch_mod.remember_channel_origin(state["thread_id"], msg) + try: # Clear the waiting ❯ prompt line sys.stdout.write("\r\033[2K") @@ -943,6 +951,11 @@ def cmd_interactive( channel_runtime=channel_runtime, ) if _slash_handled: + # A channel-issued /new or /resume rotates the thread + # inside the dispatch above; re-bind the now-current + # thread to this channel so async-notifier turns on it + # still forward back here. + _ch_mod.remember_channel_origin(state["thread_id"], msg) _print_separator() sys.stdout.write("\033[34;1m❯\033[0m ") sys.stdout.flush() @@ -1010,7 +1023,7 @@ def cmd_interactive( console.print(line_text, style=line_style, markup=False) meta = build_metadata(state["workspace_dir"], model) await _refresh_status_snapshot(text, reset_streaming_text=True) - run_streaming( + response = run_streaming( ui_backend=state["ui_backend"], agent=await _await_agent_ready(), message=text, @@ -1026,6 +1039,18 @@ def cmd_interactive( on_stream_event=_handle_stream_status_event, status_footer_builder=_stream_status_footer, ) + _notif_tid = target_thread_id or state["thread_id"] + if _ch_mod.publish_to_channel_origin(_notif_tid, response): + # Forwarded to a channel — print the same closing + # "Replied to" line a normal channel turn shows, so the + # forwarded block reads as terminated on screen. + _origin = _ch_mod.get_channel_origin(_notif_tid) + if _origin is not None: + tx = Text() + tx.append(f"[{_origin.channel_type}: Replied to ", style="dim") + tx.append(_origin.sender or _origin.chat_id, style="cyan") + tx.append("]", style="dim") + console.print(tx) await _refresh_status_snapshot(reset_streaming_text=True) console.print() _print_separator() diff --git a/EvoScientist/cli/tui_interactive.py b/EvoScientist/cli/tui_interactive.py index 61ccd41..ad4d972 100644 --- a/EvoScientist/cli/tui_interactive.py +++ b/EvoScientist/cli/tui_interactive.py @@ -590,6 +590,7 @@ def run_textual_interactive( # Clear all widgets except #welcome self.clear_chat() + _ch_mod.forget_channel_origin(self._conversation_tid) if not workspace_fixed: self._workspace_dir = create_session_workspace(run_name) self._conversation_tid = generate_thread_id() @@ -656,6 +657,11 @@ def run_textual_interactive( await sync_widget.cleanup() self._workspace_dir = workspace_dir + if thread_id != self._conversation_tid: + # Only drop the origin on a real thread change — resuming the + # already-active thread must keep its live origin so a later + # async-notifier turn still forwards to the channel. + _ch_mod.forget_channel_origin(self._conversation_tid) self._conversation_tid = thread_id # Background reload: history renders immediately; next turn awaits. self._start_background_agent_load(self._workspace_dir) @@ -895,14 +901,27 @@ def run_textual_interactive( # get viewport-follow during streaming (only after completion). # Mark busy synchronously so the next poll tick doesn't re-enter. self._busy = True - self._run_task = asyncio.ensure_future( - self._run_turn( + effective_tid = target_thread_id or self._conversation_tid + + async def _run_and_publish() -> None: + response = await self._run_turn( text, skip_user_message=True, resolve_mentions=False, thread_id_override=target_thread_id, ) - ) + if _ch_mod.publish_to_channel_origin(effective_tid, response or ""): + # Mirror a normal channel turn's closing "Replied to" line + # so the forwarded notification reads as terminated. + _origin = _ch_mod.get_channel_origin(effective_tid) + if _origin is not None: + self._append_system( + f"[{_origin.channel_type}: Replied to " + f"{_origin.sender or _origin.chat_id}]", + style="dim", + ) + + self._run_task = asyncio.ensure_future(_run_and_publish()) async def _read_async_tasks_tui( self, target_thread_id: str | None @@ -1949,9 +1968,14 @@ def run_textual_interactive( skip_user_message: bool = False, resolve_mentions: bool = True, thread_id_override: str | None = None, - ) -> None: + ) -> str: """Handle a user turn: stream agent response with widgets. + Returns the final response text from ``_stream_with_widgets`` so + callers (notably the async-notifier path) can forward it onward — + e.g. push it back to the originating channel. Returns ``""`` if + the turn was cancelled or the agent failed to load. + Args: user_text: The user's message text. skip_user_message: If True, suppress the UserMessage widget echo @@ -1968,6 +1992,7 @@ def run_textual_interactive( to the live tid when ``None``. """ cancelled = False + response = "" try: self._busy = True self._turn_started_at = datetime.now() @@ -1992,9 +2017,9 @@ def run_textual_interactive( try: await self._await_agent_ready() except Exception: - return + return "" - await self._stream_with_widgets( + response = await self._stream_with_widgets( message_to_send, display_text=user_text, file_warnings=file_warnings, @@ -2018,6 +2043,8 @@ def run_textual_interactive( self._render_queue_indicator() self._run_task = asyncio.ensure_future(self._run_turn(next_msg)) + return response + async def _process_channel_message(self, msg: ChannelMessage) -> None: """Process a channel message: stream agent response and reply. @@ -2030,6 +2057,7 @@ def run_textual_interactive( prompt_widget = None if not _ch_mod._claim_or_complete_channel_request(msg): return + _ch_mod.remember_channel_origin(self._conversation_tid, msg) try: self._busy = True self._turn_started_at = datetime.now() @@ -2135,6 +2163,10 @@ def run_textual_interactive( channel_runtime=self._channel_runtime, ) if _slash_handled: + # A channel-issued /new or /resume rotates the thread in + # the dispatch above; re-bind the now-current thread to + # this channel so async-notifier turns on it still forward. + _ch_mod.remember_channel_origin(self._conversation_tid, msg) return # outer finally handles _busy / widget cleanup # Non-slash message — streams through the agent, so wait diff --git a/tests/test_channel_notification_origin.py b/tests/test_channel_notification_origin.py new file mode 100644 index 0000000..866ed39 --- /dev/null +++ b/tests/test_channel_notification_origin.py @@ -0,0 +1,272 @@ +"""Tests for the per-thread channel-origin registry used by the async-notifier +push-back path.""" + +from __future__ import annotations + +import asyncio +import threading +from unittest.mock import MagicMock + +import pytest + +from EvoScientist.cli import channel as channel_cli +from EvoScientist.cli.channel import ChannelMessage + + +@pytest.fixture(autouse=True) +def _restore_channel_globals(): + """Restore bus globals + the origin registry between tests.""" + original = { + "_manager": channel_cli._manager, + "_bus_loop": channel_cli._bus_loop, + "_bus_thread": channel_cli._bus_thread, + } + with channel_cli._thread_channel_origins_lock: + original_origins = dict(channel_cli._thread_channel_origins) + channel_cli._thread_channel_origins.clear() + yield + channel_cli._manager = original["_manager"] + channel_cli._bus_loop = original["_bus_loop"] + channel_cli._bus_thread = original["_bus_thread"] + with channel_cli._thread_channel_origins_lock: + channel_cli._thread_channel_origins.clear() + channel_cli._thread_channel_origins.update(original_origins) + + +def _make_msg( + *, + channel_type: str = "imessage", + chat_id: str = "+15551234567", + metadata: dict | None = None, +) -> ChannelMessage: + return ChannelMessage( + msg_id="msg-1", + content="hi", + sender="alice", + channel_type=channel_type, + metadata=metadata if metadata is not None else {"foo": "bar"}, + chat_id=chat_id, + ) + + +def _install_fake_bus( + monkeypatch, +) -> tuple[asyncio.AbstractEventLoop, MagicMock, threading.Thread]: + """Spin up a real asyncio loop on a background thread + a stub bus. + + Returns (loop, publish_outbound_mock, thread). The caller is responsible + for stopping the loop at the end of the test. + """ + loop = asyncio.new_event_loop() + ready = threading.Event() + + def _runner(): + asyncio.set_event_loop(loop) + ready.set() + loop.run_forever() + + thread = threading.Thread(target=_runner, daemon=True) + thread.start() + ready.wait(timeout=2) + + publish_outbound = MagicMock() + + async def _publish_outbound(msg): + publish_outbound(msg) + + bus = MagicMock() + bus.publish_outbound = _publish_outbound + manager = MagicMock() + manager.bus = bus + + monkeypatch.setattr(channel_cli, "_bus_loop", loop) + monkeypatch.setattr(channel_cli, "_manager", manager) + return loop, publish_outbound, thread + + +def _stop_loop(loop: asyncio.AbstractEventLoop, thread: threading.Thread) -> None: + loop.call_soon_threadsafe(loop.stop) + thread.join(timeout=2) + loop.close() + + +def _wait_for_publish( + mock: MagicMock, *, expected: int = 1, timeout: float = 2.0 +) -> None: + """Block until ``mock`` has been called ``expected`` times (or timeout).""" + import time + + deadline = time.time() + timeout + while mock.call_count < expected and time.time() < deadline: + time.sleep(0.01) + + +def test_remember_and_publish_roundtrip(monkeypatch): + loop, publish_outbound, thread = _install_fake_bus(monkeypatch) + try: + msg = _make_msg(metadata={"thread_root": "root-1"}) + channel_cli.remember_channel_origin("tid-1", msg) + + assert channel_cli.publish_to_channel_origin("tid-1", "hello") is True + _wait_for_publish(publish_outbound) + + publish_outbound.assert_called_once() + sent = publish_outbound.call_args.args[0] + assert sent.channel == "imessage" + assert sent.chat_id == "+15551234567" + assert sent.content == "hello" + assert sent.metadata == {"thread_root": "root-1"} + assert sent.reply_to is None + finally: + _stop_loop(loop, thread) + + +def test_origin_remembers_sender_distinct_from_chat_id(): + """The origin stores the human-readable sender separately from chat_id, so + the notifier closer can show the same handle a normal turn shows. + + Regression: iMessage exposes an internal chat id (e.g. "1") as chat_id + while the handle (phone number) lives on sender. The closer must display + sender, not chat_id, or the user sees a confusing "Replied to 1". + """ + channel_cli.remember_channel_origin( + "tid-sender", + _make_msg(chat_id="1"), # iMessage-style internal chat id + ) + origin = channel_cli.get_channel_origin("tid-sender") + assert origin is not None + assert origin.chat_id == "1" # routing id preserved + assert origin.sender == "alice" # human handle preserved for display + + +def test_publish_records_sent_metric(monkeypatch): + """A forwarded notification records a "sent" metric, mirroring the normal + channel-reply path (``manager.record_message``).""" + loop, _publish_outbound, thread = _install_fake_bus(monkeypatch) + try: + channel_cli.remember_channel_origin( + "tid-metric", _make_msg(channel_type="telegram") + ) + assert channel_cli.publish_to_channel_origin("tid-metric", "done") is True + # record_message runs right after publish_outbound inside the same + # coroutine; sync on it before asserting. + _wait_for_publish(channel_cli._manager.record_message) + channel_cli._manager.record_message.assert_called_once_with("telegram", "sent") + finally: + _stop_loop(loop, thread) + + +def test_publish_returns_false_without_origin(monkeypatch): + loop, publish_outbound, thread = _install_fake_bus(monkeypatch) + try: + assert channel_cli.publish_to_channel_origin("unknown-tid", "hi") is False + publish_outbound.assert_not_called() + finally: + _stop_loop(loop, thread) + + +def test_publish_returns_false_when_bus_down(monkeypatch): + monkeypatch.setattr(channel_cli, "_bus_loop", None) + monkeypatch.setattr(channel_cli, "_manager", None) + + channel_cli.remember_channel_origin("tid-2", _make_msg()) + assert channel_cli.publish_to_channel_origin("tid-2", "hi") is False + + +def test_publish_returns_false_for_empty_content(monkeypatch): + loop, publish_outbound, thread = _install_fake_bus(monkeypatch) + try: + channel_cli.remember_channel_origin("tid-3", _make_msg()) + assert channel_cli.publish_to_channel_origin("tid-3", "") is False + assert channel_cli.publish_to_channel_origin("tid-3", " \n ") is False + publish_outbound.assert_not_called() + finally: + _stop_loop(loop, thread) + + +def test_forget_origin(monkeypatch): + loop, publish_outbound, thread = _install_fake_bus(monkeypatch) + try: + channel_cli.remember_channel_origin("tid-4", _make_msg()) + channel_cli.forget_channel_origin("tid-4") + assert channel_cli.publish_to_channel_origin("tid-4", "hi") is False + publish_outbound.assert_not_called() + finally: + _stop_loop(loop, thread) + + +def test_remember_overwrites_same_thread(monkeypatch): + loop, publish_outbound, thread = _install_fake_bus(monkeypatch) + try: + channel_cli.remember_channel_origin( + "tid-5", _make_msg(channel_type="telegram", chat_id="111") + ) + channel_cli.remember_channel_origin( + "tid-5", _make_msg(channel_type="imessage", chat_id="222") + ) + assert channel_cli.publish_to_channel_origin("tid-5", "hi") is True + _wait_for_publish(publish_outbound) + sent = publish_outbound.call_args.args[0] + assert sent.channel == "imessage" + assert sent.chat_id == "222" + finally: + _stop_loop(loop, thread) + + +def test_publish_swallows_bus_error(monkeypatch, caplog): + loop = asyncio.new_event_loop() + ready = threading.Event() + + def _runner(): + asyncio.set_event_loop(loop) + ready.set() + loop.run_forever() + + thread = threading.Thread(target=_runner, daemon=True) + thread.start() + ready.wait(timeout=2) + + async def _boom(_msg): + raise RuntimeError("bus dead") + + bus = MagicMock() + bus.publish_outbound = _boom + manager = MagicMock() + manager.bus = bus + monkeypatch.setattr(channel_cli, "_bus_loop", loop) + monkeypatch.setattr(channel_cli, "_manager", manager) + + channel_cli.remember_channel_origin("tid-6", _make_msg()) + try: + with caplog.at_level("WARNING", logger="EvoScientist.cli.channel"): + # The publish is scheduled (returns True); the coroutine raises + # asynchronously and the done-callback logs the failure. + assert channel_cli.publish_to_channel_origin("tid-6", "hi") is True + # Give the bus thread a moment to run the coroutine + callback. + import time + + deadline = time.time() + 2.0 + while ( + not any( + "Async notification publish" in r.message for r in caplog.records + ) + and time.time() < deadline + ): + time.sleep(0.01) + assert any("Async notification publish" in r.message for r in caplog.records) + finally: + _stop_loop(loop, thread) + + +def test_remember_with_falsy_thread_id_noop(monkeypatch): + """Defensive: a falsy thread_id must not pollute the registry.""" + msg = _make_msg() + channel_cli.remember_channel_origin(None, msg) + channel_cli.remember_channel_origin("", msg) + with channel_cli._thread_channel_origins_lock: + assert channel_cli._thread_channel_origins == {} + + +def test_get_channel_origin_returns_none_for_unknown(): + assert channel_cli.get_channel_origin("never-registered") is None + assert channel_cli.get_channel_origin(None) is None