diff --git a/EvoScientist/channels/base.py b/EvoScientist/channels/base.py index 7478adf..6ea5817 100644 --- a/EvoScientist/channels/base.py +++ b/EvoScientist/channels/base.py @@ -22,6 +22,7 @@ from .bus.events import InboundMessage, OutboundMessage from .capabilities import ChannelCapabilities from .debug import TraceMixin, debug_trace_enabled from .formatter import UnifiedFormatter +from .interaction import is_slash_command from .plugin import ChannelMeta, ChannelPlugin _logger = logging.getLogger(__name__) @@ -1057,6 +1058,43 @@ class Channel(TraceMixin, ChannelPlugin, ABC): """Buffer *msg* with debounce, then publish to bus.""" sender = msg.sender_id + if self._on_activity: + try: + self._on_activity(sender, "received") + except Exception: + pass + + # Slash commands are control messages, not prompt fragments. Flush any + # prompt already waiting for this sender, then publish the command as + # its own message so either arrival order cannot newline-merge them. + if is_slash_command(msg.content) and self._bus: + # A flush removes itself from this mapping before awaiting the bus + # publish. Therefore a task still present here has not detached + # its buffered payload yet and is safe to cancel; an in-flight, + # backpressured publish is deliberately left alone. + debounce_task = self._debounce_tasks.pop(sender, None) + if debounce_task is not None: + debounce_task.cancel() + try: + await debounce_task + except asyncio.CancelledError: + # Awaiting a cancelled child normally raises here with no + # cancellation pending on this task. If our caller also + # cancelled queue_message(), preserve that outer signal. + current = asyncio.current_task() + if current is not None and current.cancelling() > 0: + raise + try: + await self._process_buffered_messages(sender) + except Exception: + _logger.error( + f"{self.name} buffered-prompt flush failed for {sender}; " + "publishing the command anyway", + exc_info=True, + ) + await self._bus.publish_inbound(msg) + return + if sender not in self._message_buffers: self._message_buffers[sender] = [] self._message_metadata[sender] = msg.metadata @@ -1069,12 +1107,6 @@ class Channel(TraceMixin, ChannelPlugin, ABC): if msg.media: self._message_media[sender].extend(msg.media) - if self._on_activity: - try: - self._on_activity(sender, "received") - except Exception: - pass - if sender in self._debounce_tasks: self._debounce_tasks[sender].cancel() @@ -1089,8 +1121,10 @@ class Channel(TraceMixin, ChannelPlugin, ABC): await asyncio.sleep(_w) try: await self._process_buffered_messages(_s) - except Exception as e: - _logger.error(f"{self.name} debounce flush error for {_s}: {e}") + except Exception: + _logger.error( + f"{self.name} debounce flush error for {_s}", exc_info=True + ) self._debounce_tasks[sender] = asyncio.create_task(debounce_callback()) diff --git a/EvoScientist/channels/bus/events.py b/EvoScientist/channels/bus/events.py index 411006f..f08c017 100644 --- a/EvoScientist/channels/bus/events.py +++ b/EvoScientist/channels/bus/events.py @@ -45,6 +45,7 @@ class OutboundMessage: reply_to: str | None = None media: list[str] = field(default_factory=list) metadata: dict[str, Any] = field(default_factory=dict) + failure_notice: str | None = None @property def recipient(self) -> str: diff --git a/EvoScientist/channels/channel_manager.py b/EvoScientist/channels/channel_manager.py index bf010c5..4016576 100644 --- a/EvoScientist/channels/channel_manager.py +++ b/EvoScientist/channels/channel_manager.py @@ -17,7 +17,7 @@ import logging import pkgutil import time from collections.abc import Callable -from dataclasses import dataclass, field +from dataclasses import dataclass, field, replace from datetime import datetime from pathlib import Path from typing import Any @@ -29,6 +29,9 @@ from .plugin import ChannelPlugin logger = logging.getLogger(__name__) +# Best-effort failure notices must never wedge the dispatcher on a hung send. +_FAILURE_NOTICE_TIMEOUT = 15.0 + CHANNEL_STARTUP_PENDING_DETAIL = "starting (bus)" @@ -743,6 +746,12 @@ class ChannelManager: delivery_failed = True if not delivery_failed and (msg.content or msg.media): drained += 1 + elif delivery_failed: + await self._send_failure_notice( + channel, + msg, + timeout=max(1.0, deadline - time.monotonic()), + ) dropped = self.bus.outbound.qsize() if drained or dropped: logger.info(f"Outbound drain: {drained} sent, {dropped} dropped") @@ -843,6 +852,50 @@ class ChannelManager: # ── outbound routing ── + def _record_outbound_failure(self, channel_name: str, error: str) -> None: + health = self._health.get(channel_name) + if health is None: + return + health.consecutive_failures += 1 + health.total_failures += 1 + health.last_failure_time = time.time() + health.last_failure_error = error + + async def _send_failure_notice( + self, + channel: Channel, + msg: OutboundMessage, + *, + timeout: float | None = None, + ) -> None: + """Best-effort short notice when the real payload could not be sent.""" + if not msg.failure_notice: + return + fallback = replace( + msg, + content=msg.failure_notice, + media=[], + failure_notice=None, + ) + try: + coro = channel.send(fallback) + if timeout is not None: + coro = asyncio.wait_for(coro, timeout=timeout) + fallback_ok = await coro + except Exception as fallback_error: + logger.error( + "Error sending delivery failure notice to %s: %s", + msg.channel, + fallback_error, + ) + else: + if not fallback_ok: + logger.error( + "Error sending delivery failure notice to %s: " + "send() returned False", + msg.channel, + ) + async def _dispatch_outbound(self) -> None: """Route outbound messages from the bus to the correct channel.""" logger.info("Outbound dispatcher started") @@ -872,13 +925,20 @@ class ChannelManager: msg = processed delivery_failed = False + failure_error = "one or more outbound deliveries failed" if msg.content: - text_ok = await channel.send(msg) - if not text_ok: - logger.error( - f"Error sending to {msg.channel}: send() returned False" - ) + try: + text_ok = await channel.send(msg) + except Exception as e: + logger.error(f"Error sending to {msg.channel}", exc_info=True) + failure_error = str(e) delivery_failed = True + else: + if not text_ok: + logger.error( + f"Error sending to {msg.channel}: send() returned False" + ) + delivery_failed = True for media_path in msg.media: try: @@ -894,11 +954,18 @@ class ChannelManager: ) delivery_failed = True except Exception as e: - logger.error(f"Error sending media to {msg.channel}: {e}") + logger.error( + f"Error sending media to {msg.channel}", exc_info=True + ) + failure_error = str(e) delivery_failed = True if delivery_failed: - raise RuntimeError("one or more outbound deliveries failed") + await self._send_failure_notice( + channel, msg, timeout=_FAILURE_NOTICE_TIMEOUT + ) + self._record_outbound_failure(msg.channel, failure_error) + continue # Success health = self._health.get(msg.channel) @@ -906,13 +973,12 @@ class ChannelManager: health.consecutive_failures = 0 health.total_successes += 1 except Exception as e: - logger.error(f"Error sending to {msg.channel}: {e}") - health = self._health.get(msg.channel) - if health is not None: - health.consecutive_failures += 1 - health.total_failures += 1 - health.last_failure_time = time.monotonic() - health.last_failure_error = str(e) + # Unexpected internal error (pipeline, bookkeeping) — the + # transport paths above handle their own failures. + logger.error( + f"Outbound dispatch error for {msg.channel}", exc_info=True + ) + self._record_outbound_failure(msg.channel, str(e)) # ── per-account lifecycle ── diff --git a/EvoScientist/channels/interaction.py b/EvoScientist/channels/interaction.py index 4882c7e..a45fdf0 100644 --- a/EvoScientist/channels/interaction.py +++ b/EvoScientist/channels/interaction.py @@ -41,6 +41,12 @@ OTHER_PROMPT = "Please type your answer:" # ── stop / cancel helpers ────────────────────────────────────────────── +def is_slash_command(text: str | None) -> bool: + """Whether inbound content is a slash command (a control message, not a + prompt fragment).""" + return (text or "").lstrip().startswith("/") + + 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/channels/middleware.py b/EvoScientist/channels/middleware.py index ba8b60e..4813264 100644 --- a/EvoScientist/channels/middleware.py +++ b/EvoScientist/channels/middleware.py @@ -23,6 +23,7 @@ from typing import Any from .base import RawIncoming from .bus.events import InboundMessage, OutboundMessage from .debug import emit_debug_event_if +from .interaction import is_slash_command _logger = logging.getLogger(__name__) @@ -811,8 +812,21 @@ class MentionGatingMiddleware(InboundMiddleware): policy=self.require_mention, ) return None - # Strip mentions from group messages - if raw.is_group and self._strip_fn: + # A slash command's platform target belongs only to its first token; + # preserve mentions in its arguments. Ordinary group messages may + # still carry a bot mention elsewhere and use the full-message strip. + if self._strip_fn and is_slash_command(raw.text): + text = raw.text + token_start = len(text) - len(text.lstrip()) + token_end = token_start + while token_end < len(text) and not text[token_end].isspace(): + token_end += 1 + stripped_token = self._strip_fn(text[token_start:token_end]) + raw = dataclasses.replace( + raw, + text=text[:token_start] + stripped_token + text[token_end:], + ) + elif self._strip_fn and raw.is_group: raw = dataclasses.replace(raw, text=self._strip_fn(raw.text)) return raw @@ -928,6 +942,12 @@ class GroupHistoryMiddleware(InboundMiddleware): # Don't drop here — let MentionGatingMiddleware handle that return raw + # Slash commands must remain the leading content so channel command + # dispatchers can recognize them. Keep buffered chatter for the next + # normal mentioned message instead of injecting it ahead of a command. + if is_slash_command(raw.text): + return raw + # Mentioned: inject history context history_context = self._buffer.format_context(raw.chat_id) if history_context: diff --git a/EvoScientist/channels/telegram/channel.py b/EvoScientist/channels/telegram/channel.py index a6869f4..c2ef180 100644 --- a/EvoScientist/channels/telegram/channel.py +++ b/EvoScientist/channels/telegram/channel.py @@ -80,9 +80,7 @@ class TelegramChannel(Channel): | filters.LOCATION ) - self._app.add_handler( - MessageHandler(media_filter & ~filters.COMMAND, self._on_message) - ) + self._app.add_handler(MessageHandler(media_filter, self._on_message)) await self._app.initialize() # Cache bot username for @mention detection in groups @@ -166,6 +164,21 @@ class TelegramChannel(Channel): def _get_bot_identifier(self) -> str | None: return self._bot_username or None + @staticmethod + def _command_target(text: str) -> str | None: + """Return a Telegram command's target username. + + An empty string represents a bare command; ``None`` means the message + is not command-shaped. + """ + parts = text.lstrip().split(None, 1) + if not parts or not parts[0].startswith("/"): + return None + command_token = parts[0][1:] + if "@" not in command_token: + return "" + return command_token.rsplit("@", 1)[1].lower() + async def _send_ack_reaction( self, chat_id: str, message_id: str, emoji: str = "👀" ) -> None: @@ -207,10 +220,19 @@ class TelegramChannel(Channel): # Detect group and mention status for centralized gating is_group = message.chat.type in ("group", "supergroup") - was_mentioned = True # DM default - if is_group and self._bot_username: + was_mentioned = not is_group + if is_group: text_check = (message.text or message.caption or "").lower() - was_mentioned = f"@{self._bot_username}" in text_check + command_target = self._command_target(text_check) + if command_target is not None: + # A bare command that Telegram delivered to this bot is + # actionable. Commands explicitly addressed to another bot + # must remain ignored. + was_mentioned = not command_target or ( + bool(self._bot_username) and command_target == self._bot_username + ) + elif self._bot_username: + was_mentioned = f"@{self._bot_username}" in text_check content_parts: list[str] = [] media_paths: list[str] = [] diff --git a/EvoScientist/cli/channel.py b/EvoScientist/cli/channel.py index f0857a7..1e469a7 100644 --- a/EvoScientist/cli/channel.py +++ b/EvoScientist/cli/channel.py @@ -33,6 +33,7 @@ from ..channels.interaction import ( ApprovalPolicy, InteractionIO, PendingReplyRegistry, + is_slash_command, is_stop_command, resolve_approval, resolve_ask_user, @@ -73,6 +74,9 @@ _message_queue: queue.Queue[ChannelMessage] = queue.Queue() # Pending responses: # main → bus (msg_id → {"future": Future[str], "loop": loop, "response": str|None}) _pending_responses: dict[str, dict] = {} +# Sentinel response: the command's output already reached the channel via the +# command UI, so the bus consumer must not deliver a second message. +COMMAND_OUTPUT_ALREADY_SENT = "__evosci-command-output-already-sent__" _response_lock = threading.Lock() _RESPONSE_TIMEOUT = 600.0 @@ -326,7 +330,7 @@ async def dispatch_channel_slash_command( ``cli/interactive.py:1002-1030``. Headless serve passes ``None`` since it cannot hot-swap its polling-loop agent. """ - if not msg.content.strip().startswith("/"): + if not is_slash_command(msg.content): return False try: @@ -391,10 +395,17 @@ async def _dispatch_channel_slash_impl( from ..commands.channel_ui import ChannelCommandUI from ..commands.manager import manager as cmd_manager + # The wrapper only forwards slash-prefixed content, so an unresolved + # parse is always an unknown command — answer instead of feeding a typo + # to the agent. parsed = cmd_manager.resolve(msg.content) if parsed is None: - # Unknown slash command — let the agent handle it (matches TUI). - return False + bad_cmd = msg.content.split(None, 1)[0] + _set_channel_response( + msg.msg_id, + f"Unknown command: {bad_cmd}\nType /help to see available commands.", + ) + return True cmd, cmd_args = parsed agent_for_ctx = agent @@ -431,8 +442,11 @@ async def _dispatch_channel_slash_impl( if cmd_executed: if ctx.command_error is not None: - details = ctx.command_error or "(no details)" - _set_channel_response(msg.msg_id, f"Command error: {details}") + if ui.sent_to_channel: + _set_channel_response(msg.msg_id, COMMAND_OUTPUT_ALREADY_SENT) + else: + details = ctx.command_error or "(no details)" + _set_channel_response(msg.msg_id, f"Command error: {details}") return True if on_cmd_completed is not None: @@ -452,7 +466,12 @@ async def _dispatch_channel_slash_impl( f"[{msg.channel_type}: Executed command from {msg.sender}]", "dim", ) - _set_channel_response(msg.msg_id, f"Command executed: {msg.content}") + if ui.sent_to_channel: + # The user already saw the command's own output — a second + # "Command executed" message is just noise. + _set_channel_response(msg.msg_id, COMMAND_OUTPUT_ALREADY_SENT) + else: + _set_channel_response(msg.msg_id, f"Command executed: {msg.content}") return True # ``cmd_manager.execute`` returned False (empty / unparseable input). @@ -1161,16 +1180,21 @@ async def _handle_bus_message(bus, manager, msg) -> None: return response = _pop_channel_response(cm.msg_id) or "No response" - await bus.publish_outbound( - OutboundMessage( - channel=msg.channel, - chat_id=msg.chat_id, - content=response, - reply_to=msg.message_id or None, - metadata=msg.metadata, + if response != COMMAND_OUTPUT_ALREADY_SENT: + await bus.publish_outbound( + OutboundMessage( + channel=msg.channel, + chat_id=msg.chat_id, + content=response, + reply_to=msg.message_id or None, + metadata=msg.metadata, + ) ) - ) - manager.record_message(msg.channel, "sent") + manager.record_message(msg.channel, "sent") + else: + # The command UI published its own response before returning the + # sentinel, so account for that delivery without sending an ack. + manager.record_message(msg.channel, "sent") except asyncio.CancelledError: _pop_channel_response(cm.msg_id, cancel_pending=True) if _channel_request_state(cm.msg_id) != "active": diff --git a/EvoScientist/commands/channel_ui.py b/EvoScientist/commands/channel_ui.py index 0d2ba36..e507fa8 100644 --- a/EvoScientist/commands/channel_ui.py +++ b/EvoScientist/commands/channel_ui.py @@ -12,6 +12,8 @@ if TYPE_CHECKING: _logger = logging.getLogger(__name__) +_COMMAND_OUTPUT_FAILURE_NOTICE = "Command output could not be delivered." + class ChannelCommandUI(CommandUI): """CommandUI implementation for messaging channels with output buffering.""" @@ -37,6 +39,10 @@ class ChannelCommandUI(CommandUI): self.handle_session_resume_callback = handle_session_resume_callback self.graph_gateway = graph_gateway self._system_buffer: list[str] = [] + # Whether any output was delivered (or scheduled for delivery) to the + # channel. The slash dispatcher consults this to decide between a + # bare completion ack and staying silent. + self.sent_to_channel: bool = False def _queue_system( self, @@ -107,6 +113,7 @@ class ChannelCommandUI(CommandUI): content=grouped_text, reply_to=self.msg.message_id, metadata=self.msg.metadata, + failure_notice=_COMMAND_OUTPUT_FAILURE_NOTICE, ) if self.msg.bus_ref: @@ -114,6 +121,7 @@ class ChannelCommandUI(CommandUI): else: coro = self.msg.channel_ref.send(outbound) + self.sent_to_channel = True asyncio.run_coroutine_threadsafe(coro, loop) def mount_renderable(self, renderable: Any) -> None: @@ -147,6 +155,7 @@ class ChannelCommandUI(CommandUI): content=f"```\n{text}\n```", reply_to=self.msg.message_id, metadata=self.msg.metadata, + failure_notice=_COMMAND_OUTPUT_FAILURE_NOTICE, ) if self.msg.bus_ref: @@ -154,6 +163,7 @@ class ChannelCommandUI(CommandUI): else: coro = self.msg.channel_ref.send(outbound) + self.sent_to_channel = True asyncio.run_coroutine_threadsafe(coro, loop) async def wait_for_thread_pick( diff --git a/tests/test_bus_integration.py b/tests/test_bus_integration.py index f121637..3f3cc5e 100644 --- a/tests/test_bus_integration.py +++ b/tests/test_bus_integration.py @@ -110,6 +110,53 @@ class TestBusInboundConsumer: except asyncio.CancelledError: pass + async def test_already_sent_sentinel_suppresses_reply(self): + """A command whose output already reached the channel must not get a + second "Command executed" style reply from the consumer.""" + from EvoScientist.cli.channel import ( + COMMAND_OUTPUT_ALREADY_SENT, + _bus_inbound_consumer, + _message_queue, + _set_channel_response, + ) + + _drain_queue(_message_queue) + + bus = MessageBus() + manager = ChannelManager(bus) + ch = FakeChannel() + manager.register(ch) + + consumer = asyncio.create_task(_bus_inbound_consumer(bus, manager)) + + await bus.publish_inbound( + InboundMessage( + channel="fake", + sender_id="user1", + chat_id="chat1", + content="/help", + ) + ) + + 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, COMMAND_OUTPUT_ALREADY_SENT) + + # Nothing must be published for this message. + with pytest.raises(TimeoutError): + await asyncio.wait_for(bus.consume_outbound(), timeout=0.5) + assert manager._message_counts["fake"]["sent"] == 1 + + consumer.cancel() + try: + await consumer + except asyncio.CancelledError: + pass + async def test_no_response_fallback(self): """Empty response is replaced with 'No response' fallback.""" from EvoScientist.cli.channel import ( diff --git a/tests/test_channel_command_ui.py b/tests/test_channel_command_ui.py index a5c8298..c3220cb 100644 --- a/tests/test_channel_command_ui.py +++ b/tests/test_channel_command_ui.py @@ -138,3 +138,29 @@ async def test_handle_session_resume_distinguishes_non_displayable_messages(): ] text = _sent_text(bus_ref) assert "No displayable messages in this session." in text + + +class TestSentToChannelFlag: + async def test_starts_false(self): + ui, _ = _make_ui(thread_store=FakeThreadStore()) + assert ui.sent_to_channel is False + + async def test_flush_of_empty_buffer_keeps_flag_false(self, monkeypatch): + ui, _ = _make_ui(thread_store=FakeThreadStore()) + loop = asyncio.get_running_loop() + monkeypatch.setattr("EvoScientist.cli.channel._bus_loop", loop) + await ui.flush() + assert ui.sent_to_channel is False + + async def test_flush_with_output_sets_flag(self, monkeypatch): + bus = SimpleNamespace(publish_outbound=AsyncMock()) + ui, _ = _make_ui(thread_store=FakeThreadStore(), bus_ref=bus) + loop = asyncio.get_running_loop() + monkeypatch.setattr("EvoScientist.cli.channel._bus_loop", loop) + ui.append_system("hello") + await ui.flush() + await asyncio.sleep(0) + await asyncio.sleep(0) + assert ui.sent_to_channel is True + outbound = bus.publish_outbound.await_args.args[0] + assert outbound.failure_notice == "Command output could not be delivered." diff --git a/tests/test_channel_comprehensive.py b/tests/test_channel_comprehensive.py index 57f0f73..d556653 100644 --- a/tests/test_channel_comprehensive.py +++ b/tests/test_channel_comprehensive.py @@ -37,7 +37,7 @@ from EvoScientist.channels.bus.message_bus import MessageBus from EvoScientist.channels.channel_manager import ChannelManager from EvoScientist.channels.consumer import InboundConsumer from EvoScientist.channels.formatter import convert_markdown -from EvoScientist.channels.middleware import DedupCache +from EvoScientist.channels.middleware import DedupCache, MentionGatingMiddleware from EvoScientist.channels.retry import RetryConfig, RetryInfo, retry_async # ═══════════════════════════════════════════════════════════════════ @@ -590,6 +590,47 @@ class TestChannelMentionGating: ) assert ch._should_process(raw) is True + async def test_private_plain_text_keeps_mentions_but_command_strips_suffix(self): + middleware = MentionGatingMiddleware( + require_mention="group", + strip_fn=lambda text: text.replace("@botname", ""), + ) + + plain = await middleware.process_inbound( + RawIncoming( + sender_id="u1", + chat_id="c1", + text="please ask @botname about this", + is_group=False, + ), + {}, + ) + command = await middleware.process_inbound( + RawIncoming( + sender_id="u1", + chat_id="c1", + text="/help@botname", + is_group=False, + ), + {}, + ) + command_with_mention_argument = await middleware.process_inbound( + RawIncoming( + sender_id="u1", + chat_id="c1", + text=" /help@botname ask @botname for status", + is_group=False, + ), + {}, + ) + + assert plain is not None + assert plain.text == "please ask @botname about this" + assert command is not None + assert command.text == "/help" + assert command_with_mention_argument is not None + assert command_with_mention_argument.text == " /help ask @botname for status" + class TestChannelBuildInbound: def test_builds_valid_inbound(self): @@ -769,6 +810,197 @@ class TestChannelDebounce: assert "part1" in received.content assert "part2" in received.content + async def test_command_flushes_pending_prompt_as_separate_message(self): + bus = MessageBus() + ch = StubChannel() + ch.set_bus(bus) + + await ch.queue_message( + InboundMessage( + channel="stub", + sender_id="u1", + chat_id="c1", + content="do X", + message_id="m1", + ) + ) + await ch.queue_message( + InboundMessage( + channel="stub", + sender_id="u1", + chat_id="c1", + content="/stop", + message_id="m2", + ) + ) + + first = await bus.consume_inbound() + second = await bus.consume_inbound() + assert (first.content, second.content) == ("do X", "/stop") + + async def test_command_does_not_cancel_backpressured_prompt_flush(self): + """Once a prompt has detached from the debounce buffer and is waiting + for queue capacity, a later command must remain behind it without + cancelling or losing either message.""" + bus = MessageBus() + bus.inbound = asyncio.Queue(maxsize=1) + ch = StubChannel() + ch.set_bus(bus) + ch.initial_debounce = 0 + + await bus.publish_inbound( + InboundMessage( + channel="stub", + sender_id="blocker", + chat_id="blocker", + content="queue filler", + ) + ) + await ch.queue_message( + InboundMessage( + channel="stub", + sender_id="u1", + chat_id="c1", + content="do X", + message_id="m1", + ) + ) + flush_task = ch._debounce_tasks["u1"] + await asyncio.wait_for( + _wait_for_async( + lambda: ( + "u1" not in ch._message_buffers and "u1" not in ch._debounce_tasks + ) + ), + timeout=1.0, + ) + assert not flush_task.done() + + command_task = asyncio.create_task( + ch.queue_message( + InboundMessage( + channel="stub", + sender_id="u1", + chat_id="c1", + content="/help", + message_id="m2", + ) + ) + ) + await asyncio.sleep(0) + assert not command_task.done() + + filler = await asyncio.wait_for(bus.consume_inbound(), timeout=1.0) + prompt = await asyncio.wait_for(bus.consume_inbound(), timeout=1.0) + command = await asyncio.wait_for(bus.consume_inbound(), timeout=1.0) + await asyncio.wait_for(flush_task, timeout=1.0) + await asyncio.wait_for(command_task, timeout=1.0) + + assert filler.content == "queue filler" + assert (prompt.content, command.content) == ("do X", "/help") + + async def test_command_wait_preserves_outer_cancellation(self): + bus = MessageBus() + ch = StubChannel() + ch.set_bus(bus) + child_cancelling = asyncio.Event() + release_child = asyncio.Event() + + async def slow_to_cancel(): + try: + await asyncio.Future() + except asyncio.CancelledError: + child_cancelling.set() + await release_child.wait() + raise + + debounce_task = asyncio.create_task(slow_to_cancel()) + ch._debounce_tasks["u1"] = debounce_task + command_task = asyncio.create_task( + ch.queue_message( + InboundMessage( + channel="stub", + sender_id="u1", + chat_id="c1", + content="/help", + message_id="m2", + ) + ) + ) + + await asyncio.wait_for(child_cancelling.wait(), timeout=1.0) + command_task.cancel() + release_child.set() + with pytest.raises(asyncio.CancelledError): + await command_task + assert bus.inbound.empty() + + async def test_command_publishes_even_when_buffer_flush_fails(self): + """A failing buffered-prompt flush must not swallow the command — + losing /stop exactly when the pipeline misbehaves is the worst case.""" + bus = MessageBus() + ch = StubChannel() + ch.set_bus(bus) + + await ch.queue_message( + InboundMessage( + channel="stub", + sender_id="u1", + chat_id="c1", + content="do X", + message_id="m1", + ) + ) + + async def _boom(sender): + raise RuntimeError("flush broke") + + ch._process_buffered_messages = _boom + await ch.queue_message( + InboundMessage( + channel="stub", + sender_id="u1", + chat_id="c1", + content="/stop", + message_id="m2", + ) + ) + + published = await asyncio.wait_for(bus.consume_inbound(), timeout=1.0) + assert published.content == "/stop" + + async def test_prompt_after_command_starts_new_debounce_batch(self): + bus = MessageBus() + ch = StubChannel() + ch.set_bus(bus) + + await ch.queue_message( + InboundMessage( + channel="stub", + sender_id="u1", + chat_id="c1", + content="/new", + message_id="m1", + ) + ) + await ch.queue_message( + InboundMessage( + channel="stub", + sender_id="u1", + chat_id="c1", + content="summarize this paper", + message_id="m2", + ) + ) + await _flush_debounce(ch, "u1") + + first = await bus.consume_inbound() + second = await bus.consume_inbound() + assert (first.content, second.content) == ( + "/new", + "summarize this paper", + ) + async def test_dedup_skips_duplicate(self): """Dedup is now handled in _enqueue_raw pipeline, not queue_message.""" @@ -1133,6 +1365,119 @@ class TestChannelManagerDispatch: assert health.total_failures == 1 assert health.consecutive_failures == 1 + async def test_dispatch_uses_short_notice_when_command_output_fails(self): + bus = MessageBus() + mgr = ChannelManager(bus) + ch = StubChannel() + sent: list[OutboundMessage] = [] + + async def fail_content_then_send_notice(msg): + sent.append(msg) + return len(sent) > 1 + + ch.send = fail_content_then_send_notice + mgr.register(ch) + + task = asyncio.create_task(mgr._dispatch_outbound()) + await bus.publish_outbound( + OutboundMessage( + channel="stub", + chat_id="c1", + content="content rejected by platform", + failure_notice="Command output could not be delivered.", + ) + ) + await asyncio.wait_for( + _wait_for_async(lambda: mgr._health["stub"].total_failures == 1), + timeout=1.0, + ) + task.cancel() + try: + await task + except asyncio.CancelledError: + pass + + assert [message.content for message in sent] == [ + "content rejected by platform", + "Command output could not be delivered.", + ] + assert sent[1].failure_notice is None + + async def test_shutdown_drain_sends_failure_notice(self): + """The stop_all drain mirrors the dispatch fallback: a payload that + fails during shutdown still produces the short failure notice.""" + bus = MessageBus() + mgr = ChannelManager(bus) + ch = StubChannel() + sent: list[OutboundMessage] = [] + + async def fail_content_then_send_notice(msg): + sent.append(msg) + return len(sent) > 1 + + ch.send = fail_content_then_send_notice + mgr.register(ch) + ch._running = True + + await bus.publish_outbound( + OutboundMessage( + channel="stub", + chat_id="c1", + content="content rejected by platform", + failure_notice="Command output could not be delivered.", + ) + ) + + await mgr.stop_all() + + assert [message.content for message in sent] == [ + "content rejected by platform", + "Command output could not be delivered.", + ] + assert sent[1].failure_notice is None + + async def test_dispatch_sends_notice_when_content_send_raises(self): + """A raising send() must reach the failure notice, not the outer + handler — exceptions are the common transport failure mode.""" + bus = MessageBus() + mgr = ChannelManager(bus) + ch = StubChannel() + sent: list[OutboundMessage] = [] + + async def raise_then_send_notice(msg): + sent.append(msg) + if len(sent) == 1: + raise RuntimeError("network down") + return True + + ch.send = raise_then_send_notice + mgr.register(ch) + + task = asyncio.create_task(mgr._dispatch_outbound()) + await bus.publish_outbound( + OutboundMessage( + channel="stub", + chat_id="c1", + content="payload", + failure_notice="Command output could not be delivered.", + ) + ) + await asyncio.wait_for( + _wait_for_async(lambda: mgr._health["stub"].total_failures == 1), + timeout=1.0, + ) + task.cancel() + try: + await task + except asyncio.CancelledError: + pass + + assert [message.content for message in sent] == [ + "payload", + "Command output could not be delivered.", + ] + assert mgr._health["stub"].last_failure_error == "network down" + async def test_dispatch_send_media_return_false_counts_failure(self): """send_media() returning False should mark the delivery as failed.""" diff --git a/tests/test_cli_channel_slash.py b/tests/test_cli_channel_slash.py index c3d8916..18cc1fc 100644 --- a/tests/test_cli_channel_slash.py +++ b/tests/test_cli_channel_slash.py @@ -6,6 +6,8 @@ fed to the LLM as a plain prompt, on every UI surface (Rich CLI, TUI, headless ``serve``). """ +import asyncio +from types import SimpleNamespace from unittest.mock import AsyncMock, MagicMock, patch from EvoScientist.cli.channel import ( @@ -58,23 +60,28 @@ async def test_non_slash_returns_false(): append.assert_not_called() -async def test_unresolved_slash_returns_false(): - """Unknown slash commands must fall through (matches TUI behavior).""" +async def test_unresolved_slash_returns_unknown_command_response(): + """Unknown slash commands must never fall through to the agent.""" msg = _make_msg(content="/unknown-cmd") append = MagicMock() with patch( "EvoScientist.commands.manager.manager.resolve", return_value=None, ): - handled = await dispatch_channel_slash_command( - msg, - agent=None, - thread_id="t1", - workspace_dir=None, - checkpointer=None, - append_system=append, - ) - assert handled is False + with patch("EvoScientist.cli.channel._set_channel_response") as mock_response: + handled = await dispatch_channel_slash_command( + msg, + agent=None, + thread_id="t1", + workspace_dir=None, + checkpointer=None, + append_system=append, + ) + assert handled is True + mock_response.assert_called_once_with( + "msg-1", + "Unknown command: /unknown-cmd\nType /help to see available commands.", + ) async def test_successful_slash_execution_sets_response_and_breadcrumb(): @@ -112,6 +119,75 @@ async def test_successful_slash_execution_sets_response_and_breadcrumb(): assert any("Executed command from" in t for t in breadcrumbs) +async def test_slash_with_flushed_output_suppresses_executed_ack(): + """When the command's own output already reached the channel, the + response is the already-sent sentinel, not a second ack message.""" + from EvoScientist.cli.channel import COMMAND_OUTPUT_ALREADY_SENT + + msg = _make_msg() + fake_cmd = MagicMock() + fake_cmd.needs_agent.return_value = False + + async def _execute_with_output(content, ctx): + ctx.ui.sent_to_channel = True + return True + + with ( + patch( + "EvoScientist.commands.manager.manager.resolve", + return_value=(fake_cmd, ["core"]), + ), + patch( + "EvoScientist.commands.manager.manager.execute", + new=AsyncMock(side_effect=_execute_with_output), + ), + patch("EvoScientist.cli.channel._set_channel_response") as mock_set_resp, + ): + handled = await dispatch_channel_slash_command( + msg, + agent="fake-agent", + thread_id="t1", + workspace_dir="/tmp", + checkpointer=None, + append_system=MagicMock(), + ) + assert handled is True + mock_set_resp.assert_called_once_with("msg-1", COMMAND_OUTPUT_ALREADY_SENT) + + +async def test_real_help_command_publishes_help_once(): + """Exercise the registered /help command rather than a mocked command.""" + from EvoScientist.cli.channel import COMMAND_OUTPUT_ALREADY_SENT + + bus = SimpleNamespace(publish_outbound=AsyncMock()) + msg = _make_msg(content="/help") + msg.channel_type = "telegram" + msg.bus_ref = bus + + with ( + patch("EvoScientist.cli.channel._bus_loop", asyncio.get_running_loop()), + patch("EvoScientist.cli.channel._set_channel_response") as mock_set_resp, + ): + handled = await dispatch_channel_slash_command( + msg, + agent=None, + thread_id="t1", + workspace_dir=None, + checkpointer=None, + append_system=MagicMock(), + ) + await asyncio.sleep(0) + await asyncio.sleep(0) + + assert handled is True + bus.publish_outbound.assert_awaited_once() + outbound = bus.publish_outbound.await_args.args[0] + assert outbound.channel == "telegram" + assert "Available commands:" in outbound.content + assert "/help" in outbound.content + mock_set_resp.assert_called_once_with("msg-1", COMMAND_OUTPUT_ALREADY_SENT) + + async def test_slash_dispatch_passes_graph_gateway_to_command_context(): msg = _make_msg() fake_cmd = MagicMock() @@ -440,6 +516,43 @@ async def test_command_error_skips_completion_hook_and_reports_error(): mock_set_resp.assert_called_once_with("msg-1", "Command error: workspace conflict") +async def test_command_error_with_flushed_output_suppresses_second_error(): + """CommandManager already flushes its error text to channel UIs.""" + from EvoScientist.cli.channel import COMMAND_OUTPUT_ALREADY_SENT + + msg = _make_msg(content="/resume abc") + fake_cmd = MagicMock() + fake_cmd.needs_agent.return_value = False + + async def _execute(_command, ctx): + ctx.command_error = "workspace conflict" + ctx.ui.sent_to_channel = True + return True + + with ( + patch( + "EvoScientist.commands.manager.manager.resolve", + return_value=(fake_cmd, ["abc"]), + ), + patch( + "EvoScientist.commands.manager.manager.execute", + side_effect=_execute, + ), + patch("EvoScientist.cli.channel._set_channel_response") as mock_set_resp, + ): + handled = await dispatch_channel_slash_command( + msg, + agent=None, + thread_id="old-thread", + workspace_dir="/old-workspace", + checkpointer=None, + append_system=MagicMock(), + ) + + assert handled is True + mock_set_resp.assert_called_once_with("msg-1", COMMAND_OUTPUT_ALREADY_SENT) + + async def test_empty_command_error_still_reports_error(): """An empty string error is still a command failure sentinel.""" msg = _make_msg(content="/resume abc") diff --git a/tests/test_telegram_channel.py b/tests/test_telegram_channel.py index 83e58a6..e0e2b2d 100644 --- a/tests/test_telegram_channel.py +++ b/tests/test_telegram_channel.py @@ -1,6 +1,8 @@ """Tests for Telegram channel implementation.""" -from types import SimpleNamespace +import sys +from datetime import datetime +from types import ModuleType, SimpleNamespace from unittest.mock import AsyncMock import pytest @@ -99,3 +101,177 @@ class TestTelegramChannel: ) result = await channel.send(msg) assert result is False + + async def test_registered_handler_accepts_bot_commands(self, monkeypatch): + class FakeFilter: + def __init__(self, predicate): + self._predicate = predicate + + def __or__(self, other): + return FakeFilter( + lambda update: ( + self.check_update(update) or other.check_update(update) + ) + ) + + def __and__(self, other): + return FakeFilter( + lambda update: ( + self.check_update(update) and other.check_update(update) + ) + ) + + def __invert__(self): + return FakeFilter(lambda update: not self.check_update(update)) + + def check_update(self, update): + return self._predicate(update) + + class FakeMessageHandler: + def __init__(self, message_filter, callback): + self.filters = message_filter + self.callback = callback + + app = SimpleNamespace( + handlers=[], + bot=SimpleNamespace( + get_me=AsyncMock(return_value=SimpleNamespace(username="botname")) + ), + updater=SimpleNamespace(start_polling=AsyncMock()), + initialize=AsyncMock(), + start=AsyncMock(), + ) + app.add_handler = app.handlers.append + + class FakeApplicationBuilder: + def token(self, _token): + return self + + def build(self): + return app + + text_filter = FakeFilter(lambda update: update.message.text is not None) + command_filter = FakeFilter(lambda update: update.message.is_command) + false_filter = FakeFilter(lambda _update: False) + fake_filters = SimpleNamespace( + TEXT=text_filter, + COMMAND=command_filter, + PHOTO=false_filter, + VOICE=false_filter, + AUDIO=false_filter, + Document=SimpleNamespace(ALL=false_filter), + VIDEO=false_filter, + Sticker=SimpleNamespace(ALL=false_filter), + LOCATION=false_filter, + ) + telegram_module = ModuleType("telegram") + ext_module = ModuleType("telegram.ext") + ext_module.ApplicationBuilder = FakeApplicationBuilder + ext_module.MessageHandler = FakeMessageHandler + ext_module.filters = fake_filters + telegram_module.ext = ext_module + monkeypatch.setitem(sys.modules, "telegram", telegram_module) + monkeypatch.setitem(sys.modules, "telegram.ext", ext_module) + + channel = TelegramChannel( + TelegramConfig(bot_token="test", include_attachments=False) + ) + await channel.start() + + update = SimpleNamespace(message=SimpleNamespace(text="/help", is_command=True)) + assert app.handlers[0].filters.check_update(update) is True + assert app.handlers[0].callback == channel._on_message + + async def test_group_command_suffix_is_removed_before_enqueue(self): + channel = TelegramChannel( + TelegramConfig(bot_token="test", include_attachments=False) + ) + channel._bot_username = "botname" + update = self._text_update("/stop@botname", chat_type="supergroup") + + await channel._on_message(update, None) + + message = await channel._queue.get() + assert message.content == "/stop" + assert message.is_group is True + assert message.was_mentioned is True + + async def test_private_command_suffix_is_removed_before_enqueue(self): + channel = TelegramChannel( + TelegramConfig(bot_token="test", include_attachments=False) + ) + channel._bot_username = "botname" + update = self._text_update("/stop@botname") + + await channel._on_message(update, None) + + message = await channel._queue.get() + assert message.content == "/stop" + assert message.is_group is False + + async def test_bare_group_command_passes_mention_gating(self): + channel = TelegramChannel( + TelegramConfig(bot_token="test", include_attachments=False) + ) + channel._bot_username = "botname" + update = self._text_update("/stop", chat_type="supergroup") + + await channel._on_message(update, None) + + message = await channel._queue.get() + assert message.content == "/stop" + assert message.was_mentioned is True + + async def test_group_command_for_other_bot_is_ignored(self): + channel = TelegramChannel( + TelegramConfig(bot_token="test", include_attachments=False) + ) + channel._bot_username = "botname" + update = self._text_update("/stop@otherbot", chat_type="supergroup") + + await channel._on_message(update, None) + + assert channel._queue.empty() + + async def test_group_command_bypasses_buffered_history(self): + channel = TelegramChannel( + TelegramConfig(bot_token="test", include_attachments=False) + ) + channel._bot_username = "botname" + + chatter = self._text_update("background chatter", chat_type="supergroup") + await channel._on_message(chatter, None) + assert channel._queue.empty() + + command = self._text_update("/help@botname", chat_type="supergroup") + command.message.message_id = 790 + await channel._on_message(command, None) + + message = await channel._queue.get() + assert message.content == "/help" + assert message.is_group is True + assert message.was_mentioned is True + + async def test_start_command_flows_to_shared_dispatch(self): + channel = TelegramChannel( + TelegramConfig(bot_token="test", include_attachments=False) + ) + update = self._text_update("/start") + + await channel._on_message(update, None) + + message = await channel._queue.get() + assert message.content == "/start" + + @staticmethod + def _text_update(text, *, chat_type="private"): + message = SimpleNamespace( + from_user=SimpleNamespace(id=123), + chat_id=456, + chat=SimpleNamespace(type=chat_type), + text=text, + caption=None, + date=datetime(2026, 1, 1), + message_id=789, + ) + return SimpleNamespace(message=message)