fix(channels): retain live SDK workers after stop timeout (#6215)

* rebase(channels): replay retained worker fix

* rebase(channels): replay retained worker fix

* rebase(channels): replay retained worker fix

* rebase(channels): replay retained worker fix

* rebase(channels): replay stop-timeout classification

* rebase(channels): replay retained timeout handling

* rebase(channels): replay retained worker guidance

* docs(channels): keep lifecycle guidance within CI size limit

* docs(channels): fit lifecycle guidance within size budget

---------

Co-authored-by: Willem Jiang <willem.jiang@gmail.com>

backend/app/channels/base.py

@@ -39,6 +39,10 @@ class ChannelUnavailable(Exception):
     """
 
 
+class ChannelStopTimeout(RuntimeError):
+    """A channel still owns a live provider worker after bounded teardown."""
+
+
 @dataclass(eq=False, slots=True)
 class _ThreadsafeSubmission:
     coroutine: Coroutine[Any, Any, Any]

backend/app/channels/dingtalk.py

@@ -14,7 +14,7 @@ from typing import Any
 
 import httpx
 
-from app.channels.base import Channel
+from app.channels.base import Channel, ChannelStopTimeout
 from app.channels.commands import is_known_channel_command, strip_leading_mentions
 from app.channels.connection_identity import attach_connection_identity
 from app.channels.message_bus import InboundMessage, InboundMessageType, InboundReservation, MessageBus, OutboundMessage, ResolvedAttachment
@@ -230,8 +230,12 @@ class DingTalkChannel(Channel):
         if self._thread:
             # 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
+            thread = self._thread
+            await asyncio.to_thread(thread.join, timeout=5)
+            if thread.is_alive():
+                raise ChannelStopTimeout("DingTalk SDK thread is still running after stop timeout")
+            if self._thread is thread:
+                self._thread = None
         logger.info("DingTalk channel stopped")
 
     def _resolve_routing(self, msg: OutboundMessage) -> tuple[str, str, str]:

backend/app/channels/feishu.py

@@ -11,7 +11,7 @@ import threading
 import time
 from typing import Any, Literal
 
-from app.channels.base import Channel
+from app.channels.base import Channel, ChannelStopTimeout
 from app.channels.commands import is_known_channel_command, strip_leading_mentions
 from app.channels.connection_identity import attach_connection_identity
 from app.channels.message_bus import (
@@ -285,8 +285,12 @@ class FeishuChannel(Channel):
         if self._thread:
             # 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
+            thread = self._thread
+            await asyncio.to_thread(thread.join, timeout=5)
+            if thread.is_alive():
+                raise ChannelStopTimeout("Feishu SDK thread is still running after stop timeout")
+            if self._thread is thread:
+                self._thread = None
         logger.info("Feishu channel stopped")
 
     async def send(self, msg: OutboundMessage, *, _max_retries: int = 3) -> None:

backend/app/channels/service.py

@@ -10,7 +10,7 @@ from collections.abc import Callable
 from pathlib import Path
 from typing import TYPE_CHECKING, Any
 
-from app.channels.base import Channel
+from app.channels.base import Channel, ChannelStopTimeout
 from app.channels.manager import DEFAULT_CHANNEL_MAX_CONCURRENCY, DEFAULT_CHANNEL_SHUTDOWN_GRACE_PERIOD_SECONDS, DEFAULT_GATEWAY_URL, DEFAULT_LANGGRAPH_URL, ChannelManager
 from app.channels.message_bus import DEFAULT_INBOUND_QUEUE_MAXSIZE, MessageBus
 from app.channels.runtime_config_store import merge_runtime_channel_configs
@@ -300,6 +300,9 @@ class ChannelService:
                 # The Gateway deadline interrupted shutdown, so detaching them
                 # would hide resources that may still be in use.
                 raise
+            except ChannelStopTimeout as exc:
+                logger.warning("Channel %s remains retained after bounded stop timeout: %s", name, exc)
+                stop_errors.append(exc)
             except Exception as exc:
                 logger.exception("Error stopping channel")
                 stop_errors.append(exc)
@@ -428,6 +431,9 @@ class ChannelService:
             # interrupted cleanup, so detaching it here would hide resources
             # that may still be in use (mirrors ChannelService.stop()).
             raise
+        except ChannelStopTimeout as exc:
+            logger.warning("Channel %s remains retained after bounded stop timeout: %s", name, exc)
+            return
         except Exception:
             logger.exception("Error stopping channel %s during discard", name)
             return

backend/tests/test_dingtalk_channel.py

@@ -1772,6 +1772,22 @@ class TestCardMode:
 
 
 class TestStop:
+    def test_stop_retains_live_stream_thread_after_join_timeout(self):
+        async def go():
+            channel = DingTalkChannel(MessageBus(), config={})
+            thread = MagicMock()
+            thread.is_alive.return_value = True
+            channel._thread = thread
+            channel._running = True
+
+            with pytest.raises(RuntimeError, match="still running after stop timeout"):
+                await channel.stop()
+
+            thread.join.assert_called_once_with(timeout=5)
+            assert channel._thread is thread
+
+        _run(go())
+
     def test_stop_joins_stream_thread_off_the_event_loop(self):
         async def go():
             channel = DingTalkChannel(MessageBus(), config={})

backend/tests/test_feishu_parser.py

@@ -190,6 +190,23 @@ def test_feishu_stop_joins_ws_thread_off_the_event_loop():
     asyncio.run(go())
 
 
+def test_feishu_stop_retains_live_ws_thread_after_join_timeout():
+    async def go():
+        channel = FeishuChannel(MessageBus(), {"app_id": "test", "app_secret": "test"})
+        thread = MagicMock()
+        thread.is_alive.return_value = True
+        channel._thread = thread
+        channel._running = True
+
+        with pytest.raises(RuntimeError, match="still running after stop timeout"):
+            await channel.stop()
+
+        thread.join.assert_called_once_with(timeout=5)
+        assert channel._thread is thread
+
+    asyncio.run(go())
+
+
 def test_feishu_event_handler_ignores_non_content_message_events():
     import lark_oapi as lark