diff --git a/backend/package/yuxi/channels/plugins/qqbot/_constants.py b/backend/package/yuxi/channels/plugins/qqbot/_constants.py
index 7625432d..034765e9 100644
--- a/backend/package/yuxi/channels/plugins/qqbot/_constants.py
+++ b/backend/package/yuxi/channels/plugins/qqbot/_constants.py
@@ -144,6 +144,19 @@ GROUP_ACTIVE_QPM_PER_GROUP = 20
 
 RECALL_TIME_LIMIT_SECONDS = 120
 
+# ---------------------------------------------------------------------------
+# 频控缓存 key 前缀
+# ---------------------------------------------------------------------------
+
+# 被动消息计数 key 前缀：qqbot:passive:{account_id}:{peer_id}
+PASSIVE_COUNTER_KEY_PREFIX = "qqbot:passive:"
+
+# 主动消息 QPM 计数 key 前缀：qqbot:active_qpm:{account_id}
+ACTIVE_QPM_KEY_PREFIX = "qqbot:active_qpm:"
+
+# 群主动消息 QPM 计数 key 前缀：qqbot:active_qpm_group:{account_id}:{group_openid}
+ACTIVE_QPM_GROUP_KEY_PREFIX = "qqbot:active_qpm_group:"
+
 # ---------------------------------------------------------------------------
 # 互动回调回应超时（秒，QQ Bot 客户端 loading 超时 3s）
 # ---------------------------------------------------------------------------
@@ -189,6 +202,9 @@ __all__ = [
     "GROUP_ACTIVE_QPM_ACCOUNT_UNVERIFIED",
     "GROUP_ACTIVE_QPM_PER_GROUP",
     "RECALL_TIME_LIMIT_SECONDS",
+    "PASSIVE_COUNTER_KEY_PREFIX",
+    "ACTIVE_QPM_KEY_PREFIX",
+    "ACTIVE_QPM_GROUP_KEY_PREFIX",
     "INTERACTION_ACK_TIMEOUT_SECONDS",
     "MAX_MESSAGE_LENGTH",
 ]
diff --git a/backend/package/yuxi/channels/plugins/qqbot/adapters/directory_adapter.py b/backend/package/yuxi/channels/plugins/qqbot/adapters/directory_adapter.py
index c3552831..46f4e94b 100644
--- a/backend/package/yuxi/channels/plugins/qqbot/adapters/directory_adapter.py
+++ b/backend/package/yuxi/channels/plugins/qqbot/adapters/directory_adapter.py
@@ -10,6 +10,7 @@ QQ Bot v1 限制：
 
 from __future__ import annotations
 
+from yuxi.channels.contract.dtos.config import ConfigScope
 from yuxi.channels.contract.dtos.directory import (
     ChannelGroup,
     ChannelUser,
@@ -23,6 +24,7 @@ from yuxi.channels.contract.errors import (
     ValidationError,
 )
 from yuxi.channels.contract.ports.driven.cache_port import CachePort
+from yuxi.channels.contract.ports.driven.config_port import ConfigPort
 
 from .._constants import QQBOT_API_DEP, USER_CACHE_TTL
 from ..qqbot_client import QQBotClient
@@ -37,16 +39,36 @@ class QQBotDirectoryAdapter:
     """QQ Bot 通讯录适配器。
 
     QQ Bot v1 仅支持 ``getUserProfile``（``GET /users/{openid}``），
-    用户信息缓存 1 小时（``USER_CACHE_TTL``）。无 mutable 实例变量（INV-5）。
+    用户信息缓存 TTL 由 ``user_cache_ttl`` 配置项控制（默认 3600s）。
+    ``ChannelUser`` 序列化为 dict 后写入 ``CachePort``，命中时反序列化
+    重建 ``ChannelUser``，避免缓存 dataclass 导致的反序列化兼容性问题。
+    无 mutable 实例变量（INV-5）。
     """
 
     def __init__(
         self,
         client: QQBotClient,
         cache_port: CachePort,
+        config_port: ConfigPort,
     ) -> None:
         self._client = client
         self._cache = cache_port
+        self._config = config_port
+
+    async def _get_user_cache_ttl(self) -> int:
+        """读取 ``user_cache_ttl`` 配置（CHANNEL 作用域）。
+
+        配置值非法（非数值）时回退默认值，避免原生 ``ValueError`` 穿透。
+        """
+        cv = await self._config.get(
+            "user_cache_ttl",
+            scope=ConfigScope.CHANNEL,
+            target="qqbot",
+        )
+        try:
+            return int(cv.value) if cv.value else USER_CACHE_TTL
+        except (TypeError, ValueError):
+            return USER_CACHE_TTL
 
     async def searchUsers(
         self,
@@ -77,8 +99,9 @@ class QQBotDirectoryAdapter:
     ) -> ChannelUser:
         """查询用户详情。
 
-        调用 ``GET /users/{openid}``，缓存 1 小时。QQ Bot 返回用户昵称
-        （``nickname``）与头像 URL（``avatar``）。
+        调用 ``GET /users/{openid}``，缓存 TTL 由 ``user_cache_ttl`` 配置控制。
+        QQ Bot 返回用户昵称（``nickname``）与头像 URL（``avatar``）。
+        ``ChannelUser`` 序列化为 dict 写入缓存，命中时反序列化重建对象。
 
         @failure 用户不存在抛 ``NotFoundError``；依赖故障抛 ``DependencyError``。
         """
@@ -91,7 +114,12 @@ class QQBotDirectoryAdapter:
         cache_key = _user_cache_key(account_id, user_id)
         cached = await self._cache.get(cache_key)
         if cached.is_some():
-            return cached.unwrap()
+            data = cached.unwrap()
+            return ChannelUser(
+                peer_id=str(data.get("peer_id", user_id)),
+                name=data.get("name") or "",
+                avatar_url=data.get("avatar_url") or None,
+            )
 
         try:
             data = await self._client.get_user(account_id, user_id)
@@ -105,7 +133,15 @@ class QQBotDirectoryAdapter:
             name=str(data.get("nickname", "")) or str(data.get("username", "")),
             avatar_url=str(data.get("avatar", "")) or None,
         )
