fix(dingtalk): stop stream task on shutdown

nanobot/channels/dingtalk.py

@@ -6,6 +6,8 @@ import mimetypes
 import os
 import time
 import zipfile
+from contextlib import suppress
+from inspect import isawaitable
 from io import BytesIO
 from pathlib import Path
 from typing import Any
@@ -204,6 +206,7 @@ class DingTalkChannel(BaseChannel):
         self.config: DingTalkConfig = config
         self._client: Any = None
         self._http: httpx.AsyncClient | None = None
+        self._start_task: asyncio.Task | None = None
 
         # Access Token management for sending messages
         self._access_token: str | None = None
@@ -214,6 +217,8 @@ class DingTalkChannel(BaseChannel):
 
     async def start(self) -> None:
         """Start the DingTalk bot with Stream Mode."""
+        current_task = asyncio.current_task()
+        self._start_task = current_task
         try:
             if not DINGTALK_AVAILABLE:
                 self.logger.error(
@@ -255,10 +260,25 @@ class DingTalkChannel(BaseChannel):
 
         except Exception:
             self.logger.exception("Failed to start channel")
+        finally:
+            self._running = False
+            if self._start_task is current_task:
+                self._start_task = None
 
     async def stop(self) -> None:
         """Stop the DingTalk bot."""
         self._running = False
+        await self._close_stream_client()
+        start_task = self._start_task
+        if start_task and start_task is not asyncio.current_task() and not start_task.done():
+            start_task.cancel()
+            await asyncio.sleep(0)
+            if not start_task.done():
+                start_task.cancel()
+            with suppress(asyncio.CancelledError):
+                await start_task
+        self._client = None
+
         # Close the shared HTTP client
         if self._http:
             await self._http.aclose()
@@ -268,6 +288,23 @@ class DingTalkChannel(BaseChannel):
             task.cancel()
         self._background_tasks.clear()
 
+    async def _close_stream_client(self) -> None:
+        client = self._client
+        if client is None:
+            return
+        close = getattr(client, "close", None)
+        if close is None:
+            websocket = getattr(client, "websocket", None)
+            close = getattr(websocket, "close", None)
+        if close is None:
+            return
+        try:
+            result = close()
+            if isawaitable(result):
+                await result
+        except Exception:
+            self.logger.debug("DingTalk stream client close failed", exc_info=True)
+
     async def _get_access_token(self) -> str | None:
         """Get or refresh Access Token."""
         if self._access_token and time.time() < self._token_expiry: