From de2d9ab8a780aafff341f956c9bef533f29a2b50 Mon Sep 17 00:00:00 2001 From: Maziyang2 <834470850@qq.com> Date: Mon, 14 Sep 2026 13:48:18 +0800 Subject: [PATCH 1/6] feat(gateway): Telegram single-message streaming with quiet-progress gate + docs (#110564) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Keeps one editable streaming preview alive for a whole agent turn β€” text on both sides of tool calls keeps updating the same message instead of sealing a new one at every boundary. Opt-in via display.platforms.telegram.streaming_single_message; Telegram-only; active while text tool-progress is quiet (off/log). Approach prototyped in #110571 (@KoNit-K) β€” same config key and boundary skip β€” extended with: - quiet-progress gate also accepts tool_progress: log (no chat bubbles either; matches the gateway's own {"off", "log"} quiet definition); - mid-turn commentary no longer resets the evolving preview; - docs (messaging guide + cli-config.yaml.example) and boundary tests (2 tool boundaries, commentary continuity, quiet-mode gating). Tests: scripts/run_tests.sh tests/gateway/test_stream_consumer.py base: 3 failed β†’ with change: 74 passed, 1 skipped; adjacent stream/display suites: 98 passed, 0 failed. --- cli-config.yaml.example | 14 ++++ gateway/display_config.py | 5 ++ gateway/run_turn.py | 16 ++++ gateway/stream_consumer.py | 20 +++-- tests/gateway/test_stream_consumer.py | 89 ++++++++++++++++++++++ website/docs/user-guide/messaging/index.md | 6 ++ 6 files changed, 143 insertions(+), 7 deletions(-) diff --git a/cli-config.yaml.example b/cli-config.yaml.example index 629ea2ea93794..035826677a49f 100644 --- a/cli-config.yaml.example +++ b/cli-config.yaml.example @@ -1894,6 +1894,20 @@ 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 + # 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..0a9df3f968694 100644 --- a/gateway/display_config.py +++ b/gateway/display_config.py @@ -20,6 +20,10 @@ "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, # Gateway-only assistant/status chatter; mobile platforms opt down to final-answer-first. "interim_assistant_messages": True, "suppress_warning_notifications": False, @@ -202,6 +206,7 @@ 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, "interim_assistant_messages": _norm_bool, "suppress_warning_notifications": _norm_suppress_warning_notifications, "long_running_notifications": _norm_long_running, diff --git a/gateway/run_turn.py b/gateway/run_turn.py index d0f94d616153c..b9af96d397f96 100644 --- a/gateway/run_turn.py +++ b/gateway/run_turn.py @@ -2696,11 +2696,27 @@ def _pause_typing_before_finalize(_adapter=adapter, _chat_id=source.chat_id) -> 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. + # ``log`` emits no chat bubbles either β€” the gateway's own tool_progress_enabled + # treats {"off", "log"} as quiet β€” so both modes can host one evolving message. + # 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 + _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_display_setting(_user_config, _platform_key, "tool_progress") in ("off", "log") + ) _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, ) return _consumer_cfg, _pause_typing_before_finalize diff --git a/gateway/stream_consumer.py b/gateway/stream_consumer.py index 67c451b186b23..29be421de3d99 100644 --- a/gateway/stream_consumer.py +++ b/gateway/stream_consumer.py @@ -74,6 +74,10 @@ 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 @dataclass @@ -909,14 +913,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 +930,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/tests/gateway/test_stream_consumer.py b/tests/gateway/test_stream_consumer.py index 2bd9d70372c7e..3ce000ac0b622 100644 --- a/tests/gateway/test_stream_consumer.py +++ b/tests/gateway/test_stream_consumer.py @@ -299,6 +299,95 @@ 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 + + @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/website/docs/user-guide/messaging/index.md b/website/docs/user-guide/messaging/index.md index d194653a40cd4..c6bb49d36f318 100644 --- a/website/docs/user-guide/messaging/index.md +++ b/website/docs/user-guide/messaging/index.md @@ -903,8 +903,13 @@ 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`). + ### Warning and error notifications (opt-in suppression) Automatic warning and error notifications are shown by default. To suppress @@ -951,6 +956,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`: From 7784c4a57a9ed3550fb7f92ecb231a26b95581f3 Mon Sep 17 00:00:00 2001 From: Maziyang2 <834470850@qq.com> Date: Mon, 14 Sep 2026 14:41:27 +0800 Subject: [PATCH 2/6] =?UTF-8?q?harden(gateway):=20single-message=20streami?= =?UTF-8?q?ng=20=E2=80=94=20effective-mode=20gate,=20activity=20overlay,?= =?UTF-8?q?=20docs?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Follow-up hardening on the single-message streaming opt-in (#110564), found and fixed by exercising the real Telegram Bot API (10.3 docs + live probes) against the feature: - Gate on the EFFECTIVE tool-progress mode via the shared `resolve_tool_progress()` resolver β€” the same winning source the display path runs with (YAML wins, null inherits to the env bridge, tier defaults last) β€” so the single-message gate can never disagree with the mode the turn actually runs with. Previously the env bridge could leave tool bubbles ON while the gate treated progress as quiet. Red β†’ green test included. - Transient activity overlay (new, opt-in): while the single preview runs, tool starts and thinking snippets render under a `---` rule inside the same message and are replaced the moment real text arrives β€” the final edit never carries them. Switches: display.platforms.

.streaming_single_message_activity (default true; quiet tool_progress would otherwise show nothing during tool runs) and streaming_single_message_thinking (opt-in). Works on both Telegram transports: the ephemeral draft preview (sendMessageDraft) and the edit path; tool lines reuse the consumer's existing native-overlay machinery. - Wire the runner: tool.started events and `_thinking` (scratch/reasoning) events route into the overlay even though progress bubbles are quiet β€” before the progress-queue guard, since single mode runs without a queue. Docs: cli-config.yaml.example + messaging guide updated; verified against the Bot API reference (edit window: unlimited for bot-sent messages; drafts: 30s ephemeral preview, private chats only; 4096-char cap enforced server-side with codepoint counting β€” the consumer's utf16 accounting stays conservative). --- cli-config.yaml.example | 10 + gateway/display_config.py | 6 + gateway/run_turn.py | 22 +- gateway/run_turn_runner.py | 19 ++ gateway/stream_consumer.py | 54 ++++- tests/gateway/test_stream_consumer.py | 76 +++++++ .../test_stream_consumer_tool_progress.py | 214 ++++++++++++++++++ website/docs/user-guide/messaging/index.md | 2 + 8 files changed, 392 insertions(+), 11 deletions(-) diff --git a/cli-config.yaml.example b/cli-config.yaml.example index 035826677a49f..d5c79b5c961d8 100644 --- a/cli-config.yaml.example +++ b/cli-config.yaml.example @@ -1908,6 +1908,16 @@ display: # 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 + # 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 0a9df3f968694..1868e14a2fc32 100644 --- a/gateway/display_config.py +++ b/gateway/display_config.py @@ -24,6 +24,10 @@ # 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, # Gateway-only assistant/status chatter; mobile platforms opt down to final-answer-first. "interim_assistant_messages": True, "suppress_warning_notifications": False, @@ -207,6 +211,8 @@ def _norm_int(value: Any) -> int: "show_reasoning": _norm_bool, "streaming": _norm_bool, "streaming_single_message": _norm_bool, + "streaming_single_message_activity": _norm_bool, + "streaming_single_message_thinking": _norm_bool, "interim_assistant_messages": _norm_bool, "suppress_warning_notifications": _norm_suppress_warning_notifications, "long_running_notifications": _norm_long_running, diff --git a/gateway/run_turn.py b/gateway/run_turn.py index b9af96d397f96..f6db69bf952a6 100644 --- a/gateway/run_turn.py +++ b/gateway/run_turn.py @@ -2697,11 +2697,12 @@ def _pause_typing_before_finalize(_adapter=adapter, _chat_id=source.chat_id) -> if source.platform == Platform.TELEGRAM else 0.0 ) # Preserve the established text↔tool-progress chronology unless progress is quiet. - # ``log`` emits no chat bubbles either β€” the gateway's own tool_progress_enabled - # treats {"off", "log"} as quiet β€” so both modes can host one evolving message. + # 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 + 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 = ( @@ -2709,14 +2710,26 @@ def _pause_typing_before_finalize(_adapter=adapter, _chat_id=source.chat_id) -> and bool(resolve_display_setting( _user_config, _platform_key, "streaming_single_message", False, )) - and resolve_display_setting(_user_config, _platform_key, "tool_progress") in ("off", "log") + 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, + )) _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, ) return _consumer_cfg, _pause_typing_before_finalize @@ -2963,6 +2976,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_runner.py b/gateway/run_turn_runner.py index 7e29a4cdf8cc8..2400e073c497a 100644 --- a/gateway/run_turn_runner.py +++ b/gateway/run_turn_runner.py @@ -139,6 +139,25 @@ def progress_callback(self, event_type: str, tool_name: str | None = None, previ 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 != "clarify" + and ctx._run_still_current() + 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(): + 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]: diff --git a/gateway/stream_consumer.py b/gateway/stream_consumer.py index 29be421de3d99..fceb444a7c556 100644 --- a/gateway/stream_consumer.py +++ b/gateway/stream_consumer.py @@ -78,6 +78,11 @@ class StreamConsumerConfig: # 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 @dataclass @@ -242,13 +247,28 @@ 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) + @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)) @@ -576,7 +596,9 @@ 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 @@ -677,8 +699,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 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]) @@ -739,6 +770,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 @@ -829,7 +869,7 @@ 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 diff --git a/tests/gateway/test_stream_consumer.py b/tests/gateway/test_stream_consumer.py index 3ce000ac0b622..72e70bcce0192 100644 --- a/tests/gateway/test_stream_consumer.py +++ b/tests/gateway/test_stream_consumer.py @@ -341,6 +341,82 @@ def _config_for(tool_progress, platform="telegram"): ) 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 + @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 diff --git a/tests/gateway/test_stream_consumer_tool_progress.py b/tests/gateway/test_stream_consumer_tool_progress.py index 05f8784683bcf..09d389a5c2e35 100644 --- a/tests/gateway/test_stream_consumer_tool_progress.py +++ b/tests/gateway/test_stream_consumer_tool_progress.py @@ -181,3 +181,217 @@ 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: list = [] + self.edits: list = [] + self._n = 0 + + async def send(self, chat_id, content, **kw): + self._n += 1 + self.sends.append(content) + return SimpleNamespace(success=True, message_id=f"m{self._n}") + + async def edit_message(self, chat_id, message_id, content, **kw): + self.edits.append(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 asyncio.sleep(0.25) + consumer.on_delta("Final answer") + consumer.finish("Final answer") + await asyncio.sleep(0.25) + await task + + # 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()) + await asyncio.sleep(0.05) + consumer.on_delta("Part one. ") + await asyncio.sleep(0.25) + consumer.on_tool_progress("πŸ”§ web_search: query") + await asyncio.sleep(0.25) + consumer.on_delta("Part two.") + consumer.finish("Part one. Part two.") + await asyncio.sleep(0.25) + await task + + # 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 asyncio.sleep(0.25) + consumer.on_tool_progress("πŸ’­ second thought") + await asyncio.sleep(0.25) + consumer.on_delta("Answer") + consumer.finish("Answer") + await asyncio.sleep(0.25) + await task + + 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()) + await asyncio.sleep(0.2) + consumer.on_delta("x") + consumer.finish("x") + await asyncio.sleep(0.2) + await task + + 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()) + await asyncio.sleep(0.05) + consumer.on_tool_progress("πŸ”§ terminal: ls") + await asyncio.sleep(0.25) + consumer.on_delta("Full answer") + consumer.finish("Full answer") + await asyncio.sleep(0.25) + await task + + # Overlay rode the ephemeral draft preview… + assert any("πŸ”§" 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" diff --git a/website/docs/user-guide/messaging/index.md b/website/docs/user-guide/messaging/index.md index c6bb49d36f318..40a9542730ead 100644 --- a/website/docs/user-guide/messaging/index.md +++ b/website/docs/user-guide/messaging/index.md @@ -910,6 +910,8 @@ display: 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. + ### Warning and error notifications (opt-in suppression) Automatic warning and error notifications are shown by default. To suppress From 36eaf9e3e68c61e5ce3fd2e4304e6baf9c1ec1a0 Mon Sep 17 00:00:00 2001 From: Maziyang2 <834470850@qq.com> Date: Mon, 14 Sep 2026 15:03:59 +0800 Subject: [PATCH 3/6] feat(gateway): single-message 4096-split policy switch + tests + docs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Per the requested rule for the single-message streaming mode: - `streaming_single_message_4096_split: true` β€” over-limit content is cut into several ≀4096 messages as the preview fills (classic eager sealing). - Default `false` β€” 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 (rich-eligible content first rides the platform's 32,768-character cap). Nothing is ever lost β€” pagination just arrives later. Consumer gate `_eager_overflow_split()`; config plumbing in run_turn + display_config; docs in cli-config.yaml.example and the messaging guide. Tests: deferred keeps `sends == 1` with an over-limit final edit; eager seals into multiple messages; both replayed for zero lost/duplicated characters; config resolution incl. master-gate collapse. 101 passed in the two suites. --- cli-config.yaml.example | 7 +++ gateway/display_config.py | 2 + gateway/run_turn.py | 4 ++ gateway/stream_consumer.py | 22 ++++++- tests/gateway/test_stream_consumer.py | 42 +++++++++++++ .../test_stream_consumer_tool_progress.py | 60 ++++++++++++++++++- website/docs/user-guide/messaging/index.md | 2 + 7 files changed, 134 insertions(+), 5 deletions(-) diff --git a/cli-config.yaml.example b/cli-config.yaml.example index d5c79b5c961d8..180e24c528a9f 100644 --- a/cli-config.yaml.example +++ b/cli-config.yaml.example @@ -1918,6 +1918,13 @@ display: 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 + # 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 1868e14a2fc32..2d97ac04158e3 100644 --- a/gateway/display_config.py +++ b/gateway/display_config.py @@ -28,6 +28,7 @@ # 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, # Gateway-only assistant/status chatter; mobile platforms opt down to final-answer-first. "interim_assistant_messages": True, "suppress_warning_notifications": False, @@ -213,6 +214,7 @@ def _norm_int(value: Any) -> int: "streaming_single_message": _norm_bool, "streaming_single_message_activity": _norm_bool, "streaming_single_message_thinking": _norm_bool, + "streaming_single_message_4096_split": _norm_bool, "interim_assistant_messages": _norm_bool, "suppress_warning_notifications": _norm_suppress_warning_notifications, "long_running_notifications": _norm_long_running, diff --git a/gateway/run_turn.py b/gateway/run_turn.py index f6db69bf952a6..52805919a0ae8 100644 --- a/gateway/run_turn.py +++ b/gateway/run_turn.py @@ -2722,6 +2722,9 @@ def _pause_typing_before_finalize(_adapter=adapter, _chat_id=source.chat_id) -> _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, + )) _consumer_cfg = StreamConsumerConfig( edit_interval=scfg.edit_interval, buffer_threshold=scfg.buffer_threshold, cursor=_effective_cursor, @@ -2730,6 +2733,7 @@ def _pause_typing_before_finalize(_adapter=adapter, _chat_id=source.chat_id) -> 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, ) return _consumer_cfg, _pause_typing_before_finalize diff --git a/gateway/stream_consumer.py b/gateway/stream_consumer.py index fceb444a7c556..82510c76a7bf5 100644 --- a/gateway/stream_consumer.py +++ b/gateway/stream_consumer.py @@ -83,6 +83,12 @@ class StreamConsumerConfig: # 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 @dataclass @@ -252,6 +258,16 @@ 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 when the live surface renders tool progress in-stream: native streaming @@ -604,10 +620,12 @@ async def run(self) -> None: # 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(): diff --git a/tests/gateway/test_stream_consumer.py b/tests/gateway/test_stream_consumer.py index 72e70bcce0192..0f0cc1882dae8 100644 --- a/tests/gateway/test_stream_consumer.py +++ b/tests/gateway/test_stream_consumer.py @@ -417,6 +417,48 @@ def _cfg(**over): 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 + @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 diff --git a/tests/gateway/test_stream_consumer_tool_progress.py b/tests/gateway/test_stream_consumer_tool_progress.py index 09d389a5c2e35..8cca29f9a6d87 100644 --- a/tests/gateway/test_stream_consumer_tool_progress.py +++ b/tests/gateway/test_stream_consumer_tool_progress.py @@ -196,17 +196,21 @@ class _EditAdapter: draft_stream_is_message = False def __init__(self): - self.sends: list = [] - self.edits: list = [] + 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) - return SimpleNamespace(success=True, message_id=f"m{self._n}") + 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): @@ -395,3 +399,53 @@ async def _edit(chat_id=None, message_id=None, content=None, **kw): # …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()) + await asyncio.sleep(0.05) + consumer.on_delta("A" * 2500) + await asyncio.sleep(0.25) + consumer.on_delta("\n" + "B" * 2500) + await asyncio.sleep(0.25) + consumer.on_delta("\n" + "C" * 500) + consumer.finish(self._big_text()) + await asyncio.sleep(0.25) + await task + + @pytest.mark.asyncio + async def test_deferred_pagination_is_default(self): + consumer, adapter = _make_single_consumer() + await self._run_big(consumer) + + # 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 diff --git a/website/docs/user-guide/messaging/index.md b/website/docs/user-guide/messaging/index.md index 40a9542730ead..499a2f29fa487 100644 --- a/website/docs/user-guide/messaging/index.md +++ b/website/docs/user-guide/messaging/index.md @@ -912,6 +912,8 @@ With streaming enabled (`streaming.enabled` / `display.platforms.telegram.stream 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). + ### Warning and error notifications (opt-in suppression) Automatic warning and error notifications are shown by default. To suppress From d9e693157acde0930b002b749dd71cff098356ec Mon Sep 17 00:00:00 2001 From: Maziyang2 <834470850@qq.com> Date: Tue, 15 Sep 2026 00:31:29 +0800 Subject: [PATCH 4/6] =?UTF-8?q?feat(telegram):=20Bot=20API=2010.3=20platfo?= =?UTF-8?q?rm=20features=20=E2=80=94=20stop=20button,=20checklists,=20mess?= =?UTF-8?q?age=20effects,=20ephemeral=20scope?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - stop_button extra: DM draft previews carry can_stop via the raw API; a stopped_message_generation update re-dispatches as a synthetic /stop into the standard interrupt path; disabled or unauthorized presses are ignored - checklist_emoji extra + sender scope β†’ native sendChecklist (business accounts only; MarkdownV2 fallback elsewhere) - message_effects switch: validated effect ids (πŸ‘πŸ‘ŽπŸ”₯πŸŽ‰πŸŽŠ, live-verified; ❀ requires paid sends and is excluded), numeric pass-through, private chats only; attach on turn-final sends after message_effect_min_seconds (60s) - ephemeral_messages extra: group busy-acks whisper to the sender via extra_metadata; no group-wide leak on failure - formatter: stateful multi-line expandable-quote conversion (style A)\n\nTests: telegram format/adapter-extras/consumer/busy-ack suites green; 12-suite gateway regression 307 passed. .docs: messaging/index.md + cli-config extras section --- cli-config.yaml.example | 10 ++ gateway/display_config.py | 8 + gateway/run_busy.py | 20 ++- gateway/run_turn.py | 13 ++ gateway/stream_consumer.py | 28 ++- plugins/platforms/telegram/adapter.py | 163 ++++++++++++++++-- tests/gateway/test_busy_session_ack.py | 38 ++++ tests/gateway/test_stream_consumer.py | 33 ++++ .../test_stream_consumer_tool_progress.py | 40 +++++ tests/gateway/test_telegram_adapter_extras.py | 146 ++++++++++++++++ tests/gateway/test_telegram_format.py | 26 +++ website/docs/user-guide/messaging/index.md | 9 + 12 files changed, 519 insertions(+), 15 deletions(-) create mode 100644 tests/gateway/test_telegram_adapter_extras.py diff --git a/cli-config.yaml.example b/cli-config.yaml.example index 180e24c528a9f..9259485b8fb14 100644 --- a/cli-config.yaml.example +++ b/cli-config.yaml.example @@ -1925,6 +1925,16 @@ display: # 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 2d97ac04158e3..27cc14bc70a1d 100644 --- a/gateway/display_config.py +++ b/gateway/display_config.py @@ -29,6 +29,12 @@ "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, @@ -215,6 +221,8 @@ def _norm_int(value: Any) -> int: "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 52805919a0ae8..dbcae27918e2c 100644 --- a/gateway/run_turn.py +++ b/gateway/run_turn.py @@ -2725,6 +2725,17 @@ def _pause_typing_before_finalize(_adapter=adapter, _chat_id=source.chat_id) -> _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, @@ -2734,6 +2745,8 @@ def _pause_typing_before_finalize(_adapter=adapter, _chat_id=source.chat_id) -> 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 82510c76a7bf5..ff10491c0a202 100644 --- a/gateway/stream_consumer.py +++ b/gateway/stream_consumer.py @@ -89,6 +89,10 @@ class StreamConsumerConfig: # 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 @@ -170,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] = [] @@ -289,10 +294,25 @@ def on_tool_progress(self, line: str) -> None: 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 @@ -304,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). diff --git a/plugins/platforms/telegram/adapter.py b/plugins/platforms/telegram/adapter.py index 54b6eb8f70d5a..88ac93b06f9c6 100644 --- a/plugins/platforms/telegram/adapter.py +++ b/plugins/platforms/telegram/adapter.py @@ -387,9 +387,23 @@ 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 +# 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", +} + _CHUNK_INDICATOR_ON_FENCE_RE = re.compile(r'(?m)^``` (?P(?:\\)?\(\d+/\d+(?:\\)?\))$') @@ -606,6 +620,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( @@ -2749,6 +2766,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 +2790,44 @@ 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) + 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 + try: + payload = (getattr(update, "api_kwargs", None) or {}).get("stopped_message_generation") + except Exception: + payload = None + 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 _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 +3530,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 +3552,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 +3674,62 @@ async def _quiet() -> None: tracked.add(task) task.add_done_callback(tracked.discard) + 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.""" + try: + emoji = (metadata or {}).get("message_effect") + except Exception: + emoji = None + 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 Exception: + 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 + try: + receiver = (metadata or {}).get("ephemeral_for") + except Exception: + receiver = None + 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)) + 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( 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 +3777,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 +3830,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. @@ -4096,6 +4216,15 @@ async def send_draft(self, chat_id: str, draft_id: int, content: str, metadata: 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) @@ -5685,14 +5814,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/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 0f0cc1882dae8..6d87b0ff04aa3 100644 --- a/tests/gateway/test_stream_consumer.py +++ b/tests/gateway/test_stream_consumer.py @@ -459,6 +459,39 @@ def _cfg(**over): 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 diff --git a/tests/gateway/test_stream_consumer_tool_progress.py b/tests/gateway/test_stream_consumer_tool_progress.py index 8cca29f9a6d87..7492565dee2ee 100644 --- a/tests/gateway/test_stream_consumer_tool_progress.py +++ b/tests/gateway/test_stream_consumer_tool_progress.py @@ -396,6 +396,8 @@ async def _edit(chat_id=None, message_id=None, content=None, **kw): # 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" @@ -449,3 +451,41 @@ async def test_eager_split_seals_head_messages(self): 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 {}) diff --git a/tests/gateway/test_telegram_adapter_extras.py b/tests/gateway/test_telegram_adapter_extras.py new file mode 100644 index 0000000000000..c39b14beb3045 --- /dev/null +++ b/tests/gateway/test_telegram_adapter_extras.py @@ -0,0 +1,146 @@ +"""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, _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/website/docs/user-guide/messaging/index.md b/website/docs/user-guide/messaging/index.md index 499a2f29fa487..ec029c100c7db 100644 --- a/website/docs/user-guide/messaging/index.md +++ b/website/docs/user-guide/messaging/index.md @@ -914,6 +914,15 @@ While that mode is active, a transient **activity overlay** can ride inside the 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 From 34b496a76fe832a24da755c6da4d316463219712 Mon Sep 17 00:00:00 2001 From: Maziyang2 <834470850@qq.com> Date: Tue, 15 Sep 2026 14:31:36 +0800 Subject: [PATCH 5/6] feat(telegram): completion effects ride the rich send path too - _try_send_rich attaches message_effect_id from metadata under the same gate as the legacy path (turn-final + DM-only + verified id); a completed reply rendered via sendRichMessage no longer silently drops the effect - tests: rich send carries the field when requested, omits it otherwise --- plugins/platforms/telegram/adapter.py | 5 ++++ tests/gateway/test_telegram_rich_messages.py | 24 +++++++++++++++++++- 2 files changed, 28 insertions(+), 1 deletion(-) diff --git a/plugins/platforms/telegram/adapter.py b/plugins/platforms/telegram/adapter.py index 88ac93b06f9c6..db6e3a507daf8 100644 --- a/plugins/platforms/telegram/adapter.py +++ b/plugins/platforms/telegram/adapter.py @@ -1508,6 +1508,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} diff --git a/tests/gateway/test_telegram_rich_messages.py b/tests/gateway/test_telegram_rich_messages.py index 13566a19936f7..5812058a7b400 100644 --- a/tests/gateway/test_telegram_rich_messages.py +++ b/tests/gateway/test_telegram_rich_messages.py @@ -19,7 +19,7 @@ from gateway.config import PlatformConfig from gateway.stream_consumer import GatewayStreamConsumer, StreamConsumerConfig -from plugins.platforms.telegram.adapter import TelegramAdapter +from plugins.platforms.telegram.adapter import TelegramAdapter, _MESSAGE_EFFECT_IDS from telegram.error import BadRequest, NetworkError, TimedOut @@ -285,6 +285,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 From b39b9e8bd85ead73a175dced192e0ba8a33776fd Mon Sep 17 00:00:00 2001 From: Maziyang2 <834470850@qq.com> Date: Wed, 14 Oct 2026 15:59:17 +0800 Subject: [PATCH 6/6] fix(gateway): align Telegram streaming with current delivery contracts Keep upstream overflow failure handling, Telegram admission accounting, and send deadlines when replaying the single-message feature. Extract streaming configuration, progress dispatch, and Telegram draft extras into topical modules to satisfy current code health limits. Bound deferred live previews without truncating the buffered final answer. Preserve explicit thinking opt-in and suppress overlays after stop. Replace fixed waits in new streaming tests with observed transport events. Validation: 367 tests passed across 21 focused and adjacent gateway files on Python 3.14.7; all 12 scripts/check checks passed. Preview overflow and overlay gates reproduced failures before their fixes. No live Telegram send/edit/stop test was run. --- gateway/run_turn.py | 86 +-------- gateway/run_turn_progress.py | 81 +++++++++ gateway/run_turn_runner.py | 77 +------- gateway/run_turn_stream_config.py | 91 ++++++++++ gateway/stream_consumer.py | 7 +- plugins/platforms/telegram/adapter.py | 159 +---------------- .../telegram/telegram_stream_extras.py | 165 ++++++++++++++++++ .../test_stream_consumer_tool_progress.py | 76 +++++--- tests/gateway/test_telegram_adapter_extras.py | 3 +- tests/gateway/test_telegram_rich_messages.py | 3 +- 10 files changed, 407 insertions(+), 341 deletions(-) create mode 100644 gateway/run_turn_progress.py create mode 100644 gateway/run_turn_stream_config.py create mode 100644 plugins/platforms/telegram/telegram_stream_extras.py diff --git a/gateway/run_turn.py b/gateway/run_turn.py index dbcae27918e2c..af6df36e4c078 100644 --- a/gateway/run_turn.py +++ b/gateway/run_turn.py @@ -2667,88 +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 - ) - # 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 + + 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).""" 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 2400e073c497a..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,79 +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()) - # 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 != "clarify" - and ctx._run_still_current() - 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(): - 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) 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 ff10491c0a202..11a8e6245b1e9 100644 --- a/gateway/stream_consumer.py +++ b/gateway/stream_consumer.py @@ -750,7 +750,7 @@ def _drain_queue(self) -> _Tick: if not existing.startswith("πŸ’­") ] self._tool_progress_lines.append(line) - if len(self._tool_progress_lines) > 6: + 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: @@ -917,6 +917,11 @@ async def _push_update(self, tick: _Tick) -> None: 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. diff --git a/plugins/platforms/telegram/adapter.py b/plugins/platforms/telegram/adapter.py index db6e3a507daf8..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) @@ -392,18 +393,6 @@ def _strip_mdv2(text: str) -> str: return cleaned -# 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", -} - _CHUNK_INDICATOR_ON_FENCE_RE = re.compile(r'(?m)^``` (?P(?:\\)?\(\d+/\d+(?:\\)?\))$') @@ -525,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. @@ -2795,43 +2784,6 @@ 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) - 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 - try: - payload = (getattr(update, "api_kwargs", None) or {}).get("stopped_message_generation") - except Exception: - payload = None - 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 _source_for_platform_event_auth(self, update): """Route a supported update to its event-specific auth-source extractor (reactor / editor); @@ -3679,61 +3631,7 @@ async def _quiet() -> None: tracked.add(task) task.add_done_callback(tracked.discard) - 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.""" - try: - emoji = (metadata or {}).get("message_effect") - except Exception: - emoji = None - 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 Exception: - 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 - try: - receiver = (metadata or {}).get("ephemeral_for") - except Exception: - receiver = None - 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)) - 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( self, chat_id: str, content: str, reply_to: Optional[str] = None, metadata: Optional[dict[str, Any]] = None) -> SendResult: @@ -4194,57 +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 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, 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 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_stream_consumer_tool_progress.py b/tests/gateway/test_stream_consumer_tool_progress.py index 7492565dee2ee..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 @@ -264,11 +265,10 @@ async def test_overlay_only_opens_preview_and_final_is_clean(self): consumer.on_tool_progress("πŸ”§ terminal: ls") task = asyncio.create_task(consumer.run()) - await asyncio.sleep(0.25) + await async_wait_until(lambda: any("πŸ”§" in c for c in adapter.sends)) consumer.on_delta("Final answer") consumer.finish("Final answer") - await asyncio.sleep(0.25) - await task + 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 @@ -281,15 +281,13 @@ async def test_overlay_only_opens_preview_and_final_is_clean(self): async def test_text_overlay_strategy_b_then_text_clears_it(self): consumer, adapter = _make_single_consumer() task = asyncio.create_task(consumer.run()) - await asyncio.sleep(0.05) consumer.on_delta("Part one. ") - await asyncio.sleep(0.25) + await async_wait_until(lambda: adapter.sends) consumer.on_tool_progress("πŸ”§ web_search: query") - await asyncio.sleep(0.25) + 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.sleep(0.25) - await task + await asyncio.wait_for(task, timeout=10) # Strategy B frame: text + rule + tool line while the tool runs. assert any( @@ -305,13 +303,12 @@ async def test_thinking_snippet_replaces_previous(self): consumer.on_tool_progress("πŸ’­ first thought") task = asyncio.create_task(consumer.run()) - await asyncio.sleep(0.25) + await async_wait_until(lambda: adapter.sends) consumer.on_tool_progress("πŸ’­ second thought") - await asyncio.sleep(0.25) + await async_wait_until(lambda: any("second thought" in c for c in adapter.edits)) consumer.on_delta("Answer") consumer.finish("Answer") - await asyncio.sleep(0.25) - await task + 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 @@ -327,11 +324,9 @@ async def test_overlay_requires_single_mode(self): consumer.on_tool_progress("πŸ”§ nope") task = asyncio.create_task(consumer.run()) - await asyncio.sleep(0.2) consumer.on_delta("x") consumer.finish("x") - await asyncio.sleep(0.2) - await task + 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 @@ -386,13 +381,11 @@ async def _edit(chat_id=None, message_id=None, content=None, **kw): ) consumer = GatewayStreamConsumer(adapter, "chat-1", cfg) task = asyncio.create_task(consumer.run()) - await asyncio.sleep(0.05) consumer.on_tool_progress("πŸ”§ terminal: ls") - await asyncio.sleep(0.25) + await async_wait_until(lambda: any("πŸ”§" in c for c in adapter.drafts)) consumer.on_delta("Full answer") consumer.finish("Full answer") - await asyncio.sleep(0.25) - await task + 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 @@ -420,20 +413,21 @@ def _big_text(): async def _run_big(self, consumer): task = asyncio.create_task(consumer.run()) - await asyncio.sleep(0.05) consumer.on_delta("A" * 2500) - await asyncio.sleep(0.25) + await async_wait_until(lambda: consumer.adapter.sends) consumer.on_delta("\n" + "B" * 2500) - await asyncio.sleep(0.25) + 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.sleep(0.25) - await task + 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() - await self._run_big(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 @@ -489,3 +483,35 @@ def test_completion_effect_gated_by_min_seconds(self): 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 index c39b14beb3045..d1de91cdf61e5 100644 --- a/tests/gateway/test_telegram_adapter_extras.py +++ b/tests/gateway/test_telegram_adapter_extras.py @@ -14,7 +14,8 @@ import pytest from gateway.config import PlatformConfig -from plugins.platforms.telegram.adapter import TelegramAdapter, _MESSAGE_EFFECT_IDS +from plugins.platforms.telegram.adapter import TelegramAdapter +from plugins.platforms.telegram.telegram_stream_extras import _MESSAGE_EFFECT_IDS @pytest.fixture() diff --git a/tests/gateway/test_telegram_rich_messages.py b/tests/gateway/test_telegram_rich_messages.py index 5812058a7b400..4c90576b772e1 100644 --- a/tests/gateway/test_telegram_rich_messages.py +++ b/tests/gateway/test_telegram_rich_messages.py @@ -19,7 +19,8 @@ from gateway.config import PlatformConfig from gateway.stream_consumer import GatewayStreamConsumer, StreamConsumerConfig -from plugins.platforms.telegram.adapter import TelegramAdapter, _MESSAGE_EFFECT_IDS +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