Track DingTalk stream message tasks to avoid dangling tasks (#7429)

* Track DingTalk stream message tasks to avoid dangling tasks

The stream loop handed each frame to client.background_task() via a
bare asyncio.create_task(...) whose result was dropped. asyncio only
keeps a weak reference to a task once it awaits, so the task could be
garbage-collected mid-execution (silently dropping a message) and any
exception it raised was never observed.

Keep a strong reference to each handler task and discard it via a done
callback, matching the background-task tracking used elsewhere (e.g.
the subagent executor and the MCP session pool).

* fix(channels): observe DingTalk stream task failures

---------

Co-authored-by: lisry-1 <lisry-1@users.noreply.github.com>
Co-authored-by: Willem Jiang <willem.jiang@gmail.com>

backend/app/channels/AGENTS.md

@@ -4,14 +4,16 @@ Store JSON decode errors follow corrupt-file recovery; SQL import keeps the orig
 
 IM channels use `langgraph-sdk` HTTP; Gateway owns thread lifecycle.
 WeChat outbound: off-loop cap+1; overflow `None` before encryption/upload; <=0 unlimited.
-Cancel drains cursor writes; expiry drains cursor/auth reset under `_auth_lock`.
+Cancel drains cursors; expiry drains cursor/auth reset under `_auth_lock`.
 Inbound: ingest/drop drains off-loop cleanup of owned dirs, never metadata paths.
 
 Telegram retries share bootstrap/Bot cleanup. Dead loops warn once; clear
 matching futures under `_shutdown_resources_lock` (closure unconfirmed).
 Finalize asyncgens; keep primary errors; propagate lone Bot errors.
 
-**Auth**: SDK and raw-httpx `/goal` calls send process-local auth plus `_csrf_headers()`; internal auth requires CSRF.
+**Auth**: SDK/raw-httpx `/goal` use process-local auth + `_csrf_headers()`; internal auth needs CSRF.
+
+DingTalk retains tasks and logs failures.
 
 **Components**:
 - `message_bus.py` - Inbound queue → dispatcher; outbound uses a dispatch-start listener snapshot. Unsubscribe neither cancels nor drains captured callbacks.

backend/app/channels/dingtalk.py

@@ -206,6 +206,11 @@ class DingTalkChannel(Channel):
         # Serialize inbound-file writes into the uploads directory to avoid
         # racing writers clobbering one another (mirrors FeishuChannel).
         self._file_write_lock = threading.Lock()
+        # Strong references to the per-message stream handler tasks. asyncio only
+        # keeps a weak reference to a task once it awaits, so a dropped
+        # ``create_task`` result can be garbage-collected mid-execution and its
+        # exception never observed; the done callback keeps the set bounded.
+        self._background_tasks: set[asyncio.Task[Any]] = set()
 
     @property
     def supports_streaming(self) -> bool:
@@ -508,7 +513,10 @@ class DingTalkChannel(Channel):
                     keepalive = asyncio.create_task(client.keepalive(websocket))
                     try:
                         async for raw_message in websocket:
-                            asyncio.create_task(client.background_task(json.loads(raw_message)))
+                            task = asyncio.create_task(client.background_task(json.loads(raw_message)))
+                            self._background_tasks.add(task)
+                            task.add_done_callback(self._background_tasks.discard)
+                            task.add_done_callback(self._log_task_error)
                     finally:
                         keepalive.cancel()
             except websockets.exceptions.ConnectionClosedError as exc:
@@ -522,6 +530,15 @@ class DingTalkChannel(Channel):
                 logger.exception("[DingTalk] stream error")
                 await asyncio.sleep(_DINGTALK_STREAM_ERROR_RETRY_SECONDS)
 
+    @staticmethod
+    def _log_task_error(task: asyncio.Task[Any]) -> None:
+        """Retrieve and log failures from per-message stream tasks."""
+        if task.cancelled():
+            return
+        exc = task.exception()
+        if exc is not None:
+            logger.error("[DingTalk] stream message task failed: %s", exc, exc_info=exc)
+
     async def _open_stream_connection(self, client: Any) -> dict[str, Any] | None:
         """Fetch a Stream ticket: ``DingTalkStreamClient.open_connection()`` with a deadline.
 

backend/tests/AGENTS.md

@@ -178,6 +178,13 @@ contract probes execute the actual migration test setup through SQLAlchemy's
 dialect argument conversion, stopping before database acquisition; the real
 SQLite/PostgreSQL cases still exercise upgrade, downgrade and re-upgrade.
 
+## DingTalk stream tasks
+
+DingTalk stream lifecycle tests drive `_serve_stream` with a fake websocket and
+real handler tasks. Use events to pin pending-task retention, completion cleanup,
+logged failures and quiet cancellation; drain handlers and keepalive tasks even
+when an assertion fails.
+
 ## Real Compose tests
 
 `support/compose.py` probes `docker compose version --short` and requires Compose

backend/tests/test_dingtalk_channel.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import asyncio
+import json
 import logging
 import threading
 import time
@@ -1857,6 +1858,115 @@ class TestCardMode:
         _run(go())
 
 
+class TestStreamTaskLifecycle:
+    @pytest.mark.parametrize("outcome", ["success", "failure", "cancelled"])
+    def test_tracks_and_finalizes_message_tasks(self, outcome, monkeypatch, caplog):
+        async def go():
+            channel = DingTalkChannel(MessageBus(), config={})
+            channel._running = True
+            frames = [{"sequence": index} for index in range(50 if outcome == "success" else 1)]
+            received = []
+            handler_tasks = []
+            keepalive_tasks = []
+            handlers_started = asyncio.Event()
+            release_handlers = asyncio.Event()
+            handlers_finalized = asyncio.Event()
+            close_stream = asyncio.Event()
+            keepalive_started = asyncio.Event()
+            keepalive_stopped = asyncio.Event()
+            failure = RuntimeError("stream handler failed")
+            loop = asyncio.get_running_loop()
+            previous_exception_handler = loop.get_exception_handler()
+            loop_errors = []
+            loop.set_exception_handler(lambda _loop, context: loop_errors.append(context))
+
+            async def background_task(frame):
+                received.append(frame)
+                handler_tasks.append(asyncio.current_task())
+                if len(received) == len(frames):
+                    handlers_started.set()
+                await release_handlers.wait()
+                if outcome == "failure":
+                    raise failure
+
+            async def keepalive(_websocket):
+                keepalive_tasks.append(asyncio.current_task())
+                keepalive_started.set()
+                try:
+                    await asyncio.Event().wait()
+                finally:
+                    keepalive_stopped.set()
+
+            class WebSocket:
+                async def __aenter__(self):
+                    return self
+
+                async def __aexit__(self, *_args):
+                    return None
+
+                async def __aiter__(self):
+                    for frame in frames:
+                        yield json.dumps(frame)
+                    await close_stream.wait()
+                    channel._running = False
+
+            client = SimpleNamespace(pre_start=MagicMock(), background_task=background_task, keepalive=keepalive)
+            monkeypatch.setattr(channel, "_open_stream_connection", AsyncMock(return_value={"endpoint": "wss://example.invalid/stream", "ticket": "test-ticket"}))
+            monkeypatch.setattr("websockets.connect", lambda _uri: WebSocket())
+            stream_task = asyncio.create_task(channel._serve_stream(client))
+            try:
+                await asyncio.wait_for(handlers_started.wait(), timeout=1)
+                await asyncio.wait_for(keepalive_started.wait(), timeout=1)
+                assert received == frames
+                assert channel._background_tasks == set(handler_tasks)
+                assert all(not task.done() for task in handler_tasks)
+
+                remaining = set(handler_tasks)
+
+                def finalized(task):
+                    remaining.remove(task)
+                    if not remaining:
+                        handlers_finalized.set()
+
+                # These observers run after the production done callbacks.
+                for task in handler_tasks:
+                    task.add_done_callback(finalized)
+                    if outcome == "cancelled":
+                        task.cancel()
+                release_handlers.set()
+                await asyncio.wait_for(handlers_finalized.wait(), timeout=1)
+                assert channel._background_tasks == set()
+                if outcome == "cancelled":
+                    assert all(task.cancelled() for task in handler_tasks)
+
+                close_stream.set()
+                await asyncio.wait_for(stream_task, timeout=1)
+                await asyncio.wait_for(keepalive_stopped.wait(), timeout=1)
+                assert channel._background_tasks == set()
+                client.pre_start.assert_called_once_with()
+                errors = [record for record in caplog.records if record.name == "app.channels.dingtalk" and record.levelno >= logging.ERROR]
+                if outcome == "failure":
+                    assert len(errors) == 1
+                    assert "stream message task" in errors[0].getMessage()
+                    assert errors[0].exc_info is not None
+                    assert errors[0].exc_info[1] is failure
+                else:
+                    assert errors == []
+                assert loop_errors == []
+            finally:
+                channel._running = False
+                close_stream.set()
+                release_handlers.set()
+                tasks = [stream_task, *handler_tasks, *keepalive_tasks]
+                for task in tasks:
+                    if not task.done():
+                        task.cancel()
+                await asyncio.gather(*tasks, return_exceptions=True)
+                loop.set_exception_handler(previous_exception_handler)
+
+        _run(go())
+
+
 class TestStop:
     def test_stop_retains_live_stream_thread_after_join_timeout(self):
         async def go():