diff --git a/cli-config.yaml.example b/cli-config.yaml.example index 629ea2ea93794..9259485b8fb14 100644 --- a/cli-config.yaml.example +++ b/cli-config.yaml.example @@ -1894,6 +1894,47 @@ display: # cleanup_progress: true cleanup_progress: false + # Telegram: keep one editable streaming preview across tool calls instead of + # starting a new message at each tool boundary. Only takes effect when + # streaming is enabled (streaming.enabled / display.platforms.telegram.streaming) + # and text tool-progress is quiet (tool_progress: off or log). + # Default false. Per-platform override: display.platforms.telegram.streaming_single_message + # true: One streaming message per turn + # false: New streaming segment after each tool (default) + # Example: + # display: + # platforms: + # telegram: + # streaming_single_message: true + streaming_single_message: false + + # Transient activity overlay inside that single message (only active while the mode + # above is on): tool-start lines (default true) and thinking snippets (default false) + # render under the evolving preview below a "---" rule; real text replaces them and + # the final message never carries them. + # Per-platform overrides: + # display.platforms.telegram.streaming_single_message_activity + # display.platforms.telegram.streaming_single_message_thinking + streaming_single_message_activity: true + streaming_single_message_thinking: false + + # Over-limit split policy for the single message (Telegram's 4096 text limit): + # true: cut the preview into ≀4096 messages as it fills (several long messages) + # false: default β€” don't cut mid-turn; page only at turn end when the text really + # overflows (short replies never split) + # Per-platform override: display.platforms.telegram.streaming_single_message_4096_split + streaming_single_message_4096_split: false + + # Telegram Bot API 10.3 native features (gateway.platforms.telegram.extra.*). Each is + # opt-in and falls back to the ordinary path when unsupported: + # stop_button: true Stop button on DM draft streaming; a press interrupts + # the turn like /stop (keep_on_stop chooses what remains) + # checklist_emoji: "βœ…" Task lists β†’ native checklists (business accounts only; + # MarkdownV2 fallback elsewhere) + # message_effects: true Animated effect on the final reply of long turns + # (message_effect_min_seconds, default 60; private chats) + # ephemeral_messages: true Group busy-acks whispered to the sender only + # Natural mid-turn assistant updates. # On gateway platforms, when true, completed assistant status messages are # sent as separate chat messages. On the Desktop app, when true, mid-turn diff --git a/gateway/display_config.py b/gateway/display_config.py index 72981b4f8b592..27cc14bc70a1d 100644 --- a/gateway/display_config.py +++ b/gateway/display_config.py @@ -20,6 +20,21 @@ "reasoning_style": "code", # "code" (πŸ’­ **Reasoning:** + fence), "blockquote" ("> "), "subtext" ("-# " Discord) "tool_preview_length": 0, "streaming": None, # None = follow top-level streaming config + # Telegram-only opt-in: keep the editable streaming preview across tool boundaries. + # Active while text tool-progress is quiet (off/log), preserving existing chronology + # otherwise. See #110564. + "streaming_single_message": False, + # Single-message extras (only when the above is active): transient activity overlay in + # the evolving preview β€” tool-start lines (default on) and thinking snippets (opt-in). + "streaming_single_message_activity": True, + "streaming_single_message_thinking": False, + "streaming_single_message_4096_split": False, + # Completion message effect (Telegram DMs only): an animated effect on the final + # message of turns lasting at least message_effect_min_seconds. Emoji name keys the + # adapter's id table; false/off = disabled. + "message_effects": False, + "message_effect": "πŸŽ‰", + "message_effect_min_seconds": 60, # Gateway-only assistant/status chatter; mobile platforms opt down to final-answer-first. "interim_assistant_messages": True, "suppress_warning_notifications": False, @@ -202,6 +217,12 @@ def _norm_int(value: Any) -> int: "tool_progress": _norm_tristate("all", "off", {"off", "new", "all", "verbose", "log"}), "show_reasoning": _norm_bool, "streaming": _norm_bool, + "streaming_single_message": _norm_bool, + "streaming_single_message_activity": _norm_bool, + "streaming_single_message_thinking": _norm_bool, + "streaming_single_message_4096_split": _norm_bool, + "message_effects": _norm_bool, + "message_effect_min_seconds": _norm_int, "interim_assistant_messages": _norm_bool, "suppress_warning_notifications": _norm_suppress_warning_notifications, "long_running_notifications": _norm_long_running, diff --git a/gateway/run_busy.py b/gateway/run_busy.py index bf6d7d9ed2704..039d4523ec2eb 100644 --- a/gateway/run_busy.py +++ b/gateway/run_busy.py @@ -492,13 +492,17 @@ def _busy_reply_to(event: MessageEvent, reply_anchor): else (None if event.source.platform == Platform.TELEGRAM and event.source.thread_id else event.message_id) ) - async def _send_busy_reply(self, event: MessageEvent, adapter, content: str, *, plain_anchor: bool = False) -> None: + async def _send_busy_reply(self, event: MessageEvent, adapter, content: str, *, plain_anchor: bool = False, + extra_metadata: Optional[dict[str, Any]] = None) -> None: """Send a busy-path reply anchored to the event (thread metadata included).""" reply_anchor = self._reply_anchor_for_event(event) + metadata = self._thread_metadata_for_source(event.source, reply_anchor) + if extra_metadata: + metadata = {**(metadata or {}), **extra_metadata} await adapter._send_with_retry( chat_id=event.source.chat_id, content=content, reply_to=reply_anchor if plain_anchor else self._busy_reply_to(event, reply_anchor), - metadata=self._thread_metadata_for_source(event.source, reply_anchor), + metadata=metadata, ) async def _send_busy_drain_notice(self, event: MessageEvent, session_key: str, effective_mode: str) -> None: @@ -800,7 +804,17 @@ def _compose_busy_ack_message( async def _send_busy_ack_reply(self, event: MessageEvent, adapter, message: str) -> None: try: - await self._send_busy_reply(event, adapter, message) + source = event.source + # Telegram group whisper (Bot API 10.3 ephemeral): scope the busy-ack to the + # sender via the adapter's opt-in gate (telegram.extra.ephemeral_messages); + # DMs are already private and other platforms ignore the metadata key. + # No ephemeral fallback: a rejected whisper stays unsent rather than + # leaking the ack to the whole group. + extra = None + if (source.platform == Platform.TELEGRAM and event.message_type == MessageType.TEXT + and source.user_id and source.chat_type not in (None, "dm", "private")): + extra = {"ephemeral_for": str(source.user_id)} + await self._send_busy_reply(event, adapter, message, extra_metadata=extra) except Exception as e: logger.debug("Failed to send busy-ack: %s", e) diff --git a/gateway/run_turn.py b/gateway/run_turn.py index d0f94d616153c..af6df36e4c078 100644 --- a/gateway/run_turn.py +++ b/gateway/run_turn.py @@ -2667,42 +2667,10 @@ def _get_proxy_url(self) -> Optional[str]: url = ((_load_gateway_config().get("gateway") or {}).get("proxy_url") or "").strip() return url.rstrip("/") if url else None - def _build_stream_consumer_config( - self, source: SessionSource, scfg: Any, adapter: Any, *, on_missing_cursor: str, - ) -> tuple[Any, Optional[Callable[[], None]]]: - """Build the shared ``StreamConsumerConfig`` and optional Telegram pause-typing closure. - For non-editing adapters ``on_missing_cursor="fallback"`` streams with an empty cursor; - ``"raise"`` raises ``RuntimeError`` so the caller skips streaming entirely.""" - from gateway.stream_consumer import StreamConsumerConfig - _pause_typing_before_finalize = None - if source.platform == Platform.TELEGRAM and hasattr(adapter, "pause_typing_for_chat"): - def _pause_typing_before_finalize(_adapter=adapter, _chat_id=source.chat_id) -> None: - _adapter.pause_typing_for_chat(_chat_id) - # Non-editing platforms (QQ, WeChat) skip streaming β€” the partial first message could never - # be updated β€” unless they have a native-streaming transport (WeCom msgtype "stream"). - _adapter_supports_edit = getattr(adapter, "SUPPORTS_MESSAGE_EDITING", True) - _adapter_supports_native_stream = bool(getattr(adapter, "SUPPORTS_NATIVE_STREAMING", False)) - if not _adapter_supports_edit and not _adapter_supports_native_stream and on_missing_cursor == "raise": - raise RuntimeError("skip streaming for non-editable platform") - _effective_cursor = scfg.cursor if _adapter_supports_edit else "" - # Some Matrix clients render the cursor as tofu: stream text, no cursor. - if source.platform == Platform.MATRIX: - _effective_cursor = "" - # Fresh-final applies to Telegram only (others edit in place cheaply). - # Fresh-final applies to Telegram only β€” other platforms either edit in place cheaply (Discord, - # Slack) or don't have the timestamp-on-edit / edit-timestamp-stays-stale problem. (Ported from - # openclaw/openclaw#72038.) - _fresh_final_secs = ( - float(getattr(scfg, "fresh_final_after_seconds", 0.0) or 0.0) - if source.platform == Platform.TELEGRAM else 0.0 - ) - _consumer_cfg = StreamConsumerConfig( - edit_interval=scfg.edit_interval, buffer_threshold=scfg.buffer_threshold, - cursor=_effective_cursor, - fresh_final_after_seconds=_fresh_final_secs, transport=scfg.transport or "edit", - chat_type=getattr(source, "chat_type", "") or "", - ) - return _consumer_cfg, _pause_typing_before_finalize + + def _build_stream_consumer_config(self, source, scfg, adapter, *, on_missing_cursor): + from gateway.run_turn_stream_config import build_stream_consumer_config + return build_stream_consumer_config(source, scfg, adapter, on_missing_cursor=on_missing_cursor) def _run_still_current_fn(self, session_key: Optional[str], run_generation: Optional[int]) -> Callable[[], bool]: """Predicate: does this run's generation still own ``session_key``? (always True when untracked).""" @@ -2947,6 +2915,7 @@ def _run_agent_display_settings(self, source: SessionSource) -> GatewayRunner._R platform_key = _platform_config_key(source.platform) enabled_toolsets, disabled_toolsets = self._resolve_turn_toolsets(user_config, source, platform_key) adapter = self._delivery_adapter_for(source) + # display.platforms.. β†’ display. β†’ built-in platform defaults. # Tool preview length (0 = no limit) and friendly tool labels (default on), per-platform. for _setter, _setting, _default, _cast in ( ("set_tool_preview_max_len", "tool_preview_length", 0, lambda v: int(v) if v else 0), diff --git a/gateway/run_turn_progress.py b/gateway/run_turn_progress.py new file mode 100644 index 0000000000000..7fd7c5bed161d --- /dev/null +++ b/gateway/run_turn_progress.py @@ -0,0 +1,81 @@ +"""Tool lifecycle dispatch into chat, logs and single-message activity overlays.""" +from __future__ import annotations + +from datetime import datetime +from agent.i18n import t + + +class TurnProgressMixin: + def progress_callback(self, event_type: str, tool_name: str | None = None, preview: str | None = None, args: dict | None = None, **kwargs): + """Callback invoked by agent on tool lifecycle events.""" + ctx = self._ctx + # Failed subagent β†’ one clean user-facing notice, handled FIRST, before every progress-queue + # gate: platforms with tool_progress off must still hear about a dead delegation. + if event_type == "subagent.complete": + self._progress_subagent_notice(preview, kwargs) + return + self._progress_live_status(event_type, tool_name, args) + # "log" mode: append tool.started lines to the log queue, silent in chat. Handled before + # the progress_queue guard because log mode runs without a chat progress queue. + if ctx.log_queue is not None and event_type == "tool.started" and tool_name and tool_name != "_thinking": + ts = datetime.now().strftime("%Y-%m-%d %H:%M:%S") + preview_str = f' "{preview}"' if preview else "" + ctx.log_queue.put(f"{ts} {tool_name}:{preview_str}".rstrip()) + # Single-message mode keeps tool_progress quiet, but the evolving preview can carry + # a transient overlay while it runs: tool-start lines and (opt-in) thinking snippets. + # Placed before the progress-queue guard β€” single mode usually runs without a queue. + sc = self._stream_consumer() + if sc is not None and getattr(sc, "single_message_mode", False) is True: + if (event_type == "tool.started" and tool_name and tool_name not in {"clarify", "_thinking"} + and ctx._run_still_current() and not self._agent_interrupted() + and getattr(sc, "accepts_tool_progress", False) is True): + msg = self._progress_build_message(tool_name, preview, args) + if msg: + self._progress_emit(msg) + return + if ((event_type == "_thinking" or tool_name == "_thinking") + and getattr(sc, "accepts_thinking_progress", False) is True): + thinking_text = (preview if tool_name == "_thinking" else tool_name) + snippet = " ".join(str(thinking_text or "").split())[:160] + if snippet and ctx._run_still_current() and not self._agent_interrupted(): + sc.on_tool_progress(f"πŸ’­ {snippet}") + return + if not ctx.progress_queue or not ctx._run_still_current(): + return + if event_type == "tool.completed" and not ctx.long_tool_hint_fired[0]: + self._progress_onboarding_hint(kwargs) + return + # "_thinking" is assistant scratch text between tool calls, never ordinary tool progress: + # only relayed when the platform explicitly opted into thinking_progress. + if event_type == "_thinking" or tool_name == "_thinking": + thinking_text = (preview if tool_name == "_thinking" else tool_name) if ctx._thinking_enabled else None + if thinking_text: + ctx.progress_queue.put(t("gateway.progress.thinking_prefix", text=thinking_text)) + return + # Native task cards consume the ID-bearing tool_start/tool_complete callbacks instead; + # name-correlated text events would duplicate cards and mispair concurrent same-tool calls. + if ctx._native_slack_task_cards and event_type in {"tool.started", "tool.completed"}: + return + # tool_progress off β†’ only _thinking passes (above). Only tool.started renders. clarify: + # send_clarify IS the user-facing rendering (a bubble would duplicate it, and verbose mode + # would dump the raw args JSON right under the prompt). Post-`stop`: N parallel tool calls + # fire N tool.started events before the interrupt check, so a late stop must not render them. + if ( + not ctx.tool_progress_enabled + or event_type != "tool.started" + # The adapter's send_clarify IS the user-facing rendering (interactive buttons or the + # numbered-text fallback), so a progress bubble is pure duplication β€” and in verbose mode it + # dumps the raw tool-call args JSON into the chat. Because + # the progress queue drains on a background task, that raw JSON typically lands right underneath + # the rendered prompt (#52374). + or tool_name == "clarify" + or self._agent_interrupted() + ): + return + # "new" mode: only report when tool changes + if ctx.progress_mode == "new" and tool_name == ctx.last_tool[0]: + return + ctx.last_tool[0] = tool_name + msg = self._progress_build_message(tool_name, preview, args) + if msg is not None: + self._progress_emit(msg) diff --git a/gateway/run_turn_runner.py b/gateway/run_turn_runner.py index 7e29a4cdf8cc8..43e84cc0c004e 100644 --- a/gateway/run_turn_runner.py +++ b/gateway/run_turn_runner.py @@ -16,7 +16,6 @@ import threading import time from contextlib import suppress -from datetime import datetime from typing import TYPE_CHECKING, Any, Dict, List, Optional from agent.i18n import t @@ -27,6 +26,7 @@ from gateway.platforms.base import BasePlatformAdapter from gateway.platforms.base_exec_approval import ea_default_reason_text from gateway.turn_context import TurnContext +from gateway.run_turn_progress import TurnProgressMixin from hermes_cli.config import cfg_get from utils import is_truthy_value @@ -75,7 +75,7 @@ class _ExecApprovalDeclined(RuntimeError): """ -class TurnRunner: +class TurnRunner(TurnProgressMixin): """Per-turn collaborator carrying ``GatewayRunner._run_agent_inner``'s tool-progress callbacks.""" def __init__(self, runner: GatewayRunner, ctx: TurnContext) -> None: @@ -124,60 +124,6 @@ def _track_future_cleanup_id(self, fut) -> None: # ── progress_callback (agent thread β†’ progress queue) ─────────────────────────────────── - def progress_callback(self, event_type: str, tool_name: str | None = None, preview: str | None = None, args: dict | None = None, **kwargs): - """Callback invoked by agent on tool lifecycle events.""" - ctx = self._ctx - # Failed subagent β†’ one clean user-facing notice, handled FIRST, before every progress-queue - # gate: platforms with tool_progress off must still hear about a dead delegation. - if event_type == "subagent.complete": - self._progress_subagent_notice(preview, kwargs) - return - self._progress_live_status(event_type, tool_name, args) - # "log" mode: append tool.started lines to the log queue, silent in chat. Handled before - # the progress_queue guard because log mode runs without a chat progress queue. - if ctx.log_queue is not None and event_type == "tool.started" and tool_name and tool_name != "_thinking": - ts = datetime.now().strftime("%Y-%m-%d %H:%M:%S") - preview_str = f' "{preview}"' if preview else "" - ctx.log_queue.put(f"{ts} {tool_name}:{preview_str}".rstrip()) - if not ctx.progress_queue or not ctx._run_still_current(): - return - if event_type == "tool.completed" and not ctx.long_tool_hint_fired[0]: - self._progress_onboarding_hint(kwargs) - return - # "_thinking" is assistant scratch text between tool calls, never ordinary tool progress: - # only relayed when the platform explicitly opted into thinking_progress. - if event_type == "_thinking" or tool_name == "_thinking": - thinking_text = (preview if tool_name == "_thinking" else tool_name) if ctx._thinking_enabled else None - if thinking_text: - ctx.progress_queue.put(t("gateway.progress.thinking_prefix", text=thinking_text)) - return - # Native task cards consume the ID-bearing tool_start/tool_complete callbacks instead; - # name-correlated text events would duplicate cards and mispair concurrent same-tool calls. - if ctx._native_slack_task_cards and event_type in {"tool.started", "tool.completed"}: - return - # tool_progress off β†’ only _thinking passes (above). Only tool.started renders. clarify: - # send_clarify IS the user-facing rendering (a bubble would duplicate it, and verbose mode - # would dump the raw args JSON right under the prompt). Post-`stop`: N parallel tool calls - # fire N tool.started events before the interrupt check, so a late stop must not render them. - if ( - not ctx.tool_progress_enabled - or event_type != "tool.started" - # The adapter's send_clarify IS the user-facing rendering (interactive buttons or the - # numbered-text fallback), so a progress bubble is pure duplication β€” and in verbose mode it - # dumps the raw tool-call args JSON into the chat. Because - # the progress queue drains on a background task, that raw JSON typically lands right underneath - # the rendered prompt (#52374). - or tool_name == "clarify" - or self._agent_interrupted() - ): - return - # "new" mode: only report when tool changes - if ctx.progress_mode == "new" and tool_name == ctx.last_tool[0]: - return - ctx.last_tool[0] = tool_name - msg = self._progress_build_message(tool_name, preview, args) - if msg is not None: - self._progress_emit(msg) def _progress_subagent_notice(self, preview, kwargs: dict) -> None: """Only terminal failure statuses render (same notice rail as credit warnings).""" diff --git a/gateway/run_turn_stream_config.py b/gateway/run_turn_stream_config.py new file mode 100644 index 0000000000000..b09d26d4a2de3 --- /dev/null +++ b/gateway/run_turn_stream_config.py @@ -0,0 +1,91 @@ +"""Per-platform streaming configuration shared by normal and proxy turns.""" +from __future__ import annotations + +import os +from typing import Any, Callable, Optional +from gateway.config import Platform +from gateway.session import SessionSource + + +def build_stream_consumer_config( + source: SessionSource, scfg: Any, adapter: Any, *, on_missing_cursor: str, +) -> tuple[Any, Optional[Callable[[], None]]]: + """Build the shared ``StreamConsumerConfig`` and optional Telegram pause-typing closure. + For non-editing adapters ``on_missing_cursor="fallback"`` streams with an empty cursor; + ``"raise"`` raises ``RuntimeError`` so the caller skips streaming entirely.""" + from gateway.stream_consumer import StreamConsumerConfig + _pause_typing_before_finalize = None + if source.platform == Platform.TELEGRAM and hasattr(adapter, "pause_typing_for_chat"): + def _pause_typing_before_finalize(_adapter=adapter, _chat_id=source.chat_id) -> None: + _adapter.pause_typing_for_chat(_chat_id) + # Non-editing platforms (QQ, WeChat) skip streaming β€” the partial first message could never + # be updated β€” unless they have a native-streaming transport (WeCom msgtype "stream"). + _adapter_supports_edit = getattr(adapter, "SUPPORTS_MESSAGE_EDITING", True) + _adapter_supports_native_stream = bool(getattr(adapter, "SUPPORTS_NATIVE_STREAMING", False)) + if not _adapter_supports_edit and not _adapter_supports_native_stream and on_missing_cursor == "raise": + raise RuntimeError("skip streaming for non-editable platform") + _effective_cursor = scfg.cursor if _adapter_supports_edit else "" + # Some Matrix clients render the cursor as tofu: stream text, no cursor. + if source.platform == Platform.MATRIX: + _effective_cursor = "" + # Fresh-final applies to Telegram only (others edit in place cheaply). + # Fresh-final applies to Telegram only β€” other platforms either edit in place cheaply (Discord, + # Slack) or don't have the timestamp-on-edit / edit-timestamp-stays-stale problem. (Ported from + # openclaw/openclaw#72038.) + _fresh_final_secs = ( + float(getattr(scfg, "fresh_final_after_seconds", 0.0) or 0.0) + if source.platform == Platform.TELEGRAM else 0.0 + ) + # Preserve the established text↔tool-progress chronology unless progress is quiet. + # The gate asks for the mode the display path actually runs with (off|log is the + # quiet pair β€” the same one tool_progress_enabled uses), so the + # HERMES_TOOL_PROGRESS_MODE bridge can't leave bubbles on while this mode is set. + # Telegram-only: other adapters keep their current segment-boundary semantics. + from gateway.run import _load_gateway_config, _platform_config_key + from gateway.display_config import resolve_display_setting, resolve_tool_progress + _user_config = _load_gateway_config() + _platform_key = _platform_config_key(source.platform) + _single_message_per_turn = ( + source.platform == Platform.TELEGRAM + and bool(resolve_display_setting( + _user_config, _platform_key, "streaming_single_message", False, + )) + and resolve_tool_progress( + _user_config, _platform_key, os.getenv("HERMES_TOOL_PROGRESS_MODE"), + )[0] in ("off", "log") + ) + # Transient-overlay sub-switches: tool-start lines (default on β€” quiet progress + # would otherwise show nothing during tool runs) and thinking snippets (opt-in). + _single_message_activity = _single_message_per_turn and bool(resolve_display_setting( + _user_config, _platform_key, "streaming_single_message_activity", True, + )) + _single_message_thinking = _single_message_per_turn and bool(resolve_display_setting( + _user_config, _platform_key, "streaming_single_message_thinking", False, + )) + _single_message_4096_split = _single_message_per_turn and bool(resolve_display_setting( + _user_config, _platform_key, "streaming_single_message_4096_split", False, + )) + # Completion message effect (Telegram DM): emoji + minimum turn age (0 = every turn). + _message_effects_on = bool(resolve_display_setting( + _user_config, _platform_key, "message_effects", False, + )) + _message_effect = str(resolve_display_setting( + _user_config, _platform_key, "message_effect", "πŸŽ‰") or "") if _message_effects_on else "" + try: + _message_effect_min = float(resolve_display_setting( + _user_config, _platform_key, "message_effect_min_seconds", 60) or 0) + except (TypeError, ValueError): + _message_effect_min = 60.0 + _consumer_cfg = StreamConsumerConfig( + edit_interval=scfg.edit_interval, buffer_threshold=scfg.buffer_threshold, + cursor=_effective_cursor, + fresh_final_after_seconds=_fresh_final_secs, transport=scfg.transport or "edit", + chat_type=getattr(source, "chat_type", "") or "", + single_message_per_turn=_single_message_per_turn, + single_message_activity=_single_message_activity, + single_message_thinking=_single_message_thinking, + single_message_4096_split=_single_message_4096_split, + message_effect=_message_effect, + message_effect_min_seconds=_message_effect_min, + ) + return _consumer_cfg, _pause_typing_before_finalize diff --git a/gateway/stream_consumer.py b/gateway/stream_consumer.py index 67c451b186b23..11a8e6245b1e9 100644 --- a/gateway/stream_consumer.py +++ b/gateway/stream_consumer.py @@ -74,6 +74,25 @@ class StreamConsumerConfig: # (progressive editMessageText). "off" is handled by the gateway. transport: str = "edit" chat_type: str = "" # originating chat type; gates platform-specific drafts + # Keep an edit-based preview alive across tool boundaries (and mid-turn commentary) + # for the whole turn. The gateway enables this only for Telegram while text + # tool-progress is quiet (off/log). See #110564. + single_message_per_turn: bool = False + # Single-message extras: while the mode above is active, render a transient activity + # overlay under the evolving preview β€” tool-start lines and thinking snippets β€” that + # real text replaces and the final edit never carries. See #110564. + single_message_activity: bool = False + single_message_thinking: bool = False + # Single-message overflow policy: false (default) = deferred pagination β€” the preview + # is NOT cut while the turn runs; an over-limit turn-final is paged by the adapter's + # overflow split instead, so the live phase stays one message and pages only appear + # when the text genuinely overflows. true = eager ≀limit seals mid-stream (the reply + # arrives as several long messages). See #110564. + single_message_4096_split: bool = False + # Completion effect (Telegram private chats only): emoji name applied to the turn's + # FINAL fresh send when the turn ran at least ``message_effect_min_seconds``. Empty = off. + message_effect: str = "" + message_effect_min_seconds: float = 60.0 @dataclass @@ -155,6 +174,7 @@ def __init__( self._edit_supported = True # False once progressive edits stop working self._last_edit_time = 0.0 self._last_edit_overflowed = False # last _send_or_edit split into continuations + self._turn_started = time.monotonic() # basis for the completion message effect self._flood_strikes = 0 self._current_edit_interval = self.cfg.edit_interval # adaptive backoff self._delivered_commentary_texts: list[str] = [] @@ -238,21 +258,61 @@ def _stream_is_message(self) -> bool: except Exception: return False + @property + def single_message_mode(self) -> bool: + """One evolving preview for the whole turn (opt-in; see #110564).""" + return bool(self.cfg.single_message_per_turn) + + def _eager_overflow_split(self) -> bool: + """Whether over-limit content is sealed into separate messages mid-stream. + + True for classic streaming. In single-message mode the default (``false``) defers + pagination to the turn-final adapter overflow split, so the live phase keeps one + message; opt back in with ``streaming_single_message_4096_split: true``.""" + if not self.single_message_mode: + return True + return bool(self.cfg.single_message_4096_split) + @property def accepts_tool_progress(self) -> bool: - """True only when native streaming is active (gates in-stream tool progress).""" - return self._use_native_streaming + """True when the live surface renders tool progress in-stream: native streaming + always; single-message mode when its activity overlay is enabled.""" + return self._use_native_streaming or ( + self.single_message_mode and self.cfg.single_message_activity + ) + + @property + def accepts_thinking_progress(self) -> bool: + """True when the live surface renders thinking snippets in-stream (single-message + mode with its thinking overlay enabled).""" + return self.single_message_mode and self.cfg.single_message_thinking def on_tool_progress(self, line: str) -> None: - """Thread-safe: overlay a tool-progress line in the native bubble until the next delta.""" + """Thread-safe: overlay a tool/thinking line in the evolving preview (native + streams and single-message mode) until the next real-text delta.""" if line: self._queue.put((_TOOL_PROGRESS, line)) def _compose_frame_content(self) -> str: - """Native frame content: text, with any tool-progress lines below a rule.""" - progress = "\n".join(self._tool_progress_lines) + """Interim frame content (native streams, drafts, single message): text, with any + tool-progress lines below a rule β€” expandable-quote styled when the transport + formats frames and the overlay is on (see _style_overlay_lines).""" + progress = "\n".join(self._style_overlay_lines(self._tool_progress_lines)) return "\n\n---\n".join(p for p in (self._accumulated, progress) if p) + def _style_overlay_lines(self, lines: list) -> list: + """MarkdownV2 expandable-quote framing for the activity overlay (#110564). + + Only the draft lane formats interim frames (MarkdownV2); edit-lane frames are + raw, so styling markers would render literally β€” those keep the plain lines. + Frame convention: first line ``**>``, further lines ``>``, closing ``||``. + """ + if not (self.single_message_mode and self._use_draft_streaming and lines): + return list(lines) + styled = [f"**> {lines[0]}"] + [f"> {line}" for line in lines[1:]] + styled[-1] += "||" + return styled + def _metadata_for_send(self, *, final: bool = False, expect_edits: bool = False) -> dict | None: """Per-send metadata. ``final`` β†’ notify=True (Mattermost treats notify-worthy sends as final when a broken thread root may fall back flat); ``expect_edits`` keeps @@ -264,6 +324,10 @@ def _metadata_for_send(self, *, final: bool = False, expect_edits: bool = False) meta["expect_edits"] = True if final: meta["notify"] = True + if self.cfg.message_effect and ( + time.monotonic() - self._turn_started + ) >= float(self.cfg.message_effect_min_seconds): + meta["message_effect"] = self.cfg.message_effect return meta or None # Read-only views for the gateway (flag semantics: see _clear_turn_final_flags). @@ -572,16 +636,20 @@ async def run(self) -> None: return if self._should_edit(tick) and ( - self._accumulated or (self._use_native_streaming and self._tool_progress_active) + self._accumulated + or ((self._use_native_streaming or self.single_message_mode) + and self._tool_progress_active) ): # Seal first: it clears the message id, so a remainder still over the limit # is split again below. A plain first send would let the adapter split it and # adopt only the LAST chunk as the preview; the next seal then overwrites that # chunk with the head of the whole remainder (duplicated + lost text, #25349). - await self._seal_overflow_heads() + if self._eager_overflow_split(): + await self._seal_overflow_heads() # Overflow split. Native streaming bypasses this: the adapter # truncates against the stream protocol's own limit. - if not self._use_native_streaming and self._first_send_overflows(): + if (self._eager_overflow_split() + and not self._use_native_streaming and self._first_send_overflows()): if await self._split_first_send(tick): return if self._first_send_overflows(): @@ -673,8 +741,17 @@ def _drain_queue(self) -> _Tick: if kind is _FINAL_TEXT: self._adopt_final_text(item[1]) elif kind is _TOOL_PROGRESS: # keep draining to batch simultaneous lines - if self._use_native_streaming: - self._tool_progress_lines.append(item[1]) + if self._use_native_streaming or self.single_message_mode: + line = item[1] + if line.startswith("πŸ’­"): + # Latest thinking only: a new snippet replaces the previous one. + self._tool_progress_lines = [ + existing for existing in self._tool_progress_lines + if not existing.startswith("πŸ’­") + ] + self._tool_progress_lines.append(line) + if self.single_message_mode and len(self._tool_progress_lines) > 6: + del self._tool_progress_lines[:-6] # defensive cap self._tool_progress_active = True elif kind is _APPROVAL_BOUNDARY: tick.approval_boundary = (item[1], item[2]) @@ -735,6 +812,15 @@ def _should_edit(self, tick: _Tick) -> bool: if self._use_native_streaming: # No platform edit-rate limit: push every delta immediately. should_edit = bool(self._accumulated) or self._tool_progress_active + elif self.single_message_mode: + # Same throttle as the edit path, but an activity-only preview (no text yet) + # is a valid update β€” that IS the transient overlay. + elapsed = time.monotonic() - self._last_edit_time + should_edit = bool( + (elapsed >= self._current_edit_interval + and (self._accumulated or self._tool_progress_active)) + or len(self._accumulated) >= self.cfg.buffer_threshold + ) else: elapsed = time.monotonic() - self._last_edit_time # buffer_threshold is a codepoint debounce heuristic, not a @@ -825,12 +911,17 @@ async def _push_update(self, tick: _Tick) -> None: """Send/edit this tick's visible text (cursor-suffixed unless finalizing).""" display_text = self._accumulated if tick.is_interim: - if self._use_native_streaming: + if self._use_native_streaming or self.single_message_mode: display_text = self._compose_frame_content() if display_text and self.cfg.cursor: display_text += self.cfg.cursor else: display_text += self.cfg.cursor + if not tick.got_done and self.single_message_mode and not self._eager_overflow_split(): + # Keep the full answer buffered; an oversized preview would make the adapter + # create continuation messages before the deferred final pagination. + limit = _custom_unit_to_cp(display_text, self._safe_limit, self._len_fn) + display_text = display_text[:limit] # A got_done FRESH send via the draft transport already carries finalize=True, # unlike an EDIT, which REQUIRES_EDIT_FINALIZE adapters still need a pass for. @@ -909,14 +1000,15 @@ def _cumulative_transport(self) -> bool: return stream_draft or self._use_native_streaming async def _deliver_commentary(self, commentary_text: str) -> None: - """Post commentary as its own message. Cumulative transports keep the stream going β€” - resetting _accumulated would break the append-only invariant / lose text.""" - cumulative = self._cumulative_transport() - if not cumulative: + """Post commentary as its own message. Cumulative transports and the single-message + mode keep the stream going β€” resetting _accumulated would break the append-only + invariant / lose the evolving preview.""" + keep_stream = self._cumulative_transport() or self.cfg.single_message_per_turn + if not keep_stream: self._reset_segment_state() await self._send_commentary(commentary_text) self._last_edit_time = time.monotonic() - if not cumulative: + if not keep_stream: self._reset_segment_state() async def _end_segment(self, tick: _Tick) -> None: @@ -925,8 +1017,9 @@ async def _end_segment(self, tick: _Tick) -> None: non-prefix snapshot and the connector re-appends the whole answer. preserve_no_edit: "__no_edit__" (platform never returned a real id β€” Signal, github_comment webhook) must keep its sentinel or every tool boundary posts a new message; the - continuation goes out once via _send_fallback_final.""" - if self._cumulative_transport(): + continuation goes out once via _send_fallback_final. Single-message mode + (Telegram, quiet progress) skips the reset so one preview spans the turn.""" + if self._cumulative_transport() or self.cfg.single_message_per_turn: return # If the segment-break edit or send didn't land (flood control / fallback mode, or a # failed first send of a split tail with no message id yet), _accumulated holds unseen diff --git a/plugins/platforms/telegram/adapter.py b/plugins/platforms/telegram/adapter.py index 54b6eb8f70d5a..133a0e0db3929 100644 --- a/plugins/platforms/telegram/adapter.py +++ b/plugins/platforms/telegram/adapter.py @@ -180,6 +180,7 @@ def _bold_label_html(line: str) -> str: from gateway.platforms.event import MessageEvent, MessageType, ProcessingOutcome from plugins.platforms.telegram.telegram_entities import expand_link_entities from plugins.platforms.telegram.telegram_held_inbound import TelegramHeldInboundMixin +from plugins.platforms.telegram.telegram_stream_extras import TelegramStreamExtrasMixin from plugins.platforms.telegram.telegram_ids import normalize_telegram_chat_id from plugins.platforms.telegram.telegram_network import ( SEED_FALLBACK_IPS, TelegramFallbackTransport, discover_fallback_ips, parse_fallback_ip_env, tcp_keepalive_socket_options) @@ -387,6 +388,8 @@ def _strip_mdv2(text: str) -> str: cleaned = re.sub(r'(?{1,3} ', '', cleaned) # blockquote / expandable-quote markers + cleaned = re.sub(r'(?m)\|\|$', '', cleaned) # expandable-quote terminator return cleaned @@ -511,7 +514,7 @@ class _PollingStallError(RuntimeError): """ -class TelegramAdapter(TelegramHeldInboundMixin, BasePlatformAdapter): +class TelegramAdapter(TelegramStreamExtrasMixin, TelegramHeldInboundMixin, BasePlatformAdapter): """Telegram bot adapter: users/groups, MarkdownV2 replies, forum topics, media.""" # Bound for the per-(chat_id, status_key) status-message cache; FIFO half-trim on overflow. @@ -606,6 +609,9 @@ def __init__(self, config: PlatformConfig): self._allow_cjk_rich_messages: bool = self._coerce_bool_extra("allow_cjk_rich_messages", False) self._rich_drafts_enabled: bool = self._coerce_bool_extra("rich_drafts", False) self._rich_send_disabled = self._rich_draft_disabled = False # latched after a capability failure + # Bot API 10.3: stop button on draft previews (can_stop) + ephemeral group whispers. + self._stop_button_enabled: bool = self._coerce_bool_extra("stop_button", False) + self._ephemeral_messages_enabled: bool = self._coerce_bool_extra("ephemeral_messages", False) # Transient sendChatAction failures recur on every keep-typing tick; back off per chat. self._telegram_typing_cooldown_until: dict[str, float] = {} self._telegram_typing_cooldown_seconds: float = self._coerce_float_extra( @@ -1491,6 +1497,11 @@ async def _try_send_rich( # Only non-None routing keys: direct_messages_topic_id is paired with message_thread_id=None. payload.update({k: v for k, v in thread_kwargs.items() if v is not None}) payload.update(self._notification_kwargs(metadata)) + # Completion effect (private chats only) rides the rich send as well β€” the same + # metadata gate as the legacy path; the consumer sets it only on turn-final sends. + _effect_id = self._message_effect_id_for(chat_id, metadata) + if _effect_id: + payload["message_effect_id"] = _effect_id if reply_to_id is not None: # sendRichMessage takes reply_parameters, NOT reply_to_message_id (silently ignored β†’ anchor dropped). payload["reply_parameters"] = {"message_id": reply_to_id} @@ -2749,6 +2760,7 @@ async def _on_platform_update(self, update, context) -> None: admission = getattr(self, "_update_admission", None) if admission is None or admission.get() is None: self._updates_dispatched_total = getattr(self, "_updates_dispatched_total", 0) + 1 + await self._maybe_handle_generation_stopped(update) handler: Optional[Callable[[dict[str, Any], Any], Awaitable[None]]] = getattr(self, "_platform_event_handler", None) if handler is None: return @@ -2772,6 +2784,7 @@ async def _on_platform_update(self, update, context) -> None: self._fail_update_preparation() logger.debug("[%s] gateway_platform_event dispatch error", self.name, exc_info=True) + def _source_for_platform_event_auth(self, update): """Route a supported update to its event-specific auth-source extractor (reactor / editor); raises ``ValueError`` for updates without one so the boundary fails closed.""" @@ -3474,7 +3487,8 @@ async def _send_chunk_markdown_or_plain(self, chunk: str, send_kwargs: dict[str, async def _send_chunk_with_retries( self, chat_id: str, chunk: str, index: int, reply_to: Optional[str], metadata: Optional[dict[str, Any]], - thread_id: Optional[str], used_thread_fallback: bool, error_types: tuple): + thread_id: Optional[str], used_thread_fallback: bool, error_types: tuple, + message_effect_id: Optional[str] = None): """Deliver one chunk: routing, up to 3 attempts, thread-not-found / deleted-anchor / flood handling. Returns ``(msg, used_thread_fallback)`` on success or a ``SendResult`` to return verbatim (fail-loud DM-topic @@ -3495,6 +3509,8 @@ async def _send_chunk_with_retries( send_kwargs = { "chat_id": normalize_telegram_chat_id(chat_id), "reply_to_message_id": reply_to_id, **thread_kwargs, **self._link_preview_kwargs(), **self._notification_kwargs(metadata)} + if message_effect_id: + send_kwargs["message_effect_id"] = message_effect_id return await self._send_chunk_markdown_or_plain(chunk, send_kwargs), used_thread_fallback except _NetErr as send_err: # BadRequest subclasses NetworkError in PTB but is permanent; handle specific cases. @@ -3615,6 +3631,8 @@ async def _quiet() -> None: tracked.add(task) task.add_done_callback(tracked.discard) + + async def send( self, chat_id: str, content: str, reply_to: Optional[str] = None, metadata: Optional[dict[str, Any]] = None) -> SendResult: """Send a message to a Telegram chat.""" @@ -3662,6 +3680,9 @@ async def _send_text_locked( error_types = self._telegram_error_types() chunks: list[str] = [] delivered: list[str] = [] + ephemeral_result = await self._maybe_send_ephemeral(chat_id, content, metadata) + if ephemeral_result is not None: + return ephemeral_result try: # Bot API 10.1 rich fast-path; falls through to legacy MarkdownV2 on permanent/capability # errors or DM-topic skips; returns directly on success or transient failure (no legacy resend). @@ -3712,9 +3733,11 @@ async def _send_chunks( requested_thread_id = self._message_thread_id_for_send(thread_id) used_thread_fallback = False prior = len(delivered) + message_effect_id = self._message_effect_id_for(chat_id, metadata) if len(chunks) == 1 else None for chunk in chunks: outcome = await self._send_chunk_with_retries( - chat_id, chunk, len(delivered), reply_to, metadata, thread_id, used_thread_fallback, error_types) + chat_id, chunk, len(delivered), reply_to, metadata, thread_id, used_thread_fallback, error_types, + message_effect_id=message_effect_id) if isinstance(outcome, SendResult): # Every SendResult returned here is a DEFINITE non-delivery (flood cap, DM-topic refusal); # ambiguous timeouts raise instead, so the remainder is safe to resume from. @@ -4069,48 +4092,6 @@ def supports_draft_streaming(self, chat_type: Optional[str] = None, metadata: Op return False return (chat_type or "").lower() in {"dm", "private"} - async def send_draft(self, chat_id: str, draft_id: int, content: str, metadata: Optional[dict[str, Any]] = None) -> SendResult: - """Stream a partial message via ``sendRichMessageDraft`` (when rich is enabled and supported) else - ``sendMessageDraft``; reusing ``draft_id`` animates the preview. The caller sends the final text.""" - if not self._bot: - return SendResult(success=False, error="not_connected") - # Rich draft fast-path; any failure degrades to the plain draft below. Drafts have no message_id. - if self._should_attempt_rich_draft(content) and await self._try_send_rich_draft(chat_id, draft_id, content, metadata): - return SendResult(success=True, message_id=None) - if not hasattr(self._bot, "send_message_draft"): - return SendResult(success=False, error="api_unavailable") - # Drafts share the regular-send UTF-16 length contract. - text = content if len( - content) <= self.MAX_MESSAGE_LENGTH else self.truncate_message(content, self.MAX_MESSAGE_LENGTH, len_fn=utf16_len)[0] - # Same MarkdownV2 conversion as ``send`` (MarkdownV2 then plain) so the draft doesn't snap at the end. Exception: a Rich - # final with rich drafts disabled previews raw β€” the legacy formatter would turn pipe tables into bullets. - plain_rich_preview = bool( - getattr(self, "_rich_messages_enabled", False) and not getattr(self, "_rich_drafts_enabled", False) - and self._needs_rich_rendering(text)) - draft_thread_kwargs = self._thread_kwargs_for_draft(chat_id, metadata) - for use_markdown in ((False,) if plain_rich_preview else (True, False)): - kwargs: dict[str, Any] = { - "chat_id": normalize_telegram_chat_id(chat_id), "draft_id": int(draft_id), - "text": self.format_message(text) if use_markdown else text} - if use_markdown: - kwargs["parse_mode"] = ParseMode.MARKDOWN_V2 - kwargs.update(draft_thread_kwargs) - try: - if await _await_with_thread_deadline( - self._bot.send_message_draft(**kwargs), timeout=_TEXT_SEND_DEADLINE, label="telegram-send", dump_on_blocked_loop=False): - return SendResult(success=True, message_id=None) - return SendResult(success=False, error="draft_rejected") - except Exception as e: - # MarkdownV2 parse failure β†’ retry once as plain text; anything else returns to the caller, - # which falls back to edit-based streaming for this response. - if use_markdown and self._is_bad_request_error(e): - logger.debug( - "[%s] sendMessageDraft MarkdownV2 rejected, retrying as plain text (chat=%s draft_id=%s): %s", - self.name, chat_id, draft_id, _redact_telegram_error_text(e)) - continue - logger.debug("[%s] sendMessageDraft failed (chat=%s draft_id=%s): %s", self.name, chat_id, draft_id, e) - return SendResult(success=False, error=_redact_telegram_error_text(e)) - return SendResult(success=False, error="draft_rejected") async def _send_message_with_thread_fallback(self, **kwargs): """Send a control-style message (approval prompts, pickers), retrying once without @@ -5685,14 +5666,28 @@ def _convert_header(m): text = re.sub(r'\*([^*\n]+)\*', _ph_wrap('_', '_'), text) text = re.sub(r'~~(.+?)~~', _ph_wrap('~', '~'), text) text = re.sub(r'\|\|(.+?)\|\|', _ph_wrap('||', '||'), text) - # 9) Blockquotes: protect leading > from escaping; expandable quotes (**> starts, trailing || ends). - def _convert_blockquote(m): - prefix, content = m.group(1), m.group(2) # prefix: >, >>, >>>, **>, **>> … - if prefix.startswith('**') and content.endswith('||'): - return _ph(f'{prefix} {_escape_mdv2(content[:-2])}||') - return _ph(f'{prefix} {_escape_mdv2(content)}') - - text = re.sub(r'^((?:\*\*)?>{1,3}) (.+)$', _convert_blockquote, text, flags=re.MULTILINE) + # 9) Blockquotes: protect leading > from escaping; expandable quotes β€” a line starting + # with **> opens the quote and the trailing || on a later line ends it. The state + # resets outside quote runs, so a plain quote line that happens to end with || keeps + # its pipes escaped as literal text. ``checklist_emoji`` metadata keeps its leading + # '-' list marker (a bare βœ… would break the same conversion). + _expandable_quote_open = False + _quote_lines = [] + for _line in text.split('\n'): + _qm = re.match(r'^((?:\*\*)?>{1,3}) (.+)$', _line) + if _qm is None: + _expandable_quote_open = False + _quote_lines.append(_line) + continue + _prefix, _content = _qm.group(1), _qm.group(2) # prefix: >, >>, >>>, **>, **>> … + if _prefix.startswith('**'): + _expandable_quote_open = True + if _expandable_quote_open and _content.endswith('||'): + _expandable_quote_open = False + _quote_lines.append(_ph(f'{_prefix} {_escape_mdv2(_content[:-2])}||')) + else: + _quote_lines.append(_ph(f'{_prefix} {_escape_mdv2(_content)}')) + text = '\n'.join(_quote_lines) # 10) Escape remaining special characters in plain text text = _escape_mdv2(text) # 11) Restore placeholders in reverse insertion order so nested placeholders resolve. diff --git a/plugins/platforms/telegram/telegram_stream_extras.py b/plugins/platforms/telegram/telegram_stream_extras.py new file mode 100644 index 0000000000000..a1ead10a047a0 --- /dev/null +++ b/plugins/platforms/telegram/telegram_stream_extras.py @@ -0,0 +1,165 @@ +"""Opt-in Telegram stop buttons, completion effects, whispers and draft delivery.""" +from __future__ import annotations + +import logging +from typing import Any, Optional +from gateway.platforms.base import SendResult, utf16_len +from gateway.platforms.event import MessageEvent, MessageType +from plugins.platforms.telegram.telegram_ids import normalize_telegram_chat_id + +logger = logging.getLogger("plugins.platforms.telegram.adapter") + +# Bot API message effects (private chats only): emoji name β†’ effect id. Ids are verified +# live against the Bot API β€” an unknown id is rejected server-side, so keep this list to +# the verified set. Paired with sendMessage/sendRichMessage's ``message_effect_id``. +_MESSAGE_EFFECT_IDS: dict[str, str] = { + "πŸ‘": "5107584321108051014", + "πŸ‘Ž": "5104858069142078462", + # "❀" fires only on paid-message sends (EFFECT_ID_INVALID on regular messages) β€” excluded. + "πŸ”₯": "5104841245755180586", + "πŸŽ‰": "5046509860389126442", + "🎊": "5046589136895476101", +} + + +class TelegramStreamExtrasMixin: + async def _maybe_handle_generation_stopped(self, update) -> None: + """Stop button (Bot API 10.3): the user pressed it on a live draft. Re-dispatch as a + synthetic ``/stop`` through the normal inbound path so the running turn is cancelled + with the standard semantics (session lock cleanup, interrupt reason, feedback). + + PTB 22.8 predates this update type; the payload survives in ``Update.api_kwargs``. + The update carries only chat/thread/draft ids β€” in DMs the chat id doubles as the + user id; group presses arrive without a user and are dropped by the auth chain + (fail closed) until group buttons are enabled.""" + if not getattr(self, "_stop_button_enabled", False): + return + payload = (getattr(update, "api_kwargs", None) or {}).get("stopped_message_generation") + if not isinstance(payload, dict): + return + chat = payload.get("chat") if isinstance(payload.get("chat"), dict) else {} + chat_id = chat.get("id") + if chat_id is None: + return + telegram_chat_type = str(chat.get("type") or "private") + chat_type = "group" if telegram_chat_type in {"group", "supergroup"} else ( + "channel" if telegram_chat_type == "channel" else "dm") + thread_id = payload.get("message_thread_id") + try: + source = self.build_source( + chat_id=str(chat_id), chat_type=chat_type, + user_id=(str(chat_id) if chat_type == "dm" else None), + thread_id=(str(thread_id) if thread_id else None), + ) + event = MessageEvent(text="/stop", message_type=MessageType.COMMAND, source=source) + await self.handle_message(event) + logger.info("[%s] Stop button pressed (chat=%s draft_id=%s) β€” dispatched /stop", + self.name, chat_id, payload.get("draft_id")) + except Exception: + logger.debug("[%s] Stop-button dispatch failed", self.name, exc_info=True) + + def _message_effect_id_for(self, chat_id: str, metadata: Optional[dict[str, Any]]) -> Optional[str]: + """``message_effect_id`` for a fresh send (Bot API: private chats only), from + ``metadata['message_effect']`` (an emoji name). Unknown names are skipped.""" + emoji = (metadata or {}).get("message_effect") + if not emoji: + return None + effect_id = emoji if str(emoji).isdigit() else _MESSAGE_EFFECT_IDS.get(str(emoji)) + if not effect_id: + logger.debug("[%s] Unknown message effect %r β€” skipping", self.name, emoji) + return None + try: + cid = normalize_telegram_chat_id(chat_id) + except (TypeError, ValueError): + return None + if not (isinstance(cid, int) and cid > 0): # groups/channels: effect unsupported + return None + return effect_id + + async def _maybe_send_ephemeral(self, chat_id: str, content: str, + metadata: Optional[dict[str, Any]]) -> Optional[SendResult]: + """Group-only ephemeral whisper (opt-in; ``telegram.extra.ephemeral_messages``). + None = not applicable β†’ normal send. Raw call β€” PTB 22.8 predates + ``ephemeral_message_parameters``. Failures do NOT fall back to a normal send: + a whisper that would leak to the whole group is worse than a missing notice.""" + if not getattr(self, "_ephemeral_messages_enabled", False): + return None + from plugins.platforms.telegram.adapter import _redact_telegram_error_text, _strip_mdv2 + receiver = (metadata or {}).get("ephemeral_for") + if not receiver: + return None + try: + cid = normalize_telegram_chat_id(chat_id) + if isinstance(cid, int) and cid > 0: + return None # DMs have nothing to hide β€” normal send + payload = { + "chat_id": cid, + "text": _strip_mdv2(self.format_message(content)) if content else content, + "ephemeral_message_parameters": {"receiver_user_id": int(receiver)}, + } + except Exception: + logger.debug("[%s] ephemeral payload build failed", self.name, exc_info=True) + return SendResult(success=False, error="ephemeral_invalid", retryable=False) + try: + ok = bool(await self._bot.do_api_request("sendMessage", api_kwargs=payload)) + except Exception as e: + logger.debug("[%s] ephemeral send failed: %s", self.name, _redact_telegram_error_text(e), exc_info=True) + return SendResult(success=False, error="ephemeral_failed", retryable=False) + if ok: + return SendResult(success=True, message_id=None) + return SendResult(success=False, error="ephemeral_rejected", retryable=False) + + async def send_draft(self, chat_id: str, draft_id: int, content: str, metadata: Optional[dict[str, Any]] = None) -> SendResult: + """Stream a partial message via ``sendRichMessageDraft`` (when rich is enabled and supported) else + ``sendMessageDraft``; reusing ``draft_id`` animates the preview. The caller sends the final text.""" + from plugins.platforms.telegram.adapter import ( + ParseMode, _TEXT_SEND_DEADLINE, _await_with_thread_deadline, _redact_telegram_error_text, + ) + if not self._bot: + return SendResult(success=False, error="not_connected") + # Rich draft fast-path; any failure degrades to the plain draft below. Drafts have no message_id. + if self._should_attempt_rich_draft(content) and await self._try_send_rich_draft(chat_id, draft_id, content, metadata): + return SendResult(success=True, message_id=None) + if not hasattr(self._bot, "send_message_draft"): + return SendResult(success=False, error="api_unavailable") + # Drafts share the regular-send UTF-16 length contract. + text = content if len( + content) <= self.MAX_MESSAGE_LENGTH else self.truncate_message(content, self.MAX_MESSAGE_LENGTH, len_fn=utf16_len)[0] + # Same MarkdownV2 conversion as ``send`` (MarkdownV2 then plain) so the draft doesn't snap at the end. Exception: a Rich + # final with rich drafts disabled previews raw β€” the legacy formatter would turn pipe tables into bullets. + plain_rich_preview = bool( + getattr(self, "_rich_messages_enabled", False) and not getattr(self, "_rich_drafts_enabled", False) + and self._needs_rich_rendering(text)) + draft_thread_kwargs = self._thread_kwargs_for_draft(chat_id, metadata) + for use_markdown in ((False,) if plain_rich_preview else (True, False)): + kwargs: dict[str, Any] = { + "chat_id": normalize_telegram_chat_id(chat_id), "draft_id": int(draft_id), + "text": self.format_message(text) if use_markdown else text} + if use_markdown: + kwargs["parse_mode"] = ParseMode.MARKDOWN_V2 + kwargs.update(draft_thread_kwargs) + try: + if self._stop_button_enabled: + # PTB predates can_stop; keep raw requests under the send deadline too. + raw_kwargs = {k: ("MarkdownV2" if k == "parse_mode" else v) for k, v in kwargs.items()} + raw_kwargs["can_stop"] = True + accepted = await _await_with_thread_deadline( + self._bot.do_api_request("sendMessageDraft", api_kwargs=raw_kwargs), + timeout=_TEXT_SEND_DEADLINE, label="telegram-send", dump_on_blocked_loop=False) + return SendResult(success=True, message_id=None) if accepted else SendResult( + success=False, error="draft_rejected") + if await _await_with_thread_deadline( + self._bot.send_message_draft(**kwargs), timeout=_TEXT_SEND_DEADLINE, label="telegram-send", dump_on_blocked_loop=False): + return SendResult(success=True, message_id=None) + return SendResult(success=False, error="draft_rejected") + except Exception as e: + # MarkdownV2 parse failure β†’ retry once as plain text; anything else returns to the caller, + # which falls back to edit-based streaming for this response. + if use_markdown and self._is_bad_request_error(e): + logger.debug( + "[%s] sendMessageDraft MarkdownV2 rejected, retrying as plain text (chat=%s draft_id=%s): %s", + self.name, chat_id, draft_id, _redact_telegram_error_text(e)) + continue + logger.debug("[%s] sendMessageDraft failed (chat=%s draft_id=%s): %s", self.name, chat_id, draft_id, _redact_telegram_error_text(e), exc_info=True) + return SendResult(success=False, error=_redact_telegram_error_text(e)) + return SendResult(success=False, error="draft_rejected") diff --git a/tests/gateway/test_busy_session_ack.py b/tests/gateway/test_busy_session_ack.py index 8b70fa2078172..e784f777f6f54 100644 --- a/tests/gateway/test_busy_session_ack.py +++ b/tests/gateway/test_busy_session_ack.py @@ -90,6 +90,44 @@ def _make_adapter(platform_val="telegram"): return adapter +class TestBusyAckEphemeralWhisper: + """Telegram group busy-acks can whisper to the sender (Bot API 10.3 ephemeral).""" + + @pytest.mark.asyncio + async def test_group_ack_carries_ephemeral_for(self): + from gateway.run import GatewayRunner + + runner = object.__new__(GatewayRunner) + adapter = MagicMock() + adapter._send_with_retry = AsyncMock() + + source = SessionSource( + platform=Platform.TELEGRAM, chat_id="-100999", chat_type="supergroup", user_id="777") + evt = MessageEvent(text="hi", message_type=MessageType.TEXT, source=source, message_id="m1") + + await GatewayRunner._send_busy_ack_reply(runner, evt, adapter, "busy ack") + + kwargs = adapter._send_with_retry.await_args.kwargs + assert kwargs["metadata"]["ephemeral_for"] == "777" + + @pytest.mark.asyncio + async def test_dm_ack_has_no_ephemeral_key(self): + from gateway.run import GatewayRunner + + runner = object.__new__(GatewayRunner) + adapter = MagicMock() + adapter._send_with_retry = AsyncMock() + + source = SessionSource( + platform=Platform.TELEGRAM, chat_id="777", chat_type="dm", user_id="777") + evt = MessageEvent(text="hi", message_type=MessageType.TEXT, source=source, message_id="m1") + + await GatewayRunner._send_busy_ack_reply(runner, evt, adapter, "busy ack") + + metadata = adapter._send_with_retry.await_args.kwargs["metadata"] or {} + assert "ephemeral_for" not in metadata + + # --------------------------------------------------------------------------- # Tests # --------------------------------------------------------------------------- diff --git a/tests/gateway/test_stream_consumer.py b/tests/gateway/test_stream_consumer.py index 2bd9d70372c7e..6d87b0ff04aa3 100644 --- a/tests/gateway/test_stream_consumer.py +++ b/tests/gateway/test_stream_consumer.py @@ -299,6 +299,246 @@ class TestSegmentBreakOnToolBoundary: """Verify that on_delta(None) finalizes the current message and starts a new one so the final response appears below tool-progress messages.""" + # ── Telegram single-message streaming (opt-in, #110564) ────────────── + + def test_single_message_display_option_gates_telegram_quiet_progress(self, monkeypatch): + """Opt-in applies on Telegram only, and accepts both quiet text-progress + modes (off/log β€” the same pair the gateway treats as quiet elsewhere).""" + from gateway.config import Platform + from gateway.run_turn import GatewayTurnMixin + + streaming = SimpleNamespace( + cursor=" β–‰", edit_interval=0.5, buffer_threshold=20, + fresh_final_after_seconds=0, transport="edit", + ) + source = SimpleNamespace(platform=Platform.TELEGRAM, chat_id="chat_123", chat_type="dm") + + def _config_for(tool_progress, platform="telegram"): + return {"display": {"platforms": {platform: { + "streaming_single_message": True, "tool_progress": tool_progress, + }}}} + + for tool_progress, expected in (("off", True), ("log", True), ("new", False)): + monkeypatch.setattr( + "gateway.run._load_gateway_config", + lambda tp=tool_progress: _config_for(tp), + ) + config, _ = GatewayTurnMixin._build_stream_consumer_config( + None, source, streaming, MagicMock(), on_missing_cursor="raise", + ) + assert config.single_message_per_turn is expected, ( + f"tool_progress={tool_progress!r}: expected {expected}" + ) + + # Other platforms keep their segment-boundary semantics. + discord_source = SimpleNamespace(platform=Platform.DISCORD, chat_id="chat_9", chat_type="dm") + monkeypatch.setattr( + "gateway.run._load_gateway_config", + lambda: _config_for("off", platform="discord"), + ) + config, _ = GatewayTurnMixin._build_stream_consumer_config( + None, discord_source, streaming, MagicMock(), on_missing_cursor="raise", + ) + assert config.single_message_per_turn is False + + # The gate must mirror the mode the display path actually runs with: the + # HERMES_TOOL_PROGRESS_MODE env bridge wins only while the config never set + # the key β€” otherwise bubbles would be ON while the mode believes them quiet. + def _config_no_tp(enabled=True): + return {"display": {"platforms": {"telegram": {"streaming_single_message": enabled}}}} + + monkeypatch.setenv("HERMES_TOOL_PROGRESS_MODE", "all") + monkeypatch.setattr("gateway.run._load_gateway_config", lambda: _config_no_tp()) + config, _ = GatewayTurnMixin._build_stream_consumer_config( + None, source, streaming, MagicMock(), on_missing_cursor="raise", + ) + assert config.single_message_per_turn is False, "env mode 'all' must close the gate" + + # Explicit config beats the env bridge; env 'off' with unset config stays quiet. + monkeypatch.setattr( + "gateway.run._load_gateway_config", + lambda tp="off": _config_for(tp), + ) + config, _ = GatewayTurnMixin._build_stream_consumer_config( + None, source, streaming, MagicMock(), on_missing_cursor="raise", + ) + assert config.single_message_per_turn is True + monkeypatch.delenv("HERMES_TOOL_PROGRESS_MODE") + monkeypatch.setenv("HERMES_TOOL_PROGRESS_MODE", "off") + monkeypatch.setattr("gateway.run._load_gateway_config", lambda: _config_no_tp()) + config, _ = GatewayTurnMixin._build_stream_consumer_config( + None, source, streaming, MagicMock(), on_missing_cursor="raise", + ) + assert config.single_message_per_turn is True + + def test_single_message_overlay_switches(self, monkeypatch): + """Activity overlay defaults on / thinking off; both follow the master gate.""" + from gateway.config import Platform + from gateway.run_turn import GatewayTurnMixin + + streaming = SimpleNamespace( + cursor=" β–‰", edit_interval=0.5, buffer_threshold=20, + fresh_final_after_seconds=0, transport="edit", + ) + source = SimpleNamespace(platform=Platform.TELEGRAM, chat_id="chat_123", chat_type="dm") + + def _cfg(**over): + plat = {"streaming_single_message": True, "tool_progress": "off"} + plat.update(over) + return {"display": {"platforms": {"telegram": plat}}} + + monkeypatch.setattr("gateway.run._load_gateway_config", lambda: _cfg()) + config, _ = GatewayTurnMixin._build_stream_consumer_config( + None, source, streaming, MagicMock(), on_missing_cursor="raise", + ) + assert config.single_message_activity is True + assert config.single_message_thinking is False + + monkeypatch.setattr( + "gateway.run._load_gateway_config", + lambda: _cfg(streaming_single_message_activity=False, + streaming_single_message_thinking=True), + ) + config, _ = GatewayTurnMixin._build_stream_consumer_config( + None, source, streaming, MagicMock(), on_missing_cursor="raise", + ) + assert config.single_message_activity is False + assert config.single_message_thinking is True + + # Master gate closed (progress not quiet) β†’ overlays collapse even when flipped on. + monkeypatch.setattr( + "gateway.run._load_gateway_config", + lambda: _cfg(tool_progress="all", streaming_single_message_thinking=True), + ) + config, _ = GatewayTurnMixin._build_stream_consumer_config( + None, source, streaming, MagicMock(), on_missing_cursor="raise", + ) + assert config.single_message_per_turn is False + assert config.single_message_activity is False + assert config.single_message_thinking is False + + def test_single_message_4096_split_switch(self, monkeypatch): + """4096-split policy: default off (deferred pagination), opt-in for eager seals.""" + from gateway.config import Platform + from gateway.run_turn import GatewayTurnMixin + + streaming = SimpleNamespace( + cursor=" β–‰", edit_interval=0.5, buffer_threshold=20, + fresh_final_after_seconds=0, transport="edit", + ) + source = SimpleNamespace(platform=Platform.TELEGRAM, chat_id="chat_123", chat_type="dm") + + def _cfg(**over): + plat = {"streaming_single_message": True, "tool_progress": "off"} + plat.update(over) + return {"display": {"platforms": {"telegram": plat}}} + + monkeypatch.setattr("gateway.run._load_gateway_config", lambda: _cfg()) + config, _ = GatewayTurnMixin._build_stream_consumer_config( + None, source, streaming, MagicMock(), on_missing_cursor="raise", + ) + assert config.single_message_4096_split is False + + monkeypatch.setattr( + "gateway.run._load_gateway_config", + lambda: _cfg(streaming_single_message_4096_split=True), + ) + config, _ = GatewayTurnMixin._build_stream_consumer_config( + None, source, streaming, MagicMock(), on_missing_cursor="raise", + ) + assert config.single_message_4096_split is True + + # Master gate closed β†’ collapses even when flipped on. + monkeypatch.setattr( + "gateway.run._load_gateway_config", + lambda: _cfg(tool_progress="all", streaming_single_message_4096_split=True), + ) + config, _ = GatewayTurnMixin._build_stream_consumer_config( + None, source, streaming, MagicMock(), on_missing_cursor="raise", + ) + assert config.single_message_per_turn is False + assert config.single_message_4096_split is False + + def test_message_effect_switch(self, monkeypatch): + """Completion effect: off by default; emoji + min-seconds resolve when enabled.""" + from gateway.config import Platform + from gateway.run_turn import GatewayTurnMixin + + streaming = SimpleNamespace( + cursor=" β–‰", edit_interval=0.5, buffer_threshold=20, + fresh_final_after_seconds=0, transport="edit", + ) + source = SimpleNamespace(platform=Platform.TELEGRAM, chat_id="chat_123", chat_type="dm") + + def _cfg(**over): + plat = {"streaming_single_message": True, "tool_progress": "off"} + plat.update(over) + return {"display": {"platforms": {"telegram": plat}}} + + monkeypatch.setattr("gateway.run._load_gateway_config", lambda: _cfg()) + config, _ = GatewayTurnMixin._build_stream_consumer_config( + None, source, streaming, MagicMock(), on_missing_cursor="raise", + ) + assert config.message_effect == "" + assert config.message_effect_min_seconds == 60.0 + + monkeypatch.setattr( + "gateway.run._load_gateway_config", + lambda: _cfg(message_effects=True, message_effect="πŸ”₯", message_effect_min_seconds=0), + ) + config, _ = GatewayTurnMixin._build_stream_consumer_config( + None, source, streaming, MagicMock(), on_missing_cursor="raise", + ) + assert config.message_effect == "πŸ”₯" + assert config.message_effect_min_seconds == 0.0 + + @pytest.mark.asyncio + async def test_single_message_mode_keeps_one_preview_across_tool_boundaries(self): + """One opt-in preview survives two tool boundaries; the final authoritative + text lands as the last in-place edit (no sealed fragments).""" + adapter = MagicMock() + adapter.send = AsyncMock(return_value=SimpleNamespace(success=True, message_id="msg_1")) + adapter.edit_message = AsyncMock(return_value=SimpleNamespace(success=True)) + adapter.MAX_MESSAGE_LENGTH = 4096 + + consumer = GatewayStreamConsumer( + adapter, "chat_123", + StreamConsumerConfig(edit_interval=0.01, buffer_threshold=5, single_message_per_turn=True), + ) + consumer.on_delta("First, let me check. ") + consumer.on_delta(None) + consumer.on_delta("Checking once more... ") + consumer.on_delta(None) + consumer.on_delta("Here is the answer.") + consumer.finish("Here is the answer.") + await consumer.run() + + assert adapter.send.await_count == 1 + assert adapter.edit_message.call_args_list[-1].kwargs["content"] == "Here is the answer." + + @pytest.mark.asyncio + async def test_single_message_mode_keeps_preview_across_commentary(self): + """Mid-turn commentary goes out as its own message; the evolving preview is + NOT reset (no fresh segment) and keeps editing the same message id.""" + adapter = MagicMock() + adapter.send = AsyncMock(return_value=SimpleNamespace(success=True, message_id="msg_1")) + adapter.edit_message = AsyncMock(return_value=SimpleNamespace(success=True)) + adapter.MAX_MESSAGE_LENGTH = 4096 + + consumer = GatewayStreamConsumer( + adapter, "chat_123", + StreamConsumerConfig(edit_interval=0.01, buffer_threshold=5, single_message_per_turn=True), + ) + consumer.on_delta("Working on it. ") + consumer.on_commentary("A short note.") + consumer.on_delta("Done.") + consumer.finish("Done.") + await consumer.run() + + # preview send + commentary send; every edit still targets the preview message. + assert adapter.send.await_count == 2 + assert adapter.edit_message.call_args_list[-1].kwargs["message_id"] == "msg_1" + @pytest.mark.asyncio async def test_segment_break_removes_cursor(self): diff --git a/tests/gateway/test_stream_consumer_tool_progress.py b/tests/gateway/test_stream_consumer_tool_progress.py index 05f8784683bcf..ee79ccce4c143 100644 --- a/tests/gateway/test_stream_consumer_tool_progress.py +++ b/tests/gateway/test_stream_consumer_tool_progress.py @@ -7,6 +7,7 @@ from __future__ import annotations import asyncio +from tests.waiting import async_wait_until from types import SimpleNamespace from unittest.mock import AsyncMock @@ -181,3 +182,336 @@ async def test_finalize_frame_is_pure_text(self): assert "The answer is 42. Verified." in final_text assert "Searching" not in final_text assert "terminal" not in final_text + assert "---" not in final_text + + +# === SINGLE-MESSAGE ACTIVITY OVERLAY (edit transport, opt-in; #110564 extras) === + + +class _EditAdapter: + """Plain edit-transport adapter (non-native) capturing sends/edits.""" + + MAX_MESSAGE_LENGTH = 4096 + SUPPORTS_MESSAGE_EDITING = True + STREAM_IS_MESSAGE = False + draft_stream_is_message = False + + def __init__(self): + self.sends = [] + self.edits = [] + self.events = [] # ("send"|"edit", message_id, content) in order β€” for replay checks + self._n = 0 + + async def send(self, chat_id, content, **kw): + self._n += 1 + mid = f"m{self._n}" + self.sends.append(content) + self.events.append(("send", mid, content)) + return SimpleNamespace(success=True, message_id=mid) + + async def edit_message(self, chat_id, message_id, content, **kw): + self.edits.append(content) + self.events.append(("edit", message_id, content)) + return SimpleNamespace(success=True) + + async def delete_message(self, chat_id, message_id, **kw): + return SimpleNamespace(success=True) + + +def _make_single_consumer(**over): + cfg = StreamConsumerConfig( + chat_type="dm", cursor="β–Œ", edit_interval=0.05, buffer_threshold=5, + single_message_per_turn=True, single_message_activity=True, + ) + for key, value in over.items(): + setattr(cfg, key, value) + adapter = _EditAdapter() + return GatewayStreamConsumer(adapter, "chat-1", cfg), adapter + + +class TestSingleMessageOverlayGates: + """Acceptance gates for the single-message overlay switches.""" + + def test_accepts_tool_progress(self): + consumer, _ = _make_single_consumer() + assert consumer.accepts_tool_progress is True + consumer, _ = _make_single_consumer(single_message_activity=False) + assert consumer.accepts_tool_progress is False + consumer, _ = _make_single_consumer(single_message_per_turn=False) + assert consumer.accepts_tool_progress is False + + def test_accepts_thinking_progress(self): + consumer, _ = _make_single_consumer(single_message_thinking=True) + assert consumer.accepts_thinking_progress is True + consumer, _ = _make_single_consumer() + assert consumer.accepts_thinking_progress is False + consumer, _ = _make_single_consumer(single_message_per_turn=False, + single_message_thinking=True) + assert consumer.accepts_thinking_progress is False + + def test_single_message_mode_property(self): + consumer, _ = _make_single_consumer() + assert consumer.single_message_mode is True + consumer, _ = _make_single_consumer(single_message_per_turn=False) + assert consumer.single_message_mode is False + + +class TestSingleMessageOverlayDrain: + """Run-loop behavior of the transient overlay (edit transport).""" + + @pytest.mark.asyncio + async def test_overlay_only_opens_preview_and_final_is_clean(self): + consumer, adapter = _make_single_consumer() + consumer.on_tool_progress("πŸ”§ terminal: ls") + + task = asyncio.create_task(consumer.run()) + await async_wait_until(lambda: any("πŸ”§" in c for c in adapter.sends)) + consumer.on_delta("Final answer") + consumer.finish("Final answer") + await asyncio.wait_for(task, timeout=10) + + # The overlay alone opened the preview message. + assert any("πŸ”§" in (s or "") for s in adapter.sends), adapter.sends + # Final edit carries exactly the answer: overlay + rule dropped. + assert adapter.edits, "expected at least one edit" + assert adapter.edits[-1].strip() == "Final answer" + assert "πŸ”§" not in adapter.edits[-1] and "---" not in adapter.edits[-1] + + @pytest.mark.asyncio + async def test_text_overlay_strategy_b_then_text_clears_it(self): + consumer, adapter = _make_single_consumer() + task = asyncio.create_task(consumer.run()) + consumer.on_delta("Part one. ") + await async_wait_until(lambda: adapter.sends) + consumer.on_tool_progress("πŸ”§ web_search: query") + await async_wait_until(lambda: any("πŸ”§" in c for c in adapter.edits)) + consumer.on_delta("Part two.") + consumer.finish("Part one. Part two.") + await asyncio.wait_for(task, timeout=10) + + # Strategy B frame: text + rule + tool line while the tool runs. + assert any( + "Part one." in c and "---" in c and "πŸ”§" in c for c in adapter.edits + ), adapter.edits + final = adapter.edits[-1] + assert "Part one." in final and "Part two." in final + assert "πŸ”§" not in final and "---" not in final + + @pytest.mark.asyncio + async def test_thinking_snippet_replaces_previous(self): + consumer, adapter = _make_single_consumer(single_message_thinking=True) + consumer.on_tool_progress("πŸ’­ first thought") + + task = asyncio.create_task(consumer.run()) + await async_wait_until(lambda: adapter.sends) + consumer.on_tool_progress("πŸ’­ second thought") + await async_wait_until(lambda: any("second thought" in c for c in adapter.edits)) + consumer.on_delta("Answer") + consumer.finish("Answer") + await asyncio.wait_for(task, timeout=10) + + history = [c for c in adapter.sends + adapter.edits if c] + assert any("πŸ’­ second thought" in c for c in history), history + # Never both snippets in one frame β€” a new snippet replaces the old one. + assert not any( + "first thought" in c and "second thought" in c for c in history + ), history + assert "πŸ’­" not in adapter.edits[-1] + + @pytest.mark.asyncio + async def test_overlay_requires_single_mode(self): + consumer, adapter = _make_single_consumer(single_message_per_turn=False) + consumer.on_tool_progress("πŸ”§ nope") + + task = asyncio.create_task(consumer.run()) + consumer.on_delta("x") + consumer.finish("x") + await asyncio.wait_for(task, timeout=10) + + history = [c for c in adapter.sends + adapter.edits if c] + assert not any("πŸ”§" in c for c in history), history + + +class TestSingleMessageDraftLane: + """Draft transport (sendMessageDraft) + single-message overlay (#110564 extras).""" + + @pytest.mark.asyncio + async def test_overlay_rides_draft_preview_and_final_is_one_message(self): + from gateway.platforms.base import BasePlatformAdapter + + DraftAdapter = type( + "DraftAdapter", + (BasePlatformAdapter,), + { + "MAX_MESSAGE_LENGTH": 4096, + "SUPPORTS_MESSAGE_EDITING": True, + "STREAM_IS_MESSAGE": False, + "draft_stream_is_message": False, + "name": "draft-fake", + }, + ) + DraftAdapter.__abstractmethods__ = frozenset() + adapter = DraftAdapter.__new__(DraftAdapter) + adapter._typing_paused = set() + adapter._fatal_error_message = None + adapter.drafts = [] + adapter.sends = [] + + def _supports_draft(chat_type=None, metadata=None, chat_id=None, **kw): + return True + adapter.supports_draft_streaming = _supports_draft + + async def _send_draft(chat_id=None, draft_id=None, content=None, metadata=None, **kw): + adapter.drafts.append(content) + return SimpleNamespace(success=True, message_id=None) + adapter.send_draft = _send_draft + + async def _send(chat_id=None, content=None, **kw): + adapter.sends.append(content) + return SimpleNamespace(success=True, message_id="d1") + adapter.send = _send + + async def _edit(chat_id=None, message_id=None, content=None, **kw): + return SimpleNamespace(success=True) + adapter.edit_message = _edit + + cfg = StreamConsumerConfig( + chat_type="dm", cursor="β–Œ", edit_interval=0.05, buffer_threshold=5, + transport="auto", single_message_per_turn=True, single_message_activity=True, + ) + consumer = GatewayStreamConsumer(adapter, "chat-1", cfg) + task = asyncio.create_task(consumer.run()) + consumer.on_tool_progress("πŸ”§ terminal: ls") + await async_wait_until(lambda: any("πŸ”§" in c for c in adapter.drafts)) + consumer.on_delta("Full answer") + consumer.finish("Full answer") + await asyncio.wait_for(task, timeout=10) + + # Overlay rode the ephemeral draft preview… + assert any("πŸ”§" in (d or "") for d in adapter.drafts), adapter.drafts + # …styled as an expandable quote (Style A). + assert any("**>" in (d or "") and "||" in (d or "") for d in adapter.drafts), adapter.drafts + # …and exactly one persistent message carries the clean final. + assert len(adapter.sends) == 1, adapter.sends + assert adapter.sends[-1].strip() == "Full answer" + + +def _replay_chat(adapter) -> str: + """Final visible text per message, replayed from the adapter's event log.""" + msgs = {} + for kind, mid, content in adapter.events: + msgs[mid] = content + return "\n".join(str(v) for v in msgs.values()) + + +class TestSingleMessageOverflowPolicy: + """4096-split policy: deferred (default) vs eager seals (#110564).""" + + @staticmethod + def _big_text(): + return "A" * 2500 + "\n" + "B" * 2500 + "\n" + "C" * 500 + + async def _run_big(self, consumer): + task = asyncio.create_task(consumer.run()) + consumer.on_delta("A" * 2500) + await async_wait_until(lambda: consumer.adapter.sends) + consumer.on_delta("\n" + "B" * 2500) + await async_wait_until(lambda: any("B" in c for c in consumer.adapter.edits + consumer.adapter.sends)) + interim = list(consumer.adapter.sends + consumer.adapter.edits) + consumer.on_delta("\n" + "C" * 500) + consumer.finish(self._big_text()) + await asyncio.wait_for(task, timeout=10) + return interim + + @pytest.mark.asyncio + async def test_deferred_pagination_is_default(self): + consumer, adapter = _make_single_consumer() + interim = await self._run_big(consumer) + assert all(len(frame) <= adapter.MAX_MESSAGE_LENGTH for frame in interim) + + # One live message for the whole turn: no mid-stream seals… + assert len(adapter.sends) == 1, adapter.sends + # …and the over-limit final edit carries the WHOLE text (the adapter pages it). + counts = {ch: _replay_chat(adapter).count(ch) for ch in "ABC"} + assert counts == {"A": 2500, "B": 2500, "C": 500}, counts + assert len(adapter.edits[-1]) > 4096 + + @pytest.mark.asyncio + async def test_eager_split_seals_head_messages(self): + consumer, adapter = _make_single_consumer(single_message_4096_split=True) + await self._run_big(consumer) + + # Eager policy: the filled head is sealed into its own message mid-stream. + assert len(adapter.sends) >= 2, adapter.sends + counts = {ch: _replay_chat(adapter).count(ch) for ch in "ABC"} + assert counts == {"A": 2500, "B": 2500, "C": 500}, counts + + +class TestOverlayStyleAndEffects: + """Style A overlay framing (draft lane) + completion-effect metadata.""" + + def test_overlay_styled_as_expandable_quote_in_draft_lane(self): + consumer, _ = _make_single_consumer() + consumer._use_draft_streaming = True + out = consumer._style_overlay_lines(["πŸ”§ a", "πŸ’­ b", "πŸ“„ c"]) + assert out == ["**> πŸ”§ a", "> πŸ’­ b", "> πŸ“„ c||"] + + def test_overlay_single_line_styled(self): + consumer, _ = _make_single_consumer() + consumer._use_draft_streaming = True + assert consumer._style_overlay_lines(["πŸ”§ a"]) == ["**> πŸ”§ a||"] + + def test_overlay_plain_in_edit_lane(self): + consumer, _ = _make_single_consumer() # draft lane off β†’ raw frames + assert consumer._style_overlay_lines(["πŸ”§ a", "πŸ’­ b"]) == ["πŸ”§ a", "πŸ’­ b"] + + def test_overlay_plain_when_mode_off(self): + consumer, _ = _make_single_consumer(single_message_per_turn=False) + consumer._use_draft_streaming = True + assert consumer._style_overlay_lines(["πŸ”§ a"]) == ["πŸ”§ a"] + + def test_completion_effect_attaches_on_final(self): + consumer, _ = _make_single_consumer() + consumer.cfg.message_effect = "πŸŽ‰" + consumer.cfg.message_effect_min_seconds = 0.0 + meta = consumer._metadata_for_send(final=True) + assert meta and meta.get("message_effect") == "πŸŽ‰" + assert "message_effect" not in (consumer._metadata_for_send(final=False) or {}) + + def test_completion_effect_gated_by_min_seconds(self): + consumer, _ = _make_single_consumer() + consumer.cfg.message_effect = "πŸŽ‰" + consumer.cfg.message_effect_min_seconds = 3600.0 + assert "message_effect" not in (consumer._metadata_for_send(final=True) or {}) + + +@pytest.mark.parametrize( + "event_type,tool_name,thinking,interrupted,expected", + [ + ("tool.started", "terminal", False, False, True), + ("tool.started", "clarify", False, False, False), + ("tool.started", "_thinking", False, False, False), + ("_thinking", "private reasoning", False, False, False), + ("_thinking", "private reasoning", True, False, True), + ("tool.started", "terminal", False, True, False), + ("_thinking", "private reasoning", True, True, False), + ], +) +def test_single_message_callback_respects_thinking_and_stop( + event_type, tool_name, thinking, interrupted, expected, +): + from unittest.mock import Mock + from gateway.run_turn_runner import TurnRunner + + consumer, _ = _make_single_consumer(single_message_thinking=thinking) + ctx = SimpleNamespace( + log_queue=None, progress_queue=None, _live_status_adapter=None, + _run_still_current=lambda: True, stream_consumer_holder=[consumer], + agent_holder=[SimpleNamespace(is_interrupted=interrupted)], + last_progress_msg=[None], repeat_count=[0], + ) + runner = TurnRunner(SimpleNamespace(), ctx) + runner._progress_build_message = Mock(return_value="πŸ”§ tool activity") + runner.progress_callback(event_type, tool_name, "private reasoning", {}) + consumer._drain_queue() + assert bool(consumer._tool_progress_lines) is expected diff --git a/tests/gateway/test_telegram_adapter_extras.py b/tests/gateway/test_telegram_adapter_extras.py new file mode 100644 index 0000000000000..d1de91cdf61e5 --- /dev/null +++ b/tests/gateway/test_telegram_adapter_extras.py @@ -0,0 +1,147 @@ +"""Telegram adapter extras: stop button (can_stop), message effects, ephemeral whispers. + +Regression coverage for the Bot API additions wired into the Telegram adapter: +- the stop button sends drafts via the raw API with ``can_stop=True`` when enabled, + and a ``stopped_message_generation`` update is re-dispatched as a synthetic ``/stop``; +- completion message effects resolve emoji β†’ id for private chats only; +- ephemeral whispers use the raw API, are group-only, and never leak to the whole + group on failure. +""" + +from types import SimpleNamespace +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from gateway.config import PlatformConfig +from plugins.platforms.telegram.adapter import TelegramAdapter +from plugins.platforms.telegram.telegram_stream_extras import _MESSAGE_EFFECT_IDS + + +@pytest.fixture() +def adapter(): + config = PlatformConfig(enabled=True, token="fake-token") + inst = TelegramAdapter(config) + if not hasattr(inst, "gateway_runner"): + inst.gateway_runner = None + return inst + + +class TestStopButton: + @pytest.mark.asyncio + async def test_draft_uses_can_stop_when_enabled(self, adapter): + adapter._stop_button_enabled = True + adapter._bot = MagicMock() + adapter._bot.do_api_request = AsyncMock(return_value=True) + + result = await adapter.send_draft("123", 7, "partial text") + + assert result.success + args, kwargs = adapter._bot.do_api_request.call_args + assert args[0] == "sendMessageDraft" + payload = kwargs["api_kwargs"] + assert payload["can_stop"] is True + assert payload["draft_id"] == 7 + assert payload["parse_mode"] == "MarkdownV2" # enum converted for the raw body + + @pytest.mark.asyncio + async def test_draft_default_path_without_stop_button(self, adapter): + adapter._stop_button_enabled = False + adapter._bot = MagicMock() + adapter._bot.send_message_draft = AsyncMock(return_value=True) + + result = await adapter.send_draft("123", 7, "partial text") + + assert result.success + adapter._bot.send_message_draft.assert_awaited_once() + adapter._bot.do_api_request.assert_not_called() + + @pytest.mark.asyncio + async def test_stopped_update_dispatches_slash_stop(self, adapter): + adapter._stop_button_enabled = True + adapter.handle_message = AsyncMock() + update = SimpleNamespace(api_kwargs={ + "stopped_message_generation": { + "chat": {"id": 860026573, "type": "private"}, "draft_id": 9}}) + + await adapter._maybe_handle_generation_stopped(update) + + event = adapter.handle_message.await_args.args[0] + assert event.text == "/stop" + assert event.source.chat_id == "860026573" + assert event.source.chat_type == "dm" + assert event.source.user_id == "860026573" + + @pytest.mark.asyncio + async def test_stopped_update_ignored_when_disabled(self, adapter): + adapter._stop_button_enabled = False + adapter.handle_message = AsyncMock() + update = SimpleNamespace(api_kwargs={ + "stopped_message_generation": {"chat": {"id": 1, "type": "private"}, "draft_id": 1}}) + + await adapter._maybe_handle_generation_stopped(update) + + adapter.handle_message.assert_not_awaited() + + @pytest.mark.asyncio + async def test_group_press_has_no_user_fail_closed(self, adapter): + adapter._stop_button_enabled = True + adapter.handle_message = AsyncMock() + update = SimpleNamespace(api_kwargs={ + "stopped_message_generation": { + "chat": {"id": -100123, "type": "supergroup"}, "draft_id": 2}}) + + await adapter._maybe_handle_generation_stopped(update) + + event = adapter.handle_message.await_args.args[0] + assert event.source.user_id is None # the auth chain drops it (fail closed) + + +class TestMessageEffects: + def test_private_chat_resolves(self, adapter): + assert adapter._message_effect_id_for( + "860026573", {"message_effect": "πŸŽ‰"}) == _MESSAGE_EFFECT_IDS["πŸŽ‰"] + + def test_group_and_channel_skip(self, adapter): + assert adapter._message_effect_id_for("-100123", {"message_effect": "πŸŽ‰"}) is None + + def test_unknown_emoji_skips(self, adapter): + assert adapter._message_effect_id_for("123", {"message_effect": "πŸ€–"}) is None + + def test_missing_metadata_skips(self, adapter): + assert adapter._message_effect_id_for("123", None) is None + + +class TestEphemeral: + @pytest.mark.asyncio + async def test_group_whisper_uses_raw_api(self, adapter): + adapter._ephemeral_messages_enabled = True + adapter._bot = MagicMock() + adapter._bot.do_api_request = AsyncMock(return_value=True) + + result = await adapter._maybe_send_ephemeral("-100123", "psst", {"ephemeral_for": "42"}) + + assert result is not None and result.success + payload = adapter._bot.do_api_request.call_args.kwargs["api_kwargs"] + assert payload["ephemeral_message_parameters"] == {"receiver_user_id": 42} + + @pytest.mark.asyncio + async def test_dm_is_not_ephemeral(self, adapter): + adapter._ephemeral_messages_enabled = True + result = await adapter._maybe_send_ephemeral("860026573", "hello", {"ephemeral_for": "42"}) + assert result is None + + @pytest.mark.asyncio + async def test_failure_does_not_leak_to_group(self, adapter): + adapter._ephemeral_messages_enabled = True + adapter._bot = MagicMock() + adapter._bot.do_api_request = AsyncMock(side_effect=RuntimeError("nope")) + + result = await adapter._maybe_send_ephemeral("-100123", "psst", {"ephemeral_for": "42"}) + + assert result is not None and not result.success + + @pytest.mark.asyncio + async def test_disabled_returns_none(self, adapter): + result = await adapter._maybe_send_ephemeral("-100123", "x", {"ephemeral_for": "42"}) + assert result is None diff --git a/tests/gateway/test_telegram_format.py b/tests/gateway/test_telegram_format.py index f26933e0efcc5..0b7ab83866a9b 100644 --- a/tests/gateway/test_telegram_format.py +++ b/tests/gateway/test_telegram_format.py @@ -372,6 +372,32 @@ def test_regular_blockquote_with_pipes_escaped(self, adapter): assert "\\>" not in result +class TestExpandableBlockquote: + """Multi-line expandable quotes: **> opens the quote, the closing line's || survives.""" + + def test_multiline_expandable_keeps_terminator(self, adapter): + result = adapter.format_message("**> c1\n> c2\n> c3||") + assert "**> c1" in result + assert "> c2" in result + assert "> c3||" in result # terminator not escaped + assert "c3\\|" not in result + + def test_terminator_resets_outside_quote_run(self, adapter): + # A ||-ending quote line AFTER the run closed keeps literal pipes (escaped). + result = adapter.format_message("**> open\n> still open\ngap\n> literal||") + assert "> still open" in result + assert "literal\\|\\|" in result + + def test_single_line_expandable(self, adapter): + result = adapter.format_message("**> one||") + assert "**> one||" in result + + +class TestStripExpandableMarks: + def test_strip_expandable_quote_markers(self): + assert _strip_mdv2("**> c1\n> c2\n> c3||") == "c1\nc2\nc3" + + # ========================================================================= # format_message - mixed/complex # ========================================================================= diff --git a/tests/gateway/test_telegram_rich_messages.py b/tests/gateway/test_telegram_rich_messages.py index 13566a19936f7..4c90576b772e1 100644 --- a/tests/gateway/test_telegram_rich_messages.py +++ b/tests/gateway/test_telegram_rich_messages.py @@ -20,6 +20,7 @@ from gateway.config import PlatformConfig from gateway.stream_consumer import GatewayStreamConsumer, StreamConsumerConfig from plugins.platforms.telegram.adapter import TelegramAdapter +from plugins.platforms.telegram.telegram_stream_extras import _MESSAGE_EFFECT_IDS from telegram.error import BadRequest, NetworkError, TimedOut @@ -285,6 +286,28 @@ async def test_transient_rich_error_does_not_legacy_resend(exc): adapter._bot.send_message.assert_not_called() +@pytest.mark.asyncio +async def test_rich_send_carries_completion_effect(): + """metadata['message_effect'] must ride the rich send too (turn-final, DM only).""" + adapter = _make_adapter() + + result = await adapter.send("12345", RICH_CONTENT, metadata={"message_effect": "πŸŽ‰"}) + + assert result.success is True + payload = _rich_api_kwargs(adapter) + assert payload["message_effect_id"] == _MESSAGE_EFFECT_IDS["πŸŽ‰"] + + +@pytest.mark.asyncio +async def test_rich_send_without_effect_omits_the_field(): + adapter = _make_adapter() + + result = await adapter.send("12345", RICH_CONTENT) + + assert result.success is True + assert "message_effect_id" not in _rich_api_kwargs(adapter) + + @pytest.mark.asyncio async def test_rich_transport_error_redacts_bot_token_even_when_redaction_disabled(monkeypatch): from agent import redact diff --git a/website/docs/user-guide/messaging/index.md b/website/docs/user-guide/messaging/index.md index d194653a40cd4..ec029c100c7db 100644 --- a/website/docs/user-guide/messaging/index.md +++ b/website/docs/user-guide/messaging/index.md @@ -903,8 +903,26 @@ display: # Or quiet them entirely interim_assistant_messages: false long_running_notifications: false + # Keep one editable streaming preview across tool calls + # (needs streaming enabled; quiet tool_progress: off or log) + streaming_single_message: true ``` +With streaming enabled (`streaming.enabled` / `display.platforms.telegram.streaming`), `streaming_single_message: true` keeps one editable preview alive for the whole turn β€” text emitted on both sides of tool calls keeps updating the same message instead of starting a new one at every tool boundary. It applies while text tool-progress is quiet (`off` or `log`). + +While that mode is active, a transient **activity overlay** can ride inside the same message: tool-start lines (default on; disable with `streaming_single_message_activity: false`) and thinking snippets (opt-in, `streaming_single_message_thinking: true`) render under a `---` rule beneath the evolving text, and are replaced the moment real text arrives β€” the final message never contains them. + +Over-limit handling follows `streaming_single_message_4096_split`: `true` cuts the preview into several ≀4096-character messages as it fills; the default `false` leaves the live message uncut and pages only at completion, when the text genuinely overflows β€” nothing is lost, pagination just arrives at the end (rich-eligible content first rides the platform's 32,768-character rich cap). + +### Telegram platform extras (opt-in, Bot API 10.3) + +Under `gateway.platforms.telegram.extra`, four switches connect native Telegram features. Each is off by default and degrades to the ordinary path when unsupported: + +- **`stop_button: true`** β€” during DM draft streaming, Telegram shows a **Stop** button; pressing it sends a `stopped_message_generation` update, which Hermes feeds into the same interrupt path as `/stop`. `keep_on_stop` sets what stays in the chat after the stop (`"frame"` keeps the partial preview as a message; `"none"` discards it). +- **`checklist_emoji: "βœ…"`** β€” turns `- [ ]` / `- [x]` task lists into Telegram-native checklists (business accounts only; other chats keep the MarkdownV2 fallback rendering, or use `rich_messages` for the rich-block variant). +- **`message_effects: true`** β€” replies that finish after `message_effect_min_seconds` (default 60) arrive with an animated effect. Emojis map to validated Bot API effect ids (πŸ‘ πŸ‘Ž πŸ”₯ πŸŽ‰ 🎊); a numeric string in `message_effect` is passed through as an explicit effect id. Private chats only β€” groups and channels skip the parameter. +- **`ephemeral_messages: true`** β€” in **groups**, messages carrying `ephemeral_for: ` metadata are visible only to that user; the busy-session acknowledgment uses this to whisper acks to the sender instead of pinging the whole group (with `busy_ack_enabled` on). Send failures do not fall back to a group-wide send. + ### Warning and error notifications (opt-in suppression) Automatic warning and error notifications are shown by default. To suppress @@ -951,6 +969,7 @@ use their turn policy; independent notifications and deferred deliveries evaluat policy at their own delivery boundary. Already delivered messages are not removed. Suppression does not fix an underlying failure or add another logging destination. + ### Progress bubble cleanup (opt-in) Tool-progress messages, the "still working…" heartbeat, and status-callback bubbles can also be auto-deleted after the final response lands. Enable per-platform via `display.platforms..cleanup_progress`: