fix(telegram): preserve group reply conversation context (#7654)

* fix(telegram): preserve group reply conversation context

* fix(telegram): log mapping outages once per streamed answer

---------

Co-authored-by: haotianqi666 <250384991+haotianqi666@users.noreply.github.com>
Co-authored-by: Willem Jiang <willem.jiang@gmail.com>

README.md

@@ -1345,6 +1345,7 @@ Notes:
 - Telegram shutdown retains ongoing cleanup across a timeout or cancellation so a retry can finish it. Stopping during startup also cancels initialization and closes the partially initialized bot's HTTP clients before retiring its worker loop. If the Telegram loop has already stopped, shutdown warns that download clients may remain open and releases the abandoned cleanup state; later stops do not repeat that warning.
 - Slack stop waits up to five seconds for an in-flight Socket Mode connection and its cleanup. If cleanup is still pending, the previous instance stays tracked and a restart can be retried after it finishes. SDK cleanup failures report their original error and remain retryable.
 - In Telegram private chats, text, attachment and streaming replies (including "Working on it...") can still be delivered if their referenced user or bot message was deleted, such as after clearing chat history during a run. Group reply-target requirements remain unchanged to preserve forum routing.
+- In Telegram groups, replying to a delivered bot response or attachment continues its DeerFlow conversation. Reply mappings survive restarts and stay scoped to the user's connection when one is bound. New, unrelated group messages still start separate conversations; private-chat routing is unchanged.
 - Feishu/Lark now queues rapid follow-up messages per mapped DeerFlow `thread_id` instead of immediately surfacing the generic busy reply, and topic replies keep a per-message card with a compact source-message preview across queued/running/final patches.
 - Streaming IM channels treat backend `error` events like transport failures and follow the same reply and retry path. The channel logs the error type and message for diagnosis.
 

backend/app/channels/AGENTS.md

@@ -47,7 +47,7 @@ Shared attachment feedback uses public `attachment_delivery_text`, preserves rou
 6. Slack/Discord chat: `runs.wait()` → extract final response → publish outbound
 6b. GitHub chat (`ChannelRunPolicy.fire_and_forget=True`): `runs.create()` returns once the run is `pending`; the manager does not wait for the final state and does not publish an outbound; the agent posts its own reply via sandbox `gh`. `ConflictError` on a busy thread still trips the standard `THREAD_BUSY_MESSAGE` path (log-only on GitHub); with `buffer_followups_on_busy=True` (GitHub's default) the triggering message is captured into a per-thread buffer so a concurrent comment is not silently dropped (see "Follow-up buffering while busy").
 7. Feishu patches one card per source (`config.update_multi=true`), with topic previews and queued → running → final states. Opt-in cards show context/question; bounded tokens preserve request ID/source. Originator/option checks consume once; consumption fails closed across workers (`feishu_card_interaction.py`). Card replies keep bound aliases in fields/metadata; ordinary replies preserve `/new`. Accepted intake retires only that card's buttons/text hint before terminal PATCH; rejected intake keeps text fallback. Non-final chunks store each thread alias once per reply; the final rewrites all.
-8. Telegram edits the "Working on it..." stream target: 1s private/3s group throttle (20 msg/min), 4096-char truncation, rate-limited updates dropped; final edits split overflow. Text/preview/photo/document reply sends use `allow_sending_without_reply` only for `chat_id > 0`: missing private targets degrade atomically; group/forum routing and transport retries stay unchanged.
+8. Telegram previews: 1s private/3s group, 20/min, 4096 chars; drop rate-limited updates, split final overflow. Only private sends (`chat_id > 0`) use `allow_sending_without_reply`; degrade missing targets atomically. Preserve group/forum routing/retries and durable owner aliases. Stream mapping errors log one traceback/answer, then debug; delivery/retries continue.
 9. DingTalk AI Cards (`card_template_id`): `runs.stream()` → create card with initial text → `PUT /v1.0/card/streaming` → finalize (`is_final=True`). Creation/streaming failure uses `sampleMarkdown`; tables preserve escaped pipes and empty edge cells. Card calls get the channel token, not the SDK's.
 10. For commands (`/new`, `/status`, `/models`, `/model`, `/memory`, `/goal`, `/agent`, `/help`): handle locally or query Gateway API. `/agent list` reads the effective owner's Custom Agents; `/agent use <name>` validates in the same owner bucket, creates a new thread, and persists `channel_agent_name` in its metadata. A Custom Agent writes canonical `metadata.agent_name` so thread-search routes Web continuation to `/workspace/agents/<name>/chats/<thread_id>`; `lead_agent` omits it and stays on the ordinary chat route. 
 `/model <name>` persists `channel_model_name` in the *current* thread's metadata (in place, no fresh thread); `/model` reports the effective model, `/model default` clears it (stale pins self-clear).

backend/app/channels/telegram.py

@@ -323,14 +323,20 @@ class TelegramChannel(Channel):
 
         if not msg.is_final:
             await self._send_stream_update(chat_id, key, msg.text, reply_to=self._parse_message_id(msg.thread_ts))
+            state = self._stream_messages.get(key)
+            if state is not None:
+                binding = (msg.thread_id, msg.connection_id, msg.owner_user_id, msg.thread_ts, state["message_id"])
+                if state.get("thread_binding") != binding and await self._remember_reply_thread(msg, state["message_id"], stream_state=state):
+                    state["thread_binding"] = binding
             return
 
         state = self._stream_messages.pop(key, None)
         if state is not None:
+            await self._remember_reply_thread(msg, state["message_id"], stream_state=state)
             if self._can_send_rich(msg.text) and await self._edit_rich_message(chat_id, state["message_id"], msg.text):
                 self._last_bot_message[msg.chat_id] = state["message_id"]
                 return
-            await self._finalize_stream_message(chat_id, msg.chat_id, state, msg.text)
+            await self._finalize_stream_message(chat_id, msg.chat_id, state, msg.text, msg)
             return
 
         if self._can_send_rich(msg.text):
@@ -340,10 +346,51 @@ class TelegramChannel(Channel):
                 logger.warning("[Telegram] Rich Message send failed in chat=%s; falling back to plain text: %s", chat_id, exc)
                 message_id = None
             if message_id is not None:
+                await self._remember_reply_thread(msg, message_id)
                 return
 
         for chunk in self._split_message(msg.text):
-            await self._send_new_message(chat_id, msg.chat_id, chunk, _max_retries=_max_retries)
+            message_id = await self._send_new_message(chat_id, msg.chat_id, chunk, _max_retries=_max_retries)
+            await self._remember_reply_thread(msg, message_id)
+
+    async def _remember_reply_thread(self, msg: OutboundMessage, message_id: int | None, *, stream_state: dict[str, Any] | None = None) -> bool:
+        """Keep group replies on their source thread, without undoing a later /new."""
+        if not msg.thread_id or message_id is None or int(msg.chat_id) >= 0:
+            return True
+        store = self.config.get("channel_store")
+        if msg.connection_id:
+            # Bound conversations must never fall back to the shared legacy store.
+            if self._connection_repo is None or not msg.owner_user_id:
+                return False
+        elif store is None:
+            return False
+        topic_ids = {str(message_id)}
+        if msg.thread_ts:
+            topic_ids.add(msg.thread_ts)
+        try:
+            for topic_id in topic_ids:
+                if msg.connection_id:
+                    await self._connection_repo.claim_thread_id(
+                        connection_id=msg.connection_id,
+                        owner_user_id=msg.owner_user_id,
+                        provider=self.name,
+                        external_conversation_id=msg.chat_id,
+                        external_topic_id=topic_id,
+                        thread_id=msg.thread_id,
+                    )
+                else:
+                    await store.claim_thread_id(self.name, msg.chat_id, msg.thread_id, topic_id=topic_id)
+        except Exception:
+            # Delivery has already succeeded. Retry the binding on the next update,
+            # not the Telegram send, which could duplicate a visible reply.
+            if stream_state is not None and stream_state.get("thread_binding_error_logged"):
+                logger.debug("[Telegram] reply thread mapping retry failed in chat=%s", msg.chat_id)
+            else:
+                logger.exception("[Telegram] failed to remember reply thread mapping in chat=%s", msg.chat_id)
+                if stream_state is not None:
+                    stream_state["thread_binding_error_logged"] = True
+            return False
+        return True
 
     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.
@@ -409,7 +456,7 @@ class TelegramChannel(Channel):
         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:
+    async def _finalize_stream_message(self, chat_id: int, chat_key: str, state: dict[str, Any], text: str, msg: OutboundMessage) -> None:
         """Apply the final text: edit the streamed message, splitting overflow into follow-ups."""
         bot = self._application.bot
         chunks = self._split_message(text or "")
@@ -423,10 +470,12 @@ class TelegramChannel(Channel):
         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])
