fix(telegram): retain polling worker on stop timeout (#6333)

* fix(telegram): retain polling ownership on stop timeout

* test(telegram): cover stop timeout ownership retry

* docs(channels): pin Telegram stop ownership

backend/app/channels/telegram.py

@@ -10,7 +10,7 @@ import time
 from collections.abc import Coroutine
 from typing import Any
 
-from app.channels.base import Channel
+from app.channels.base import Channel, ChannelStopTimeout
 from app.channels.connection_identity import attach_connection_identity
 from app.channels.message_bus import (
     INBOUND_FILE_CONTENT_KEY,
@@ -269,8 +269,9 @@ class TelegramChannel(Channel):
                             logger.warning("[Telegram] polling thread did not stop within the shutdown budget")
             finally:
                 if worker_thread is not None and worker_thread.is_alive():
-                    logger.warning("[Telegram] polling thread is still exiting after bounded shutdown")
-                self._thread = None
+                    raise ChannelStopTimeout("Telegram polling thread is still running after stop timeout")
+                if self._thread is worker_thread:
+                    self._thread = None
                 self._application = None
         logger.info("Telegram channel stopped")
 

backend/tests/test_telegram_channel_connections.py

@@ -9,6 +9,7 @@ from unittest.mock import AsyncMock, MagicMock
 
 import pytest
 
+from app.channels.base import ChannelStopTimeout
 from app.channels.message_bus import MessageBus
 from app.channels.telegram import TelegramChannel
 
@@ -220,3 +221,40 @@ async def test_bind_on_main_replies_via_telegram_loop(repo):
     assert "connected" in update.message.reply_text.call_args.args[0].lower()
     connections = await repo.list_connections("deerflow-user-1")
     assert len(connections) == 1
+
+
+@pytest.mark.anyio
+async def test_stop_retains_polling_worker_ownership_until_thread_exits():
+    """A bounded Telegram stop must retain a still-live polling worker for retry."""
+
+    class PollingThread:
+        def __init__(self):
+            self.alive = True
+            self.join_calls = 0
+
+        def is_alive(self):
+            return self.alive
+
+        def join(self, timeout=None):
+            self.join_calls += 1
+
+    channel = TelegramChannel(bus=MessageBus(), config={"bot_token": "test-token"})
+    worker = PollingThread()
+    application = object()
+    channel._thread = worker
+    channel._application = application
+    channel._tg_loop = None
+    channel._running = True
+
+    with pytest.raises(ChannelStopTimeout, match="Telegram polling thread is still running"):
+        await channel.stop()
+
+    assert channel._thread is worker
+    assert channel._application is application
+    assert worker.join_calls == 1
+
+    worker.alive = False
+    await channel.stop()
+
+    assert channel._thread is None
+    assert channel._application is None