diff --git a/README.md b/README.md index 8ac9aee710e..bbda0be277a 100644 --- a/README.md +++ b/README.md @@ -1042,6 +1042,7 @@ Notes: - If `assistant_id` is set to a custom agent name, DeerFlow still routes through `lead_agent` and injects that value as `agent_name`, so the custom agent's SOUL/config takes effect for IM channels. - IM channel workers call Gateway's LangGraph-compatible API internally and automatically attach process-local internal auth plus the CSRF cookie/header pair required for thread and run creation. - Inbound work is bounded to `inbound_queue_maxsize` pending messages plus `max_concurrency` active workers. When capacity is exhausted, socket/polling providers drop new messages before sending DeerFlow's working acknowledgment and emit a rate-limited warning. Buzz leaves its replay cursor unchanged and reconnects for relay replay; GitHub webhooks return `503`, marking the delivery failed for manual/API redelivery. Shutdown closes admission immediately, keeps channel transports available while accepted messages drain for up to `shutdown_grace_period_seconds`, then cancels and awaits active handlers before closing provider resources; the Gateway's outer timeout can cancel an incomplete shutdown without detaching those resources. +- Telegram shutdown retains ongoing cleanup across a timeout or cancellation so a retry can finish it. 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. - 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. diff --git a/backend/app/channels/AGENTS.md b/backend/app/channels/AGENTS.md index 27ba7e5598d..83159682a22 100644 --- a/backend/app/channels/AGENTS.md +++ b/backend/app/channels/AGENTS.md @@ -1,11 +1,15 @@ ### IM Channels System (`app/channels/`) -Bridges external messaging platforms (Feishu, Slack, Telegram, Discord, DingTalk, GitHub) to the DeerFlow agent via Gateway's LangGraph-compatible API. +IM channels reach DeerFlow through `langgraph-sdk` HTTP; Gateway owns thread lifecycle. WeChat `_read_outbound_bytes` reads cap+1 off-loop; overflow returns `None` before encryption/upload; caps <=0 are unlimited. Cursor writes drain before cancellation; token expiry drains cursor/auth clearing under `_auth_lock`. -**Architecture**: Channels use `langgraph-sdk` HTTP to reach Gateway; Gateway owns thread lifecycle. The SDK client and raw-httpx `/goal` calls send process-local auth plus `_csrf_headers()`; internal auth is not CSRF-exempt. +Telegram stop retries share live cleanup. If its loop dies, warn once and +drop only the matching future under `_shutdown_resources_lock`; client closure +remains unconfirmed. + +**Auth**: SDK and raw-httpx `/goal` calls send process-local auth plus `_csrf_headers()`; internal auth requires CSRF. **Components**: - `message_bus.py` - Inbound queue → dispatcher; outbound uses a dispatch-start listener snapshot. Unsubscribe neither cancels nor drains captured callbacks. diff --git a/backend/app/channels/telegram.py b/backend/app/channels/telegram.py index 8f0c74f710d..32df62b59b8 100644 --- a/backend/app/channels/telegram.py +++ b/backend/app/channels/telegram.py @@ -8,6 +8,7 @@ import threading import time from collections.abc import Coroutine +from concurrent.futures import Future from typing import Any from app.channels.allowed_users import parse_allowed_users @@ -149,6 +150,10 @@ def __init__(self, bus: MessageBus, config: dict[str, Any]) -> None: # its pool on the loop that performs the download. self._download_bot: Any = None self._download_bot_lock = asyncio.Lock() + # The first stop() owns this Telegram-loop cleanup. A bounded or + # cancelled caller leaves it running so a retry can join the same work. + self._shutdown_resources_future: Future[None] | None = None + self._shutdown_resources_lock = threading.Lock() # Tasks submitted from the main dispatcher loop back to PTB's loop. # Only the Telegram loop mutates this set. self._tg_bridge_tasks: set[asyncio.Task[Any]] = set() @@ -224,44 +229,66 @@ async def stop(self) -> None: deadline = shutdown_loop.time() + TELEGRAM_SHUTDOWN_TIMEOUT_SECONDS telegram_loop = self._tg_loop worker_thread = self._thread + telegram_loop_running = telegram_loop is not None and telegram_loop.is_running() + + # Bridge tasks and the download Bot's HTTPX clients may only be released + # on the Telegram loop, so that loop has to outlive the cleanup. Exactly + # one cleanup generation is scheduled per channel: a bounded or + # cancelled caller leaves it running, and a retried stop() joins the + # same work instead of scheduling a second shutdown next to it. + cleanup_future = self._shutdown_resources_future + if cleanup_future is None and telegram_loop_running: + with self._shutdown_resources_lock: + cleanup_future = self._shutdown_resources_future + if cleanup_future is None: + cleanup_future = asyncio.run_coroutine_threadsafe(self._shutdown_telegram_resources(), telegram_loop) + self._shutdown_resources_future = cleanup_future + if cleanup_future is not None and not telegram_loop_running: + # Its owner loop is gone, so nothing can finish this cleanup. + logger.warning("[Telegram] Telegram loop stopped before its resource cleanup finished; the download bot's HTTPX clients may not have been closed") + with self._shutdown_resources_lock: + if self._shutdown_resources_future is cleanup_future: + self._shutdown_resources_future = None + cleanup_future = None + if cleanup_future is not None: + try: + remaining = max(0.0, deadline - shutdown_loop.time()) + # shield(): the deadline and caller cancellation must interrupt + # only this wait, never the cleanup the Telegram loop owns. + await asyncio.wait_for(asyncio.shield(asyncio.wrap_future(cleanup_future)), timeout=remaining) + except asyncio.CancelledError: + # Retain the loop and its cleanup: ChannelService keeps this + # instance, and its next stop() converges on the same generation. + raise + except TimeoutError: + logger.warning("[Telegram] timed out shutting down Telegram loop resources during shutdown") + raise ChannelStopTimeout("Telegram loop cleanup is still running after stop timeout") from None + except Exception as exc: + logger.warning("[Telegram] failed to shut down Telegram loop resources during shutdown: %s", type(exc).__name__) - try: - if telegram_loop and telegram_loop.is_running(): - drain_future = asyncio.run_coroutine_threadsafe(self._shutdown_telegram_resources(), telegram_loop) + if telegram_loop_running: + try: + telegram_loop.call_soon_threadsafe(telegram_loop.stop) + except RuntimeError: + pass + if worker_thread is not None and worker_thread.is_alive(): + remaining = max(0.0, deadline - shutdown_loop.time()) + if remaining: try: - remaining = max(0.0, deadline - shutdown_loop.time()) - await asyncio.wait_for(asyncio.wrap_future(drain_future), timeout=remaining) - except asyncio.CancelledError: - drain_future.cancel() - raise + await asyncio.wait_for( + asyncio.to_thread(worker_thread.join, remaining), + timeout=remaining, + ) except TimeoutError: - drain_future.cancel() - logger.warning("[Telegram] timed out shutting down Telegram loop resources during shutdown") - except Exception as exc: - logger.warning("[Telegram] failed to shut down Telegram loop resources during shutdown: %s", type(exc).__name__) - finally: - if telegram_loop and telegram_loop.is_running(): - try: - telegram_loop.call_soon_threadsafe(telegram_loop.stop) - except RuntimeError: - pass - try: - if worker_thread is not None and worker_thread.is_alive(): - remaining = max(0.0, deadline - shutdown_loop.time()) - if remaining: - try: - await asyncio.wait_for( - asyncio.to_thread(worker_thread.join, remaining), - timeout=remaining, - ) - except TimeoutError: - logger.warning("[Telegram] polling thread did not stop within the shutdown budget") - finally: - if worker_thread is not None and worker_thread.is_alive(): - raise ChannelStopTimeout("Telegram polling thread is still running after stop timeout") - if self._thread is worker_thread: - self._thread = None - self._application = None + logger.warning("[Telegram] polling thread did not stop within the shutdown budget") + if worker_thread is not None and worker_thread.is_alive(): + raise ChannelStopTimeout("Telegram polling thread is still running after stop timeout") + if self._thread is worker_thread: + self._thread = None + self._application = None + with self._shutdown_resources_lock: + if self._shutdown_resources_future is cleanup_future: + self._shutdown_resources_future = None logger.info("Telegram channel stopped") async def send(self, msg: OutboundMessage, *, _max_retries: int = 3) -> None: diff --git a/backend/tests/test_channels.py b/backend/tests/test_channels.py index c460262e316..c898bf9202a 100644 --- a/backend/tests/test_channels.py +++ b/backend/tests/test_channels.py @@ -11354,7 +11354,60 @@ async def get_file(self, file_id): _run(go()) - def test_cancelled_stop_still_stops_thread_and_clears_state(self): + @pytest.mark.parametrize("owner_loop", ["missing", "stopped", "closed"]) + def test_stop_releases_abandoned_cleanup_generation(self, caplog, owner_loop): + from app.channels.telegram import TelegramChannel + + async def go(): + ch = TelegramChannel(bus=MessageBus(), config={"bot_token": "test-token"}) + telegram_loop = asyncio.new_event_loop() if owner_loop != "missing" else None + if owner_loop == "closed": + telegram_loop.close() + cleanup_future: Future[None] = Future() + ch._tg_loop = telegram_loop + ch._shutdown_resources_future = cleanup_future + ch._download_bot = SimpleNamespace(shutdown=AsyncMock()) + try: + with caplog.at_level(logging.WARNING, logger="app.channels.telegram"): + await ch.stop() + await ch.stop() + + assert ch._shutdown_resources_future is None + assert caplog.text.count("Telegram loop stopped before its resource cleanup finished") == 1 + # Dropping an abandoned generation is not evidence that its + # loop-owned clients were closed, nor a reason to cancel it. + ch._download_bot.shutdown.assert_not_awaited() + assert not cleanup_future.done() + finally: + if telegram_loop is not None and not telegram_loop.is_closed(): + telegram_loop.close() + + _run(go()) + + def test_stop_abandons_only_the_matching_cleanup_generation(self, monkeypatch): + from app.channels import telegram as telegram_module + + async def go(): + ch = telegram_module.TelegramChannel(bus=MessageBus(), config={"bot_token": "test-token"}) + abandoned: Future[None] = Future() + replacement: Future[None] = Future() + ch._shutdown_resources_future = abandoned + + def replace_generation_before_release(*args, **kwargs): + with ch._shutdown_resources_lock: + ch._shutdown_resources_future = replacement + + monkeypatch.setattr(telegram_module.logger, "warning", replace_generation_before_release) + await ch.stop() + + assert ch._shutdown_resources_future is replacement + assert not abandoned.done() + assert not replacement.done() + + _run(go()) + + def test_cancelled_stop_keeps_owned_cleanup_and_retry_converges(self): + """A cancelled stop must not cancel the cleanup the Telegram loop owns.""" from app.channels.telegram import TelegramChannel async def go(): @@ -11362,16 +11415,23 @@ async def go(): ch = TelegramChannel(bus=bus, config={"bot_token": "test-token"}) telegram_loop = asyncio.new_event_loop() loop_started: Future[None] = Future() - drain_started = threading.Event() + cleanup_started = threading.Event() + cleanup_finished = threading.Event() + release_cleanup = threading.Event() + shutdown_calls = 0 def run_telegram_loop(): asyncio.set_event_loop(telegram_loop) telegram_loop.call_soon(loop_started.set_result, None) telegram_loop.run_forever() - async def slow_drain(): - drain_started.set() - await asyncio.Event().wait() + class DownloadBot: + async def shutdown(self): + nonlocal shutdown_calls + shutdown_calls += 1 + cleanup_started.set() + await asyncio.to_thread(release_cleanup.wait) + cleanup_finished.set() loop_thread = threading.Thread(target=run_telegram_loop, daemon=True) loop_thread.start() @@ -11381,18 +11441,173 @@ async def slow_drain(): ch._thread = loop_thread ch._running = True ch._application = SimpleNamespace(bot=SimpleNamespace()) - ch._cancel_telegram_bridge_tasks = slow_drain + ch._download_bot = DownloadBot() stop_task = asyncio.create_task(ch.stop()) - assert await asyncio.to_thread(drain_started.wait, 2) + assert await asyncio.to_thread(cleanup_started.wait, 2) stop_task.cancel() with pytest.raises(asyncio.CancelledError): await stop_task + # Cancellation reached the caller only: the loop and the cleanup + # it owns stay with the channel so a retry can finish them. + assert not cleanup_finished.is_set() + assert telegram_loop.is_running() + assert loop_thread.is_alive() + assert ch._thread is loop_thread + assert ch._application is not None + assert ch._download_bot is not None + + release_cleanup.set() + assert await asyncio.to_thread(cleanup_finished.wait, 2) + await ch.stop() + + assert shutdown_calls == 1 + assert ch._download_bot is None + assert ch._thread is None + assert ch._application is None + assert not loop_thread.is_alive() + finally: + if telegram_loop.is_running(): + telegram_loop.call_soon_threadsafe(telegram_loop.stop) + await asyncio.to_thread(loop_thread.join, 2) + if loop_thread.is_alive(): + pytest.fail("Telegram test event loop did not stop") + telegram_loop.close() + + _run(go()) + + def test_concurrent_stops_share_one_telegram_cleanup_generation(self): + """Concurrent stops must not schedule a second Telegram-loop cleanup.""" + from app.channels.telegram import TelegramChannel + + async def go(): + bus = MessageBus() + ch = TelegramChannel(bus=bus, config={"bot_token": "test-token"}) + telegram_loop = asyncio.new_event_loop() + loop_started: Future[None] = Future() + cleanup_started = threading.Event() + cleanup_finished = threading.Event() + release_cleanup = threading.Event() + shutdown_calls = 0 + + def run_telegram_loop(): + asyncio.set_event_loop(telegram_loop) + telegram_loop.call_soon(loop_started.set_result, None) + telegram_loop.run_forever() + + class DownloadBot: + async def shutdown(self): + nonlocal shutdown_calls + shutdown_calls += 1 + cleanup_started.set() + await asyncio.to_thread(release_cleanup.wait) + cleanup_finished.set() + + loop_thread = threading.Thread(target=run_telegram_loop, daemon=True) + loop_thread.start() + try: + loop_started.result(timeout=2) + ch._tg_loop = telegram_loop + ch._thread = loop_thread + ch._running = True + ch._application = SimpleNamespace(bot=SimpleNamespace()) + ch._download_bot = DownloadBot() + + first = asyncio.create_task(ch.stop()) + assert await asyncio.to_thread(cleanup_started.wait, 2) + second = asyncio.create_task(ch.stop()) + await asyncio.sleep(0.05) + + assert not first.done() + assert not second.done() + + release_cleanup.set() + await asyncio.gather(first, second) + + assert shutdown_calls == 1 + assert await asyncio.to_thread(cleanup_finished.wait, 2) + assert ch._download_bot is None + assert ch._thread is None + assert ch._application is None assert not loop_thread.is_alive() + finally: + if telegram_loop.is_running(): + telegram_loop.call_soon_threadsafe(telegram_loop.stop) + await asyncio.to_thread(loop_thread.join, 2) + if loop_thread.is_alive(): + pytest.fail("Telegram test event loop did not stop") + telegram_loop.close() + + _run(go()) + + def test_shutdown_timeout_keeps_download_bot_cleanup_running(self, monkeypatch): + """The Gateway deadline must not cancel Telegram-loop-owned Bot cleanup.""" + from app.channels import telegram as telegram_module + from app.channels.base import ChannelStopTimeout + + async def go(): + bus = MessageBus() + ch = telegram_module.TelegramChannel(bus=bus, config={"bot_token": "test-token"}) + telegram_loop = asyncio.new_event_loop() + loop_started: Future[None] = Future() + cleanup_started = threading.Event() + cleanup_finished = threading.Event() + release_cleanup = threading.Event() + shutdown_calls = 0 + + def run_telegram_loop(): + asyncio.set_event_loop(telegram_loop) + telegram_loop.call_soon(loop_started.set_result, None) + telegram_loop.run_forever() + + class DownloadBot: + async def shutdown(self): + nonlocal shutdown_calls + shutdown_calls += 1 + cleanup_started.set() + await asyncio.to_thread(release_cleanup.wait) + cleanup_finished.set() + + loop_thread = threading.Thread(target=run_telegram_loop, daemon=True) + loop_thread.start() + try: + loop_started.result(timeout=2) + ch._tg_loop = telegram_loop + ch._thread = loop_thread + ch._running = True + ch._application = SimpleNamespace(bot=SimpleNamespace()) + ch._download_bot = DownloadBot() + monkeypatch.setattr(telegram_module, "TELEGRAM_SHUTDOWN_TIMEOUT_SECONDS", 0.01) + + with pytest.raises(ChannelStopTimeout, match="Telegram loop cleanup is still running"): + await ch.stop() + + # The deadline interrupted the wait, not the cleanup: the loop + # still owns the unfinished download Bot shutdown. + assert await asyncio.to_thread(cleanup_started.wait, 2) + assert not cleanup_finished.is_set() + assert telegram_loop.is_running() + assert loop_thread.is_alive() + assert ch._thread is loop_thread + assert ch._application is not None + assert ch._download_bot is not None + + release_cleanup.set() + assert await asyncio.to_thread(cleanup_finished.wait, 2) + + # A retried stop joins that same cleanup and only then tears the + # channel down. + monkeypatch.setattr(telegram_module, "TELEGRAM_SHUTDOWN_TIMEOUT_SECONDS", 5.0) + await ch.stop() + + assert shutdown_calls == 1 + assert ch._download_bot is None assert ch._thread is None assert ch._application is None + assert not telegram_loop.is_running() + assert not loop_thread.is_alive() finally: if telegram_loop.is_running(): telegram_loop.call_soon_threadsafe(telegram_loop.stop)