diff --git a/nanobot/channels/dingtalk/runtime.py b/nanobot/channels/dingtalk/runtime.py index f00d75e49..a66924f5c 100644 --- a/nanobot/channels/dingtalk/runtime.py +++ b/nanobot/channels/dingtalk/runtime.py @@ -196,7 +196,7 @@ class NanobotDingTalkHandler(_CallbackHandlerBase): ) ) self.channel._background_tasks.add(task) - task.add_done_callback(self.channel._background_tasks.discard) + task.add_done_callback(self.channel._on_background_task_done) return AckMessage.STATUS_OK, "OK" @@ -257,6 +257,16 @@ class DingTalkChannel(BaseChannel): # Hold references to background tasks to prevent GC self._background_tasks: set[asyncio.Task[None]] = set() + def _on_background_task_done(self, task: asyncio.Task[None]) -> None: + self._background_tasks.discard(task) + if task.cancelled(): + return + exception = task.exception() + if exception is not None: + self.logger.opt(exception=exception).error( + "DingTalk inbound message task failed" + ) + async def start(self) -> None: """Start the DingTalk bot with Stream Mode.""" current_task = asyncio.current_task() @@ -326,8 +336,11 @@ class DingTalkChannel(BaseChannel): await self._http.aclose() self._http = None # Cancel outstanding background tasks - for task in self._background_tasks: + background_tasks = tuple(self._background_tasks) + for task in background_tasks: task.cancel() + if background_tasks: + await asyncio.gather(*background_tasks, return_exceptions=True) self._background_tasks.clear() async def _close_stream_client(self) -> None: diff --git a/nanobot/channels/dingtalk/tests/test_dingtalk_channel.py b/nanobot/channels/dingtalk/tests/test_dingtalk_channel.py index 2721a646d..aa69c061c 100644 --- a/nanobot/channels/dingtalk/tests/test_dingtalk_channel.py +++ b/nanobot/channels/dingtalk/tests/test_dingtalk_channel.py @@ -3,7 +3,7 @@ import json import zipfile from io import BytesIO from types import SimpleNamespace -from unittest.mock import AsyncMock +from unittest.mock import AsyncMock, MagicMock import httpx import pytest @@ -402,6 +402,61 @@ async def test_handler_uses_voice_recognition_text_when_text_is_empty(monkeypatc assert msg.chat_id == "group:conv123" +@pytest.mark.asyncio +async def test_handler_retrieves_background_message_failure(monkeypatch) -> None: + bus = MessageBus() + channel = DingTalkChannel( + DingTalkConfig(client_id="app", client_secret="secret", allow_from=["user1"]), + bus, + ) + handler = NanobotDingTalkHandler(channel) + failure = RuntimeError("inbound dispatch failed") + mock_logger = MagicMock() + channel.logger = mock_logger + + class _FakeChatbotMessage: + text = SimpleNamespace(content="hello") + extensions = {} + sender_staff_id = "user1" + sender_id = "fallback-user" + sender_nick = "Alice" + message_type = "text" + + @staticmethod + def from_dict(_data): + return _FakeChatbotMessage() + + async def fail(*_args) -> None: + raise failure + + monkeypatch.setattr(dingtalk_module, "ChatbotMessage", _FakeChatbotMessage) + monkeypatch.setattr(dingtalk_module, "AckMessage", SimpleNamespace(STATUS_OK="OK")) + monkeypatch.setattr(channel, "_on_message", fail) + event_loop = asyncio.get_running_loop() + previous_handler = event_loop.get_exception_handler() + loop_errors: list[dict[str, object]] = [] + event_loop.set_exception_handler(lambda _loop, context: loop_errors.append(context)) + + try: + status, body = await handler.process( + SimpleNamespace(data={"conversationType": "1", "text": {"content": "hello"}}) + ) + for _ in range(10): + await asyncio.sleep(0) + if not channel._background_tasks: + break + finally: + event_loop.set_exception_handler(previous_handler) + + assert (status, body) == ("OK", "OK") + assert not channel._background_tasks + assert not loop_errors + mock_logger.opt.assert_called_once_with(exception=failure) + mock_logger.opt.return_value.error.assert_called_once_with( + "DingTalk inbound message task failed" + ) + + @pytest.mark.asyncio async def test_handler_processes_file_message(monkeypatch) -> None: """Test that file messages are handled and forwarded with downloaded path.""" @@ -650,6 +705,41 @@ async def test_stop_cancels_stream_client_after_sdk_swallows_first_cancel(monkey assert start_task.cancelled() +@pytest.mark.asyncio +async def test_stop_waits_for_background_message_tasks() -> None: + channel = DingTalkChannel( + DingTalkConfig(client_id="app", client_secret="secret", allow_from=["*"]), + MessageBus(), + ) + mock_logger = MagicMock() + channel.logger = mock_logger + started = asyncio.Event() + cancelled = asyncio.Event() + + async def wait_forever() -> None: + started.set() + try: + await asyncio.Future() + finally: + cancelled.set() + + task = asyncio.create_task(wait_forever()) + channel._background_tasks.add(task) + task.add_done_callback(channel._on_background_task_done) + await started.wait() + + try: + await channel.stop() + assert task.done() + assert cancelled.is_set() + assert not channel._background_tasks + mock_logger.opt.assert_not_called() + finally: + if not task.done(): + task.cancel() + await asyncio.gather(task, return_exceptions=True) + + @pytest.mark.asyncio async def test_download_dingtalk_file(tmp_path, monkeypatch) -> None: """Test the two-step file download flow (get URL then download content)."""