fix(channels): retain Slack connect work through shutdown (#6808)

* fix(channels): retain Slack connect work through shutdown

* fix(channels): distinguish Slack cleanup failures from shutdown deadlines

---------

Co-authored-by: Totoro-qaq <279883115+Totoro-qaq@users.noreply.github.com>

README.md

@@ -1050,6 +1050,7 @@ Notes:
 - 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.
+- 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.
 - 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/slack.py

@@ -10,12 +10,13 @@ from typing import Any
 
 from markdown_to_mrkdwn import SlackMarkdownConverter
 
-from app.channels.base import Channel
+from app.channels.base import Channel, ChannelStopTimeout
 from app.channels.commands import is_known_channel_command
 from app.channels.connection_identity import attach_connection_identity
 from app.channels.message_bus import InboundMessageType, InboundReservation, MessageBus, OutboundMessage, ResolvedAttachment
 
 logger = logging.getLogger(__name__)
+SLACK_STOP_TIMEOUT_SECONDS = 5.0
 
 _slack_md_converter = SlackMarkdownConverter()
 
@@ -103,6 +104,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_connect_future: asyncio.Future[None] | None = None
         self._socket_close_task: asyncio.Task[None] | None = None
         self._web_client = None
         self._loop: asyncio.AbstractEventLoop | None = None
@@ -115,6 +117,10 @@ class SlackChannel(Channel):
     async def start(self) -> None:
         if self._running:
             return
+        if self._socket_client is not None or self._socket_close_task is not None:
+            # A timed-out or cancelled stop still owns its SDK work. Reuse its
+            # bounded wait rather than replacing a client that can still dial.
+            await self.stop()
 
         try:
             from slack_sdk import WebClient
@@ -152,12 +158,11 @@ class SlackChannel(Channel):
         self._running = True
         self.bus.subscribe_outbound(self._on_outbound)
 
-        # Start socket mode in background thread. Go through
-        # ``_connect_socket_mode`` so a connect failure gets logged: nothing
-        # awaits the future ``run_in_executor`` returns, so an exception stored
-        # on it would never surface.
+        # Start socket mode in a background thread.
+        # ``_connect_socket_mode`` logs startup failures. Retain the future so
+        # stop() can wait for an in-flight dial before closing its socket.
         socket_client = self._socket_client
-        asyncio.get_event_loop().run_in_executor(None, self._connect_socket_mode, socket_client)
+        self._socket_connect_future = asyncio.get_event_loop().run_in_executor(None, self._connect_socket_mode, socket_client)
         logger.info("Slack channel started")
 
     def _connect_socket_mode(self, socket_client: Any) -> None:
@@ -166,7 +171,7 @@ class SlackChannel(Channel):
         Mirrors the Telegram and Discord channels, whose background threads log
         their errors instead of letting them vanish with the thread.
         """
-        if socket_client is not self._socket_client:
+        if not self._running or socket_client is not self._socket_client:
             return
         try:
             socket_client.connect()
@@ -177,24 +182,35 @@ class SlackChannel(Channel):
         self._running = False
         self.bus.unsubscribe_outbound(self._on_outbound)
         await self._close_and_drain_threadsafe_futures()
-        # ``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))
+        # The SDK can publish a session after close() if connect() is already
+        # waiting for its URL. One owned task drains that connect and then
+        # closes off-loop. A timeout or cancellation keeps it available for
+        # retry; cancelling an executor future would not stop its thread.
+        if self._socket_close_task is None and self._socket_client is not None:
+            self._socket_close_task = asyncio.create_task(self._close_socket_client(self._socket_client, self._socket_connect_future))
         close_task = self._socket_close_task
         if close_task is not None:
             try:
-                await asyncio.shield(close_task)
-            finally:
+                await asyncio.wait_for(asyncio.shield(close_task), timeout=SLACK_STOP_TIMEOUT_SECONDS)
+            except TimeoutError as exc:
                 if close_task.done():
+                    close_task.result()
+                else:
+                    raise ChannelStopTimeout("Slack Socket Mode cleanup is still running") from exc
+            finally:
+                if close_task.done() and self._socket_close_task is close_task:
                     self._socket_close_task = None
         logger.info("Slack channel stopped")
 
+    async def _close_socket_client(self, socket_client: Any, connect_future: asyncio.Future[None] | None) -> None:
+        if connect_future is not None:
+            await asyncio.shield(connect_future)
+        await asyncio.to_thread(socket_client.close)
+        if self._socket_client is socket_client:
+            self._socket_client = None
+        if self._socket_connect_future is connect_future:
+            self._socket_connect_future = None
+
     async def send(self, msg: OutboundMessage, *, _max_retries: int = 3) -> None:
         web_client = await self._get_web_client_for_message(msg)
         if not web_client:

backend/tests/blocking_io/test_channel_stop_offloop.py

@@ -18,10 +18,12 @@ gate raises ``BlockingError`` from the thread-join lock wait.
 
 from __future__ import annotations
 
+import asyncio
 import threading
 
 import pytest
 
+from app.channels.base import ChannelStopTimeout
 from app.channels.dingtalk import DingTalkChannel
 from app.channels.discord import DiscordChannel
 from app.channels.feishu import FeishuChannel
@@ -47,6 +49,45 @@ async def test_slack_stop_closes_real_socket_mode_client_off_loop() -> None:
     assert channel._socket_client is None
 
 
+async def test_slack_stop_keeps_connect_cleanup_off_loop(monkeypatch) -> None:
+    started, release = threading.Event(), threading.Event()
+    loop_thread = threading.get_ident()
+    close_threads = []
+
+    class SocketClient:
+        def connect(self):
+            started.set()
+            release.wait(5)
+
+        def close(self):
+            close_threads.append(threading.get_ident())
+
+    monkeypatch.setattr("app.channels.slack.SLACK_STOP_TIMEOUT_SECONDS", 0.02)
+    channel = SlackChannel(MessageBus(), {})
+    client = SocketClient()
+    channel._socket_client = client
+    channel._running = True
+    future = asyncio.get_running_loop().run_in_executor(None, channel._connect_socket_mode, client)
+    channel._socket_connect_future = future
+    try:
+        assert await asyncio.to_thread(started.wait, 2)
+        with pytest.raises(ChannelStopTimeout):
+            await channel.stop()
+        assert not future.cancelled()
+        assert close_threads == []
+        release.set()
+        await asyncio.wait_for(asyncio.shield(channel._socket_close_task), 2)
+        await channel.stop()
+        assert len(close_threads) == 1
+        assert close_threads[0] != loop_thread
+        assert channel._socket_client is None
+    finally:
+        release.set()
+        await asyncio.wait_for(asyncio.shield(future), 5)
+        monkeypatch.setattr("app.channels.slack.SLACK_STOP_TIMEOUT_SECONDS", 3)
+        await channel.stop()
+
+
 @pytest.mark.parametrize("channel_cls", [FeishuChannel, DingTalkChannel, DiscordChannel])
 async def test_stop_joins_sdk_thread_off_loop(channel_cls) -> None:
     release = threading.Event()

backend/tests/test_slack_channel_connections.py

@@ -179,8 +179,8 @@ def test_slack_socket_mode_connect_failure_is_logged(monkeypatch, caplog):
             },
         )
 
-        # start() drops the future for the executor-backed connect; capture it
-        # here so the test can wait for the connect attempt to finish.
+        # Capture the executor-backed connect so the test can wait for the
+        # failing attempt before checking its logged error.
         loop = asyncio.get_running_loop()
         real_run_in_executor = loop.run_in_executor
         connect_futures = []
@@ -252,25 +252,31 @@ def test_slack_socket_mode_queued_connect_does_not_target_replacement_client(mon
             # 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()
+            future = loop.create_future()
+            queued_callbacks.append((func, args, future))
+            return future
 
         monkeypatch.setattr(loop, "run_in_executor", queue_run_in_executor)
 
         await channel.start()
-        await channel.stop()
+        stopping = asyncio.create_task(channel.stop())
+        await asyncio.sleep(0)
+        stale_func, stale_args, stale_future = queued_callbacks[0]
+        stale_func(*stale_args)
+        stale_future.set_result(None)
+        await stopping
         await channel.start()
 
         assert len(clients) == 2
         assert len(queued_callbacks) == 2
 
-        stale_func, stale_args = queued_callbacks[0]
         stale_func(*stale_args)
         assert clients[0].connect_calls == 0
         assert clients[1].connect_calls == 0
 
-        current_func, current_args = queued_callbacks[1]
+        current_func, current_args, current_future = queued_callbacks[1]
         current_func(*current_args)
+        current_future.set_result(None)
         assert clients[0].connect_calls == 0
         assert clients[1].connect_calls == 1
 

backend/tests/test_slack_channel_lifecycle.py

@@ -0,0 +1,243 @@
+"""Slack stop retains executor-backed connects until their sockets are closed."""
+
+from __future__ import annotations
+
+import asyncio
+import contextlib
+import socket
+import threading
+from contextlib import asynccontextmanager
+from unittest.mock import MagicMock
+
+import pytest
+
+from app.channels.base import ChannelStopTimeout
+from app.channels.message_bus import MessageBus
+from app.channels.service import ChannelService
+from app.channels.slack import SlackChannel
+
+pytestmark = pytest.mark.asyncio
+
+
+def _capture_connect(monkeypatch, channel):
+    loop = asyncio.get_running_loop()
+    run_in_executor = loop.run_in_executor
+    futures = []
+
+    def capture(executor, func, *args):
+        future = run_in_executor(executor, func, *args)
+        if func == channel._connect_socket_mode:
+            futures.append(future)
+        return future
+
+    monkeypatch.setattr(loop, "run_in_executor", capture)
+    return futures
+
+
+@asynccontextmanager
+async def _blocked_connect(monkeypatch):
+    import slack_sdk.socket_mode
+
+    started, release = threading.Event(), threading.Event()
+    clients = []
+
+    class SocketClient:
+        def __init__(self, **kwargs):
+            self.socket_mode_request_listeners = []
+            self.connect_finished = False
+            self.connected = False
+            self.close_calls = 0
+            clients.append(self)
+
+        def connect(self):
+            started.set()
+            if not release.wait(5):
+                raise TimeoutError("test did not release connection")
+            self.connected = True
+            self.connect_finished = True
+
+        def close(self):
+            self.close_calls += 1
+            self.connected = False
+
+    monkeypatch.setattr(slack_sdk.socket_mode, "SocketModeClient", SocketClient)
+    channel = SlackChannel(MessageBus(), {"bot_token": "test", "app_token": "test", "bot_user_id": "UBOT", "web_client_factory": lambda **kwargs: object()})
+    futures = _capture_connect(monkeypatch, channel)
+    try:
+        await channel.start()
+        assert await asyncio.to_thread(started.wait, 2)
+        yield channel, clients, futures[0], release
+    finally:
+        release.set()
+        for future in futures:
+            await asyncio.wait_for(asyncio.shield(future), 5)
+        monkeypatch.setattr("app.channels.slack.SLACK_STOP_TIMEOUT_SECONDS", 3, raising=False)
+        await asyncio.wait_for(channel.stop(), 4)
+
+
+async def test_stop_preserves_sdk_close_timeout_and_retries():
+    error = TimeoutError("SDK close failed")
+    socket_client = MagicMock()
+    socket_client.close.side_effect = [error, None]
+    channel = SlackChannel(MessageBus(), {})
+    channel._socket_client = socket_client
+
+    stopping = asyncio.create_task(channel.stop())
+    await asyncio.sleep(0)
+    cleanup = channel._socket_close_task
+    assert cleanup is not None
+    with pytest.raises(TimeoutError, match="SDK close failed") as caught:
+        await stopping
+    # Python may copy TimeoutError across the executor boundary; stop must
+    # preserve the exception actually produced by its owned cleanup task.
+    assert caught.value is cleanup.exception()
+    assert type(caught.value) is TimeoutError
+    assert caught.value.__cause__ is not caught.value
+    assert channel._socket_client is socket_client
+    assert channel._socket_close_task is None
+    assert socket_client.close.call_count == 1
+
+    await channel.stop()
+    assert socket_client.close.call_count == 2
+    assert channel._socket_client is None
+    assert channel._socket_close_task is None
+
+
+async def test_stop_timeout_retains_connect_and_same_cleanup_for_retry(monkeypatch):
+    monkeypatch.setattr("app.channels.slack.SLACK_STOP_TIMEOUT_SECONDS", 0.02, raising=False)
+    async with _blocked_connect(monkeypatch) as (channel, clients, connect, release):
+        with pytest.raises(ChannelStopTimeout, match="cleanup is still running"):
+            await channel.stop()
+        cleanup = channel._socket_close_task
+        assert cleanup is not None and not cleanup.done()
+        assert channel._socket_client is clients[0]
+        assert channel._socket_connect_future is connect
+        assert not connect.done()
+        assert clients[0].close_calls == 0
+
+        with pytest.raises(ChannelStopTimeout):
+            await channel.start()
+        assert channel._socket_close_task is cleanup
+        assert len(clients) == 1
+        assert not connect.cancelled()
+
+        release.set()
+        await asyncio.wait_for(asyncio.shield(cleanup), 2)
+        await channel.stop()
+        assert clients[0].connect_finished
+        assert clients[0].close_calls == 1
+        assert not clients[0].connected
+        assert channel._socket_client is None
+        assert channel._socket_connect_future is None
+        assert channel._socket_close_task is None
+
+
+async def test_cancelling_stop_does_not_cancel_connect_or_duplicate_cleanup(monkeypatch):
+    async with _blocked_connect(monkeypatch) as (channel, clients, connect, release):
+        stopping = asyncio.create_task(channel.stop())
+        await asyncio.sleep(0)
+        stopping.cancel()
+        with pytest.raises(asyncio.CancelledError):
+            await stopping
+        cleanup = channel._socket_close_task
+        assert cleanup is not None and not cleanup.done()
+        assert channel._socket_client is clients[0]
+        assert not connect.cancelled()
+        assert clients[0].close_calls == 0
+
+        release.set()
+        await channel.stop()
+        assert clients[0].close_calls == 1
+        assert channel._socket_close_task is None
+
+
+async def test_service_retains_slack_while_connect_is_still_owned(monkeypatch):
+    monkeypatch.setattr("app.channels.slack.SLACK_STOP_TIMEOUT_SECONDS", 0.02, raising=False)
+    async with _blocked_connect(monkeypatch) as (channel, clients, connect, release):
+        service = ChannelService({"slack": {"enabled": True}}, store=MagicMock())
+        service._running = True
+        service._channels["slack"] = channel
+        assert await service.restart_channel("slack", reload_config=False) is False
+        assert service._channels["slack"] is channel
+        assert channel._socket_client is clients[0]
+        assert not connect.cancelled()
+        assert len(clients) == 1
+
+        release.set()
+        await asyncio.wait_for(asyncio.shield(channel._socket_close_task), 2)
+        assert await service.remove_channel("slack") is True
+        assert "slack" not in service._channels
+        assert clients[0].close_calls == 1
+
+
+async def test_real_sdk_late_connection_is_closed_before_stop_completes(monkeypatch):
+    from websockets.asyncio.server import serve
+
+    for key in ("HTTP_PROXY", "HTTPS_PROXY", "ALL_PROXY", "http_proxy", "https_proxy", "all_proxy"):
+        monkeypatch.delenv(key, raising=False)
+    monkeypatch.setattr("app.channels.slack.SLACK_STOP_TIMEOUT_SECONDS", 0.02, raising=False)
+    discovery_started, release_discovery = threading.Event(), threading.Event()
+    accepted, peer_closed = asyncio.Event(), asyncio.Event()
+    peers = []
+
+    async def handler(peer):
+        peers.append(peer)
+        accepted.set()
+        await peer.wait_closed()
+        peer_closed.set()
+
+    async with serve(handler, "127.0.0.1", 0, ping_interval=None, close_timeout=0.2) as server:
+        port = server.sockets[0].getsockname()[1]
+
+        class WebClient:
+            ssl = None
+
+            def __init__(self, **kwargs):
+                pass
+
+            def apps_connections_open(self, **kwargs):
+                discovery_started.set()
+                if not release_discovery.wait(5):
+                    raise TimeoutError("test did not release local URL discovery")
+                return {"url": f"ws://127.0.0.1:{port}/socket-mode"}
+
+        channel = SlackChannel(MessageBus(), {"bot_token": "test", "app_token": "test", "bot_user_id": "UBOT", "web_client_factory": WebClient})
+        futures = _capture_connect(monkeypatch, channel)
+        sdk = None
+        try:
+            await channel.start()
+            sdk = channel._socket_client
+            assert await asyncio.to_thread(discovery_started.wait, 2)
+            with pytest.raises(ChannelStopTimeout):
+                await channel.stop()
+            assert channel._socket_client is sdk
+            assert not futures[0].cancelled()
+            release_discovery.set()
+            monkeypatch.setattr("app.channels.slack.SLACK_STOP_TIMEOUT_SECONDS", 5)
+            await asyncio.wait_for(channel.stop(), 6)
+            await asyncio.wait_for(accepted.wait(), 2)
+            await asyncio.wait_for(peer_closed.wait(), 2)
+            assert not sdk.is_connected()
+            assert sdk.closed is True
+            assert sdk.auto_reconnect_enabled is False
+            assert channel._socket_client is None
+            assert channel._socket_connect_future is None
+        finally:
+            release_discovery.set()
+            for future in futures:
+                await asyncio.wait_for(asyncio.shield(future), 5)
+            if sdk is not None:
+                # slack-sdk 3.41.0 leaves its receiver interval runner alive
+                # after close; explicitly reap the test-owned runner as well.
+                sdk.auto_reconnect_enabled = False
+                sdk.default_auto_reconnect_enabled = False
+                if sdk.current_session is not None:
+                    sdk.current_session_state.terminated = True
+                    sock = sdk.current_session.sock
+                    if sock is not None:
+                        with contextlib.suppress(OSError):
+                            sock.shutdown(socket.SHUT_RDWR)
+                await asyncio.wait_for(asyncio.to_thread(sdk.close), 5)
+                await asyncio.wait_for(asyncio.to_thread(sdk.current_session_runner.shutdown), 3)
+            for peer in peers:
+                peer.transport.abort()