+            message_id = await self._send_new_message(chat_id, chat_key, chunks[0])
+            await self._remember_reply_thread(msg, message_id, stream_state=state)
 
         for chunk in chunks[1:]:
-            await self._send_new_message(chat_id, chat_key, chunk)
+            message_id = await self._send_new_message(chat_id, chat_key, chunk)
+            await self._remember_reply_thread(msg, message_id, stream_state=state)
 
     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."""
@@ -561,6 +610,7 @@ class TelegramChannel(Channel):
                 sent = await bot.send_document(**kwargs)
 
             self._last_bot_message[msg.chat_id] = sent.message_id
+            await self._remember_reply_thread(msg, sent.message_id)
             logger.info("[Telegram] file sent: %s to chat=%s", attachment.filename, msg.chat_id)
             return True
         except Exception:

backend/tests/test_telegram_reply_threads.py

@@ -0,0 +1,256 @@
+"""Telegram reply routing through real stores, with an offline Bot API double."""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+from itertools import count
+from types import SimpleNamespace
+from unittest.mock import AsyncMock
+
+import pytest
+
+from app.channels.manager import ChannelManager
+from app.channels.message_bus import MessageBus, OutboundMessage, ResolvedAttachment
+from app.channels.store import JsonChannelStore, SqlChannelStore
+from app.channels.telegram import TelegramChannel
+
+pytestmark = pytest.mark.anyio
+
+
+@pytest.fixture(params=["json", "sqlite", "bound"])
+async def routing(request, tmp_path):
+    repo = None
+    connection = None
+    session_factory = None
+    if request.param != "json":
+        from deerflow.persistence.channel_connections import ChannelConnectionRepository, ChannelCredentialCipher
+        from deerflow.persistence.engine import close_engine, get_session_factory, init_engine
+
+        await init_engine("sqlite", url=f"sqlite+aiosqlite:///{tmp_path / 'channel.db'}", sqlite_dir=str(tmp_path))
+        session_factory = get_session_factory()
+        if request.param == "bound":
+            repo = ChannelConnectionRepository(session_factory, cipher=ChannelCredentialCipher.from_key("test-key"))
+            connection = await repo.upsert_connection(
+                owner_user_id="owner-a",
+                provider="telegram",
+                external_account_id="42",
+                workspace_id="-10042",
+            )
+    store = SqlChannelStore(session_factory) if session_factory else JsonChannelStore(tmp_path / "store.json")
+    bus = MessageBus()
+    channel = TelegramChannel(bus=bus, config={"bot_token": "test-token", "channel_store": store, "connection_repo": repo})
+    channel._main_loop = asyncio.get_running_loop()
+    message_ids = count(100)
+    sent_ids = []
+
+    async def send(**kwargs):
+        message_id = next(message_ids)
+        sent_ids.append(message_id)
+        return SimpleNamespace(message_id=message_id)
+
+    async def rich(endpoint, api_kwargs):
+        if endpoint == "editMessageText":
+            return {"message_id": api_kwargs["message_id"]}
+        return {"message_id": (await send()).message_id}
+
+    bot = SimpleNamespace(send_message=AsyncMock(side_effect=send), edit_message_text=AsyncMock(), do_api_request=AsyncMock(side_effect=rich), send_photo=AsyncMock(side_effect=send), send_document=AsyncMock(side_effect=send))
+    channel._application = SimpleNamespace(bot=bot)
+    thread_ids = count(1)
+    client = SimpleNamespace(threads=SimpleNamespace(create=AsyncMock(side_effect=lambda **kw: {"thread_id": f"thread-{next(thread_ids)}"})))
+    manager = ChannelManager(bus=bus, store=store, connection_repo=repo)
+
+    async def incoming(message_id, reply_id=None, *, chat_type="supergroup", user_id=42):
+        chat_id = 42 if chat_type == "private" else -10042
+        update = SimpleNamespace(
+            effective_chat=SimpleNamespace(id=chat_id, type=chat_type),
+            effective_user=SimpleNamespace(id=user_id),
+            message=SimpleNamespace(text="question", caption=None, photo=[], document=None, message_id=message_id, reply_to_message=SimpleNamespace(message_id=reply_id) if reply_id else None),
+        )
+        await channel._on_text(update, None)
+        return await asyncio.wait_for(bus.get_inbound(), timeout=5)
+
+    def outbound(inbound, thread_id, **kwargs):
+        return OutboundMessage(
+            channel_name="telegram",
+            chat_id=inbound.chat_id,
+            thread_id=thread_id,
+            thread_ts=inbound.thread_ts,
+            connection_id=inbound.connection_id,
+            owner_user_id=inbound.owner_user_id,
+            metadata=inbound.metadata,
+            text=kwargs.pop("text", "Answer"),
+            **kwargs,
+        )
+
+    harness = SimpleNamespace(channel=channel, bot=bot, bus=bus, store=store, repo=repo, connection=connection, manager=manager, client=client, incoming=incoming, outbound=outbound, sent_ids=sent_ids)
+    try:
+        yield harness
+    finally:
+        await channel._close_and_drain_threadsafe_futures()
+        if session_factory:
+            await close_engine()
+
+
+@pytest.mark.parametrize("chat_type", ["private", "group", "supergroup"])
+@pytest.mark.parametrize("reply_to_bot", [False, True])
+async def test_reply_reuses_conversation(routing, chat_type, reply_to_bot):
+    h = routing
+    first = await h.incoming(10, chat_type=chat_type)
+    thread_id, _ = await h.manager._get_or_create_thread(h.client, first)
+    await h.channel.send(h.outbound(first, thread_id))
+    response_id = h.bot.edit_message_text.await_args.kwargs["message_id"]
+    followup = await h.incoming(11, response_id if reply_to_bot else 10, chat_type=chat_type)
+    assert await h.manager._get_or_create_thread(h.client, followup) == (thread_id, False)
+    h.client.threads.create.assert_awaited_once()
+
+
+async def test_stream_aliases_are_not_rewritten_and_new_keeps_its_binding(routing):
+    h = routing
+    first = await h.incoming(10)
+    thread_id, _ = await h.manager._get_or_create_thread(h.client, first)
+    repository = h.repo or h.store
+    claim = repository.claim_thread_id
+    repository.claim_thread_id = AsyncMock(wraps=claim)
+    await h.channel.send(h.outbound(first, thread_id, text="Partial", is_final=False))
+    calls = repository.claim_thread_id.await_count
+    assert calls == 2  # Source message and bot preview.
+    await h.channel.send(h.outbound(first, thread_id, text="Partial", is_final=False))
+    assert repository.claim_thread_id.await_count == calls
+
+    command = await h.incoming(11, h.sent_ids[0])
+    new_thread = await h.manager._create_thread(h.client, command)
+    assert new_thread != thread_id
+    await h.channel.send(h.outbound(first, thread_id, text="Old run finished"))
+    assert await h.manager._lookup_thread_id(command) == new_thread
+
+
+@pytest.mark.parametrize("delivery", ["plain", "rich", "split", "fallback"])
+async def test_stream_binding_outage_logs_once_per_answer(routing, caplog, delivery):
+    h = routing
+    first = await h.incoming(10)
+    thread_id, _ = await h.manager._get_or_create_thread(h.client, first)
+    repository = h.repo or h.store
+    repository.claim_thread_id = AsyncMock(side_effect=RuntimeError("store unavailable"))
+
+    with caplog.at_level(logging.DEBUG, logger="app.channels.telegram"):
+        for index in range(20):
+            await h.channel.send(h.outbound(first, thread_id, text=f"Partial {index}", is_final=False))
+        if delivery == "rich":
+            h.channel.config["rich_messages"] = True
+        if delivery == "fallback":
+            h.bot.edit_message_text.side_effect = RuntimeError("edit unavailable")
+        text = "x" * 9000 if delivery == "split" else "Final answer"
+        await h.channel.send(h.outbound(first, thread_id, text=text))
+
+        failures = [record for record in caplog.records if record.name == "app.channels.telegram" and "thread mapping" in record.getMessage()]
+        assert len(failures) == repository.claim_thread_id.await_count
+        assert len(failures) >= 21, "every failed update must keep retrying the mapping"
+        assert sum(record.levelno == logging.ERROR for record in failures) == 1
+        assert failures[0].exc_info is not None
+        assert all(record.levelno == logging.DEBUG and record.exc_info is None for record in failures[1:])
+        assert len(h.sent_ids) == {"plain": 1, "rich": 1, "split": 3, "fallback": 2}[delivery]
+
+        # Finalization retires the per-answer log state. A subsequent answer
+        # must report its first outage even on the same conversation key.
+        await h.channel.send(h.outbound(first, thread_id, text="Next answer", is_final=False))
+        errors = [record for record in caplog.records if record.name == "app.channels.telegram" and "thread mapping" in record.getMessage() and record.levelno == logging.ERROR]
+        assert len(errors) == 2
+        await h.channel.send(h.outbound(first, thread_id, text="Next answer"))
+
+
+async def test_binding_failure_does_not_fail_delivery_and_stream_retries(routing, caplog):
+    h = routing
+    first = await h.incoming(10)
+    thread_id, _ = await h.manager._get_or_create_thread(h.client, first)
+    repository = h.repo or h.store
+    claim = repository.claim_thread_id
+    repository.claim_thread_id = AsyncMock(side_effect=RuntimeError("store unavailable"))
+    await h.channel.send(h.outbound(first, thread_id, is_final=False))
+    assert h.bot.edit_message_text.await_count == 1
+    assert "thread mapping" in caplog.text
+    if h.repo:
+        assert await h.store.list_entries() == []
+    repository.claim_thread_id = claim
+    await h.channel.send(h.outbound(first, thread_id, is_final=False))
+    followup = await h.incoming(11, h.sent_ids[0])
+    assert await h.manager._get_or_create_thread(h.client, followup) == (thread_id, False)
+
+
+@pytest.mark.parametrize("routing", ["bound"], indirect=True)
+async def test_bound_reply_does_not_expose_thread_to_another_owner(routing):
+    h = routing
+    first = await h.incoming(10)
+    thread_id, _ = await h.manager._get_or_create_thread(h.client, first)
+    await h.channel.send(h.outbound(first, thread_id))
+    response_id = h.sent_ids[0]
+    assert await h.store.list_entries() == []
+    await h.repo.upsert_connection(owner_user_id="owner-b", provider="telegram", external_account_id="43", workspace_id="-10042")
+    other = await h.incoming(11, response_id, user_id=43)
+    assert other.owner_user_id == "owner-b"
+    assert await h.manager._lookup_thread_id(other) is None
+    unbound = await h.incoming(12, response_id, user_id=44)
+    assert unbound.connection_id is None
+    assert await h.manager._lookup_thread_id(unbound) is None
+    h.client.threads.create.assert_awaited_once()
+
+
+@pytest.mark.parametrize("delivery", ["split", "fallback", "rich", "rich-new", "rich-fallback", "plain-new", "preview", "preview-fallback"])
+async def test_every_delivered_reply_keeps_thread(routing, delivery):
+    h = routing
+    first = await h.incoming(10)
+    thread_id, _ = await h.manager._get_or_create_thread(h.client, first)
+    placeholder_id = h.sent_ids[0]
+    if delivery.endswith("new"):
+        h.channel._stream_messages.clear()
+    if delivery == "fallback":
+        h.bot.edit_message_text.side_effect = RuntimeError("edit unavailable")
+    if delivery.startswith("rich"):
+        h.channel.config["rich_messages"] = True
+    if delivery == "rich-fallback":
+        from telegram.error import BadRequest
+
+        h.bot.do_api_request.side_effect = BadRequest("unsupported")
+    text = "x" * 9000 if delivery == "split" else "**Answer**"
+    await h.channel.send(h.outbound(first, thread_id, text=text, is_final=not delivery.startswith("preview")))
+    if delivery == "preview-fallback":
+        h.channel._stream_messages[f"{first.chat_id}:{first.thread_ts}"]["last_edit_at"] = 0.0
+        h.bot.edit_message_text.side_effect = RuntimeError("edit unavailable")
+        await h.channel.send(h.outbound(first, thread_id, text="Next chunk", is_final=False))
+    reply_ids = h.sent_ids[1:] if delivery.endswith("new") or delivery == "fallback" else list(h.sent_ids)
+    assert reply_ids
+    for reply_id in reply_ids:
+        followup = await h.incoming(200 + reply_id, reply_id)
+        assert await h.manager._get_or_create_thread(h.client, followup) == (thread_id, False)
+    if delivery == "split":
+        assert placeholder_id in reply_ids and len(reply_ids) == 3
+
+
+async def test_followup_source_alias_and_restart(routing):
+    h = routing
+    first = await h.incoming(10)
+    thread_id, _ = await h.manager._get_or_create_thread(h.client, first)
+    await h.channel.send(h.outbound(first, thread_id))
+    followup = await h.incoming(11, h.sent_ids[0])
+    assert await h.manager._get_or_create_thread(h.client, followup) == (thread_id, False)
+    await h.channel.send(h.outbound(followup, thread_id))
+    store = JsonChannelStore(h.store.path) if isinstance(h.store, JsonChannelStore) else h.store
+    restarted = ChannelManager(bus=h.bus, store=store, connection_repo=h.repo)
+    for reply_id in (11, h.sent_ids[-1]):
+        incoming = await h.incoming(300 + reply_id, reply_id)
+        assert await restarted._get_or_create_thread(h.client, incoming) == (thread_id, False)
+    unrelated = await h.incoming(900)
+    assert (await restarted._get_or_create_thread(h.client, unrelated))[0] != thread_id
+
+
+@pytest.mark.parametrize("image", [False, True])
+async def test_reply_to_attachment_keeps_thread(routing, tmp_path, image):
+    h = routing
+    first = await h.incoming(10)
+    thread_id, _ = await h.manager._get_or_create_thread(h.client, first)
+    path = tmp_path / ("chart.png" if image else "report.txt")
+    path.write_bytes(b"fixture")
+    attachment = ResolvedAttachment(virtual_path=f"/mnt/user-data/outputs/{path.name}", actual_path=path, filename=path.name, size=7, mime_type="image/png" if image else "text/plain", is_image=image)
+    assert await h.channel.send_file(h.outbound(first, thread_id), attachment)
+    followup = await h.incoming(11, h.sent_ids[-1])
+    assert await h.manager._get_or_create_thread(h.client, followup) == (thread_id, False)