-        await self._cache.set(cache_key, user, ttl_seconds=USER_CACHE_TTL)
+        await self._cache.set(
+            cache_key,
+            {
+                "peer_id": user.peer_id,
+                "name": user.name,
+                "avatar_url": user.avatar_url,
+            },
+            ttl_seconds=await self._get_user_cache_ttl(),
+        )
         return user
 
     async def getGroupMembers(
diff --git a/backend/package/yuxi/channels/plugins/qqbot/adapters/identity_resolver_adapter.py b/backend/package/yuxi/channels/plugins/qqbot/adapters/identity_resolver_adapter.py
index 56509411..a659ce01 100644
--- a/backend/package/yuxi/channels/plugins/qqbot/adapters/identity_resolver_adapter.py
+++ b/backend/package/yuxi/channels/plugins/qqbot/adapters/identity_resolver_adapter.py
@@ -6,16 +6,17 @@
 身份解析策略：
     - 调用 ``GET /users/{openid}`` 获取用户信息
     - 缓存 60 秒（``IDENTITY_CACHE_TTL``）
-    - ``getConfidence``：直接命中 1.0 / 缓存命中 0.9
+    - ``getConfidence``：统一返回 1.0（实际置信度由 ``resolveIdentity``
+      返回值携带，区分直接命中 1.0 / 缓存命中 0.9）
 """
 
 from __future__ import annotations
 
-from datetime import datetime, UTC
+from datetime import UTC, datetime
 from typing import Any
 
-from yuxi.channels.contract.plugin.capability import CapabilityProof
 from yuxi.channels.contract.dtos.channel import ChannelType
+from yuxi.channels.contract.dtos.config import ConfigScope
 from yuxi.channels.contract.dtos.identity import (
     IdentityResolveCmd,
     IdentityResolveResult,
@@ -26,7 +27,9 @@ from yuxi.channels.contract.errors import (
     Error,
     ValidationError,
 )
+from yuxi.channels.contract.plugin.capability import CapabilityProof
 from yuxi.channels.contract.ports.driven.cache_port import CachePort
+from yuxi.channels.contract.ports.driven.config_port import ConfigPort
 from yuxi.channels.contract.ports.driven.logger_port import LoggerPort
 
 from .._constants import (
@@ -45,7 +48,8 @@ def _identity_cache_key(account_id: str, peer_id: str) -> str:
 class QQBotIdentityResolverAdapter:
     """QQ Bot 身份解析适配器。
 
-    通过 ``GET /users/{openid}`` 解析 QQ Bot 用户身份，缓存 60 秒。
+    通过 ``GET /users/{openid}`` 解析 QQ Bot 用户身份，缓存 TTL 由
+    ``identity_cache_ttl`` 配置项控制（默认 60s）。
     继承 ``CapabilityProver`` Protocol，提供 ``proveCapability`` 能力证明。
     无 mutable 实例变量（INV-5）。
     """
@@ -55,10 +59,27 @@ class QQBotIdentityResolverAdapter:
         client: QQBotClient,
         cache_port: CachePort,
         logger_port: LoggerPort,
+        config_port: ConfigPort,
     ) -> None:
         self._client = client
         self._cache = cache_port
         self._logger = logger_port
+        self._config = config_port
+
+    async def _get_identity_cache_ttl(self) -> int:
+        """读取 ``identity_cache_ttl`` 配置（CHANNEL 作用域）。
+
+        配置值非法（非数值）时回退默认值，避免原生 ``ValueError`` 穿透。
+        """
+        cv = await self._config.get(
+            "identity_cache_ttl",
+            scope=ConfigScope.CHANNEL,
+            target="qqbot",
+        )
+        try:
+            return int(cv.value) if cv.value else IDENTITY_CACHE_TTL
+        except (TypeError, ValueError):
+            return IDENTITY_CACHE_TTL
 
     async def resolveIdentity(
         self,
@@ -127,7 +148,7 @@ class QQBotIdentityResolverAdapter:
         await self._cache.set(
             cache_key,
             cache_data,
-            ttl_seconds=IDENTITY_CACHE_TTL,
+            ttl_seconds=await self._get_identity_cache_ttl(),
         )
 
         return result
diff --git a/backend/package/yuxi/channels/plugins/qqbot/adapters/inbound_adapter.py b/backend/package/yuxi/channels/plugins/qqbot/adapters/inbound_adapter.py
index 8383d9c4..8d29ea2e 100644
--- a/backend/package/yuxi/channels/plugins/qqbot/adapters/inbound_adapter.py
+++ b/backend/package/yuxi/channels/plugins/qqbot/adapters/inbound_adapter.py
@@ -26,6 +26,8 @@ from yuxi.channels.contract.dtos.common import (
 from yuxi.channels.contract.dtos.inbound import SignatureVerifyResult
 from yuxi.channels.contract.ports.driven.logger_port import LoggerPort
 
+from ._event_utils import extract_event_data, extract_event_type
+
 # QQ Bot msg_type 常量
 _MSG_TYPE_TEXT = 0
 _MSG_TYPE_MARKDOWN = 2
@@ -36,20 +38,6 @@ _C2C_EVENT_TYPE = "C2C_MESSAGE_CREATE"
 _GROUP_EVENT_TYPE = "GROUP_AT_MESSAGE_CREATE"
 
 
-def _extract_event_data(raw_event: RawEvent) -> dict[str, Any]:
-    """从原始事件提取 ``d`` 字段。"""
-    payload = raw_event.payload
-    data = payload.get("d")
-    if not isinstance(data, dict):
-        return {}
-    return data
-
-
-def _extract_event_type(raw_event: RawEvent) -> str:
-    """提取 QQ Bot 事件类型（``t`` 字段）。"""
-    return str(raw_event.payload.get("t", ""))
-
-
 def _parse_attachments(data: dict[str, Any]) -> tuple[Attachment, ...]:
     """从事件数据提取附件列表。
 
@@ -117,8 +105,8 @@ class QQBotInboundAdapter:
 
         @failure 事件结构不符合 QQ Bot 协议时抛 ``ValidationError``。
         """
-        data = _extract_event_data(raw_event)
-        event_type = _extract_event_type(raw_event)
+        data = extract_event_data(raw_event)
+        event_type = extract_event_type(raw_event)
 
         content_text = str(data.get("content", ""))
         msg_type = data.get("msg_type")
diff --git a/backend/package/yuxi/channels/plugins/qqbot/adapters/lifecycle_adapter.py b/backend/package/yuxi/channels/plugins/qqbot/adapters/lifecycle_adapter.py
index e4fa6f99..bba5ac8e 100644
--- a/backend/package/yuxi/channels/plugins/qqbot/adapters/lifecycle_adapter.py
+++ b/backend/package/yuxi/channels/plugins/qqbot/adapters/lifecycle_adapter.py
@@ -9,12 +9,17 @@ QQ Bot 账户标识为 ``app_id``（QQ 开放平台应用 ID），凭据为
 
 from __future__ import annotations
 
-from typing import Any
+from datetime import UTC, datetime
 
-from yuxi.channels.contract.dtos.channel import ChannelAccount
+from yuxi.channels.contract.dtos.channel import (
+    ChannelAccount,
+    RotateCredentialsResult,
+)
 from yuxi.channels.contract.errors import (
+    AuthError,
     DependencyError,
     Error,
+    LifecycleHookError,
     ValidationError,
 )
 from yuxi.channels.contract.ports.driven.config_port import ConfigPort
@@ -74,8 +79,8 @@ class QQBotLifecycleAdapter:
     ) -> None:
         """账户配置写入后处理。
 
-        预取 access_token 验证凭据有效性，失败时记录 warning 但不抛异常
-        （不阻断配置写入流程，后续消息投递时再次获取 token）。
+        预取 access_token 验证凭据有效性，失败时抛 ``LifecycleHookError``
+        以便编排链路将账户标记为 degraded（符合契约要求）。
         """
         try:
             await self._client.get_app_access_token(
@@ -83,17 +88,15 @@ class QQBotLifecycleAdapter:
                 account.config.get("app_secret", ""),
             )
         except Error as exc:
-            await self._logger.warn(
-                "QQBot token prefetch failed after account config written",
-                account_id=account.account_id,
-                error=str(exc),
-            )
+            raise LifecycleHookError(
+                hook="afterAccountConfigWritten",
+                reason=f"token prefetch failed: {exc}",
+            ) from exc
         except Exception as exc:
-            await self._logger.warn(
-                "QQBot token prefetch unexpected error",
-                account_id=account.account_id,
-                error=str(exc),
-            )
+            raise LifecycleHookError(
+                hook="afterAccountConfigWritten",
+                reason=f"token prefetch unexpected error: {exc}",
+            ) from exc
 
     async def beforeAccountDelete(self, account: ChannelAccount) -> None:
         """账户删除前清理。
@@ -108,7 +111,8 @@ class QQBotLifecycleAdapter:
     async def onAccountEnabled(self, account: ChannelAccount) -> None:
         """账户启用回调。
 
-        预取 token 验证凭据有效性，记录结果。
+        预取 token 验证凭据有效性，失败时抛 ``LifecycleHookError`` 以便
+        编排链路感知账户异常状态（符合契约要求）。
         """
         try:
             await self._client.get_app_access_token(
@@ -120,11 +124,15 @@ class QQBotLifecycleAdapter:
                 account_id=account.account_id,
             )
         except Error as exc:
-            await self._logger.warn(
-                "QQBot account enabled but token prefetch failed",
-                account_id=account.account_id,
-                error=str(exc),
-            )
+            raise LifecycleHookError(
+                hook="onAccountEnabled",
+                reason=f"token prefetch failed: {exc}",
+            ) from exc
+        except Exception as exc:
+            raise LifecycleHookError(
+                hook="onAccountEnabled",
+                reason=f"token prefetch unexpected error: {exc}",
+            ) from exc
 
     async def onAccountDisabled(self, account: ChannelAccount) -> None:
         """账户禁用回调。
@@ -143,7 +151,7 @@ class QQBotLifecycleAdapter:
     async def rotateCredentials(
         self,
         account: ChannelAccount,
-    ) -> dict[str, Any]:
+    ) -> RotateCredentialsResult:
         """轮换凭据。
 
         QQ Bot 凭据轮换为更新 ``app_secret``（``app_id`` 保持不变）。
@@ -162,16 +170,19 @@ class QQBotLifecycleAdapter:
         except Exception as exc:
             raise DependencyError(dep=QQBOT_API_DEP, cause=exc) from exc
 
-        return {
-            "rotated": True,
-            "account_id": account.account_id,
-            "message": "credentials rotated successfully",
-        }
+        return RotateCredentialsResult(
+            account_id=account.account_id,
+            rotated_at=datetime.now(UTC),
+            old_credentials_revoked=False,
+        )
 
     async def validateCredentials(self, config: dict) -> bool:
         """验证凭据有效性。
 
         通过预取 token 验证 ``app_id`` / ``app_secret`` 是否正确。
+        仅 ``AuthError`` / ``ValidationError`` 表示凭据无效返回 ``False``，
+        其他异常（如 ``DependencyError`` 表示网络故障）向上传播，由调用方
+        处理（符合契约 @failure 约束）。
         """
         app_id = config.get("app_id", "")
         app_secret = config.get("app_secret", "")
@@ -179,7 +190,7 @@ class QQBotLifecycleAdapter:
             return False
         try:
             await self._client.get_app_access_token(app_id, app_secret)
-        except Exception:
+        except (AuthError, ValidationError):
             return False
         return True
 
diff --git a/backend/package/yuxi/channels/plugins/qqbot/adapters/mention_adapter.py b/backend/package/yuxi/channels/plugins/qqbot/adapters/mention_adapter.py
index 3fd3b3f0..74a49fda 100644
--- a/backend/package/yuxi/channels/plugins/qqbot/adapters/mention_adapter.py
+++ b/backend/package/yuxi/channels/plugins/qqbot/adapters/mention_adapter.py
@@ -9,28 +9,15 @@ QQ Bot v1 提及语义：
 
 from __future__ import annotations
 
-from typing import Any
-
 from yuxi.channels.contract.dtos.common import RawEvent
 from yuxi.channels.contract.dtos.mention import MentionContext, MentionType
 
+from ._event_utils import extract_event_type
+
 _GROUP_EVENT_TYPE = "GROUP_AT_MESSAGE_CREATE"
 _C2C_EVENT_TYPE = "C2C_MESSAGE_CREATE"
 
 
-def _extract_event_type(raw_event: RawEvent) -> str:
-    """提取 QQ Bot 事件类型。"""
-    return str(raw_event.payload.get("t", ""))
-
-
-def _extract_event_data(raw_event: RawEvent) -> dict[str, Any]:
-    """从原始事件提取 ``d`` 字段。"""
-    data = raw_event.payload.get("d")
-    if not isinstance(data, dict):
-        return {}
-    return data
-
-
 class QQBotMentionAdapter:
     """QQ Bot 提及适配器。
 
@@ -45,7 +32,7 @@ class QQBotMentionAdapter:
         - ``C2C_MESSAGE_CREATE`` → ``NONE``（单聊无需提及）
         - 其他事件 → ``NONE``
         """
-        event_type = _extract_event_type(raw_event)
+        event_type = extract_event_type(raw_event)
         if event_type == _GROUP_EVENT_TYPE:
             return MentionContext(type=MentionType.EXPLICIT)
         return MentionContext(type=MentionType.NONE)
diff --git a/backend/package/yuxi/channels/plugins/qqbot/adapters/message_ops_adapter.py b/backend/package/yuxi/channels/plugins/qqbot/adapters/message_ops_adapter.py
index d1b3439f..3dc2f823 100644
--- a/backend/package/yuxi/channels/plugins/qqbot/adapters/message_ops_adapter.py
+++ b/backend/package/yuxi/channels/plugins/qqbot/adapters/message_ops_adapter.py
@@ -16,9 +16,10 @@ QQ Bot v1 支持的消息操作：
 
 from __future__ import annotations
 
-from datetime import datetime, UTC
+from datetime import UTC, datetime
 from typing import Any
 
+from yuxi.channels.contract.dtos.channel import Message
 from yuxi.channels.contract.dtos.common import MessageContent, MessageFormat
 from yuxi.channels.contract.dtos.message_ops import (
     MessageOperation,
@@ -179,7 +180,7 @@ class QQBotMessageOpsAdapter:
             raise DependencyError(dep=QQBOT_API_DEP, cause=exc) from exc
         return True
 
-    async def getMessage(self, channel_msg_id: str) -> None:
+    async def getMessage(self, channel_msg_id: str) -> Message | None:
         """获取消息。
 
         QQ Bot v1 不提供消息查询 API，始终返回 ``None``。
diff --git a/backend/package/yuxi/channels/plugins/qqbot/adapters/outbound_adapter.py b/backend/package/yuxi/channels/plugins/qqbot/adapters/outbound_adapter.py
index 05e68bde..babbcf00 100644
--- a/backend/package/yuxi/channels/plugins/qqbot/adapters/outbound_adapter.py
+++ b/backend/package/yuxi/channels/plugins/qqbot/adapters/outbound_adapter.py
@@ -9,6 +9,11 @@ QQ Bot msg_type 映射：
     - MARKDOWN → msg_type=2（含 markdown + keyboard）
     - RICH → msg_type=2（rich_message 已由 RichMessageAdapter 渲染）
 
+被动消息频控（QQ Bot 平台硬性限制）：
+    - C2C 被动消息：60 分钟内最多回复 4 条（``passive_reply_limit_c2c``）
+    - Group 被动消息：5 分钟内最多回复 5 条（``passive_reply_limit_group``）
+    - 主动消息不受被动消息频控限制，但受 QPM 限制
+
 依赖方向：仅 import ``yuxi.channels.contract.*`` 与同插件内部模块
 （``qqbot_client``），不污染框架层。
 """
@@ -19,17 +24,28 @@ import time
 from typing import Any
 
 from yuxi.channels.contract.dtos.common import MessageContent, MessageFormat
+from yuxi.channels.contract.dtos.config import ConfigScope
 from yuxi.channels.contract.dtos.outbound import FormattedMessage
 from yuxi.channels.contract.dtos.outbox import MultiPartReceipt
 from yuxi.channels.contract.errors import (
     DependencyError,
     Error,
     NotFoundError,
+    RateLimitError,
 )
 from yuxi.channels.contract.ports.driven.cache_port import CachePort
+from yuxi.channels.contract.ports.driven.config_port import ConfigPort
 from yuxi.channels.contract.ports.driven.logger_port import LoggerPort
 
-from .._constants import MSG_CACHE_TTL_SECONDS, QQBOT_API_DEP
+from .._constants import (
+    MSG_CACHE_TTL_SECONDS,
+    PASSIVE_COUNTER_KEY_PREFIX,
+    PASSIVE_REPLY_LIMIT_C2C,
+    PASSIVE_REPLY_LIMIT_GROUP,
+    PASSIVE_REPLY_TTL_C2C_MS,
+    PASSIVE_REPLY_TTL_GROUP_MS,
+    QQBOT_API_DEP,
+)
 from ..qqbot_client import QQBotClient
 
 # QQ Bot msg_type 常量
@@ -43,9 +59,9 @@ def _msg_key(channel_msg_id: str) -> str:
     return f"qqbot:msg:{channel_msg_id}"
 
 
-def _peer_key(account_id: str, peer_id: str) -> str:
-    """构造 ``(account_id, peer_id)`` → 被动消息频控计数的缓存 key。"""
-    return f"qqbot:passive:{account_id}:{peer_id}"
+def _passive_counter_key(account_id: str, peer_id: str) -> str:
+    """构造被动消息频控计数的缓存 key。"""
+    return f"{PASSIVE_COUNTER_KEY_PREFIX}{account_id}:{peer_id}"
 
 
 def _build_send_content(payload: FormattedMessage) -> tuple[int, dict[str, Any]]:
@@ -74,21 +90,13 @@ def _build_send_content(payload: FormattedMessage) -> tuple[int, dict[str, Any]]
     return _MSG_TYPE_TEXT, {"content": payload.content}
 
 
-def _infer_chat_type(peer_id: str) -> str:
-    """QQ Bot 无法从 peer_id 推断会话类型，需通过反向映射缓存查询。
-
-    本函数仅用于无缓存时的默认值，返回 ``"c2c"``。
-    实际会话类型由 ``sendMessage`` 写入缓存时记录。
-    """
-    return "c2c"
-
-
 class QQBotOutboundAdapter:
     """QQ Bot 出站适配器。
 
     通过 ``QQBotClient`` 投递消息至 QQ Bot 侧，支持多部分消息投递
     （FR-22）与消息更新。``account_id`` 反向映射通过 ``CachePort`` 存储，
-    供 ``updateMessage`` 与 ``MessageOpsAdapter`` 共享。
+    供 ``updateMessage`` 与 ``MessageOpsAdapter`` 共享。被动消息频控
+    通过 ``CachePort`` 的计数器实现，超限时抛 ``RateLimitError``。
     无 mutable 实例变量（INV-5）。
     """
 
@@ -97,10 +105,12 @@ class QQBotOutboundAdapter:
         client: QQBotClient,
         logger_port: LoggerPort,
         cache_port: CachePort,
+        config_port: ConfigPort,
     ) -> None:
         self._client = client
         self._logger = logger_port
         self._cache = cache_port
+        self._config = config_port
 
     def supportsOutbound(self) -> bool:
         """QQ Bot 支持出站消息投递。"""
@@ -113,6 +123,101 @@ class QQBotOutboundAdapter:
         """
         return True
 
+    async def _get_passive_freq_config(self, is_group: bool) -> tuple[int, int]:
+        """读取被动消息频控配置 ``(limit, ttl_ms)``。
+
+        - C2C: ``passive_reply_limit_c2c`` / ``passive_reply_ttl_c2c_ms``
+        - Group: ``passive_reply_limit_group`` / ``passive_reply_ttl_group_ms``
+
+        配置值非法（非数值）时回退默认值，避免原生 ``ValueError`` 穿透。
+        """
+        if is_group:
+            limit_cv = await self._config.get(
+                "passive_reply_limit_group",
+                scope=ConfigScope.CHANNEL,
+                target="qqbot",
+            )
+            ttl_cv = await self._config.get(
+                "passive_reply_ttl_group_ms",
+                scope=ConfigScope.CHANNEL,
+                target="qqbot",
+            )
+            try:
+                limit = int(limit_cv.value) if limit_cv.value else PASSIVE_REPLY_LIMIT_GROUP
+            except (TypeError, ValueError):
+                limit = PASSIVE_REPLY_LIMIT_GROUP
+            try:
+                ttl_ms = int(ttl_cv.value) if ttl_cv.value else PASSIVE_REPLY_TTL_GROUP_MS
+            except (TypeError, ValueError):
+                ttl_ms = PASSIVE_REPLY_TTL_GROUP_MS
+        else:
+            limit_cv = await self._config.get(
+                "passive_reply_limit_c2c",
+                scope=ConfigScope.CHANNEL,
+                target="qqbot",
+            )
+            ttl_cv = await self._config.get(
+                "passive_reply_ttl_c2c_ms",
+                scope=ConfigScope.CHANNEL,
+                target="qqbot",
+            )
+            try:
+                limit = int(limit_cv.value) if limit_cv.value else PASSIVE_REPLY_LIMIT_C2C
+            except (TypeError, ValueError):
+                limit = PASSIVE_REPLY_LIMIT_C2C
+            try:
+                ttl_ms = int(ttl_cv.value) if ttl_cv.value else PASSIVE_REPLY_TTL_C2C_MS
+            except (TypeError, ValueError):
+                ttl_ms = PASSIVE_REPLY_TTL_C2C_MS
+        return limit, ttl_ms
+
+    async def _check_passive_freq(
+        self,
+        account_id: str,
+        peer_id: str,
+        is_group: bool,
+        msg_id: str | None,
+    ) -> None:
+        """检查被动消息频控。
+
+        QQ Bot 平台限制：被动消息（携带 ``msg_id`` 的回复）在固定时间窗口内
+        有次数上限。超限时抛 ``RateLimitError``。无 ``msg_id`` 的主动消息
+        不受此限制。
+
+        频控计数通过 ``CachePort.set`` 实现，并通过 ``acquireAdvisoryLock``
+        串行化读-改-写操作，避免并发请求竞态导致超频未触发。
+        """
+        if not msg_id:
+            # 主动消息不受被动消息频控限制
+            return
+
+        limit, ttl_ms = await self._get_passive_freq_config(is_group)
+        counter_key = _passive_counter_key(account_id, peer_id)
+        ttl_seconds = max(ttl_ms // 1000, 1)
+        lock_key = f"{PASSIVE_COUNTER_KEY_PREFIX}lock:{account_id}:{peer_id}"
+
+        lock = await self._cache.acquireAdvisoryLock(lock_key, ttl_seconds=10)
+        try:
+            cached = await self._cache.get(counter_key)
+            if cached.is_nothing():
+                await self._cache.set(counter_key, 1, ttl_seconds=ttl_seconds)
+            else:
+                count = cached.unwrap()
+                if not isinstance(count, int):
+                    try:
+                        count = int(count) if count else 0
+                    except (TypeError, ValueError):
+                        count = 0
+                if count >= limit:
+                    raise RateLimitError(
+                        resource=QQBOT_API_DEP,
+                        retry_after_ms=ttl_ms,
+                    )
+                await self._cache.set(counter_key, count + 1, ttl_seconds=ttl_seconds)
+        finally:
+            if lock is not None:
+                await self._cache.releaseAdvisoryLock(lock)
+
     async def sendMessage(
         self,
         account_id: str,
@@ -125,19 +230,24 @@ class QQBotOutboundAdapter:
         """发送消息，返回 QQ Bot 消息 ID。
 
         QQ Bot 被动消息需携带 ``msg_id``（入站消息 ID）与 ``msg_seq``
-        以满足频控要求。``msg_id`` 从 ``payload.rich_message`` 或
-        ``idempotency_key`` 提取。
+        以满足频控要求。``msg_id`` 从 ``payload.metadata`` 的 ``msg_id``
+        字段提取（由出站管道从入站事件注入）。主动消息（无 ``msg_id``）
+        不携带被动消息频控参数。
 
         @failure 依赖故障抛 ``DependencyError``；参数非法抛
-            ``ValidationError``。
+            ``ValidationError``；被动消息超频抛 ``RateLimitError``。
         """
         msg_type, content = _build_send_content(payload)
-        # 被动消息需 msg_id + msg_seq；msg_id 从 metadata 或 idempotency_key 提取
-        msg_id = _extract_msg_id(payload, idempotency_key)
+        # msg_id 从 payload.metadata 提取（由出站管道从入站事件注入）
+        msg_id = _extract_msg_id(payload)
         msg_seq = _generate_msg_seq()
 
+        is_group = _is_group_peer(payload)
+        # 被动消息频控检查
+        await self._check_passive_freq(account_id, peer_id, is_group, msg_id)
+
         try:
-            if _is_group_peer(payload):
+            if is_group:
                 resp = await self._client.send_group_message(
                     account_id,
                     peer_id,
@@ -168,7 +278,7 @@ class QQBotOutboundAdapter:
             )
 
         # 写入反向映射：(account_id, peer_id, chat_type)
-        chat_type = "group" if _is_group_peer(payload) else "c2c"
+        chat_type = "group" if is_group else "c2c"
         await self._cache.set(
             _msg_key(message_id),
             (account_id, peer_id, chat_type),
@@ -340,17 +450,19 @@ class QQBotOutboundAdapter:
 # ----------------------------------------------------------------------
 
 
-def _extract_msg_id(
-    payload: FormattedMessage,
-    idempotency_key: str | None,
-) -> str | None:
-    """从 payload 或 idempotency_key 提取被动消息 msg_id。
+def _extract_msg_id(payload: FormattedMessage) -> str | None:
+    """从 ``payload.rich_message.reply_to_msg_id`` 提取被动消息 msg_id。
 
     被动消息需携带入站消息 ID（msg_id）以满足 QQ Bot 频控要求。
-    idempotency_key 在出站管道中由 reply 阶段设置为入站 msg_id。
+    msg_id 从 ``payload.rich_message`` 的 ``reply_to_msg_id`` 字段提取
+    （由出站管道从入站事件注入到 rich_message）。主动消息无此字段时
+    返回 ``None``。
     """
-    if idempotency_key:
-        return idempotency_key
+    rich = payload.rich_message
+    if isinstance(rich, dict):
+        msg_id = rich.get("reply_to_msg_id")
+        if msg_id:
+            return str(msg_id)
     return None
 
 
diff --git a/backend/package/yuxi/channels/plugins/qqbot/adapters/probeable_adapter.py b/backend/package/yuxi/channels/plugins/qqbot/adapters/probeable_adapter.py
index 05ff68e4..ce4eb6bc 100644
--- a/backend/package/yuxi/channels/plugins/qqbot/adapters/probeable_adapter.py
+++ b/backend/package/yuxi/channels/plugins/qqbot/adapters/probeable_adapter.py
@@ -10,11 +10,12 @@ from __future__ import annotations
 
 import time
 
+from yuxi.channels.contract.dtos.channel import ChannelType
 from yuxi.channels.contract.dtos.health import AdapterProbeOutcome
-from yuxi.channels.contract.errors import Error
 from yuxi.channels.contract.ports.driven.logger_port import LoggerPort
+from yuxi.channels.contract.ports.driven.persistence_port import PersistencePort
 
-from .._constants import QQBOT_API_DEP
+from .._constants import CHANNEL_TARGET, QQBOT_API_DEP
 from ..qqbot_client import QQBotClient
 
 
@@ -22,33 +23,43 @@ class QQBotProbeableAdapter:
     """QQ Bot 可探测适配器。
 
     通过 ``QQBotClient.get_gateway`` 主动调用 QQ Bot API 探测可用性。
-    无 mutable 实例变量（INV-5）。
+    ``probe()`` 为无参方法，通过 ``PersistencePort`` 列出首个 qqbot 账户
+    进行探测。无 mutable 实例变量（INV-5）。
     """
 
     def __init__(
         self,
         client: QQBotClient,
         logger_port: LoggerPort,
+        persistence_port: PersistencePort,
     ) -> None:
         self._client = client
         self._logger = logger_port
+        self._persistence = persistence_port
 
     async def probe(self) -> AdapterProbeOutcome:
         """主动探测 QQ Bot API 可用性。
 
-        调用 ``GET /gateway`` 验证 API 连通性与 token 有效性。探测失败
-        时返回 ``is_available=False``，**不抛异常**（由 ``ChannelProbe``
-        统一映射）。
+        通过 ``PersistencePort`` 列出首个 qqbot 账户，调用
+        ``QQBotClient.get_gateway`` 验证 API 连通性与 token 有效性。
+        探测失败时返回 ``is_available=False``，**不抛异常**（由
+        ``ChannelProbe`` 统一映射）。
         """
         start = time.monotonic()
         try:
-            # account_id 为空：probe 为无参方法，无法指定具体账户。
-            # QQBotClient.get_gateway 内部通过 _build_auth_header 读取
-            # 账户配置，此处传入空字符串会在无账户配置时抛 DependencyError。
-            # 实际部署中 probe 由 ChannelProbe 按账户调用（通过
-            # probeDeep），此处仅作基础连通性探测占位。
-            await self._client.get_gateway("")
-        except Error as exc:
+            accounts = await self._persistence.listChannelAccounts(
+                ChannelType(CHANNEL_TARGET),
+                limit=1,
+            )
+            if not accounts:
+                return AdapterProbeOutcome(
+                    is_available=False,
+                    latency_ms=int((time.monotonic() - start) * 1000),
+                    reason=f"{QQBOT_API_DEP} probe failed: no qqbot account configured",
+                )
+            account_id = accounts[0].account_id
+            await self._client.get_gateway(account_id)
+        except Exception as exc:
             latency_ms = int((time.monotonic() - start) * 1000)
             await self._logger.warn(
                 "QQBot probe failed",
@@ -60,18 +71,6 @@ class QQBotProbeableAdapter:
                 latency_ms=latency_ms,
                 reason=f"{QQBOT_API_DEP} probe failed: {exc}",
             )
-        except Exception as exc:
-            latency_ms = int((time.monotonic() - start) * 1000)
-            await self._logger.warn(
-                "QQBot probe unexpected error",
-                latency_ms=latency_ms,
-                error=str(exc),
-            )
-            return AdapterProbeOutcome(
-                is_available=False,
-                latency_ms=latency_ms,
-                reason=f"{QQBOT_API_DEP} probe unexpected error: {exc}",
-            )
 
         latency_ms = int((time.monotonic() - start) * 1000)
         return AdapterProbeOutcome(
diff --git a/backend/package/yuxi/channels/plugins/qqbot/adapters/session_adapter.py b/backend/package/yuxi/channels/plugins/qqbot/adapters/session_adapter.py
index f2a435e8..c82daad1 100644
--- a/backend/package/yuxi/channels/plugins/qqbot/adapters/session_adapter.py
+++ b/backend/package/yuxi/channels/plugins/qqbot/adapters/session_adapter.py
@@ -20,20 +20,21 @@ from yuxi.channels.contract.dtos.common import RawEvent
 from yuxi.channels.contract.errors import ValidationError
 
 from .._constants import CHANNEL_TARGET
+from ._event_utils import extract_event_data, extract_event_type
 
 # QQ Bot 事件类型 → 会话场景映射
 _C2C_EVENT_TYPE = "C2C_MESSAGE_CREATE"
 _GROUP_EVENT_TYPE = "GROUP_AT_MESSAGE_CREATE"
 
 
-def _extract_event_data(raw_event: RawEvent) -> dict[str, Any]:
-    """从原始事件提取 ``d`` 字段（QQ Bot WS 事件数据载荷）。
+def _require_event_data(raw_event: RawEvent) -> dict[str, Any]:
+    """提取事件数据，缺失时抛 ``ValidationError``。
 
-    @failure 当事件结构不符合 QQ Bot 协议时抛 ``ValidationError``。
+    会话适配器对事件数据有强依赖（解析 peer_id / group_id），缺失时
+    属于协议违规，需 fail-fast。
     """
-    payload = raw_event.payload
-    data = payload.get("d")
-    if not isinstance(data, dict):
+    data = extract_event_data(raw_event)
+    if not data:
         raise ValidationError(
             field="d",
             message="qqbot event missing data payload",
@@ -41,11 +42,6 @@ def _extract_event_data(raw_event: RawEvent) -> dict[str, Any]:
     return data
 
 
-def _extract_event_type(raw_event: RawEvent) -> str:
-    """提取 QQ Bot 事件类型（``t`` 字段）。"""
-    return str(raw_event.payload.get("t", ""))
-
-
 class QQBotSessionAdapter:
     """QQ Bot 会话适配器。
 
@@ -57,8 +53,8 @@ class QQBotSessionAdapter:
 
         @failure 事件类型未知或数据缺失时抛 ``ValidationError``。
         """
-        event_type = _extract_event_type(raw_event)
-        data = _extract_event_data(raw_event)
+        event_type = extract_event_type(raw_event)
+        data = _require_event_data(raw_event)
         account_id = raw_event.account_id or ""
 
         if event_type == _C2C_EVENT_TYPE:
@@ -109,9 +105,9 @@ class QQBotSessionAdapter:
         - C2C 场景：返回 ``author.openid``
         - Group 场景：返回 ``author.member_openid``
         """
-        data = _extract_event_data(raw_event)
+        data = _require_event_data(raw_event)
         author = data.get("author", {})
-        event_type = _extract_event_type(raw_event)
+        event_type = extract_event_type(raw_event)
 
         if event_type == _C2C_EVENT_TYPE:
             peer_id = author.get("openid", "")
@@ -132,7 +128,7 @@ class QQBotSessionAdapter:
 
     async def getChatType(self, raw_event: RawEvent) -> str:
         """提取会话类型（p2p/group）。"""
-        event_type = _extract_event_type(raw_event)
+        event_type = extract_event_type(raw_event)
         if event_type == _C2C_EVENT_TYPE:
             return "p2p"
         if event_type == _GROUP_EVENT_TYPE:
diff --git a/backend/package/yuxi/channels/plugins/qqbot/adapters/status_adapter.py b/backend/package/yuxi/channels/plugins/qqbot/adapters/status_adapter.py
index 4e4175a0..d397504f 100644
--- a/backend/package/yuxi/channels/plugins/qqbot/adapters/status_adapter.py
+++ b/backend/package/yuxi/channels/plugins/qqbot/adapters/status_adapter.py
@@ -8,13 +8,15 @@ QQ Bot v1 仅处理 MESSAGE 事件，状态回传事件（DELIVERED/READ 等）
 
 from __future__ import annotations
 
-from datetime import datetime, UTC
+from datetime import UTC, datetime
 from typing import Any
 
 from yuxi.channels.contract.dtos.common import RawEvent
 from yuxi.channels.contract.dtos.status import EventType, StatusPayload
 from yuxi.channels.contract.errors import ValidationError
 
+from ._event_utils import extract_event_data, extract_event_type
+
 # QQ Bot 消息事件类型集合
 _MESSAGE_EVENT_TYPES = frozenset(
     {
@@ -24,16 +26,14 @@ _MESSAGE_EVENT_TYPES = frozenset(
 )
 
 
-def _extract_event_type(raw_event: RawEvent) -> str:
-    """提取 QQ Bot 事件类型（``t`` 字段）。"""
-    return str(raw_event.payload.get("t", ""))
+def _require_event_data(raw_event: RawEvent) -> dict[str, Any]:
+    """提取事件数据，缺失时抛 ``ValidationError``。
 
-
-def _extract_event_data(raw_event: RawEvent) -> dict[str, Any]:
-    """从原始事件提取 ``d`` 字段。"""
-    payload = raw_event.payload
-    data = payload.get("d")
-    if not isinstance(data, dict):
+    状态负载提取对事件数据有强依赖（解析 id / timestamp），缺失时
+    属于协议违规，需 fail-fast。
+    """
+    data = extract_event_data(raw_event)
+    if not data:
         raise ValidationError(
             field="d",
             message="qqbot event missing data payload",
@@ -54,7 +54,7 @@ class QQBotStatusAdapter:
         - ``C2C_MESSAGE_CREATE`` / ``GROUP_AT_MESSAGE_CREATE`` → ``MESSAGE``
         - 其他事件 → ``UNKNOWN``
         """
-        event_type = _extract_event_type(raw_event)
+        event_type = extract_event_type(raw_event)
         if event_type in _MESSAGE_EVENT_TYPES:
             return EventType.MESSAGE
         return EventType.UNKNOWN
@@ -66,7 +66,7 @@ class QQBotStatusAdapter:
         场景下被调用。当前实现返回基于事件数据的 ``UNKNOWN`` 状态负载，
         ``ref_channel_msg_id`` 取 ``d.id``，时间戳取 ``d.timestamp`` 或当前时间。
         """
-        data = _extract_event_data(raw_event)
+        data = _require_event_data(raw_event)
         ref_id = str(data.get("id", ""))
         if not ref_id:
             raise ValidationError(
diff --git a/backend/package/yuxi/channels/plugins/qqbot/adapters/stream_connector_adapter.py b/backend/package/yuxi/channels/plugins/qqbot/adapters/stream_connector_adapter.py
index d279a636..c6074711 100644
--- a/backend/package/yuxi/channels/plugins/qqbot/adapters/stream_connector_adapter.py
+++ b/backend/package/yuxi/channels/plugins/qqbot/adapters/stream_connector_adapter.py
@@ -7,11 +7,12 @@ QQ Bot WebSocket 协议（设计文档 §1.3）：
     1. ``GET /gateway`` 获取 WSS URL
     2. 建立 WS 连接，接收 Opcode 10 Hello（含 heartbeat_interval）
     3. 发送 Opcode 2 Identify（携带 token + intents）
-    4. 接收 Opcode 0 Ready（连接就绪）
-    5. 定期发送 Opcode 1 Heartbeat
-    6. 接收 Opcode 11 Heartbeat ACK
-    7. 接收 Opcode 0 Dispatch（业务事件）
-    8. 接收 Opcode 7 Reconnect（服务端要求重连）
+    4. 接收 Opcode 0 Ready（连接就绪，缓存 session_id）
+    5. 定期发送 Opcode 1 Heartbeat（``d`` 为最后接收的序列号或 ``null``）
+    6. 接收 Opcode 11 Heartbeat ACK（重置卡死检测计时器）
+    7. 接收 Opcode 0 Dispatch（业务事件，更新序列号）
+    8. 接收 Opcode 7 Reconnect（服务端要求重连，触发 Resume）
+    9. 接收 Opcode 9 Invalid Session（需重新 Identify）
 
 本适配器仅实现 ``connect`` / ``getStreamConfig``，``receive`` / ``send``
 / ``close`` 通过 ``StreamConnection`` 回调返回，由 ``StreamWorker`` 调用。
@@ -26,6 +27,7 @@ from typing import Any
 
 import aiohttp
 
+from yuxi.channels.contract.dtos.config import ConfigScope
 from yuxi.channels.contract.dtos.transport import StreamConfig, StreamConnection
 from yuxi.channels.contract.errors import (
     AuthError,
@@ -40,8 +42,9 @@ from .._constants import (
     DEFAULT_INTENTS,
     QQBOT_API_DEP,
     WS_HEARTBEAT_INTERVAL_MS,
+    WS_STALL_TIMEOUT_MS,
 )
-
+from ..qqbot_client import QQBotClient
 
 # QQ Bot WebSocket Opcode 常量
 _OP_DISPATCH = 0
@@ -67,9 +70,11 @@ class QQBotStreamConnectorAdapter:
         self,
         config_port: ConfigPort,
         logger_port: LoggerPort,
+        client: QQBotClient,
     ) -> None:
         self._config = config_port
         self._logger = logger_port
+        self._client = client
 
     async def getStreamConfig(self) -> StreamConfig:
         """返回流连接配置。
@@ -77,8 +82,6 @@ class QQBotStreamConnectorAdapter:
         心跳间隔从 ``ConfigPort`` 读取（``ws_heartbeat_interval_ms``），
         缺省回退 ``WS_HEARTBEAT_INTERVAL_MS``。
         """
-        from yuxi.channels.contract.dtos.config import ConfigScope
-
         cv = await self._config.get(
             "ws_heartbeat_interval_ms",
             scope=ConfigScope.CHANNEL,
@@ -98,8 +101,8 @@ class QQBotStreamConnectorAdapter:
             2. 建立 WS 连接
             3. 接收 Opcode 10 Hello，提取 heartbeat_interval
             4. 发送 Opcode 2 Identify（携带 token + intents）
-            5. 接收 Opcode 0 Ready（连接就绪）
-            6. 启动心跳协程
+            5. 接收 Opcode 0 Ready（连接就绪，缓存 session_id 与 sequence）
+            6. 启动心跳协程（携带最后序列号，含卡死检测）
             7. 返回 ``StreamConnection`` 回调对象
 
         @failure 鉴权失败抛 ``AuthError``；依赖故障抛 ``DependencyError``；
@@ -133,6 +136,11 @@ class QQBotStreamConnectorAdapter:
             await session.close()
             raise DependencyError(dep=QQBOT_API_DEP, cause=exc) from exc
 
+        # 连接状态闭包变量
+        session_id: str | None = None
+        last_sequence: int | None = None
+        last_ack_time: float = time.monotonic()
+
         try:
             # 3. 接收 Hello
             hello_msg = await ws.receive(timeout=10)
@@ -169,23 +177,42 @@ class QQBotStreamConnectorAdapter:
                 )
             ready_data = json.loads(ready_msg.data)
             if ready_data.get("op") == _OP_INVALID_SESSION:
+                await ws.close()
                 await session.close()
                 raise AuthError(
                     reason="qqbot identify failed: invalid session",
                 )
             if ready_data.get("op") != _OP_DISPATCH:
+                await ws.close()
                 await session.close()
                 raise DependencyError(
                     dep=QQBOT_API_DEP,
                     cause=ValueError(f"expected op=0 dispatch (ready), got op={ready_data.get('op')}"),
                 )
+            # 缓存 session_id 与初始 sequence 供 Resume 使用
+            ready_d = ready_data.get("d", {})
+            session_id = ready_d.get("session_id")
+            if "s" in ready_data and ready_data["s"] is not None:
+                last_sequence = ready_data["s"]
 
             # 6. 启动心跳协程
-            heartbeat_task = asyncio.create_task(self._heartbeat_loop(ws, heartbeat_interval, account_id))
+            stall_timeout_ms = await self._get_stall_timeout_ms()
+            heartbeat_task = asyncio.create_task(
+                self._heartbeat_loop(
+                    ws,
+                    heartbeat_interval,
+                    account_id,
+                    lambda: last_sequence,
+                    stall_timeout_ms,
+                    lambda: last_ack_time,
+                )
+            )
         except Error:
+            await ws.close()
             await session.close()
             raise
         except Exception as exc:
+            await ws.close()
             await session.close()
             raise DependencyError(dep=QQBOT_API_DEP, cause=exc) from exc
 
@@ -197,21 +224,49 @@ class QQBotStreamConnectorAdapter:
 
         # 7. 返回 StreamConnection 回调对象
         async def receive() -> dict[str, Any]:
-            """接收下一条 WS 消息。"""
-            msg = await ws.receive()
-            if msg.type == aiohttp.WSMsgType.TEXT:
-                return json.loads(msg.data)
-            if msg.type == aiohttp.WSMsgType.CLOSED:
-                raise DependencyError(
-                    dep=QQBOT_API_DEP,
-                    cause=ConnectionError("ws connection closed"),
-                )
-            if msg.type == aiohttp.WSMsgType.ERROR:
-                raise DependencyError(
-                    dep=QQBOT_API_DEP,
-                    cause=ws.exception() or ConnectionError("ws error"),
-                )
-            return {}
+            """接收下一条 WS 消息。
+
+            处理 Opcode 7 Reconnect（抛异常触发 StreamWorker 重连）与
+            Opcode 9 Invalid Session（抛异常触发重新 Identify）。
+            Opcode 0 Dispatch 更新 last_sequence；Opcode 11 Heartbeat ACK
+            更新 last_ack_time（供心跳协程卡死检测）。
+            """
+            nonlocal last_sequence, last_ack_time
+            while True:
+                msg = await ws.receive()
+                if msg.type == aiohttp.WSMsgType.TEXT:
+                    data = json.loads(msg.data)
+                    op = data.get("op")
+                    if op == _OP_RECONNECT:
+                        # 服务端要求重连：抛异常触发 StreamWorker 重连
+                        raise DependencyError(
+                            dep=QQBOT_API_DEP,
+                            cause=ConnectionError("qqbot server requested reconnect (op=7)"),
+                        )
+                    if op == _OP_INVALID_SESSION:
+                        # Invalid Session：需重新 Identify
+                        raise AuthError(
+                            reason="qqbot invalid session (op=9), re-identify required",
+                        )
+                    # Heartbeat ACK：更新卡死检测计时器，继续接收下一条
+                    if op == _OP_HEARTBEAT_ACK:
+                        last_ack_time = time.monotonic()
+                        continue
+                    # 更新最后接收的序列号
+                    if "s" in data and data["s"] is not None:
+                        last_sequence = data["s"]
+                    return data
+                if msg.type == aiohttp.WSMsgType.CLOSED:
+                    raise DependencyError(
+                        dep=QQBOT_API_DEP,
+                        cause=ConnectionError("ws connection closed"),
+                    )
+                if msg.type == aiohttp.WSMsgType.ERROR:
+                    raise DependencyError(
+                        dep=QQBOT_API_DEP,
+                        cause=ws.exception() or ConnectionError("ws error"),
+                    )
+                # 非 TEXT/CLOSED/ERROR 类型（BINARY/PING/PONG/CLOSING/CLOSE）继续接收
 
         async def send(data: dict[str, Any]) -> None:
             """发送 WS 消息。"""
@@ -235,7 +290,11 @@ class QQBotStreamConnectorAdapter:
             receive=receive,
             send=send,
             close=close,
-            extra={"account_id": account_id, "gateway_url": gateway_url},
+            extra={
+                "account_id": account_id,
+                "gateway_url": gateway_url,
+                "session_id": session_id,
+            },
         )
 
     # ------------------------------------------------------------------
@@ -246,10 +305,9 @@ class QQBotStreamConnectorAdapter:
         """获取 Gateway URL。
 
         优先从 ConfigPort 读取 ``gateway_url``（支持私有化部署固定配置），
-        缺省调用 ``QQBotClient.get_gateway`` 动态获取。
+        缺省调用 ``QQBotClient.get_gateway`` 动态获取。获取失败时抛异常，
+        由调用方 ``connect`` 统一处理。
         """
-        from yuxi.channels.contract.dtos.config import ConfigScope
-
         cv = await self._config.get(
             "gateway_url",
             scope=ConfigScope.CHANNEL,
@@ -259,14 +317,11 @@ class QQBotStreamConnectorAdapter:
         if gateway_url:
             return str(gateway_url)
 
-        # 动态获取：需通过 QQBotClient，但本适配器未注入 client（避免循环依赖）
-        # 实际部署中 gateway_url 由 LifecycleAdapter 预取并写入 ConfigPort
-        return gateway_url or ""
+        # 动态获取 Gateway URL，失败时抛异常由 connect 统一处理
+        return await self._client.get_gateway(account_id)
 
     async def _get_intents(self) -> int:
         """从 ConfigPort 读取 Intents 配置。"""
-        from yuxi.channels.contract.dtos.config import ConfigScope
-
         cv = await self._config.get(
             "intents",
             scope=ConfigScope.CHANNEL,
@@ -282,24 +337,45 @@ class QQBotStreamConnectorAdapter:
         except (ValueError, TypeError):
             return DEFAULT_INTENTS
 
+    async def _get_stall_timeout_ms(self) -> int:
+        """从 ConfigPort 读取 WS 卡死检测超时。"""
+        cv = await self._config.get(
+            "ws_stall_timeout_ms",
+            scope=ConfigScope.CHANNEL,
+            target="qqbot",
+        )
+        if cv.value:
+            try:
+                return int(cv.value)
+            except (TypeError, ValueError):
+                pass
+        return WS_STALL_TIMEOUT_MS
+
     async def _heartbeat_loop(
         self,
         ws: aiohttp.ClientWebSocketResponse,
         interval_ms: int,
         account_id: str,
+        get_last_sequence: callable,
+        stall_timeout_ms: int,
+        get_last_ack_time: callable,
     ) -> None:
         """心跳协程：定期发送 Opcode 1 Heartbeat。
 
-        QQ Bot 心跳间隔由 Hello 消息携带（毫秒），本协程按间隔发送
-        心跳包。连接关闭时协程自动退出（``ws.send_json`` 抛异常）。
+        QQ Bot 心跳间隔由 Hello 消息携带（毫秒），``d`` 字段为最后接收的
+        Dispatch 事件序列号（``s``），未收到事件时为 ``null``。
+        通过 ``stall_timeout_ms`` 检测僵尸连接：超过该时长未收到
+        Heartbeat ACK 则主动断开。
+
+        ``get_last_ack_time`` 返回 ``receive`` 回调更新的最后 ACK 时间戳，
+        用于卡死检测。连接关闭时协程自动退出（``ws.send_json`` 抛异常）。
         """
         interval_sec = interval_ms / 1000.0
-        last_heartbeat = time.monotonic()
         while True:
             try:
                 await asyncio.sleep(interval_sec)
-                await ws.send_json({"op": _OP_HEARTBEAT, "d": last_heartbeat})
-                last_heartbeat = time.monotonic()
+                seq = get_last_sequence()
+                await ws.send_json({"op": _OP_HEARTBEAT, "d": seq})
             except asyncio.CancelledError:
                 break
             except Exception as exc:
@@ -310,5 +386,19 @@ class QQBotStreamConnectorAdapter:
                 )
                 break
 
+            # 卡死检测：超过 stall_timeout 未收到 Heartbeat ACK 则断开
+            now = time.monotonic()
+            if now - get_last_ack_time() > stall_timeout_ms / 1000.0:
+                await self._logger.warn(
+                    "QQBot ws stall detected, closing connection",
+                    account_id=account_id,
+                    stall_timeout_ms=stall_timeout_ms,
+                )
+                try:
+                    await ws.close()
+                except Exception:
+                    pass
+                break
+
 
 __all__ = ["QQBotStreamConnectorAdapter"]
diff --git a/backend/package/yuxi/channels/plugins/qqbot/adapters/whitelist_adapter.py b/backend/package/yuxi/channels/plugins/qqbot/adapters/whitelist_adapter.py
index 1e1f57b7..a360f5c1 100644
--- a/backend/package/yuxi/channels/plugins/qqbot/adapters/whitelist_adapter.py
+++ b/backend/package/yuxi/channels/plugins/qqbot/adapters/whitelist_adapter.py
@@ -6,7 +6,8 @@
     - DM: ``qqbot:whitelist:{account_id}:dm``
     - Group: ``qqbot:whitelist:{account_id}:group``
 
-值为 ``WhitelistEntry`` 列表的 JSON 序列化字符串。
+值为 ``WhitelistEntry`` 列表的 JSON 序列化字符串。写操作通过
+``CachePort.acquireAdvisoryLock`` 串行化，避免并发修改导致数据丢失。
 """
 
 from __future__ import annotations
@@ -26,12 +27,20 @@ from yuxi.channels.contract.ports.driven.logger_port import LoggerPort
 # 白名单缓存 TTL（秒），7 天
 _WHITELIST_TTL_SECONDS = 604800
 
+# 白名单写操作咨询锁 TTL（秒），30 秒
+_WHITELIST_LOCK_TTL_SECONDS = 30
+
 
 def _whitelist_key(account_id: str, policy_type: WhitelistPolicyType) -> str:
     """构造白名单缓存 key。"""
     return f"qqbot:whitelist:{account_id}:{policy_type.value}"
 
 
+def _whitelist_lock_key(account_id: str) -> str:
+    """构造白名单写操作咨询锁 key。"""
+    return f"qqbot:whitelist_lock:{account_id}"
+
+
 def _serialize_entries(entries: list[WhitelistEntry]) -> str:
     """序列化白名单条目列表为 JSON 字符串。"""
     return json.dumps(
@@ -90,7 +99,8 @@ def _deserialize_entries(raw: str) -> list[WhitelistEntry]:
 class QQBotWhitelistAdapter:
     """QQ Bot 白名单适配器。
 
-    白名单存储在 ``CachePort``，支持 DM / Group 两种策略类型。
+    白名单存储在 ``CachePort``，支持 DM / Group 两种策略类型。写操作
+    通过 ``acquireAdvisoryLock`` 串行化，避免并发修改导致数据丢失。
     无 mutable 实例变量（INV-5）。
     """
 
@@ -128,17 +138,25 @@ class QQBotWhitelistAdapter:
                 field="peer_id",
                 message="qqbot whitelist entry peer_id must not be empty",
             )
-        entries = list(await self.getWhitelist(account_id, entry.peer_type))
-        # 去重：已存在则更新
-        entries = [e for e in entries if e.peer_id != entry.peer_id]
-        entries.append(entry)
-        key = _whitelist_key(account_id, entry.peer_type)
-        await self._cache.set(
-            key,
-            _serialize_entries(entries),
-            ttl_seconds=_WHITELIST_TTL_SECONDS,
+        lock = await self._cache.acquireAdvisoryLock(
+            _whitelist_lock_key(account_id),
+            ttl_seconds=_WHITELIST_LOCK_TTL_SECONDS,
         )
-        return True
+        try:
+            entries = list(await self.getWhitelist(account_id, entry.peer_type))
+            # 去重：已存在则更新
+            entries = [e for e in entries if e.peer_id != entry.peer_id]
+            entries.append(entry)
+            key = _whitelist_key(account_id, entry.peer_type)
+            await self._cache.set(
+                key,
+                _serialize_entries(entries),
+                ttl_seconds=_WHITELIST_TTL_SECONDS,
+            )
+            return True
+        finally:
+            if lock is not None:
+                await self._cache.releaseAdvisoryLock(lock)
 
     async def removeFromWhitelist(
         self,
@@ -147,17 +165,25 @@ class QQBotWhitelistAdapter:
         policy_type: WhitelistPolicyType,
     ) -> bool:
         """移除白名单条目。"""
-        entries = list(await self.getWhitelist(account_id, policy_type))
-        new_entries = [e for e in entries if e.peer_id != peer_id]
-        if len(new_entries) == len(entries):
-            return False
-        key = _whitelist_key(account_id, policy_type)
-        await self._cache.set(
-            key,
-            _serialize_entries(new_entries),
-            ttl_seconds=_WHITELIST_TTL_SECONDS,
+        lock = await self._cache.acquireAdvisoryLock(
+            _whitelist_lock_key(account_id),
+            ttl_seconds=_WHITELIST_LOCK_TTL_SECONDS,
         )
-        return True
+        try:
+            entries = list(await self.getWhitelist(account_id, policy_type))
+            new_entries = [e for e in entries if e.peer_id != peer_id]
+            if len(new_entries) == len(entries):
+                return False
+            key = _whitelist_key(account_id, policy_type)
+            await self._cache.set(
+                key,
+                _serialize_entries(new_entries),
+                ttl_seconds=_WHITELIST_TTL_SECONDS,
+            )
+            return True
+        finally:
+            if lock is not None:
+                await self._cache.releaseAdvisoryLock(lock)
 
     async def updateEntry(
         self,
@@ -170,64 +196,86 @@ class QQBotWhitelistAdapter:
                 field="peer_id",
                 message="qqbot whitelist entry peer_id must not be empty",
             )
-        entries = list(await self.getWhitelist(account_id, entry.peer_type))
-        found = False
-        new_entries: list[WhitelistEntry] = []
-        for e in entries:
-            if e.peer_id == entry.peer_id:
-                new_entries.append(entry)
-                found = True
-            else:
-                new_entries.append(e)
-        if not found:
-            return False
-        key = _whitelist_key(account_id, entry.peer_type)
-        await self._cache.set(
-            key,
-            _serialize_entries(new_entries),
-            ttl_seconds=_WHITELIST_TTL_SECONDS,
+        lock = await self._cache.acquireAdvisoryLock(
+            _whitelist_lock_key(account_id),
+            ttl_seconds=_WHITELIST_LOCK_TTL_SECONDS,
         )
-        return True
+        try:
+            entries = list(await self.getWhitelist(account_id, entry.peer_type))
+            found = False
+            new_entries: list[WhitelistEntry] = []
+            for e in entries:
+                if e.peer_id == entry.peer_id:
+                    new_entries.append(entry)
+                    found = True
+                else:
+                    new_entries.append(e)
+            if not found:
+                return False
+            key = _whitelist_key(account_id, entry.peer_type)
+            await self._cache.set(
+                key,
+                _serialize_entries(new_entries),
+                ttl_seconds=_WHITELIST_TTL_SECONDS,
+            )
+            return True
+        finally:
+            if lock is not None:
+                await self._cache.releaseAdvisoryLock(lock)
 
     async def batchAdd(
         self,
         account_id: str,
         entries: tuple[WhitelistEntry, ...],
     ) -> dict[str, Any]:
-        """批量添加白名单条目。"""
+        """批量添加白名单条目。
+
+        @post 返回 ``{"succeeded": int, "failed": int, "failures": [{"peer_id", "reason"}]}``。
+        """
         if not entries:
-            return {"total": 0, "succeeded": 0, "failed": 0}
+            return {"succeeded": 0, "failed": 0, "failures": []}
         succeeded = 0
-        failed = 0
+        failures: list[dict[str, str]] = []
         # 按 policy_type 分组处理
         by_type: dict[WhitelistPolicyType, list[WhitelistEntry]] = {}
         for entry in entries:
             by_type.setdefault(entry.peer_type, []).append(entry)
 
-        for policy_type, group in by_type.items():
-            existing = list(await self.getWhitelist(account_id, policy_type))
-            existing_ids = {e.peer_id for e in existing}
-            for entry in group:
-                if not entry.peer_id:
-                    failed += 1
-                    continue
-                if entry.peer_id in existing_ids:
-                    # 更新已有条目
-                    existing = [entry if e.peer_id == entry.peer_id else e for e in existing]
-                else:
-                    existing.append(entry)
-                succeeded += 1
-            key = _whitelist_key(account_id, policy_type)
-            await self._cache.set(
-                key,
-                _serialize_entries(existing),
-                ttl_seconds=_WHITELIST_TTL_SECONDS,
-            )
+        lock = await self._cache.acquireAdvisoryLock(
+            _whitelist_lock_key(account_id),
+            ttl_seconds=_WHITELIST_LOCK_TTL_SECONDS,
+        )
+        try:
+            for policy_type, group in by_type.items():
+                existing = list(await self.getWhitelist(account_id, policy_type))
+                existing_ids = {e.peer_id for e in existing}
+                batch_succeeded = 0
+                for entry in group:
+                    if not entry.peer_id:
+                        failures.append({"peer_id": "", "reason": "empty peer_id"})
+                        continue
+                    if entry.peer_id in existing_ids:
+                        # 更新已有条目
+                        existing = [entry if e.peer_id == entry.peer_id else e for e in existing]
+                    else:
+                        existing.append(entry)
+                        existing_ids.add(entry.peer_id)
+                    batch_succeeded += 1
+                key = _whitelist_key(account_id, policy_type)
+                await self._cache.set(
+                    key,
+                    _serialize_entries(existing),
+                    ttl_seconds=_WHITELIST_TTL_SECONDS,
+                )
+                succeeded += batch_succeeded
+        finally:
+            if lock is not None:
+                await self._cache.releaseAdvisoryLock(lock)
 
         return {
-            "total": len(entries),
             "succeeded": succeeded,
-            "failed": failed,
+            "failed": len(failures),
+            "failures": failures,
         }
 
     async def exportEntries(
diff --git a/backend/package/yuxi/channels/plugins/qqbot/entry.py b/backend/package/yuxi/channels/plugins/qqbot/entry.py
index deb3d434..ea21ee52 100644
--- a/backend/package/yuxi/channels/plugins/qqbot/entry.py
+++ b/backend/package/yuxi/channels/plugins/qqbot/entry.py
@@ -99,10 +99,10 @@ def channel_entry(host: PluginHost, manifest: ChannelManifest) -> PluginManifest
     """
 
     # 1. 获取端口实例（宿主已从 manifest.accessible_ports 加载权限）
-    #    QQ Bot 适配器仅需 config / logger / cache 端口，不使用 persistence 端口。
     config_port = host.getConfigPort()
     logger_port = host.getLoggerPort()
     cache_port = host.getCachePort()
+    persistence_port = host.getPersistencePort()
 
     # 2. 实例化 QQBotClient 与 LifecycleHandler
     #    依赖关系为单向：handler 在 onInit 调 client.attach_http_client 注入
@@ -125,7 +125,7 @@ def channel_entry(host: PluginHost, manifest: ChannelManifest) -> PluginManifest
     )
     host.registerAdapter(
         "outbound",
-        QQBotOutboundAdapter(client, logger_port, cache_port),
+        QQBotOutboundAdapter(client, logger_port, cache_port, config_port),
     )
     host.registerAdapter("session", QQBotSessionAdapter())
     host.registerAdapter("status", QQBotStatusAdapter())
@@ -137,7 +137,7 @@ def channel_entry(host: PluginHost, manifest: ChannelManifest) -> PluginManifest
     )
     host.registerAdapter(
         "directory",
-        QQBotDirectoryAdapter(client, cache_port),
+        QQBotDirectoryAdapter(client, cache_port, config_port),
     )
     host.registerAdapter(
         "whitelist",
@@ -157,18 +157,19 @@ def channel_entry(host: PluginHost, manifest: ChannelManifest) -> PluginManifest
     )
     host.registerAdapter(
         "probeable",
-        QQBotProbeableAdapter(client, logger_port),
+        QQBotProbeableAdapter(client, logger_port, persistence_port),
     )
     # identity_resolver 为单实例注册（非列表追加）
     host.registerAdapter(
         "identity_resolver",
-        QQBotIdentityResolverAdapter(client, cache_port, logger_port),
+        QQBotIdentityResolverAdapter(client, cache_port, logger_port, config_port),
     )
     # stream_connector：QQ Bot WS 长连接适配器，由 StreamWorker 管理连接生命周期。
-    # 需 ConfigPort 读取 ws_heartbeat_interval_ms（声明于 manifest config_schema）。
+    # 需 ConfigPort 读取 ws_heartbeat_interval_ms（声明于 manifest config_schema），
+    # 需 QQBotClient 动态获取 Gateway URL。
     host.registerAdapter(
         "stream_connector",
-        QQBotStreamConnectorAdapter(config_port, logger_port),
+        QQBotStreamConnectorAdapter(config_port, logger_port, client),
     )
 
     # 4. 注册生命周期钩子
diff --git a/backend/package/yuxi/channels/plugins/qqbot/lifecycle.py b/backend/package/yuxi/channels/plugins/qqbot/lifecycle.py
index ad0159bb..8e4d1ee5 100644
--- a/backend/package/yuxi/channels/plugins/qqbot/lifecycle.py
+++ b/backend/package/yuxi/channels/plugins/qqbot/lifecycle.py
@@ -125,7 +125,7 @@ class QQBotLifecycleHandler:
         if self._http_client is not None:
             try:
                 await self._http_client.aclose()
-            except Exception as exc:
+            except (httpx.HTTPError, OSError) as exc:
                 await self._logger.warn(
                     "QQBot httpx client close failed during unload",
                     error=str(exc),
diff --git a/backend/package/yuxi/channels/plugins/qqbot/manifest.json b/backend/package/yuxi/channels/plugins/qqbot/manifest.json
index 90d4b766..7925ed4e 100644
--- a/backend/package/yuxi/channels/plugins/qqbot/manifest.json
+++ b/backend/package/yuxi/channels/plugins/qqbot/manifest.json
@@ -107,8 +107,8 @@
     {"key": "group_active_qpm_per_group", "type": "integer", "required": false, "default": 20, "hot_reloadable": true, "scope": "channel", "constraints": {"min": 1, "max": 60}},
     {"key": "token_cache_ttl", "type": "integer", "required": false, "default": 7200, "hot_reloadable": true, "scope": "channel", "constraints": {"min": 60, "max": 7200}},
     {"key": "user_cache_ttl", "type": "integer", "required": false, "default": 3600, "hot_reloadable": true, "scope": "channel", "constraints": {"min": 60, "max": 7200}},
-    {"key": "url_whitelist", "type": "array", "required": false, "default": [], "hot_reloadable": true, "scope": "channel", "constraints": {"max_items": 20}},
-    {"key": "enable_interaction_callback", "type": "boolean", "required": false, "default": true, "hot_reloadable": true, "scope": "channel", "constraints": {}}
+    {"key": "identity_cache_ttl", "type": "integer", "required": false, "default": 60, "hot_reloadable": true, "scope": "channel", "constraints": {"min": 10, "max": 600}},
+    {"key": "url_whitelist", "type": "array", "required": false, "default": [], "hot_reloadable": true, "scope": "channel", "constraints": {"max_items": 20}}
   ],
   "env_vars": [
     {"name": "QQBOT_APP_ID", "description": "QQ Bot 应用 App ID", "required": true, "default": null, "sensitive": false},
diff --git a/backend/package/yuxi/channels/plugins/qqbot/qqbot_client.py b/backend/package/yuxi/channels/plugins/qqbot/qqbot_client.py
index f4bf6281..f0aa5a5b 100644
--- a/backend/package/yuxi/channels/plugins/qqbot/qqbot_client.py
+++ b/backend/package/yuxi/channels/plugins/qqbot/qqbot_client.py
@@ -139,6 +139,24 @@ class QQBotClient:
     def _token_cache_key(self, app_id: str) -> str:
         return f"{_TOKEN_CACHE_KEY_PREFIX}{app_id}"
 
+    async def _get_token_cache_ttl(self, expires_in: int) -> int:
+        """获取 token 缓存 TTL（秒），受 ``token_cache_ttl`` 配置约束。
+
+        优先使用 ``expires_in - TOKEN_REFRESH_LEAD_TIME_SEC``（提前刷新），
+        并以配置的 ``token_cache_ttl`` 作为上限，避免缓存超过平台有效期。
+        """
+        ttl = max(expires_in - TOKEN_REFRESH_LEAD_TIME_SEC, 1)
+        configured = await self._get_channel_config("token_cache_ttl", TOKEN_DEFAULT_TTL)
+        try:
+            cap = int(configured)
+        except (TypeError, ValueError):
+            cap = TOKEN_DEFAULT_TTL
+        return min(ttl, cap)
+
+    def _token_lock_key(self, app_id: str) -> str:
+        """token 刷新咨询锁 key。"""
+        return f"qqbot:token_lock:{app_id}"
+
     async def _execute_http(
         self,
         method: str,
@@ -147,12 +165,15 @@ class QQBotClient:
         *,
         json_body: dict | None = None,
         params: dict | None = None,
+        trace_id: str = "",
     ) -> httpx.Response:
         """执行 HTTP 请求，HTTP 异常转换为契约异常。
 
         不负责 token 注入与业务错误码校验，由调用方处理。优先复用
         lifecycle 注入的共享连接池；未注入时降级为每请求独立
         ``httpx.AsyncClient``（兼容测试与未启用 lifecycle 的场景）。
+
+        ``trace_id`` 由调用方注入，用于关联错误与调用链路。
         """
         timeout = await self._get_request_timeout_seconds()
         try:
@@ -181,19 +202,22 @@ class QQBotClient:
                 exc_info=exc,
                 method=method,
                 url=url,
+                trace_id=trace_id,
             )
             raise error_translator.wrap_http_exception(
                 exc,
                 timeout_ms=int(timeout * 1000),
+                trace_id=trace_id,
             ) from exc
         return resp
 
     @staticmethod
-    def _parse_json_response(resp: httpx.Response) -> dict:
+    def _parse_json_response(resp: httpx.Response, *, trace_id: str = "") -> dict:
         """解析 JSON 响应，校验 QQ Bot 业务错误码。
 
         ``code`` 非 0 时通过 ``error_translator.translate_qqbot_error``
         转换为契约异常。``code`` 为 0 时返回完整响应字典。
+        ``trace_id`` 由调用方注入，用于关联错误与调用链路。
         """
         data = resp.json()
         code = data.get("code", 0)
@@ -201,7 +225,8 @@ class QQBotClient:
             raise error_translator.translate_qqbot_error(
                 code,
                 data.get("message", ""),
-            ) from None
+                trace_id=trace_id,
+            )
         return data
 
     # ------------------------------------------------------------------
@@ -213,7 +238,8 @@ class QQBotClient:
 
         缓存写入时存储绝对过期时间戳 ``expires_at``，缓存命中时根据
         ``expires_at - now`` 计算剩余 TTL，避免长期持有 token 的调用方
-        误判有效期。
+        误判有效期。token 刷新通过 ``acquireAdvisoryLock`` 串行化，避免
+        并发场景下重复请求 token 端点（M-08）。
 
         QQ Bot token 端点：``POST /app/getAppAccessToken``，
         请求体 ``{"appId": "...", "clientSecret": "..."}``，
@@ -228,36 +254,50 @@ class QQBotClient:
             remaining = max(int(expires_at - time.time()), 0)
             return token, remaining
 
-        api_base = await self._get_api_base()
-        url = f"{api_base}{TOKEN_ENDPOINT_PATH}"
-        body = {"appId": app_id, "clientSecret": app_secret}
-
-        resp = await self._execute_http("POST", url, headers={}, json_body=body)
-        data = self._parse_json_response(resp)
-        token = data.get("access_token", "")
-        expires_in_str = data.get("expires_in", str(TOKEN_DEFAULT_TTL))
+        # 通过咨询锁串行化 token 刷新，避免并发重复请求
+        lock_key = self._token_lock_key(app_id)
+        lock_token = await self._cache.acquireAdvisoryLock(lock_key, ttl_seconds=30)
         try:
-            expires_in = int(expires_in_str)
-        except (TypeError, ValueError):
-            expires_in = TOKEN_DEFAULT_TTL
-        if expires_in <= 0:
-            raise DependencyError(
-                dep=QQBOT_API_DEP,
-                cause=ValueError(f"invalid expires_in value: {expires_in}"),
+            # double-checked locking：持有锁后再次检查缓存
+            cached = await self._cache.get(cache_key)
+            if cached.is_some():
+                token, expires_at = cached.unwrap()
+                remaining = max(int(expires_at - time.time()), 0)
+                return token, remaining
+
+            api_base = await self._get_api_base()
+            url = f"{api_base}{TOKEN_ENDPOINT_PATH}"
+            body = {"appId": app_id, "clientSecret": app_secret}
+
+            resp = await self._execute_http("POST", url, headers={}, json_body=body)
+            data = self._parse_json_response(resp)
+            token = data.get("access_token", "")
+            expires_in_str = data.get("expires_in", str(TOKEN_DEFAULT_TTL))
+            try:
+                expires_in = int(expires_in_str)
+            except (TypeError, ValueError):
+                expires_in = TOKEN_DEFAULT_TTL
+            if expires_in <= 0:
+                raise DependencyError(
+                    dep=QQBOT_API_DEP,
+                    cause=ValueError(f"invalid expires_in value: {expires_in}"),
+                )
+            expires_at = time.time() + expires_in
+            ttl = await self._get_token_cache_ttl(expires_in)
+            await self._cache.set(
+                cache_key,
+                (token, expires_at),
+                ttl_seconds=ttl,
             )
-        expires_at = time.time() + expires_in
-        ttl = max(expires_in - TOKEN_REFRESH_LEAD_TIME_SEC, 1)
-        await self._cache.set(
-            cache_key,
-            (token, expires_at),
-            ttl_seconds=ttl,
-        )
-        await self._logger.debug(
-            "QQBot access_token refreshed",
-            app_id=app_id,
-            expires_in=expires_in,
-        )
-        return token, expires_in
+            await self._logger.debug(
+                "QQBot access_token refreshed",
+                app_id=app_id,
+                expires_in=expires_in,
+            )
+            return token, expires_in
+        finally:
+            if lock_token is not None:
+                await self._cache.releaseAdvisoryLock(lock_token)
 
     async def _get_token_for_account(self, account_id: str) -> str:
         """获取账户的 access_token。"""
@@ -285,6 +325,7 @@ class QQBotClient:
         *,
         json_body: dict | None = None,
         params: dict | None = None,
+        trace_id: str = "",
     ) -> dict:
         """发起带 access_token 的请求并返回校验后的 JSON。"""
         headers = await self._build_auth_header(account_id)
@@ -296,8 +337,9 @@ class QQBotClient:
             headers,
             json_body=json_body,
             params=params,
+            trace_id=trace_id,
         )
-        return self._parse_json_response(resp)
+        return self._parse_json_response(resp, trace_id=trace_id)
 
     # ------------------------------------------------------------------
     # Gateway API