feat(telegram): single-message streaming + Bot API 10.3 extras (stop button, effects, ephemerals), all opt-in (#110564)

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

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,

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)
 

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.<platform>.<key> → display.<key> → 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),

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)