feat(telegram): stream agent replies by editing the placeholder message in place (#3534)

* docs(spec): telegram streaming output design Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * docs(plan): telegram streaming implementation plan Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * feat(telegram): report streaming support for telegram channel Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * test(channels): use slack as the non-streaming sample channel in manager tests Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * feat(telegram): register running-reply placeholder as stream target Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * test(telegram): pin last_edit_at sentinel in placeholder registration test Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * refactor(telegram): extract _send_new_message from send() Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * feat(telegram): edit streamed message in place for non-final updates Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * feat(telegram): finalize streamed message with overflow splitting When is_final=True arrives and stream state exists, pop the state, edit the streamed placeholder with the final text, split overflow into follow-up send_message calls, update _last_bot_message, and clear stream state. Falls back to _send_new_message when no stream state is registered. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * test(telegram): exercise the not-modified handler in final edit path Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * docs: telegram channel now streams replies via message editing Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * fix(telegram): harden final-delivery path with guarded retry and chunk retries Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * fix(channels): accept runtime 'messages' SSE event for streaming text accumulation The embedded runtime (matching LangGraph Platform semantics) emits SSE event name 'messages' for the requested 'messages-tuple' stream mode, so the manager never accumulated token deltas and streaming channels only updated from end-of-step 'values' snapshots — on Telegram this looked like 'Working on it...' followed by the full answer in one block. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * feat(telegram): widen stream-edit throttle to 3s in group chats Telegram caps bots at 20 messages/minute per group, stricter than the 1 msg/s per-chat guideline. Groups have negative chat ids, so pick the interval by sign. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * fix(telegram): address review findings — thread fallback messages, bound stream registry, share stream-event constants - Fallback/new stream messages now carry reply_to_message_id parsed from thread_ts so they stay nested under the user's message (finding 1) - STREAM_MODES / MESSAGE_STREAM_EVENTS constants link the requested stream modes to the SSE event names they arrive under (finding 2) - _register_stream_message bounds the in-flight registry at 256 entries, evicting oldest, guarding against leaks when a final never arrives (finding 4) Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> --------- Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
2026-06-13 10:55:59 +00:00 · 2026-06-13 08:38:28 +08:00
parent 3475f7cdad
commit 839fa99237
6 changed files with 1557 additions and 23 deletions
@@ -384,10 +384,10 @@ Bridges external messaging platforms (Feishu, Slack, Telegram, Discord, DingTalk
 **Components**:
 - `message_bus.py` - Async pub/sub hub (`InboundMessage` → queue → dispatcher; `OutboundMessage` → callbacks → channels)
 - `store.py` - JSON-file persistence mapping `channel_name:chat_id[:topic_id]` → `thread_id` (keys are `channel:chat` for root conversations and `channel:chat:topic` for threaded conversations)
- `manager.py` - Core dispatcher: creates threads via `client.threads.create()`, routes commands, keeps Slack/Telegram on `client.runs.wait()`, and uses `client.runs.stream(["messages-tuple", "values"])` for Feishu incremental outbound updates
+- `manager.py` - Core dispatcher: creates threads via `client.threads.create()`, routes commands, keeps Slack/Discord on `client.runs.wait()`, and uses `client.runs.stream(["messages-tuple", "values"])` for Feishu/Telegram incremental outbound updates
 - `base.py` - Abstract `Channel` base class (start/stop/send lifecycle)
 - `service.py` - Manages lifecycle of all configured channels from `config.yaml`
- `slack.py` / `feishu.py` / `telegram.py` / `discord.py` / `dingtalk.py` - Platform-specific implementations (`feishu.py` tracks the running card `message_id` in memory and patches the same card in place; `dingtalk.py` optionally uses AI Card streaming for in-place updates when `card_template_id` is configured)
+- `slack.py` / `feishu.py` / `telegram.py` / `discord.py` / `dingtalk.py` - Platform-specific implementations (`feishu.py` tracks the running card `message_id` in memory and patches the same card in place; `telegram.py` registers the "Working on it..." placeholder as the stream target and edits it in place via `editMessageText`; `dingtalk.py` optionally uses AI Card streaming for in-place updates when `card_template_id` is configured)
 - `app/gateway/routers/channel_connections.py` - Browser-facing user connection and disconnect APIs
 - `deerflow.persistence.channel_connections` - SQL-backed user-owned connection, optional credential, connect state, and conversation store

@@ -396,12 +396,13 @@ Bridges external messaging platforms (Feishu, Slack, Telegram, Discord, DingTalk
 2. `ChannelManager._dispatch_loop()` consumes from queue
 3. For user-owned channel connections, incoming messages carry `connection_id`, `owner_user_id`, and `workspace_id`; `owner_user_id` becomes the DeerFlow run `user_id`, while the raw platform user id remains `channel_user_id`
 4. For chat: look up/create thread through Gateway's LangGraph-compatible API
-5. Feishu chat: `runs.stream()` → accumulate AI text → publish multiple outbound updates (`is_final=False`) → publish final outbound (`is_final=True`)
-6. Slack/Telegram chat: `runs.wait()` → extract final response → publish outbound
+5. Feishu/Telegram chat: `runs.stream()` → accumulate AI text → publish multiple outbound updates (`is_final=False`) → publish final outbound (`is_final=True`)
+6. Slack/Discord chat: `runs.wait()` → extract final response → publish outbound
 7. Feishu channel sends one running reply card up front, then patches the same card for each outbound update (card JSON sets `config.update_multi=true` for Feishu's patch API requirement)
-8. DingTalk AI Card mode (when `card_template_id` configured): `runs.stream()` → create card with initial text → stream updates via `PUT /v1.0/card/streaming` → finalize on `is_final=True`. Falls back to `sampleMarkdown` if card creation or streaming fails
-9. For commands (`/new`, `/status`, `/models`, `/memory`, `/help`): handle locally or query Gateway API
-10. Outbound → channel callbacks → platform reply
+8. Telegram streaming: the "Working on it..." placeholder message is registered as the stream target; non-final updates `editMessageText` it in place (channel-side throttle: 1s in private chats, 3s in groups due to Telegram's 20 msg/min group cap; 4096-char truncation; rate-limited updates dropped); the final update performs the last edit and splits >4096 texts into follow-up messages
+9. DingTalk AI Card mode (when `card_template_id` configured): `runs.stream()` → create card with initial text → stream updates via `PUT /v1.0/card/streaming` → finalize on `is_final=True`. Falls back to `sampleMarkdown` if card creation or streaming fails
+10. For commands (`/new`, `/status`, `/models`, `/memory`, `/help`): handle locally or query Gateway API
+11. Outbound → channel callbacks → platform reply

 **Configuration** (`config.yaml` -> `channels`):
 - `langgraph_url` - LangGraph-compatible Gateway API base URL (default: `http://localhost:8001/api`)
@@ -49,6 +49,11 @@ DEFAULT_RUN_CONTEXT: dict[str, Any] = {
    "subagent_enabled": False,
 }
 STREAM_UPDATE_MIN_INTERVAL_SECONDS = 0.35
+# Stream modes requested from the runtime, and the SSE event names under which
+# the message-tuple stream may arrive: the embedded runtime (and LangGraph
+# Platform) deliver the requested "messages-tuple" mode as event "messages".
+STREAM_MODES = ["messages-tuple", "values"]
+MESSAGE_STREAM_EVENTS = ("messages-tuple", "messages")
 THREAD_BUSY_MESSAGE = "This conversation is already processing another request. Please wait for it to finish and try again."

 CHANNEL_CAPABILITIES = {
@@ -56,7 +61,7 @@ CHANNEL_CAPABILITIES = {
    "discord": {"supports_streaming": False},
    "feishu": {"supports_streaming": True},
    "slack": {"supports_streaming": False},
-    "telegram": {"supports_streaming": False},
+    "telegram": {"supports_streaming": True},
    "wechat": {"supports_streaming": False},
    "wecom": {"supports_streaming": True},
 }
@@ -1135,7 +1140,7 @@ class ChannelManager:
            "input": {"messages": [human_message]},
            "config": run_config,
            "context": run_context,
-            "stream_mode": ["messages-tuple", "values"],
+            "stream_mode": list(STREAM_MODES),
            "multitask_strategy": "reject",
        }
        if owner_headers := _owner_headers(msg):
@@ -1150,7 +1155,7 @@ class ChannelManager:
                event = getattr(chunk, "event", "")
                data = getattr(chunk, "data", None)

-                if event == "messages-tuple":
+                if event in MESSAGE_STREAM_EVENTS:
                    accumulated_text, current_message_id = _accumulate_stream_text(streamed_buffers, current_message_id, data)
                    if accumulated_text:
                        latest_text = accumulated_text
@@ -5,6 +5,7 @@ from __future__ import annotations
 import asyncio
 import logging
 import threading
+import time
 from typing import Any

 from app.channels.base import Channel
@@ -13,6 +14,18 @@ from app.channels.message_bus import InboundMessage, InboundMessageType, Message

 logger = logging.getLogger(__name__)

+TELEGRAM_MAX_MESSAGE_LENGTH = 4096
+STREAM_EDIT_MIN_INTERVAL_SECONDS = 1.0
+# Groups (negative chat_id) are capped at 20 messages/minute by Telegram,
+# so stream edits there must pace well below the private-chat 1 msg/s guideline.
+STREAM_EDIT_GROUP_MIN_INTERVAL_SECONDS = 3.0
+# Bound on tracked in-flight streamed messages; entries normally clear on the
+# final update, this only guards against leaks when a final never arrives.
+MAX_TRACKED_STREAM_MESSAGES = 256
+
+# Indirection so tests can patch the clock without touching the global time module.
+_monotonic = time.monotonic
+

 class TelegramChannel(Channel):
    """Telegram bot channel using long-polling.
@@ -36,8 +49,15 @@ class TelegramChannel(Channel):
                pass
        # chat_id -> last sent message_id for threaded replies
        self._last_bot_message: dict[str, int] = {}
+        # stream_key ("chat_id:thread_ts") -> state of the in-flight streamed
+        # bot message being edited in place: {"message_id", "last_edit_at", "last_text"}
+        self._stream_messages: dict[str, dict[str, Any]] = {}
        self._connection_repo = config.get("connection_repo")

+    @property
+    def supports_streaming(self) -> bool:
+        return True
+
    async def start(self) -> None:
        if self._running:
            return
@@ -104,10 +124,117 @@ class TelegramChannel(Channel):
            logger.error("Invalid Telegram chat_id: %s", msg.chat_id)
            return

-        kwargs: dict[str, Any] = {"chat_id": chat_id, "text": msg.text}
+        key = self._stream_key(msg.chat_id, msg.thread_ts)
+
+        if not msg.is_final:
+            await self._send_stream_update(chat_id, key, msg.text, reply_to=self._parse_message_id(msg.thread_ts))
+            return
+
+        state = self._stream_messages.pop(key, None)
+        if state is not None:
+            await self._finalize_stream_message(chat_id, msg.chat_id, state, msg.text)
+            return
+
+        await self._send_new_message(chat_id, msg.chat_id, msg.text, _max_retries=_max_retries)
+
+    async def _send_stream_update(self, chat_id: int, key: str, text: str, reply_to: int | None = None) -> None:
+        """Edit the in-flight streamed message with accumulated text.
+
+        Updates are best-effort: throttled, rate-limit drops are silent.  The
+        manager always publishes a final message afterwards, which guarantees
+        delivery of the complete text.
+        """
+        if not text:
+            return
+
+        display = text
+        if len(display) > TELEGRAM_MAX_MESSAGE_LENGTH:
+            display = display[: TELEGRAM_MAX_MESSAGE_LENGTH - 1] + "…"
+
+        bot = self._application.bot
+        state = self._stream_messages.get(key)
+
+        send_kwargs: dict[str, Any] = {"chat_id": chat_id, "text": display}
+        if reply_to:
+            send_kwargs["reply_to_message_id"] = reply_to
+
+        if state is None:
+            try:
+                sent = await bot.send_message(**send_kwargs)
+            except Exception:
+                logger.exception("[Telegram] failed to start stream message in chat=%s", chat_id)
+                return
+            self._register_stream_message(key, message_id=sent.message_id, last_text=display, last_edit_at=_monotonic())
+            return
+
+        now = _monotonic()
+        min_interval = STREAM_EDIT_GROUP_MIN_INTERVAL_SECONDS if chat_id < 0 else STREAM_EDIT_MIN_INTERVAL_SECONDS
+        if now - state["last_edit_at"] < min_interval:
+            return
+        if display == state["last_text"]:
+            return
+
+        try:
+            await bot.edit_message_text(chat_id=chat_id, message_id=state["message_id"], text=display)
+        except Exception as exc:
+            if self._is_not_modified(exc):
+                state["last_text"] = display
+                return
+            if self._is_retry_after(exc):
+                logger.debug("[Telegram] stream edit rate-limited in chat=%s, dropping update", chat_id)
+                return
+            logger.warning("[Telegram] stream edit failed in chat=%s, sending new message: %s", chat_id, exc)
+            try:
+                sent = await bot.send_message(**send_kwargs)
+            except Exception:
+                logger.exception("[Telegram] failed to send fallback stream message in chat=%s", chat_id)
+                return
+            state["message_id"] = sent.message_id
+
+        state["last_edit_at"] = _monotonic()
+        state["last_text"] = display
+
+    async def _finalize_stream_message(self, chat_id: int, chat_key: str, state: dict[str, Any], text: str) -> None:
+        """Apply the final text: edit the streamed message, splitting overflow into follow-ups."""
+        bot = self._application.bot
+        chunks = self._split_message(text or "")
+
+        edited = True
+        if chunks[0] != state["last_text"]:
+            edited = await self._edit_final_chunk(bot, chat_id, state["message_id"], chunks[0])
+
+        if edited:
+            self._last_bot_message[chat_key] = state["message_id"]
+        else:
+            # Edit could not be applied (e.g. message deleted) — deliver the
+            # first chunk as a fresh message with the standard retry policy.
+            await self._send_new_message(chat_id, chat_key, chunks[0])
+
+        for chunk in chunks[1:]:
+            await self._send_new_message(chat_id, chat_key, chunk)
+
+    async def _edit_final_chunk(self, bot, chat_id: int, message_id: int, text: str) -> bool:
+        """Edit with one rate-limit retry. Returns False if the edit could not be applied."""
+        for attempt in range(2):
+            try:
+                await bot.edit_message_text(chat_id=chat_id, message_id=message_id, text=text)
+                return True
+            except Exception as exc:
+                if self._is_not_modified(exc):
+                    return True
+                if self._is_retry_after(exc) and attempt == 0:
+                    await asyncio.sleep(self._retry_after_seconds(exc))
+                    continue
+                logger.warning("[Telegram] final edit failed in chat=%s: %s", chat_id, exc)
+                return False
+        return False
+
+    async def _send_new_message(self, chat_id: int, chat_key: str, text: str, *, _max_retries: int = 3) -> int | None:
+        """Send a fresh message with retry/backoff. Returns the sent message_id."""
+        kwargs: dict[str, Any] = {"chat_id": chat_id, "text": text}

        # Reply to the last bot message in this chat for threading
-        reply_to = self._last_bot_message.get(msg.chat_id)
+        reply_to = self._last_bot_message.get(chat_key)
        if reply_to:
            kwargs["reply_to_message_id"] = reply_to

@@ -116,8 +243,8 @@ class TelegramChannel(Channel):
        for attempt in range(_max_retries):
            try:
                sent = await bot.send_message(**kwargs)
-                self._last_bot_message[msg.chat_id] = sent.message_id
-                return
+                self._last_bot_message[chat_key] = sent.message_id
+                return sent.message_id
            except Exception as exc:
                last_exc = exc
                if attempt < _max_retries - 1:
@@ -180,17 +307,63 @@ class TelegramChannel(Channel):

    # -- helpers -----------------------------------------------------------

+    @staticmethod
+    def _stream_key(chat_id: str, thread_ts: str | None) -> str:
+        return f"{chat_id}:{thread_ts or ''}"
+
+    @staticmethod
+    def _parse_message_id(value: str | None) -> int | None:
+        try:
+            return int(value) if value else None
+        except (TypeError, ValueError):
+            return None
+
+    def _register_stream_message(self, key: str, *, message_id: int, last_text: str, last_edit_at: float) -> None:
+        self._stream_messages.pop(key, None)
+        while len(self._stream_messages) >= MAX_TRACKED_STREAM_MESSAGES:
+            self._stream_messages.pop(next(iter(self._stream_messages)))
+        self._stream_messages[key] = {
+            "message_id": message_id,
+            "last_edit_at": last_edit_at,
+            "last_text": last_text,
+        }
+
+    @staticmethod
+    def _is_retry_after(exc: Exception) -> bool:
+        return getattr(exc, "retry_after", None) is not None
+
+    @staticmethod
+    def _retry_after_seconds(exc: Exception) -> float:
+        value = getattr(exc, "retry_after", 0)
+        if hasattr(value, "total_seconds"):
+            return float(value.total_seconds())
+        return float(value)
+
+    @staticmethod
+    def _is_not_modified(exc: Exception) -> bool:
+        return "message is not modified" in str(exc).lower()
+
+    @staticmethod
+    def _split_message(text: str) -> list[str]:
+        return [text[i : i + TELEGRAM_MAX_MESSAGE_LENGTH] for i in range(0, len(text), TELEGRAM_MAX_MESSAGE_LENGTH)] or [text]
+
    async def _send_running_reply(self, chat_id: str, reply_to_message_id: int) -> None:
-        """Send a 'Working on it...' reply to the user's message."""
+        """Send a 'Working on it...' reply and register it as the stream target."""
        if not self._application:
            return
        try:
            bot = self._application.bot
-            await bot.send_message(
+            sent = await bot.send_message(
                chat_id=int(chat_id),
                text="Working on it...",
                reply_to_message_id=reply_to_message_id,
            )
+            self._register_stream_message(
+                self._stream_key(chat_id, str(reply_to_message_id)),
+                message_id=sent.message_id,
+                last_text="Working on it...",
+                last_edit_at=0.0,
+            )
            logger.info("[Telegram] 'Working on it...' reply sent in chat=%s", chat_id)
        except Exception:
            logger.exception("[Telegram] failed to send running reply in chat=%s", chat_id)
@@ -873,7 +873,7 @@ class TestChannelManager:
                bus=bus,
                store=store,
                channel_sessions={
-                    "telegram": {
+                    "slack": {
                        "assistant_id": "mobile_agent",
                        "config": {"recursion_limit": 55},
                        "context": {
@@ -896,7 +896,7 @@ class TestChannelManager:

            await manager.start()

-            inbound = InboundMessage(channel_name="telegram", chat_id="chat1", user_id="user1", text="hi")
+            inbound = InboundMessage(channel_name="slack", chat_id="chat1", user_id="user1", text="hi")
            await bus.publish_inbound(inbound)
            await _wait_for(lambda: len(outbound_received) >= 1)
            await manager.stop()
@@ -1047,7 +1047,7 @@ class TestChannelManager:
                store=store,
                default_session={"context": {"is_plan_mode": True}},
                channel_sessions={
-                    "telegram": {
+                    "slack": {
                        "assistant_id": "mobile_agent",
                        "config": {"recursion_limit": 55},
                        "context": {
@@ -1080,7 +1080,7 @@ class TestChannelManager:

            await manager.start()

-            inbound = InboundMessage(channel_name="telegram", chat_id="chat1", user_id="vip-user", text="hi")
+            inbound = InboundMessage(channel_name="slack", chat_id="chat1", user_id="vip-user", text="hi")
            await bus.publish_inbound(inbound)
            await _wait_for(lambda: len(outbound_received) >= 1)
            await manager.stop()
@@ -1202,6 +1202,76 @@ class TestChannelManager:

        _run(go())

+    def test_handle_streaming_chat_accepts_runtime_messages_event(self, monkeypatch):
+        """The embedded runtime emits SSE event name "messages" (LangGraph
+        Platform semantics) for the requested "messages-tuple" stream mode —
+        the manager must accumulate text from those events too."""
+        from app.channels.manager import ChannelManager
+
+        monkeypatch.setattr("app.channels.manager.STREAM_UPDATE_MIN_INTERVAL_SECONDS", 0.0)
+
+        async def go():
+            bus = MessageBus()
+            store = ChannelStore(path=Path(tempfile.mkdtemp()) / "store.json")
+            manager = ChannelManager(bus=bus, store=store)
+
+            outbound_received = []
+
+            async def capture_outbound(msg):
+                outbound_received.append(msg)
+
+            bus.subscribe_outbound(capture_outbound)
+
+            stream_events = [
+                _make_stream_part(
+                    "messages",
+                    [
+                        {"id": "ai-1", "content": "Hello", "type": "AIMessageChunk"},
+                        {"langgraph_node": "agent"},
+                    ],
+                ),
+                _make_stream_part(
+                    "messages",
+                    [
+                        {"id": "ai-1", "content": " world", "type": "AIMessageChunk"},
+                        {"langgraph_node": "agent"},
+                    ],
+                ),
+                _make_stream_part(
+                    "values",
+                    {
+                        "messages": [
+                            {"type": "human", "content": "hi"},
+                            {"type": "ai", "content": "Hello world"},
+                        ],
+                        "artifacts": [],
+                    },
+                ),
+            ]
+
+            mock_client = _make_mock_langgraph_client()
+            mock_client.runs.stream = MagicMock(return_value=_make_async_iterator(stream_events))
+            manager._client = mock_client
+
+            await manager.start()
+
+            inbound = InboundMessage(
+                channel_name="telegram",
+                chat_id="chat1",
+                user_id="user1",
+                text="hi",
+                thread_ts="42",
+            )
+            await bus.publish_inbound(inbound)
+            await _wait_for(lambda: len(outbound_received) >= 3)
+            await manager.stop()
+
+            mock_client.runs.stream.assert_called_once()
+            assert [msg.text for msg in outbound_received] == ["Hello", "Hello world", "Hello world"]
+            assert [msg.is_final for msg in outbound_received] == [False, False, True]
+
+        _run(go())
+
    def test_handle_feishu_streaming_marks_only_final_clarification_outbound(self, monkeypatch):
        from app.channels.manager import ChannelManager

@@ -2044,7 +2114,7 @@ class TestChannelManager:
        _run(go())

    def test_none_topic_reuses_thread(self):
-        """Messages with topic_id=None should reuse the same thread (e.g. Telegram private chat)."""
+        """Messages with topic_id=None should reuse the same thread (e.g. a private/direct chat)."""
        from app.channels.manager import ChannelManager

        async def go():
@@ -2063,10 +2133,10 @@ class TestChannelManager:
            bus.subscribe_outbound(capture)
            await manager.start()

-            # Send two messages with topic_id=None (simulates Telegram private chat)
+            # Send two messages with topic_id=None (simulates a private/direct chat)
            for text in ["hello", "what did I just say?"]:
                msg = InboundMessage(
-                    channel_name="telegram",
+                    channel_name="slack",
                    chat_id="chat1",
                    user_id="user1",
                    text=text,
@@ -4766,3 +4836,439 @@ class TestSlackMarkdownConversion:
        result = _slack_md_converter.convert("# Title")
        assert "*Title*" in result
        assert "#" not in result
+
+
+# ---------------------------------------------------------------------------
+# Telegram streaming tests
+# ---------------------------------------------------------------------------
+
+
+class TestTelegramStreaming:
+    @staticmethod
+    def _make_channel_with_bot():
+        from app.channels.telegram import TelegramChannel
+
+        bus = MessageBus()
+        ch = TelegramChannel(bus=bus, config={"bot_token": "test-token"})
+
+        mock_app = MagicMock()
+        bot = SimpleNamespace()
+        bot.sent = []
+        bot.edited = []
+        bot.next_message_id = 100
+
+        async def send_message(**kwargs):
+            bot.sent.append(kwargs)
+            result = MagicMock()
+            result.message_id = bot.next_message_id
+            bot.next_message_id += 1
+            return result
+
+        async def edit_message_text(**kwargs):
+            bot.edited.append(kwargs)
+            result = MagicMock()
+            result.message_id = kwargs["message_id"]
+            return result
+
+        bot.send_message = send_message
+        bot.edit_message_text = edit_message_text
+        mock_app.bot = bot
+        ch._application = mock_app
+        return ch, bot
+
+    def test_stream_updates_edit_placeholder_in_place(self, monkeypatch):
+        async def go():
+            ch, bot = self._make_channel_with_bot()
+
+            clock = {"now": 1000.0}
+            monkeypatch.setattr("app.channels.telegram._monotonic", lambda: clock["now"])
+
+            await ch._send_running_reply("12345", 42)
+            placeholder_id = ch._stream_messages["12345:42"]["message_id"]
+
+            update1 = OutboundMessage(channel_name="telegram", chat_id="12345", thread_id="t1", text="Hello", is_final=False, thread_ts="42")
+            await ch.send(update1)
+
+            clock["now"] += 2.0
+            update2 = OutboundMessage(channel_name="telegram", chat_id="12345", thread_id="t1", text="Hello world", is_final=False, thread_ts="42")
+            await ch.send(update2)
+
+            assert len(bot.sent) == 1  # only the placeholder
+            assert [e["message_id"] for e in bot.edited] == [placeholder_id, placeholder_id]
+            assert [e["text"] for e in bot.edited] == ["Hello", "Hello world"]
+
+        _run(go())
+
+    def test_stream_updates_throttled_within_interval(self, monkeypatch):
+        async def go():
+            ch, bot = self._make_channel_with_bot()
+
+            clock = {"now": 1000.0}
+            monkeypatch.setattr("app.channels.telegram._monotonic", lambda: clock["now"])
+
+            await ch._send_running_reply("12345", 42)
+
+            await ch.send(OutboundMessage(channel_name="telegram", chat_id="12345", thread_id="t1", text="a", is_final=False, thread_ts="42"))
+            clock["now"] += 0.3  # within 1s window -> dropped
+            await ch.send(OutboundMessage(channel_name="telegram", chat_id="12345", thread_id="t1", text="ab", is_final=False, thread_ts="42"))
+            clock["now"] += 1.0  # past window -> edited
+            await ch.send(OutboundMessage(channel_name="telegram", chat_id="12345", thread_id="t1", text="abc", is_final=False, thread_ts="42"))
+
+            assert [e["text"] for e in bot.edited] == ["a", "abc"]
+
+        _run(go())
+
+    def test_stream_updates_in_group_chat_use_wider_throttle(self, monkeypatch):
+        """Telegram groups (negative chat_id) are capped at 20 messages/minute,
+        so group-chat stream edits throttle at 3s instead of 1s."""
+
+        async def go():
+            ch, bot = self._make_channel_with_bot()
+
+            clock = {"now": 1000.0}
+            monkeypatch.setattr("app.channels.telegram._monotonic", lambda: clock["now"])
+
+            await ch._send_running_reply("-100123", 42)
+
+            await ch.send(OutboundMessage(channel_name="telegram", chat_id="-100123", thread_id="t1", text="a", is_final=False, thread_ts="42"))
+            clock["now"] += 1.2  # past the 1s private window, within the 3s group window -> dropped
+            await ch.send(OutboundMessage(channel_name="telegram", chat_id="-100123", thread_id="t1", text="ab", is_final=False, thread_ts="42"))
+            clock["now"] += 2.0  # 3.2s since last edit -> edited
+            await ch.send(OutboundMessage(channel_name="telegram", chat_id="-100123", thread_id="t1", text="abc", is_final=False, thread_ts="42"))
+
+            assert [e["text"] for e in bot.edited] == ["a", "abc"]
+
+        _run(go())
+
+    def test_stream_update_without_placeholder_sends_new_message(self):
+        async def go():
+            ch, bot = self._make_channel_with_bot()
+
+            await ch.send(OutboundMessage(channel_name="telegram", chat_id="12345", thread_id="t1", text="Hi", is_final=False, thread_ts="42"))
+
+            assert len(bot.sent) == 1
+            assert bot.sent[0]["text"] == "Hi"
+            # Threads under the user's message that started this turn
+            assert bot.sent[0]["reply_to_message_id"] == 42
+            assert ch._stream_messages["12345:42"]["message_id"] == 100
+
+        _run(go())
+
+    def test_stream_edit_fallback_message_threads_under_user_message(self, monkeypatch):
+        async def go():
+            ch, bot = self._make_channel_with_bot()
+
+            clock = {"now": 1000.0}
+            monkeypatch.setattr("app.channels.telegram._monotonic", lambda: clock["now"])
+
+            await ch._send_running_reply("12345", 42)
+
+            async def edit_gone(**kwargs):
+                raise Exception("Bad Request: message to edit not found")
+
+            bot.edit_message_text = edit_gone
+            await ch.send(OutboundMessage(channel_name="telegram", chat_id="12345", thread_id="t1", text="Hi", is_final=False, thread_ts="42"))
+
+            # Fallback message threads under the user's message and becomes the new stream target
+            assert bot.sent[1]["text"] == "Hi"
+            assert bot.sent[1]["reply_to_message_id"] == 42
+            assert ch._stream_messages["12345:42"]["message_id"] == 101
+
+        _run(go())
+
+    def test_stream_message_registry_is_bounded(self):
+        from app.channels.telegram import MAX_TRACKED_STREAM_MESSAGES
+
+        async def go():
+            ch, _bot = self._make_channel_with_bot()
+
+            for i in range(MAX_TRACKED_STREAM_MESSAGES + 1):
+                ch._register_stream_message(f"chat:{i}", message_id=i, last_text="x", last_edit_at=0.0)
+
+            assert len(ch._stream_messages) == MAX_TRACKED_STREAM_MESSAGES
+            assert "chat:0" not in ch._stream_messages  # oldest evicted
+            assert f"chat:{MAX_TRACKED_STREAM_MESSAGES}" in ch._stream_messages
+
+        _run(go())
+
+    def test_stream_update_truncates_long_text(self, monkeypatch):
+        async def go():
+            ch, bot = self._make_channel_with_bot()
+
+            clock = {"now": 1000.0}
+            monkeypatch.setattr("app.channels.telegram._monotonic", lambda: clock["now"])
+
+            await ch._send_running_reply("12345", 42)
+            long_text = "x" * 5000
+            await ch.send(OutboundMessage(channel_name="telegram", chat_id="12345", thread_id="t1", text=long_text, is_final=False, thread_ts="42"))
+
+            assert len(bot.edited) == 1
+            assert len(bot.edited[0]["text"]) == 4096
+            assert bot.edited[0]["text"].endswith("…")
+
+        _run(go())
+
+    def test_stream_update_retry_after_is_dropped(self, monkeypatch):
+        async def go():
+            ch, bot = self._make_channel_with_bot()
+
+            clock = {"now": 1000.0}
+            monkeypatch.setattr("app.channels.telegram._monotonic", lambda: clock["now"])
+
+            await ch._send_running_reply("12345", 42)
+
+            async def edit_rate_limited(**kwargs):
+                exc = Exception("Flood control exceeded")
+                exc.retry_after = 5
+                raise exc
+
+            bot.edit_message_text = edit_rate_limited
+            # Must not raise, must not send a new message
+            await ch.send(OutboundMessage(channel_name="telegram", chat_id="12345", thread_id="t1", text="Hi", is_final=False, thread_ts="42"))
+            assert len(bot.sent) == 1  # placeholder only
+
+        _run(go())
+
+    def test_telegram_reports_streaming_support(self):
+        from app.channels.manager import CHANNEL_CAPABILITIES
+        from app.channels.telegram import TelegramChannel
+
+        bus = MessageBus()
+        ch = TelegramChannel(bus=bus, config={"bot_token": "test-token"})
+        assert ch.supports_streaming is True
+        assert CHANNEL_CAPABILITIES["telegram"]["supports_streaming"] is True
+
+    def test_running_reply_registers_stream_placeholder(self):
+        from app.channels.telegram import TelegramChannel
+
+        async def go():
+            bus = MessageBus()
+            ch = TelegramChannel(bus=bus, config={"bot_token": "test-token"})
+
+            mock_app = MagicMock()
+            mock_bot = AsyncMock()
+            sent = MagicMock()
+            sent.message_id = 777
+            mock_bot.send_message = AsyncMock(return_value=sent)
+            mock_app.bot = mock_bot
+            ch._application = mock_app
+
+            await ch._send_running_reply("12345", 42)
+
+            state = ch._stream_messages["12345:42"]
+            assert state["message_id"] == 777
+            assert state["last_edit_at"] == 0.0
+            assert state["last_text"] == "Working on it..."
+            mock_bot.send_message.assert_awaited_once_with(
+                chat_id=12345,
+                text="Working on it...",
+                reply_to_message_id=42,
+            )
+
+        _run(go())
+
+    def test_final_message_edits_stream_message_and_clears_state(self, monkeypatch):
+        async def go():
+            ch, bot = self._make_channel_with_bot()
+
+            clock = {"now": 1000.0}
+            monkeypatch.setattr("app.channels.telegram._monotonic", lambda: clock["now"])
+
+            await ch._send_running_reply("12345", 42)
+            placeholder_id = ch._stream_messages["12345:42"]["message_id"]
+
+            await ch.send(OutboundMessage(channel_name="telegram", chat_id="12345", thread_id="t1", text="partial", is_final=False, thread_ts="42"))
+            await ch.send(OutboundMessage(channel_name="telegram", chat_id="12345", thread_id="t1", text="full answer", is_final=True, thread_ts="42"))
+
+            assert [e["text"] for e in bot.edited] == ["partial", "full answer"]
+            assert len(bot.sent) == 1  # placeholder only — final edited, not re-sent
+            assert "12345:42" not in ch._stream_messages
+            assert ch._last_bot_message["12345"] == placeholder_id
+
+        _run(go())
+
+    def test_final_message_splits_long_text(self, monkeypatch):
+        async def go():
+            ch, bot = self._make_channel_with_bot()
+
+            clock = {"now": 1000.0}
+            monkeypatch.setattr("app.channels.telegram._monotonic", lambda: clock["now"])
+
+            await ch._send_running_reply("12345", 42)
+            long_text = "a" * 4096 + "b" * 100
+
+            await ch.send(OutboundMessage(channel_name="telegram", chat_id="12345", thread_id="t1", text=long_text, is_final=True, thread_ts="42"))
+
+            assert len(bot.edited) == 1
+            assert bot.edited[0]["text"] == "a" * 4096
+            follow_ups = bot.sent[1:]  # bot.sent[0] is the placeholder
+            assert [m["text"] for m in follow_ups] == ["b" * 100]
+            # Fake bot assigns ids sequentially: placeholder=100, follow-up chunk=101
+            assert ch._last_bot_message["12345"] == 101
+            assert "12345:42" not in ch._stream_messages
+
+        _run(go())
+
+    def test_final_message_not_modified_error_is_ignored(self, monkeypatch):
+        async def go():
+            ch, bot = self._make_channel_with_bot()
+
+            clock = {"now": 1000.0}
+            monkeypatch.setattr("app.channels.telegram._monotonic", lambda: clock["now"])
+
+            await ch._send_running_reply("12345", 42)
+            await ch.send(OutboundMessage(channel_name="telegram", chat_id="12345", thread_id="t1", text="done", is_final=False, thread_ts="42"))
+
+            async def edit_not_modified(**kwargs):
+                raise Exception("Bad Request: message is not modified")
+
+            bot.edit_message_text = edit_not_modified
+            # Same text again as final — skipped via the equal-text guard:
+            # must not raise, must not send a new message
+            await ch.send(OutboundMessage(channel_name="telegram", chat_id="12345", thread_id="t1", text="done", is_final=True, thread_ts="42"))
+
+            assert len(bot.sent) == 1  # placeholder only
+            assert "12345:42" not in ch._stream_messages
+
+        _run(go())
+
+    def test_final_edit_raising_not_modified_is_swallowed(self, monkeypatch):
+        async def go():
+            ch, bot = self._make_channel_with_bot()
+
+            clock = {"now": 1000.0}
+            monkeypatch.setattr("app.channels.telegram._monotonic", lambda: clock["now"])
+
+            await ch._send_running_reply("12345", 42)
+            placeholder_id = ch._stream_messages["12345:42"]["message_id"]
+
+            async def edit_not_modified(**kwargs):
+                raise Exception("Bad Request: message is not modified")
+
+            bot.edit_message_text = edit_not_modified
+            # Final text differs from last_text, so the edit IS attempted and
+            # raises not-modified — must be swallowed, no fallback send.
+            await ch.send(OutboundMessage(channel_name="telegram", chat_id="12345", thread_id="t1", text="done", is_final=True, thread_ts="42"))
+
+            assert len(bot.sent) == 1  # placeholder only
+            assert "12345:42" not in ch._stream_messages
+            assert ch._last_bot_message["12345"] == placeholder_id
+
+        _run(go())
+
+    def test_final_without_stream_state_sends_plain_message(self):
+        async def go():
+            ch, bot = self._make_channel_with_bot()
+
+            await ch.send(OutboundMessage(channel_name="telegram", chat_id="12345", thread_id="t1", text="direct", is_final=True, thread_ts=None))
+
+            assert len(bot.sent) == 1
+            assert bot.sent[0]["text"] == "direct"
+            assert len(bot.edited) == 0
+
+        _run(go())
+
+    def test_final_edit_retries_once_after_rate_limit(self, monkeypatch):
+        async def go():
+            ch, bot = self._make_channel_with_bot()
+
+            clock = {"now": 1000.0}
+            monkeypatch.setattr("app.channels.telegram._monotonic", lambda: clock["now"])
+
+            sleeps = []
+
+            async def fake_sleep(delay):
+                sleeps.append(delay)
+
+            monkeypatch.setattr("app.channels.telegram.asyncio.sleep", fake_sleep)
+
+            await ch._send_running_reply("12345", 42)
+            placeholder_id = ch._stream_messages["12345:42"]["message_id"]
+
+            real_edit = bot.edit_message_text
+            calls = {"n": 0}
+
+            async def edit_flaky(**kwargs):
+                calls["n"] += 1
+                if calls["n"] == 1:
+                    exc = Exception("Flood control exceeded")
+                    exc.retry_after = 3
+                    raise exc
+                return await real_edit(**kwargs)
+
+            bot.edit_message_text = edit_flaky
+            await ch.send(OutboundMessage(channel_name="telegram", chat_id="12345", thread_id="t1", text="final", is_final=True, thread_ts="42"))
+
+            assert sleeps == [3.0]
+            assert [e["text"] for e in bot.edited] == ["final"]
+            assert len(bot.sent) == 1  # placeholder only
+            assert ch._last_bot_message["12345"] == placeholder_id
+            assert "12345:42" not in ch._stream_messages
+
+        _run(go())
+
+    def test_final_edit_double_rate_limit_falls_back_to_new_message(self, monkeypatch):
+        async def go():
+            ch, bot = self._make_channel_with_bot()
+
+            clock = {"now": 1000.0}
+            monkeypatch.setattr("app.channels.telegram._monotonic", lambda: clock["now"])
+
+            sleeps = []
+
+            async def fake_sleep(delay):
+                sleeps.append(delay)
+
+            monkeypatch.setattr("app.channels.telegram.asyncio.sleep", fake_sleep)
+
+            await ch._send_running_reply("12345", 42)
+
+            async def edit_rate_limited(**kwargs):
+                exc = Exception("Flood control exceeded")
+                exc.retry_after = 2
+                raise exc
+
+            bot.edit_message_text = edit_rate_limited
+            await ch.send(OutboundMessage(channel_name="telegram", chat_id="12345", thread_id="t1", text="final", is_final=True, thread_ts="42"))
+
+            # Fallback delivered the final text as a new message (after the placeholder)
+            assert [m["text"] for m in bot.sent] == ["Working on it...", "final"]
+            assert ch._last_bot_message["12345"] == 101
+            assert "12345:42" not in ch._stream_messages
+
+        _run(go())
+
+    def test_final_overflow_chunk_send_is_retried(self, monkeypatch):
+        async def go():
+            ch, bot = self._make_channel_with_bot()
+
+            clock = {"now": 1000.0}
+            monkeypatch.setattr("app.channels.telegram._monotonic", lambda: clock["now"])
+
+            sleeps = []
+
+            async def fake_sleep(delay):
+                sleeps.append(delay)
+
+            monkeypatch.setattr("app.channels.telegram.asyncio.sleep", fake_sleep)
+
+            await ch._send_running_reply("12345", 42)
+
+            real_send = bot.send_message
+            failures = {"left": 1}
+
+            async def send_flaky(**kwargs):
+                if failures["left"] > 0:
+                    failures["left"] -= 1
+                    raise ConnectionError("transient")
+                return await real_send(**kwargs)
+
+            bot.send_message = send_flaky
+            long_text = "a" * 4096 + "b" * 10
+            await ch.send(OutboundMessage(channel_name="telegram", chat_id="12345", thread_id="t1", text=long_text, is_final=True, thread_ts="42"))
+
+            assert bot.edited[0]["text"] == "a" * 4096
+            assert [m["text"] for m in bot.sent] == ["Working on it...", "b" * 10]
+            assert ch._last_bot_message["12345"] == 101
+
+        _run(go())