fix(qqbot): track inbound dispatch tasks so received messages aren't dropped

gateway/platforms/qqbot/adapter.py

@@ -791,18 +791,27 @@ async def _send_resume(self) -> None:
             self._session_id = None
             self._last_seq = None
 
-    @staticmethod
-    def _create_task(coro):
+    def _create_task(self, coro):
         """Schedule a coroutine, silently skipping if no event loop is running.
 
         This avoids ``RuntimeError: no running event loop`` when tests call
         ``_dispatch_payload`` synchronously outside of ``asyncio.run()``.
+
+        The task is held in ``self._background_tasks`` until it completes and
+        the done-callback discards it. The event loop only keeps a *weak*
+        reference to a bare task, so without this strong reference the GC may
+        cancel a still-running handler mid-flight — for the inbound-message
+        path that means a received QQ message is silently dropped. Mirrors the
+        tracking the base adapter already does for its own spawned tasks.
         """
         try:
             loop = asyncio.get_running_loop()
-            return loop.create_task(coro)
         except RuntimeError:
             return None
+        task = loop.create_task(coro)
+        self._background_tasks.add(task)
+        task.add_done_callback(self._background_tasks.discard)
+        return task
 
     def _dispatch_payload(self, payload: Dict[str, Any]) -> None:
         """Route inbound WebSocket payloads (dispatch synchronously, spawn async handlers)."""
@@ -846,7 +855,7 @@ def _dispatch_payload(self, payload: Dict[str, Any]) -> None:
                     "GUILD_MESSAGE_CREATE",
                     "GUILD_AT_MESSAGE_CREATE",
             }:
-                asyncio.create_task(self._on_message(t, d))
+                self._create_task(self._on_message(t, d))
             elif t == "INTERACTION_CREATE":
                 self._create_task(self._on_interaction(d))
             else:

tests/gateway/test_qqbot.py

@@ -446,6 +446,45 @@ def test_seq_increments(self):
         adapter._dispatch_payload({"op": 0, "t": "SOME_EVENT", "s": 10, "d": {}})
         assert adapter._last_seq == 10
 
+    def test_message_dispatch_without_event_loop_does_not_raise(self):
+        # A message op must be routed through the loop-safe _create_task helper,
+        # exactly like every other op _dispatch_payload handles. The inbound
+        # branch previously used a bare asyncio.create_task(), which raises
+        # "no running event loop" whenever _dispatch_payload runs outside
+        # asyncio.run() — as every other synchronous dispatch test above does.
+        adapter = self._make_adapter(app_id="a", client_secret="b")
+        adapter._on_message = mock.Mock()  # sync stub: not scheduled without a loop
+
+        adapter._dispatch_payload(
+            {"op": 0, "t": "C2C_MESSAGE_CREATE", "s": 7, "d": {"id": "m1"}}
+        )
+
+        # Sequence tracking still advances and the message was routed on.
+        assert adapter._last_seq == 7
+        adapter._on_message.assert_called_once_with("C2C_MESSAGE_CREATE", {"id": "m1"})
+
+    def test_message_dispatch_task_is_tracked(self):
+        # The spawned handler task must be held in _background_tasks so the GC
+        # cannot cancel it mid-flight (the loop only keeps a weak reference),
+        # which would silently drop a received QQ message.
+        adapter = self._make_adapter(app_id="a", client_secret="b")
+        adapter._on_message = mock.AsyncMock()
+
+        async def run():
+            adapter._dispatch_payload(
+                {"op": 0, "t": "GROUP_AT_MESSAGE_CREATE", "s": 9, "d": {"id": "m2"}}
+            )
+            assert len(adapter._background_tasks) == 1
+            task = next(iter(adapter._background_tasks))
+            await task
+            await asyncio.sleep(0)  # let the done-callback discard the entry
+            assert adapter._background_tasks == set()
+            adapter._on_message.assert_awaited_once_with(
+                "GROUP_AT_MESSAGE_CREATE", {"id": "m2"}
+            )
+
+        asyncio.run(run())
+
 
 # ---------------------------------------------------------------------------
 # READY / RESUMED handling