fix(channels): keep blocking SDK teardown off the event loop in stop() (#6134)

* fix(channels): keep blocking SDK teardown off the event loop in stop()

SlackChannel.stop() called SocketModeClient.close() inline. close() joins
the SDK's message-processor thread (~0.7s every call, measured) and waits
for in-flight listeners, whose blocking Web API calls can run up to the
WebClient timeout, so every run, stream and channel on the Gateway stalled
during shutdown and POST /api/channels/{name}/restart.

Feishu and DingTalk had the same shape: they joined their SDK threads
inline with a 5s timeout, and since those threads only return on a fatal
error, the join normally froze the loop for the full 5s.

The teardown now runs through asyncio.to_thread. Slack detaches the client
first so a still-queued connect skips it, and shields and tracks the close
task so a cancelled stop() leaves it running and a retried stop() awaits
the same close instead of closing twice.

tests/blocking_io/test_channel_stop_offloop.py anchors all three stop()
paths to the strict Blockbuster gate, driving a real slack-sdk client.

* docs(changelog): reference #6134

* fix(channels): join the Discord client thread off the event loop in stop()

DiscordChannel.stop() still joined its client thread inline with a 10s
timeout. The thread usually exits right after the cross-loop client close,
but a timed-out close or a slow _run_client() drain keeps it alive for the
whole timeout, stalling the shared Gateway loop the same way the Slack,
Feishu and DingTalk teardown did. Run the join through asyncio.to_thread so
stop() matches the channels/AGENTS.md teardown rule, and add Discord to the
strict Blockbuster anchor.

CHANGELOG.md

@@ -433,6 +433,19 @@ This release closes that milestone with **301 merged pull requests**.
 
 ### Fixed
 
+- **channels:** Stopping or restarting the Slack, Feishu, DingTalk or Discord
+  channel no longer freezes the Gateway event loop. `SlackChannel.stop()` called
+  `SocketModeClient.close()` inline, which joins the SDK's message-processor
+  thread (about 0.7s every time) and waits for in-flight event listeners, whose
+  blocking Slack Web API calls can run up to the client timeout; Feishu and
+  DingTalk joined their SDK threads inline with a 5s timeout, and because those
+  threads only exit on a fatal error the join normally waited the full 5s;
+  Discord joined its client thread inline with a 10s timeout, which a timed-out
+  client close or a slow drain could use up. Every run, stream and channel on
+  the Gateway stalled meanwhile, including on
+  `POST /api/channels/{name}/restart`. The teardown now runs in a worker thread;
+  Slack's close is shielded and tracked, so a cancelled shutdown leaves it
+  running and a retried `stop()` awaits it instead of closing twice. ([#6134])
 - **middleware:** `_externalize_to_sandbox` now validates the full byte count
   of externalized tool outputs instead of only testing non-emptiness with
   `test -s`. When a remote sandbox write truncated the file part-way (e.g. disk
@@ -7489,4 +7502,5 @@ with **180 merged pull requests** since the first 2.0 milestone tag.
 [#6091]: https://github.com/bytedance/deer-flow/pull/6091
 [#6093]: https://github.com/bytedance/deer-flow/pull/6093
 [#6112]: https://github.com/bytedance/deer-flow/pull/6112
+[#6134]: https://github.com/bytedance/deer-flow/pull/6134
 

CHANGELOG_zh.md

@@ -412,6 +412,15 @@
 
 ### 修复
 
+- **渠道：** 停止或重启 Slack、飞书、钉钉、Discord 渠道不再冻结 Gateway 事件循环。
+  `SlackChannel.stop()` 直接调用 `SocketModeClient.close()`，它会 join SDK 的消息处理线程
+  （每次约 0.7 秒），并等待正在执行的事件监听器，而监听器中阻塞的 Slack Web API 调用
+  最长可持续到客户端超时；飞书与钉钉则直接以 5 秒超时 join SDK 线程，由于这些线程
+  只在致命错误时退出，这次 join 通常会等满 5 秒；Discord 直接以 10 秒超时 join 客户端线程，
+  客户端关闭超时或清理缓慢时会用满这段时间。期间 Gateway 上的所有运行、流和渠道
+  都会停顿，`POST /api/channels/{name}/restart` 也不例外。现在这些清理都在工作线程中执行；
+  Slack 的 close 会被 shield 并跟踪，因此被取消的关闭流程不会中断它，重试的 `stop()`
+  会等待同一个 close，而不会再关闭一次。([#6134])
 - **中间件：** `_externalize_to_sandbox` 现在校验外部化工具输出的完整字节大小，
   而不再仅使用 `test -s` 检查文件是否非空。此前当远程沙箱写入中途被截断时（如磁盘满
   或管道故障），残缺的文件也会通过校验并把错误路径交付给模型；现在会严格核对字节数与负载一致，
@@ -6289,3 +6298,4 @@ DeerFlow 2.0 是围绕"超级智能体"框架的彻底重写，核心包含子
 [#6091]: https://github.com/bytedance/deer-flow/pull/6091
 [#6093]: https://github.com/bytedance/deer-flow/pull/6093
 [#6112]: https://github.com/bytedance/deer-flow/pull/6112
+[#6134]: https://github.com/bytedance/deer-flow/pull/6134

backend/app/channels/dingtalk.py

@@ -228,7 +228,9 @@ class DingTalkChannel(Channel):
         self._card_repliers.clear()
         self._card_track_ids.clear()
         if self._thread:
-            self._thread.join(timeout=5)
+            # The SDK thread only returns on a fatal error, so this join
+            # normally waits out its full timeout; keep it off the event loop.
+            await asyncio.to_thread(self._thread.join, timeout=5)
             self._thread = None
         logger.info("DingTalk channel stopped")
 

backend/app/channels/discord.py

@@ -264,7 +264,10 @@ class DiscordChannel(Channel):
                     logger.exception("[Discord] error while closing client")
 
         if self._thread:
-            self._thread.join(timeout=10)
+            # The client thread normally exits right after the close above, but
+            # a timed-out close or a slow ``_run_client()`` drain can keep it
+            # alive for the whole join timeout; keep that wait off the loop.
+            await asyncio.to_thread(self._thread.join, timeout=10)
             self._thread = None
 
         # _run_client() normally drains these tasks in its finally block.  If

backend/app/channels/feishu.py

@@ -283,7 +283,9 @@ class FeishuChannel(Channel):
         self._background_tasks.clear()
         self._running_card_tasks.clear()
         if self._thread:
-            self._thread.join(timeout=5)
+            # The SDK thread only returns on a fatal error, so this join
+            # normally waits out its full timeout; keep it off the event loop.
+            await asyncio.to_thread(self._thread.join, timeout=5)
             self._thread = None
         logger.info("Feishu channel stopped")
 

backend/app/channels/slack.py

@@ -103,6 +103,7 @@ class SlackChannel(Channel):
     def __init__(self, bus: MessageBus, config: dict[str, Any]) -> None:
         super().__init__(name="slack", bus=bus, config=config)
         self._socket_client = None
+        self._socket_close_task: asyncio.Task[None] | None = None
         self._web_client = None
         self._loop: asyncio.AbstractEventLoop | None = None
         self._allowed_users = _normalize_allowed_users(config.get("allowed_users", []))
@@ -176,9 +177,22 @@ class SlackChannel(Channel):
         self._running = False
         self.bus.unsubscribe_outbound(self._on_outbound)
         await self._close_and_drain_threadsafe_futures()
-        if self._socket_client:
-            self._socket_client.close()
-            self._socket_client = None
+        # ``SocketModeClient.close()`` joins the SDK's message-processor thread
+        # (up to ~1s) and waits for in-flight listeners, whose blocking Web API
+        # calls run up to the WebClient timeout, so it runs off the event loop.
+        # The client is detached first so a still-queued connect skips it. The
+        # close task is shielded and tracked: a cancelled stop() leaves it
+        # running, and a retried stop() awaits it instead of closing twice.
+        socket_client, self._socket_client = self._socket_client, None
+        if socket_client is not None:
+            self._socket_close_task = asyncio.create_task(asyncio.to_thread(socket_client.close))
+        close_task = self._socket_close_task
+        if close_task is not None:
+            try:
+                await asyncio.shield(close_task)
+            finally:
+                if close_task.done():
+                    self._socket_close_task = None
         logger.info("Slack channel stopped")
 
     async def send(self, msg: OutboundMessage, *, _max_retries: int = 3) -> None:

backend/tests/blocking_io/test_channel_stop_offloop.py

@@ -0,0 +1,63 @@
+"""Regression anchor: channel ``stop()`` must not block the event loop on SDK teardown.
+
+``stop()`` runs on the Gateway event loop, both at shutdown and from
+``POST /api/channels/{name}/restart``, so a blocking teardown call there stalls
+every run, stream and channel the Gateway serves:
+
+- Slack: ``SocketModeClient.close()`` joins the SDK's message-processor thread
+  and waits for in-flight listeners.
+- Feishu / DingTalk: ``stop()`` joins the SDK thread with a 5s timeout, and
+  that thread only returns on a fatal error.
+- Discord: ``stop()`` joins the client thread with a 10s timeout, which a
+  timed-out client close or a slow ``_run_client()`` drain can use up.
+
+Each test drives the real ``stop()`` (and, for Slack, a real slack-sdk client).
+If a join or close regresses back onto the event loop, the strict Blockbuster
+gate raises ``BlockingError`` from the thread-join lock wait.
+"""
+
+from __future__ import annotations
+
+import threading
+
+import pytest
+
+from app.channels.dingtalk import DingTalkChannel
+from app.channels.discord import DiscordChannel
+from app.channels.feishu import FeishuChannel
+from app.channels.message_bus import MessageBus
+from app.channels.slack import SlackChannel
+
+pytestmark = pytest.mark.asyncio
+
+
+async def test_slack_stop_closes_real_socket_mode_client_off_loop() -> None:
+    from slack_sdk.socket_mode import SocketModeClient
+
+    # Never connected: construction alone starts the SDK's worker threads,
+    # which is what close() has to join.
+    socket_client = SocketModeClient(app_token="xapp-test")
+    channel = SlackChannel(bus=MessageBus(), config={})
+    channel._socket_client = socket_client
+    channel._running = True
+
+    await channel.stop()
+
+    assert socket_client.closed is True
+    assert channel._socket_client is None
+
+
+@pytest.mark.parametrize("channel_cls", [FeishuChannel, DingTalkChannel, DiscordChannel])
+async def test_stop_joins_sdk_thread_off_loop(channel_cls) -> None:
+    release = threading.Event()
+    sdk_thread = threading.Thread(target=release.wait, daemon=True)
+    sdk_thread.start()
+    channel = channel_cls(MessageBus(), {})
+    channel._thread = sdk_thread
+    channel._running = True
+    release.set()
+
+    await channel.stop()
+
+    assert not sdk_thread.is_alive()
+    assert channel._thread is None

backend/tests/test_dingtalk_channel.py

@@ -4,6 +4,7 @@ from __future__ import annotations
 
 import asyncio
 import logging
+import threading
 from types import SimpleNamespace
 from unittest.mock import AsyncMock, MagicMock
 
@@ -1770,6 +1771,31 @@ class TestCardMode:
         _run(go())
 
 
+class TestStop:
+    def test_stop_joins_stream_thread_off_the_event_loop(self):
+        async def go():
+            channel = DingTalkChannel(MessageBus(), config={})
+            release = threading.Event()
+            # dingtalk-stream's start_forever() never returns, so the join in
+            # stop() waits out its timeout; this thread blocks the same way,
+            # bounded so a join run on the event loop fails the assertion
+            # below instead of hanging the test.
+            stream_thread = threading.Thread(target=release.wait, args=(2,), daemon=True)
+            stream_thread.start()
+            channel._thread = stream_thread
+            channel._running = True
+
+            stop_task = asyncio.create_task(channel.stop())
+            await asyncio.sleep(0.05)
+            assert not stop_task.done()
+
+            release.set()
+            await stop_task
+            assert channel._thread is None
+
+        _run(go())
+
+
 # ---------------------------------------------------------------------------
 # Inbound file support
 # ---------------------------------------------------------------------------

backend/tests/test_discord_channel.py

@@ -115,6 +115,34 @@ async def test_discord_stop_does_not_clobber_store_before_load(tmp_path) -> None
     assert json.loads(store_path.read_text()) == {"chan-9": "thread-9"}
 
 
+@pytest.mark.asyncio
+async def test_discord_stop_joins_client_thread_off_the_event_loop() -> None:
+    """stop() must not join the client thread on the shared Gateway loop.
+
+    The thread usually exits right after the cross-loop client close, but a
+    timed-out close or a slow ``_run_client()`` drain keeps it alive for up to
+    the 10s join timeout, and the Gateway loop must keep running meanwhile.
+    """
+    channel = DiscordChannel(bus=MessageBus(), config={"bot_token": "token"})
+    release = threading.Event()
+    # Bounded so a join run on the event loop fails the assertion below
+    # instead of hanging the test.
+    client_thread = threading.Thread(target=release.wait, args=(2,), daemon=True)
+    client_thread.start()
+    channel._discord_loop = None
+    channel._client = None
+    channel._thread = client_thread
+    channel._cancel_ephemeral_tasks = AsyncMock()
+
+    stop_task = asyncio.create_task(channel.stop())
+    await asyncio.sleep(0.05)
+    assert not stop_task.done()
+
+    release.set()
+    await stop_task
+    assert channel._thread is None
+
+
 def _make_discord_message(text: str):
     return SimpleNamespace(
         id=111,

backend/tests/test_feishu_parser.py

@@ -1,6 +1,7 @@
 import asyncio
 import json
 import tempfile
+import threading
 from io import BytesIO
 from pathlib import Path
 from unittest.mock import AsyncMock, MagicMock
@@ -165,6 +166,30 @@ def test_feishu_is_not_running_when_ws_thread_exits():
     assert channel.is_running is False
 
 
+def test_feishu_stop_joins_ws_thread_off_the_event_loop():
+    async def go():
+        channel = FeishuChannel(MessageBus(), {"app_id": "test", "app_secret": "test"})
+        release = threading.Event()
+        # lark's ws client never returns from start(), so the join in stop()
+        # waits out its timeout; this thread blocks the same way, bounded so a
+        # join run on the event loop fails the assertion below instead of
+        # hanging the test.
+        ws_thread = threading.Thread(target=release.wait, args=(2,), daemon=True)
+        ws_thread.start()
+        channel._thread = ws_thread
+        channel._running = True
+
+        stop_task = asyncio.create_task(channel.stop())
+        await asyncio.sleep(0.05)
+        assert not stop_task.done()
+
+        release.set()
+        await stop_task
+        assert channel._thread is None
+
+    asyncio.run(go())
+
+
 def test_feishu_event_handler_ignores_non_content_message_events():
     import lark_oapi as lark
 

backend/tests/test_slack_channel_connections.py

@@ -4,10 +4,13 @@ from __future__ import annotations
 
 import asyncio
 import sys
+import threading
 from datetime import UTC, datetime, timedelta
 from types import ModuleType
 from unittest.mock import AsyncMock, MagicMock
 
+import pytest
+
 from app.channels.message_bus import MessageBus, OutboundMessage
 
 
@@ -242,7 +245,13 @@ def test_slack_socket_mode_queued_connect_does_not_target_replacement_client(mon
         queued_callbacks = []
         loop = asyncio.get_running_loop()
 
+        real_run_in_executor = loop.run_in_executor
+
         def queue_run_in_executor(executor, func, *args):
+            # Hold back only the connect; stop() offloads close() through the
+            # same executor path and must still run.
+            if func != channel._connect_socket_mode:
+                return real_run_in_executor(executor, func, *args)
             queued_callbacks.append((func, args))
             return loop.create_future()
 
@@ -268,3 +277,84 @@ def test_slack_socket_mode_queued_connect_does_not_target_replacement_client(mon
         await channel.stop()
 
     anyio.run(go)
+
+
+class _BlockingSocketModeClient:
+    """Socket Mode stand-in whose ``close()`` blocks like the SDK's thread joins."""
+
+    def __init__(self):
+        self.close_started = threading.Event()
+        self.release = threading.Event()
+        self.close_calls = 0
+        self.connect_calls = 0
+
+    def connect(self):
+        self.connect_calls += 1
+
+    def close(self):
+        self.close_calls += 1
+        self.close_started.set()
+        # Bounded so a close() run on the event loop fails the test's
+        # assertions instead of deadlocking it.
+        self.release.wait(timeout=2)
+
+
+def test_slack_stop_closes_socket_client_off_the_event_loop():
+    import anyio
+
+    from app.channels.slack import SlackChannel
+
+    async def go():
+        channel = SlackChannel(bus=MessageBus(), config={})
+        socket_client = _BlockingSocketModeClient()
+        channel._socket_client = socket_client
+        channel._running = True
+
+        stop_task = asyncio.create_task(channel.stop())
+        assert await asyncio.to_thread(socket_client.close_started.wait, 2)
+        # close() is blocked in a worker thread, so the loop keeps running and
+        # stop() is still waiting for it.
+        assert not stop_task.done()
+        # A connect still queued in the executor must not reopen the client
+        # that is being closed.
+        channel._connect_socket_mode(socket_client)
+        assert socket_client.connect_calls == 0
+
+        socket_client.release.set()
+        await stop_task
+        assert socket_client.close_calls == 1
+        assert channel._socket_client is None
+
+    anyio.run(go)
+
+
+def test_slack_stop_retried_after_cancellation_awaits_the_same_close():
+    import anyio
+
+    from app.channels.slack import SlackChannel
+
+    async def go():
+        channel = SlackChannel(bus=MessageBus(), config={})
+        socket_client = _BlockingSocketModeClient()
+        channel._socket_client = socket_client
+        channel._running = True
+
+        first_stop = asyncio.create_task(channel.stop())
+        assert await asyncio.to_thread(socket_client.close_started.wait, 2)
+        first_stop.cancel()
+        with pytest.raises(asyncio.CancelledError):
+            await first_stop
+
+        # The Gateway retries a cancelled shutdown on the same channel. The
+        # retry must wait for the close still running in the worker rather
+        # than return early or start a second close.
+        retry = asyncio.create_task(channel.stop())
+        await asyncio.sleep(0.05)
+        assert not retry.done()
+
+        socket_client.release.set()
+        await retry
+        assert socket_client.close_calls == 1
+        assert channel._socket_client is None
+
+    anyio.run(go)