diff --git a/backend/app/channels/AGENTS.md b/backend/app/channels/AGENTS.md index 8b95a6dc37..13d29d6709 100644 --- a/backend/app/channels/AGENTS.md +++ b/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. diff --git a/backend/app/channels/dingtalk.py b/backend/app/channels/dingtalk.py index 9c5a7bdda2..2c81b7f522 100644 --- a/backend/app/channels/dingtalk.py +++ b/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. diff --git a/backend/tests/AGENTS.md b/backend/tests/AGENTS.md index a956a4cd2b..fa3975c415 100644 --- a/backend/tests/AGENTS.md +++ b/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 diff --git a/backend/tests/test_dingtalk_channel.py b/backend/tests/test_dingtalk_channel.py index 8d58d9d5dd..8682c191f8 100644 --- a/backend/tests/test_dingtalk_channel.py +++ b/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():