fix(channels): telegram slash commands (#364)

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

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:

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 ──
 

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

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:

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] = []

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":

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(

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 (

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."

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."""
 

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

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)